通过一个管道在六个AI答案引擎中跟踪您的品牌

export SCRAPELESS_API_KEY=your_api_token_here

# capture.py — 在六个AI问答引擎上运行一个提示，存储原始答案
import json
import os
import time

import requests

ENDPOINT = "https://api.scrapeless.com/api/v2/scraper/execute"
HEADERS = {
    "Content-Type": "application/json",
    "x-api-token": os.environ["SCRAPELESS_API_KEY"],
}

PROMPT = "对于JavaScript密集型网站，最佳的网页抓取API是什么？"
COUNTRY = "US"

ENGINES = {
    "chatgpt": {"actor": "scraper.chatgpt", "extra": {}},
    "grok": {"actor": "scraper.grok", "extra": {"mode": "MODEL_MODE_FAST"}},
    "gemini": {"actor": "scraper.gemini", "extra": {}},
    "perplexity": {"actor": "scraper.perplexity", "extra": {"web_search": True}},
    "copilot": {"actor": "scraper.copilot", "extra": {"mode": "smart"}},
    "google-ai-overview": {"actor": "scraper.overview", "extra": {}},
}

with open("answers.jsonl", "w", encoding="utf-8") as out:
    for platform, spec in ENGINES.items():
        payload = {
            "actor": spec["actor"],
            "input": {"prompt": PROMPT, "country": COUNTRY, **spec["extra"]},
        }
        resp = requests.post(ENDPOINT, headers=HEADERS, json=payload, timeout=300)
        resp.raise_for_status()
        data = resp.json()
        out.write(json.dumps({

"平台": platform,
            "提示": PROMPT,
            "国家": COUNTRY,
            "捕获时间": int(time.time()),
            "状态": data.get("status"),
            "任务ID": data.get("task_id"),
            "任务结果": data.get("task_result"),
        }) + "\n")
        print(f"{platform}: {data.get('status')}")

# normalize.py — answers.jsonl -> citations.jsonl (每个引用来源一行)
import json
from urllib.parse import urlparse

# 平台 -> task_result内的(array_field, url_key)对列表
CITATION_FIELDS = {
    "chatgpt": [("content_references", "url")],
    "grok": [("web_search_results", "url"), ("x_search_results", "url")],
    "gemini": [("citations", "url")],
    "perplexity": [("web_results", "url")],
    "copilot": [("citations", "url")],
    "google-ai-overview": [("source", "url")],
}

with open("answers.jsonl", encoding="utf-8") as inp, \
     open("citations.jsonl", "w", encoding="utf-8") as out:
    for line in inp:
        row = json.loads(line)
        result = row.get("task_result") or {}
        for field, url_key in CITATION_FIELDS[row["platform"]]:
            for entry in result.get(field) or []:
                url = entry.get(url_key) or ""
                if not url.startswith("http"):
                    continue
                out.write(json.dumps({
                    "平台": row["platform"],
                    "提示": row["prompt"],
                    "国家": row["country"],
                    "捕获时间": row["captured_at"],
                    "面板": field,
                    "域名": urlparse(url).netloc.removeprefix("www."),
                    "URL": url,
                    "标题": entry.get("title") or entry.get("name") or "",
                }) + "\n")

print(sum(1 for _ in open("citations.jsonl", encoding="utf-8")), "引用规范化完成")

# report.py — citations.jsonl -> 每个平台的引用份额表
import json
import os
from collections import Counter, defaultdict

品牌 = os.environ.get("BRAND_DOMAIN", "scrapeless.com")

per_platform = defaultdict(Counter)
with open("citations.jsonl", encoding="utf-8") as inp:
    for line in inp:
        row = json.loads(line)
        per_platform[row["platform"]][row["domain"]] += 1

for platform, counts in per_platform.items():
    total = sum(counts.values())
    brand_hits = counts.get(品牌, 0)
    print(f"\n{platform} — {total} 篇引用 · {品牌}: {brand_hits}")
    for domain, n in counts.most_common(5):
        marker = " ←" if domain == 品牌 else ""
        print(f"  {n:>3}  {domain}{marker}")