从零构建 Google Trends 爬虫 API：单文件架构的工程实践

1
def _parse_trend_response(raw_text: str) -> list[dict]:
2
    """解析 batchexecute RPC 响应中的趋势数据（旧版 Google 接口）"""
3
    sentinel = 'DqDTgb'
4
    idx = raw_text.find(sentinel)
5
    if idx == -1:
6
        return []
7
    # 截取 JSON 载荷并反序列化
8
    ...

1
def _parse_dom_row(row_element) -> dict | None:
2
    """解析 DOM 渲染的热搜行（新版 Google Trends 页面）"""
3
    title = row_element.query_selector('div.title-and-articles > div.title')
4
    searches = row_element.query_selector('div.search-count-title')
5
    ...

1
def _evaluate_push_response(response_json: dict, batch_size: int) -> tuple[bool, str]:
2
    """
3
    在 HTTP 200 之上再校验 ODS 业务层响应。
4
    ODS 在数据校验失败时仍返回 HTTP 200，必须检查业务字段。
5
    """
6
    code = str(response_json.get('code', ''))
7
    success_flag = response_json.get('success', False)
8

9
    if code == '200' or success_flag is True:
10
        return True, 'ok'
11

12
    msg = response_json.get('msg') or response_json.get('message', '未知业务错误')
13
    return False, f"ODS 业务层拒绝: code={code}, msg={msg}"

1
def _save_pending_rows(rows: list[dict]) -> None:
2
    """将推送失败的数据行持久化到本地 pending 目录"""
3
    _PENDING_DIR.mkdir(parents=True, exist_ok=True)
4
    ts = datetime.now().strftime('%Y%m%d_%H%M%S_%f')
5
    fpath = _PENDING_DIR / f'pending_{ts}.json'
6
    fpath.write_text(json.dumps(rows, ensure_ascii=False), encoding='utf-8')

1
def _flush_pending_rows(_skip_flush: bool = False) -> None:
2
    """重推 pending 目录中所有历史失败文件，成功后删除文件。
3
    _skip_flush=True 防止递归调用导致无限循环。
4
    """
5
    if _skip_flush:
6
        return
7
    for fpath in sorted(_PENDING_DIR.glob('pending_*.json')):
8
        rows = json.loads(fpath.read_text(encoding='utf-8'))
9
        ok, _ = push_to_server_api(rows, _skip_flush=True)
10
        if ok:
11
            fpath.unlink()   # 推送成功，删除缓存文件

1
def with_retry(fn, max_retries: int = 3, base_delay: float = 10.0):
2
    """
3
    带指数退避的重试包装器。
4
    等待时间序列：10s → 20s → 40s → 80s（上限 120s）
5
    """
6
    for attempt in range(1, max_retries + 2):
7
        try:
8
            return fn()
9
        except Exception as e:
10
            if attempt > max_retries:
11
                raise
12
            delay = min(base_delay * (2 ** (attempt - 1)), 120.0)
13
            logger.warning(f"第 {attempt} 次失败，{delay:.0f}s 后重试: {e}")
14
            time.sleep(delay)

1
def _build_row(item: dict, geo: str, gather_time_ms: int, rank: int) -> dict:
2
    # Step 1：将 url + 时间戳 + rank + 标题 纳入 mid 生成
3
    # 关键：必须包含 rank 和 title，否则同一毫秒内多行会产生相同 mid
4
    mid_src = f"{item['url']}|{gather_time_ms}|{rank}|{item['title']}"
5
    mid = hashlib.md5(mid_src.encode()).hexdigest()
6

7
    # Step 2：对域名做归一化后生成 site_id
8
    domain = re.sub(r'^www\.', '', urlparse(item['url']).netloc)
9
    site_id = hashlib.md5(domain.encode()).hexdigest()
10

11
    # Step 3：复合主键
12
    row_id = hashlib.md5(f"{mid}{site_id}".encode()).hexdigest()
13

14
    return {"id": row_id, "mid": mid, "site_id": site_id, ...}

1
def _init_scheduler():
2
    scheduler = BackgroundScheduler()
3
    scheduler.add_job(
4
        func=_scheduled_crawl_all,
5
        trigger=CronTrigger(minute=0),   # 每小时整点
6
        id='crawl_all_hourly',
7
        max_instances=1,                 # 防止任务重叠
8
        misfire_grace_time=300,          # 5分钟内补执行
9
        replace_existing=True,
10
    )
11
    scheduler.start()