抖音搜索页数据批量爬取，多关键词同步采集实现

import requests
import json
import time
import random
import pandas as pd
from concurrent.futures import ThreadPoolExecutor

# -------------------------- 配置区（必须修改） --------------------------
# 请求头：替换为你自己浏览器的参数
HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36",
    "Cookie": "你的抖音Cookie",
    "X-Sec-Token": "你的X-Sec-Token",
    "Referer": "https://www.douyin.com/",
    "Accept": "application/json, text/plain, */*"
}

# 基础配置
BASE_URL = "https://www.douyin.com/aweme/v1/web/search/item/"
MAX_PAGE = 5  # 每个关键词爬取的最大页数
THREAD_NUM = 3  # 并发线程数，控制采集速度
# ------------------------------------------------------------------------

# 存储最终采集的所有数据
ALL_DATA = []

def crawl_single_page(keyword, page):
    """
    爬取单个关键词的单页数据
    :param keyword: 搜索关键词
    :param page: 页码
    :return: 解析后的视频数据列表
    """
    # 构造请求参数（抖音搜索接口标准参数）
    params = {
        "keyword": keyword,
        "cursor": (page - 1) * 10,  # 分页游标，每页10条数据
        "search_source": "normal",
        "aid": "6383",
        "channel": "douyin_web"
    }

    try:
        # 发送GET请求，设置超时时间
        response = requests.get(BASE_URL, headers=HEADERS, params=params, timeout=10)
        response.raise_for_status()  # 抛出请求异常
        data = response.json()

        # 判断数据是否存在
        if "data" not in data or len(data["data"]) == 0:
            print(f"【{keyword}】第{page}页无数据")
            return []

        # 数据清洗：提取核心字段
        result_list = []
        for item in data["data"]:
            # 基础视频信息
            video_info = {
                "关键词": keyword,
                "视频ID": item.get("aweme_id", ""),
                "视频标题": item.get("desc", ""),
                "博主昵称": item.get("author", {}).get("nickname", ""),
                "博主ID": item.get("author", {}).get("unique_id", ""),
                "点赞数": item.get("statistics", {}).get("digg_count", 0),
                "评论数": item.get("statistics", {}).get("comment_count", 0),
                "转发数": item.get("statistics", {}).get("share_count", 0),
                "发布时间": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(item.get("create_time", 0))),
                "视频链接": f"https://www.douyin.com/video/{item.get('aweme_id', '')}"
            }
            result_list.append(video_info)

        print(f"【{keyword}】第{page}页采集成功，共{len(result_list)}条数据")
        return result_list

    except Exception as e:
        print(f"【{keyword}】第{page}页爬取失败：{str(e)}")
        return []

def crawl_keyword(keyword):
    """
    爬取单个关键词的所有分页数据
    """
    for page in range(1, MAX_PAGE + 1):
        page_data = crawl_single_page(keyword, page)
        if page_data:
            ALL_DATA.extend(page_data)
        # 随机延时，规避反爬（1-3秒）
        time.sleep(random.uniform(1, 3))
    print(f"===== 【{keyword}】全部页码采集完成 =====")

def crawl_keyword(keyword):
    """
    爬取单个关键词的所有分页数据
    """
    for page in range(1, MAX_PAGE + 1):
        page_data = crawl_single_page(keyword, page)
        if page_data:
            ALL_DATA.extend(page_data)
        # 随机延时，规避反爬（1-3秒）
        time.sleep(random.uniform(1, 3))
    print(f"===== 【{keyword}】全部页码采集完成 =====")

if __name__ == "__main__":
    # 自定义多关键词列表
    KEYWORDS = ["Python爬虫", "短视频运营", "AI工具推荐", "抖音变现技巧"]
    # 启动批量采集
    batch_crawl(KEYWORDS)