短视频平台关键词搜索爬虫技术方案与实现

2026年2月7日 互联网

一、技术方案概述

短视频平台的关键词搜索结果包含丰富的用户行为数据，通过自动化手段采集这些数据可应用于舆情分析、内容推荐等场景。本方案采用浏览器自动化控制技术，结合网络请求监控实现数据采集，相比传统API调用方式具有更强的灵活性和抗封禁能力。

二、开发环境准备

1. 核心依赖库

  1. import time
  2. import json
  3. import re
  4. import threading
  5. from typing import List, Dict, Optional
  6. from datetime import datetime
  7. from urllib.parse import quote
  8. from openpyxl import Workbook
  9. from DrissionPage import ChromiumPage
  10. from DrissionPage.common import Actions
  • DrissionPage：基于Selenium和Requests的混合驱动库，支持无头/有头模式切换
  • OpenPyXL：用于将采集结果导出为Excel格式
  • Re/Json：处理API返回的JSON数据和正则匹配

2. 浏览器配置策略

  1. class ShortVideoCrawler:
  2.     def __init__(self):
  3.         self.browser = None
  4.         self.stop_event = threading.Event()
  6.     def _init_browser(self):
  7.         """多策略浏览器初始化"""
  8.         strategies = [
  9.             self._default_init,
  10.             self._custom_config_init,
  11.             self._connect_existing_browser
  12.         ]
  14.         for strategy in strategies:
  15.             if strategy():
  16.                 return True
  17.         return False

初始化策略优先级：

  1. 默认配置启动：直接创建新浏览器实例
  2. 自定义参数启动：配置无头模式、禁用GPU等参数
  3. 连接已有实例：通过调试端口复用已打开浏览器

三、核心实现模块

1. 浏览器自动化控制

  1. def _custom_config_init(self):
  2.     """自定义浏览器配置"""
  3.     try:
  4.         from DrissionPage import ChromiumOptions
  5.         co = ChromiumOptions()
  6.         config = {
  7.             'headless': False,
  8.             'arguments': [
  9.                 '--no-sandbox',
  10.                 '--disable-dev-shm-usage',
  11.                 '--disable-gpu',
  12.                 '--remote-debugging-port=9222'
  13.             ]
  14.         }
  15.         # 动态配置参数
  16.         for k, v in config.items():
  17.             if k == 'arguments':
  18.                 for arg in v:
  19.                     co.set_argument(arg)
  20.             else:
  21.                 getattr(co, k)(v)
  23.         self.browser = ChromiumPage(addr_or_opts=co)
  24.         return self._test_browser()
  25.     except Exception as e:
  26.         print(f"配置初始化失败: {str(e)}")
  27.         return False

关键配置说明：

  • --no-sandbox：解决Linux环境权限问题
  • --remote-debugging-port：启用调试端口便于实例复用
  • 动态参数配置模式支持灵活扩展

2. 网络请求监控与分析

通过抓包工具（如Fiddler/Wireshark）分析搜索请求流程：

  1. 访问搜索入口页获取基础Cookie
  2. 构造带关键词的搜索请求URL
  3. 解析返回的JSON格式搜索结果

典型请求参数结构：

  1. {
  2.     "keyword": "Python教程",
  3.     "offset": 0,
  4.     "count": 20,
  5.     "sort_type": 0,
  6.     "device_id": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
  7. }

3. 数据采集完整流程

  1. def search_keyword(self, keyword: str, max_results: int = 100):
  2.     """执行关键词搜索采集"""
  3.     if not self.browser:
  4.         self._init_browser()
  6.     base_url = "https://www.example-platform.com/search"
  7.     results = []
  8.     offset = 0
  10.     while len(results) < max_results:
  11.         # 构造请求参数
  12.         params = {
  13.             'keyword': keyword,
  14.             'offset': offset,
  15.             'count': min(20, max_results - len(results))
  16.         }
  18.         # 执行搜索请求
  19.         try:
  20.             # 模拟浏览器行为
  21.             self.browser.get(base_url)
  22.             self.browser.ele('input[name="q"]').input(keyword)
  23.             self.browser.ele('button[type="submit"]').click()
  24.             time.sleep(3)  # 等待页面加载
  26.             # 获取网络请求（实际开发中建议直接调用API）
  27.             html = self.browser.html
  28.             # 此处应替换为实际API调用或DOM解析
  29.             items = self._parse_search_results(html)
  31.             if not items:
  32.                 break
  34.             results.extend(items)
  35.             offset += len(items)
  36.             print(f"已采集 {len(results)} 条数据")
  38.         except Exception as e:
  39.             print(f"采集异常: {str(e)}")
  40.             break
  42.     return results[:max_results]

