基于 Python 爬取 TikTok 搜索数据 Tiktok爬虫(2025.3.17)

打印 上一主题 下一主题

主题 1003|帖子 1003|积分 3009

1. 前言

在数据分析和网络爬虫的应用场景中,我们经常必要获取社交媒体平台的数据,例如 TikTok。本篇文章介绍如何利用 Python 爬取 TikTok 用户搜索数据,并解析其返回的数据。

效果截图

2. 项目环境准备

在正式运行代码之前,我们必要安装相干的 Python 库:
  1. pip install requests pandas execjs loguru
复制代码
别的,我们必要一个 JavaScript 运行环境(如 Node.js),用于执行加密签名代码。
3. 代码解析

3.1 初始化爬虫类

我们创建 TiktokUserSearch 类,并在初始化方法 __init__ 中设置请求头信息,并初始化输出文件。
  1. class TiktokUserSearch:
  2.     def __init__(self, output_file=None):
  3.         self.headers = {  # 设置请求头
  4.             "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ...",
  5.             "referer": "https://www.tiktok.com/"
  6.         }
  7.         self.cookies = None
  8.         self.output_file = output_file if output_file else f'tiktok_videos_{datetime.now().strftime("%Y%m%d_%H%M%S")}.csv'
复制代码
3.2 处理 Cookie

我们必要将 TikTok 的 cookie 从字符串转换成字典格式,以便后续请求利用。
  1.     def cookie_str_to_dict(self, cookie_str) -> dict:
  2.         cookie_dict = {}
  3.         cookies = [i.strip() for i in cookie_str.split('; ') if i.strip() != ""]
  4.         for cookie in cookies:
  5.             key, value = cookie.split('=', 1)
  6.             cookie_dict[key] = value
  7.         return cookie_dict
复制代码
3.3 发送请求

TikTok 必要利用 X-Bogus 进行签名,我们必要执行 JavaScript 代码来获取该参数。
为了防止网络不稳固,设置三次重试机制。
可根据自己需求设置代理。
  1.     def get(self, keyword, cursor, search_id, cookie_str):
  2.         self.cookies = self.cookie_str_to_dict(cookie_str)
  3.         url = "https://www.tiktok.com/api/search/general/full/"
  4.         if cursor == "0":
  5.             focus_state = "true"
  6.         else:
  7.             focus_state = "false"
  8.         params = {
  9.             "WebIdLastTime": f"{int(time.time())}",
  10.             "aid": "1988",
  11.             "app_language": "zh-Hans",
  12.             "app_name": "tiktok_web",
  13.             "browser_language": "zh-CN",
  14.             "browser_name": "Mozilla",
  15.             "browser_online": "true",
  16.             "browser_platform": "Win32",
  17.             "browser_version": self.headers['user-agent'].replace('Mozilla/', ''),
  18.             "channel": "tiktok_web",
  19.             "cookie_enabled": "true",
  20.             "cursor": cursor,
  21.             "device_id": "7339506347602019870",
  22.             "device_platform": "web_pc",
  23.             "focus_state": focus_state,
  24.             "from_page": "search",
  25.             "history_len": "7",
  26.             "is_fullscreen": "false",
  27.             "is_page_visible": "true",
  28.             "keyword": keyword,
  29.             "os": "windows",
  30.             "priority_region": "",
  31.             "referer": "",
  32.             "region": "KR",
  33.             "screen_height": "1080",
  34.             "screen_width": "1920",
  35.             "tz_name": "Asia/Shanghai",
  36.             "web_search_code": "{"tiktok":{"client_params_x":{"search_engine":{"ies_mt_user_live_video_card_use_libra":1,"mt_search_general_user_live_card":1}},"search_server":{}}}",
  37.             "webcast_language": "zh-Hans",
  38.             "msToken": self.cookies["msToken"],
  39.         }
  40.         if cursor != "0":
  41.             params.update({"search_id": search_id})
  42.         x_b = execjs.compile(open('./encrypt.js', encoding='utf-8').read()).call("sign", urlencode(params), self.headers["user-agent"])
  43.         params.update({"X-Bogus": x_b})
  44.         headers = self.headers.copy()
  45.         headers.update({"referer": "https://www.tiktok.com/search?q=" + keyword})
  46.         max_retries = 3
  47.         for attempt in range(max_retries):
  48.             try:
  49.                 response = requests.get(
  50.                     url,
  51.                     headers=headers,
  52.                     cookies=self.cookies,
  53.                     params=params,
  54.                     timeout=(3, 10),
  55.                     proxies=None
  56.                 )
  57.                 return response.json()
  58.             except (ex1, ex2, ex3) as e:
  59.                 print(f"尝试 {attempt + 1}/{max_retries} 发生网络错误:{e}")
  60.                 if attempt < max_retries - 1:
  61.                     time.sleep(2)
  62.                 else:
  63.                     return {"error": f"Network error after {max_retries} attempts: {str(e)}"}
  64.             except Exception as e:
  65.                 print(f"发生其他错误:{e}")
  66.                 return {"error": str(e)}
