将 Google 搜索 JSON 导出为 CSV 进行分析

import argparse
import csv
import json
from pathlib import Path

FIELDS = ["run_id", "requested_at", "received_at", "q", "request_json", "ordinal",
          "position", "position_state", "title", "title_state", "link", "link_state",
          "snippet", "snippet_state"]

def spreadsheet_text(value):
    text = "" if value is None else str(value)
    stripped = text.lstrip()
    if (stripped.startswith(("=", "+", "-", "@", "＝", "＋", "－", "＠"))
            or text.startswith(("\t", "\r", "\n"))):
        return "'" + text
    return text

def field(row, name):
    if name not in row:
        return "", "missing"
    value = row[name]
    if value is None:
        return "", "null"
    if name == "position":
        return (value, "value") if type(value) is int and value > 0 else ("", "invalid")
    if isinstance(value, str):
        return spreadsheet_text(value), "value"
    return spreadsheet_text(json.dumps(value, ensure_ascii=False)), "unexpected_type"

def export(source, target):
    source, target = Path(source), Path(target)
    sidecar = target.with_suffix(target.suffix + ".run.json")
    if source.resolve() in (target.resolve(), sidecar.resolve()):
        raise ValueError("Output paths must differ from input")
    record = json.loads(source.read_text(encoding="utf-8"))
    request = record.get("request")
    if not isinstance(request, dict) or not isinstance(request.get("input"), dict):
        raise ValueError("Expected a capture record with request.input")
    payload = record.get("response")
    rows = payload.get("organic_results") if isinstance(payload, dict) else None
    status = record.get("http_status")
    if status == 201:
        state, rows = "pending", []
    elif status != 200:
        state, rows = ("transport_error" if status is None else "http_error"), []
    elif not isinstance(rows, list) or any(not isinstance(x, dict) for x in rows):
        state, rows = "unmapped", []
    else:
        state = "observed" if rows else "empty"
    context = {
        "run_id": spreadsheet_text(record.get("run_id")),
        "requested_at": spreadsheet_text(record.get("requested_at")),
        "received_at": spreadsheet_text(record.get("received_at")),
        "q": spreadsheet_text(request["input"].get("q")),
        "request_json": spreadsheet_text(json.dumps(request, ensure_ascii=False, sort_keys=True)),
    }
    with target.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=FIELDS)
        writer.writeheader()
        for ordinal, item in enumerate(rows):
            output = dict(context, ordinal=ordinal)
            for name in ("position", "title", "link", "snippet"):
                output[name], output[name + "_state"] = field(item, name)
            writer.writerow(output)
    sidecar.write_text(json.dumps({"source": str(source), "run_id": record.get("run_id"),
        "state": state, "rows": len(rows), "request": request,
        "requested_at": record.get("requested_at"), "received_at": record.get("received_at"),
        "export_policy": "spreadsheet_text_prefix_v1; original values remain in source JSON"},
        ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"Exported {len(rows)} organic rows; state={state}; metadata={sidecar}")
    return state, len(rows)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("source")
    parser.add_argument("target")
    args = parser.parse_args()
    export(args.source, args.target)