设计一个带有质量检查的Google搜索数据管道

Michael Lee

Expert Network Defense Engineer

15-Sep-2026

TL;DR:

一个谷歌搜索数据管道需要记录运行情况以及其结果。 空的、待处理的、失败的和未映射的观察都可能没有预计的自然行。
在应用分析规则之前保留原始捕获。 派生表和质量报告可以重建；原始证据应保持不变。
质量规则必须与报告匹配。 一个可用的响应容器并不保证有效的位置或可比的搜索上下文。

搜索管道可以成功写入文件，但仍产生误导性分析。响应可能处于待处理状态，映射器可能会丢弃格式不正确的行，或者报告可能会将不同市场合并到同一个关键词下。仅靠存储的成功并不能证明所得到的观察能回答预期的问题。

无刮痕谷歌搜索API 提供了收集数据。这里描述的谷歌搜索数据管道在其周围增加了应用程序拥有的存储、验证和报告。调度、历史保留、数据库和质量检查是管道的责任，而不是搜索API所声称的内置服务。

Pipeline at a Glance

该管道经历请求规划、收集、原始存储、投影、质量审查和报告。在这些阶段之间保持稳定的参考，以便分析师可以追踪图表回到确切的请求和响应。

在响应到达之前，计划的工作就已经存在。其收集结果随后变为运行记录的证据，即使没有自然项目可以投影。将原始捕获单独存储，不与派生结果行混合，并保留用于将记录纳入报告的质量决定。

下面的本地程序检查保存的捕获并打印质量报告。它不收集数据，不调度工作，不创建数据仓库，也不纠正源值。这样的有限责任使其输出更易于检查和替换。

Stage 1 — Define the Request and Comparison Scope

请求定义了观察的上下文。每当提供时，请保留查询、国家、语言、位置、输入模式和页面偏移。谷歌搜索参数解释了仅根据关键词不足以识别可比观察的原因。

精确序列化的请求是保守的比较关键。更改偏移量会更改收集的切片；更改国家或措辞会改变研究上下文。如果管道后来将相似的配置进行分组，请记录规范化规则并保留原始请求。

规划运行标识符、调度的工作和已完成观察之间的关系。错过的计划收集即使没有API响应，也应在操作覆盖中可见。本地检查器无法发现它从未给出过的工作；调度器或工作账本必须提供该清单。

Stage 2 — Capture the Outcome Before Projecting Rows

谷歌搜索请求工作流将行为者 scraper.google.search 发送至 POST https://api.scrapeless.com/api/v1/scraper/request 并在 x-api-token 中包含API密钥。HTTP 200携带任务数据，而HTTP 201表示任务待处理。在读取自然数组之前，请保留这种区分。

使用一个捕获信封，其中包含提交的请求、原始响应、记录的HTTP状态、运行标识符和客户端收到时间。保持身份验证头部不在共享证据文件中。该信封是您的应用程序存储合同，而不是对服务原生响应包装的声明。

进行实时收集需要一个账户密钥，而这篇文章中并未执行实时收集。待处理任务的完成还需要一个单独验证的工作流。本地检查器在保存的捕获上运行而不需要凭证，使用合成输入来测试其规则。

Stage 3 — Preserve History and Build Derived Tables

原始快照在接受归档后应保持不变。后来的响应或修订解析器应创建新的记录或投影版本，而不是覆盖早期报告背后的证据。

关系模型可以使用一个运行表加上以运行标识符和源序号为键的子自然结果行。返回的位置保持为一个单独的属性。SQLite的 外键规则 描述了在使用该存储选项时如何对相关记录进行约束。
提交一个运行及其派生行一起，当数据库模型要求它们保持一致时。 事务模型 提供相关的数据库行为。您的摄取应用程序仍然必须定义冲突处理、连接配置和每个事务的边界。

即使便利列无法表示，仍然在原始 JSON 中保留不寻常的结果值。这让未来的映射器能够恢复细节，而无需重新收集其输出可能已更改的搜索。