复制代码
3.4 解析数据并存储

解析 TikTok 返回的视频数据,并生存到 CSV 文件。
  1.     def parse_data(self, data_list):
  2.         resultList = []
  3.         video_data = []
  4.         
  5.         for u in data_list:
  6.             try:
  7.                 item = u['item']
  8.                 author = item['author']
  9.                 stats = item['stats']
  10.                 author_stats = item['authorStats']  # 添加作者统计信息
  11.                
  12.                 # 提取需要的数据
  13.                 video_info = {
  14.                     'video_id': item['id'],
  15.                     'desc': item['desc'],
  16.                     'create_time': datetime.fromtimestamp(item['createTime']).strftime('%Y-%m-%d %H:%M:%S'),
  17.                     'duration': item['video']['duration'],
  18.                     # 作者基本信息
  19.                     'author_id': author['id'],
  20.                     'author_name': author['uniqueId'],
  21.                     'author_nickname': author['nickname'],
  22.                     'author_signature': author['signature'],
  23.                     'author_verified': author['verified'],
  24.                     # 作者统计信息
  25.                     'author_following_count': author_stats['followingCount'],  # 关注数
  26.                     'author_follower_count': author_stats['followerCount'],    # 粉丝数
  27.                     'author_heart_count': author_stats['heartCount'],          # 获赞总数
  28.                     'author_video_count': author_stats['videoCount'],          # 视频总数
  29.                     'author_digg_count': author_stats['diggCount'],           # 点赞数
  30.                     # 视频统计信息
  31.                     'digg_count': stats['diggCount'],
  32.                     'share_count': stats['shareCount'],
  33.                     'comment_count': stats['commentCount'],
  34.                     'play_count': stats['playCount'],
  35.                     'collect_count': stats.get('collectCount', 0),
  36.                     'video_url': item['video']['playAddr']
  37.                 }
  38.                
  39.                 # 添加标签信息
  40.                 if 'challenges' in item:
  41.                     video_info['hashtags'] = ','.join([tag['title'] for tag in item['challenges']])
  42.                 else:
  43.                     video_info['hashtags'] = ''
  44.                
  45.                 # 添加音乐信息
  46.                 if 'music' in item:
  47.                     music = item['music']
  48.                     video_info.update({
  49.                         'music_id': music['id'],
  50.                         'music_title': music['title'],
  51.                         'music_author': music['authorName'],
  52.                         'music_original': music['original']
  53.                     })
  54.                
  55.                 video_data.append(video_info)
  56.                 resultList.append(f"https://www.tiktok.com/@{author['uniqueId']}")
  57.             except Exception as e:
  58.                 logger.error(f"解析视频数据时出错: {str(e)}")
  59.                 continue
  60.         
  61.         # 将数据保存到CSV文件
  62.         try:
  63.             df = pd.DataFrame(video_data)
  64.             
  65.             # 检查文件是否存在
  66.             file_exists = os.path.exists(self.output_file)
  67.             
  68.             # 如果文件不存在,创建新文件并写入表头
  69.             # 如果文件存在,追加数据不写入表头
  70.             df.to_csv(self.output_file,
  71.                      mode='a',
  72.                      header=not file_exists,
  73.                      index=False,
  74.                      encoding='utf-8-sig')
  75.             
  76.             logger.info(f"数据已{'追加' if file_exists else '保存'}到文件: {self.output_file}")
  77.         except Exception as e:
  78.             logger.error(f"保存CSV文件时出错: {str(e)}")
  79.         return resultList
复制代码
3.5 运行爬虫