四、数据解析与存储

1. 结构化数据提取

  1. def _parse_search_results(self, html: str) -> List[Dict]:
  2.     """解析搜索结果HTML"""
  3.     # 实际开发中建议直接解析API返回的JSON
  4.     pattern = re.compile(
  5.         r'<div([^"]+)".*?'
  6.         r'<h3>([^<]+)</h3>.*?'
  7.         r'<span>(\d+)</span>.*?'
  8.         r'<span>(\d+)</span>',
  9.         re.S
  10.     )
  12.     items = []
  13.     for match in pattern.finditer(html):
  14.         items.append({
  15.             'video_id': match.group(1),
  16.             'title': match.group(2).strip(),
  17.             'play_count': int(match.group(3)),
  18.             'comment_count': int(match.group(4)),
  19.             'collect_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
  20.         })
  21.     return items

2. 多格式数据存储

  1. def save_to_excel(self, data: List[Dict], filename: str):
  2.     """保存为Excel文件"""
  3.     wb = Workbook()
  4.     ws = wb.active
  5.     ws.append(['视频ID', '标题', '播放量', '评论数', '采集时间'])
  7.     for item in data:
  8.         ws.append([
  9.             item['video_id'],
  10.             item['title'],
  11.             item['play_count'],
  12.             item['comment_count'],
  13.             item['collect_time']
  14.         ])
  16.     wb.save(filename)
  17.     print(f"数据已保存至 {filename}")

五、进阶优化技巧

1. 反爬策略应对

  • IP轮换：结合代理池实现请求IP多样化
  • 行为模拟：随机延迟+鼠标轨迹模拟
  • Cookie管理：持久化存储有效会话

2. 分布式采集架构

  1. class DistributedCrawler:
  2.     """分布式采集控制器"""
  3.     def __init__(self, worker_count: int = 3):
  4.         self.task_queue = Queue()
  5.         self.workers = []
  6.         for _ in range(worker_count):
  7.             w = ShortVideoCrawler()
  8.             self.workers.append(w)
  10.     def distribute_tasks(self, keywords: List[str]):
  11.         """任务分发"""
  12.         for kw in keywords:
  13.             self.task_queue.put(kw)
  15.     def start(self):
  16.         """启动工作线程"""
  17.         threads = []
  18.         for w in self.workers:
  19.             t = threading.Thread(target=self._worker_loop, args=(w,))
  20.             threads.append(t)
  21.             t.start()
  23.     def _worker_loop(self, crawler):
  24.         """工作线程循环"""
  25.         while not self.stop_event.is_set():
  26.             try:
  27.                 keyword = self.task_queue.get(timeout=10)
  28.                 results = crawler.search_keyword(keyword)
  29.                 self._save_results(keyword, results)
  30.             except Exception as e:
  31.                 print(f"工作线程异常: {str(e)}")

六、法律与伦理规范

  1. 合规性要求：

    • 严格遵守《网络安全法》相关条款
    • 避免采集用户隐私数据
    • 控制采集频率防止对目标平台造成负担

  2. 推荐实践：

    • 设置合理的请求间隔（建议3-5秒/次）
    • 仅采集公开可访问数据
    • 建立数据使用内部审批流程

本方案通过模块化设计实现了可扩展的数据采集框架，开发者可根据实际需求调整浏览器配置、解析规则和存储方式。对于大规模采集场景，建议结合消息队列和分布式任务调度系统进行优化升级。

最新文章