使用 Scrapeless 开始抓取

利用 Scrapeless 提升您的网页抓取和自动化工作流程！
今天注册并获得 5 美元的免费积分 — 无需信用卡。

现在在 Scrapeless 仪表板兑现您的免费积分。

第 4 阶段 — 应用明确的质量合同

质量检查应回答报告特定的问题。示例检查捕获是否适合保守的位置报告：运行身份、请求结构、收据时间存在、自然容器形状、链接字符串、正整数位置和完全重复的链接。

检查器独立于注释记录集合状态。一个存在的有机数组可以是 observed，同时由于位置缺失而不符合位置报告资格。这将收集的内容与特定报告可以安全使用的内容分开。

缺失的时间戳被标记，但代码不验证时间戳语法或时效性。链接检查建立非空字符串，而非可到达或可信的目的地。上下文等价性、日期范围覆盖和跨文件运行 ID 的唯一性也需要在周围的管道中进行单独检查。

JSON 值模型 是数组、对象、空值和标量之间区分的基础。不要将不支持的容器强制转换为空数组，仅仅为了让质量报告成功。

第 5 阶段 — 运行本地捕获检查器

将此程序保存为 quality_check.py。它只需要 Python 和捕获文件。使用实际的保存文件名运行 python3 quality_check.py capture-a.json capture-b.json；程序将 JSON 打印到标准输出，并不会更改其输入。

python Copy

import argparse
import json
from collections import Counter
from pathlib import Path


def inspect(record):
    notes = []
    if not isinstance(record, dict):
        return {'state': 'invalid_capture', 'notes': ['capture_not_object'], 'rows': None, 'eligible_for_position_report': False}
    request = record.get('request')
    if not isinstance(record.get('run_id'), str) or not record['run_id'].strip():
        notes.append('run_id_missing')
    if (not isinstance(request, dict) or request.get('actor') != 'scraper.google.search'
            or not isinstance(request.get('input'), dict)):
        notes.append('request_contract_invalid')
    if not isinstance(record.get('received_at'), str) or not record['received_at'].strip():
        notes.append('receipt_time_missing')
    status, payload = record.get('http_status'), record.get('response')
    rows = payload.get('organic_results') if isinstance(payload, dict) else None
    count = None
    if status == 201:
        state = 'pending'
        if not isinstance(payload, dict) or not isinstance(payload.get('taskId'), str) or not payload['taskId'].strip():
            notes.append('task_id_missing')
    elif status is None:
        state = 'transport_error'
    elif status != 200:
        state = 'http_error'
    elif not isinstance(rows, list) or any(not isinstance(row, dict) for row in rows):
        state = 'unmapped'
    else:
        state, count = ('observed' if rows else 'empty'), len(rows)
        links = []
        for ordinal, row in enumerate(rows):
            link, position = row.get('link'), row.get('position')
            if not isinstance(link, str) or not link.strip():
                notes.append(f'row_{ordinal}_link_missing_or_invalid')
            else:
                links.append(link)
            if type(position) is not int or position < 1:
                notes.append(f'row_{ordinal}_position_missing_or_invalid')
        if len(links) != len(set(links)):
            notes.append('duplicate_exact_links')
    return {'run_id': record.get('run_id'), 'state': state, 'rows': count,
            'notes': notes, 'eligible_for_position_report': state in ('observed', 'empty') and not notes}


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('captures', nargs='+')
    args = parser.parse_args()
    reports = []
    for filename in args.captures:
        try:
            report = inspect(json.loads(Path(filename).read_text(encoding='utf-8')))
        except (OSError, json.JSONDecodeError, UnicodeError) as error:
            report = {'state': 'unreadable_capture', 'rows': None, 'notes': [type(error).__name__],
                      'eligible_for_position_report': False}
        reports.append(dict(report, source=filename))
    print(json.dumps({'captures': reports, 'states': dict(Counter(x['state'] for x in reports))},
                     ensure_ascii=False, indent=2))