我们定义 main 方法,负责调用 get 方法获取数据并解析。
  1.     def main(self, keyword, cookie_str, cursor="0", search_id=None):
  2.         dataJson = self.get(keyword, cursor, search_id, cookie_str)
  3.         if dataJson:
  4.             if "error" in dataJson:
  5.                 return {"cursor": cursor, "search_id": search_id, "data": [], "status": "-2", "error": dataJson["error"]}
  6.             elif "verify_event" in str(dataJson):
  7.                 return {"cursor": cursor, "search_id": search_id, "data": [], "status": "-1"}
  8.             else:
  9.                 # 解析数据并保存到CSV
  10.                 if 'data' in dataJson:
  11.                     self.parse_data(dataJson['data'])
  12.                 return dataJson
复制代码
3.6 运行入口

最后,我们编写 if __name__ == '__main__' 逻辑,定义要爬取的关键词,并进行循环爬取。
  1. if __name__ == '__main__':
  2.     os.makedirs('results1', exist_ok=True)
  3.     topics = [
  4.     "Chen Duxiu",
  5.     "Li Dazhao",
  6. ]
  7.     for keyword in topics:
  8.         logger.info(f"开始爬取 {keyword} 的视频")
  9.         output_file = f'results1/{keyword}_videos.csv'  # 你可以自定义文件名
  10.         tiktok = TiktokUserSearch(output_file=output_file)
  11.         cookie_str = '_ttp=2ZzUB37CLclhWsrgyW56Erox1XM; tiktok_webapp_theme_auto_dark_ab=1; delay_guest_mode_vid=5; passport_csrf_token=d8e4d28ec7abdf12a7829d524dca64de; passport_csrf_token_default=d8e4d28ec7abdf12a7829d524dca64de; tt_chain_token=SSmpjX/0in/IP8BYwawD+Q==; multi_sids=7361707798058615814%3A53b730c284c4eaaa9bb2157eef01d70d; cmpl_token=AgQQAPNoF-RO0rYU5JqLsx0__dmghl8Nv5IhYNkWMA; passport_auth_status=a8a7a1e1c4b96a994a45acb38dc83509%2C; passport_auth_status_ss=a8a7a1e1c4b96a994a45acb38dc83509%2C; uid_tt=7857882a3366539dc1d9ca226b3fdc91f76b1b072c7da11dd4120368d88bf861; uid_tt_ss=7857882a3366539dc1d9ca226b3fdc91f76b1b072c7da11dd4120368d88bf861; sid_tt=53b730c284c4eaaa9bb2157eef01d70d; sessionid=53b730c284c4eaaa9bb2157eef01d70d; sessionid_ss=53b730c284c4eaaa9bb2157eef01d70d; store-idc=maliva; store-country-code=ca; store-country-code-src=uid; tt-target-idc=useast1a; tt-target-idc-sign=t3pz21FprSb2qc1ucJWFQbxzCKwgoBX9PKUWEbPHh7_4mpPThOuO0EN9pm2ORzFqk0bLFt6MtI9-gofvcVtQFoGSTOI_JvUWIAAUSHz1mM1A9jP1kRk_qucQnxEMOLvir3s4ffm0hJSh62RyKNO5LBTlT-fsqbi2tQVUwrgIGF-2HFT04S52ciyRnKAXr_0NyD3Aa0lM4J4hUGplo46wKRfId1DwwajXudUfjqJ3rvAuA8qURTsSHCKuDjLbcdfhcC0WKqemrmHFBJ11hGFJxiL4VEOClIoJGrF1_S9jvlx0H0Nph9BHlHNA-wzwi3NF6hPK17WL3TSvsqfEiKclZ5ScpHMv7ATYfOK4BVOzKXrq6fCxzNBT5kCNc4-ImuvjBNqpY8yL2s2KusWxslveOyIq3gwU3Dhxl084w5Tsp13xzuFOGNVHK5ZPeS5ERmykYFB6uTIHty9W_Z6pwN1tT9yQ-34qyZRZB7WONZn_NAFsywU6Hj4wcHLQkJ-tIiAO; last_login_method=google; tiktok_webapp_theme_source=auto; tiktok_webapp_theme=dark; sid_guard=53b730c284c4eaaa9bb2157eef01d70d%7C1740207607%7C15551996%7CThu%2C+21-Aug-2025+07%3A00%3A03+GMT; sid_ucp_v1=1.0.0-KDc2NDMxMTQzMTkwNTY2NTJiOWZhMmZhM2ZlMDg3ZDE0YzNiOGU5NTUKGQiGiIec0MeClWYQ9-vlvQYYsws4CEASSAQQAxoGbWFsaXZhIiA1M2I3MzBjMjg0YzRlYWFhOWJiMjE1N2VlZjAxZDcwZA; ssid_ucp_v1=1.0.0-KDc2NDMxMTQzMTkwNTY2NTJiOWZhMmZhM2ZlMDg3ZDE0YzNiOGU5NTUKGQiGiIec0MeClWYQ9-vlvQYYsws4CEASSAQQAxoGbWFsaXZhIiA1M2I3MzBjMjg0YzRlYWFhOWJiMjE1N2VlZjAxZDcwZA; odin_tt=a7027d0b8a102be6dd20600ca35291f4aee8c003895d8913d5c2f8276f16d6b974345357a22ed0bdcb57930d326afded3e5bd3105e061197876c80a92bbdbbfc29402634b8e7439ba178a1c7ed9ca552; tt_csrf_token=qs6ncqIZ-SbUZVzUbkUZ2SViJyY7VzYRki0M; perf_feed_cache={%22expireTimestamp%22:1742302800000%2C%22itemIds%22:[%227481325874978000134%22%2C%227463463290065161505%22%2C%227466633420098112799%22]}; msToken=BnkIjkPpJEc1i9jiiwT_paC5FW-NL62UVF7-lzpHYki9WIA_KpLrplpY-qlZfuG7V12rbCDHiyQYNrZcOnTzZLk1cvnH3_E_89nfOpqpVquKbSR-Nqr6bGDmL220vjBHdutm4R-gfVnYIG7fvWOJUkZ7yg==; ttwid=1%7Cv5j4n07c_G3ZtA91KIuree-ptnDLwgTwFuM8BnZINnQ%7C1742131441%7Ce88a85fcc36fd7e79815fddb10d16ef553b5e2e4a51c65e2c40098ade19023e2; msToken=iYDKRqCM8rSqc_9ZDzQnWcQiv_iJqPk15-6Y-iFBUmk4uIzb61dM13b9fWHcg4hxGkl9L3n56glok05TllvGurkwpgBYEF8N76ZRIii7OvNEkrk004dagNuoqQVeV9Bzd0_9naXjFXtEiMRi330G5Jdakw==; passport_fe_beating_status=false'
  12.         has_more = 1
  13.         cursor = '0'
  14.         search_id = None
  15.         while has_more:
  16.             data = tiktok.main(keyword, cookie_str, cursor, search_id)
  17.             logger.info(data)
  18.             if data and isinstance(data, dict):
  19.                 has_more = data.get('has_more', 0)
  20.                 logger.info(has_more)
  21.                 cursor = data.get('cursor', '0')
  22.                 search_id = data.get('log_pb', {}).get('impr_id')
  23.                 if 'data' in data:
  24.                     data = data['data']
  25.                 else:
  26.                     logger.error("No data found in response")
  27.                     break
  28.             else:
  29.                 logger.error("Invalid response format")
  30.                 break
  31.             time.sleep(1)  # 添加延时避免请求过快
  32.         logger.info(f"爬取 {keyword} 的视频完成")
  33.         time.sleep(30)
  34.         logger.info(f"等待30秒后继续爬取下一个主题")
  35.     logger.info("所有主题的视频爬取完成")
复制代码
4. 关键题目与办理方案


  • TikTok 反爬措施

    • 必要定期更新 Cookie
    • 得当增加请求间隔
    • 利用代理提高稳固性

  • X-Bogus 参数

    • 必要利用 JavaScript 盘算签名
    • 依赖 encrypt.js 进行加密

  • 数据存储

    • 利用 pandas 处理数据
    • 追加模式写入 CSV,避免数据丢失

6. 总结

本文介绍了如何利用 Python 爬取 TikTok 用户搜索数据,包罗如何构造请求、解析数据并存储到 CSV 文件。盼望对有类似需求的读者有所资助!
资助与咨询:私信博主或在评论区留言

 

免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看,没有账号?立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览

快速回复

您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

河曲智叟

论坛元老
这个人很懒什么都没写!
快速回复 返回顶部 返回列表