仿模板电影网站,实训做网站收获,容桂做网站,网站备案备注信息#x1f64b;♀️Tiktok APP的基于关键字检索的视频及评论信息爬虫共分为两期#xff0c;希望对大家有所帮助。 第一期见下文。 第二期#xff1a;基于视频URL的评论信息爬取 1. Node.js环境配置
首先配置 JavaScript 运行环境#xff08;如 Node.js#xff09;#x… ♀️Tiktok APP的基于关键字检索的视频及评论信息爬虫共分为两期希望对大家有所帮助。 第一期见下文。 第二期基于视频URL的评论信息爬取 1. Node.js环境配置
首先配置 JavaScript 运行环境如 Node.js用于执行加密签名代码。 Node.js下载网址https://nodejs.org/en Node.js的安装方法环境配置非常关键决定了后面的程序是否可以使用https://blog.csdn.net/liufeifeihuawei/article/details/132425239
2. Py环境配置
import time
import requests
import execjs
import os
from datetime import datetime
from urllib.parse import urlencode
from loguru import logger
import json
import random
from typing import Optional, Dict, List, Any
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading3. 基于关键字检索的视频信息爬取
1. 主程序设定爬取的关键字 通过文件topics.csv导入你希望爬取的关键字。 通过文件videosInfo.json存储爬取的结果以字典格式存储。
if __name__ __main__:os.makedirs(../results, exist_okTrue)keywords, fields read_csv(file_pathtopics.csv) # 设定爬取的关键字output_file f../results/videosInfo.json # 保存结果的文件cookie_str read_cookie()# 使用多线程并发爬取with ThreadPoolExecutor(max_workers1) as executor:futures []for i in range(len(keywords)):futures.append(executor.submit(crawl_keyword, keywords[i], output_file, cookie_str, fields[i], 20))for future in as_completed(futures):try:future.result()except Exception as e:logger.error(f爬取过程中发生错误: {str(e)})logger.info(所有主题的视频爬取完成)2. 多线程爬取单个关键词限制最大请求次数 通过request_count 设定爬取的请求次数。
def crawl_keyword(keyword: str, output_file: str, cookie_str: str, field: str, max_requests: int 10):tiktok TiktokUserSearch(output_fileoutput_file)has_more 1cursor 0search_id Nonerequest_count 0 # 初始化请求计数器while has_more and request_count max_requests:data tiktok.main(keyword, field, cookie_str, cursor, search_id)logger.info(fRequest {request_count 1}: {data})if data and isinstance(data, dict):# has_more data.get(has_more, 0)cursor data.get(cursor, 0)search_id data.get(log_pb, {}).get(impr_id)if data in data:data data[data]request_count 1 # 更新请求计数else:logger.error(No data found in response)breakelse:logger.error(Invalid response format)breaktime.sleep(random.randint(0, 5)) # 随机延时避免请求过快write_csv(keyword, request_count, file_path../results/records.csv)logger.info(f爬取 {keyword} 的视频完成共请求 {request_count} 次)3. 定义TiktokUserSearch类
允许获得24类字段包括 ️视频的URL、视频时长、标题等 视频的发布者个人简介、获赞数据、视频数据等 视频的点赞信息、分享次数、评论数量、播放次数、收藏次数等 视频的背景音乐ID音乐来源等… …
class TiktokUserSearch:def __init__(self, output_file: Optional[str] None):self.config read_config()self.headers self.config.get(headers, {})self.cookies Noneself.output_file output_file if output_file else ftiktok_videos_{datetime.now().strftime(%Y%m%d_%H%M%S)}.csvself.proxies self.config.get(proxies, None) # 代理配置self.lock threading.Lock() # 线程锁def cookie_str_to_dict(self, cookie_str: str) - Dict[str, str]:将cookie字符串转换为字典cookie_dict {}try:cookies [i.strip() for i in cookie_str.split(; ) if i.strip() ! ]for cookie in cookies:key, value cookie.split(, 1)cookie_dict[key] valueexcept Exception as e:logger.error(f转换cookie时出错: {str(e)})raisereturn cookie_dictdef get(self, keyword: str, cursor: str, search_id: Optional[str], cookie_str: str) - Dict[str, Any]:发送请求并获取数据self.cookies self.cookie_str_to_dict(cookie_str)url https://www.tiktok.com/api/search/general/full/focus_state true if cursor 0 else falseparams {WebIdLastTime: f{int(time.time())},aid: 1988,app_language: zh-Hans,app_name: tiktok_web,browser_language: zh-CN,# ... 略webcast_language: zh-Hans,msToken: self.cookies[msToken],}if cursor ! 0:params.update({search_id: search_id})try:x_b execjs.compile(open(../configs/encrypt.js, encodingutf-8).read()).call(sign, urlencode(params),self.headers[user-agent])params.update({X-Bogus: x_b})except Exception as e:logger.error(f生成X-Bogus时出错: {str(e)})return {error: str(e)}headers self.headers.copy()headers.update({referer: https://www.tiktok.com/search?q keyword})max_retries 3for attempt in range(max_retries):try:response requests.get(url,headersheaders,cookiesself.cookies,paramsparams,timeout(3, 10),proxiesself.proxies)response.raise_for_status()return response.json()except (ex1, ex2, ex3) as e:logger.warning(f尝试 {attempt 1}/{max_retries} 发生网络错误{e})if attempt max_retries - 1:time.sleep(2)else:return {error: fNetwork error after {max_retries} attempts: {str(e)}}except Exception as e:logger.error(f发生其他错误{e})return {error: str(e)}def parse_data(self, data_list: List[Dict[str, Any]], keyword: str, field: str) - List[str]:解析数据并保存到json文件resultList []video_data []for u in data_list:try:item u[item]author item[author]stats item[stats]author_stats item[authorStats]video_id str(item[id]), # 视频的唯一标识符TikTok 视频 IDauthor_name str(author[uniqueId]), # 作者的 TikTok 账号video_url fhttps://www.tiktok.com/{author_name[0]}/video/{video_id[0]}video_info {search_keyword: keyword,video_field: field,video_id: video_id[0], # 视频的唯一标识符TikTok 视频 IDdesc: item[desc], # 视频的文字描述caption/标题create_time: datetime.fromtimestamp(item[createTime]).strftime(%Y-%m-%d %H:%M:%S), # 视频的发布时间duration: item[video][duration], # 视频时长单位秒video_url: video_url, # 视频播放地址author_id: author[id], # 作者的唯一 IDauthor_name: author_name[0], # 作者的 TikTok 账号uniqueId即用户名#... 略author_following_count: author_stats[followingCount], # 作者关注的人数digg_count: stats[diggCount], # 视频的点赞like数量share_count: stats[shareCount], # 视频的分享次数comment_count: stats[commentCount], # 视频的评论数量play_count: stats[playCount], # 视频的播放次数collect_count: stats.get(collectCount, 0), # 视频的收藏次数}# video_info[comments] self.get_comments(video_url)if challenges in item:video_info[hashtags] ,.join([tag[title] for tag in item[challenges]])else:video_info[hashtags] # 背景音乐if music in item:music item[music]video_info.update({music_id: music[id],music_title: music[title],music_author: music[authorName],music_original: music[original]})video_data.append(video_info)resultList.append(fhttps://www.tiktok.com/{author[uniqueId]})except Exception as e:logger.error(f解析视频数据时出错: {str(e)})continue# **追加写入 JSON 文件**try:# 如果文件存在读取已有数据if os.path.exists(self.output_file):with open(self.output_file, r, encodingutf-8) as f:try:existing_data json.load(f)except json.JSONDecodeError:existing_data [] # 如果 JSON 解析失败重置为空列表else:existing_data []# 追加新数据existing_data.extend(video_data)# 保存回 JSON 文件with open(self.output_file, w, encodingutf-8) as f:json.dump(existing_data, f, ensure_asciiFalse, indent4)logger.info(f数据已{追加 if existing_data else 保存}到文件: {self.output_file})except Exception as e:logger.error(f保存 JSON 文件时出错: {str(e)})return resultListdef main(self, keyword: str, field: str, cookie_str: str, cursor: str 0, search_id: Optional[str] None) - Dict[str, Any]:主函数执行搜索并解析数据dataJson self.get(keyword, cursor, search_id, cookie_str)if dataJson:if error in dataJson:return {cursor: cursor, search_id: search_id, data: [], status: -2,error: dataJson[error]}elif verify_event in str(dataJson):return {cursor: cursor, search_id: search_id, data: [], status: -1}else:if data in dataJson:self.parse_data(dataJson[data], keyword, field)return dataJson
