构建关键词社交监听工具

requests==2.32.3

from __future__ import annotations

import os
import sys
import urllib.parse

import requests

SERP_URL = "https://api.brightdata.com/request"
TRIGGER_URL = "https://api.brightdata.com/datasets/v3/trigger"

PLATFORMS = ["instagram.com", "tiktok.com", "x.com"]


def discover(keyword: str, api_key: str, zone: str) -> dict[str, list[dict]]:
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    results_by_platform: dict[str, list[dict]] = {}

    for site in PLATFORMS:
        query = f'site:{site} "{keyword}"'
        google_url = (
            "https://www.google.com/search"
            f"?q={urllib.parse.quote_plus(query)}&hl=en&gl=us"
        )

        response = requests.post(
            SERP_URL,
            headers=headers,
            json={
                "zone": zone,
                "url": google_url,
                "format": "raw",
                "data_format": "parsed_light",
            },
            timeout=180,
        )
        response.raise_for_status()
        organic = response.json().get("organic", [])
        results_by_platform[site] = organic

    return results_by_platform

import re

INSTAGRAM_POST_RE = re.compile(r"https://(?:www\.)?instagram\.com/(?:p|reel)/[^/?#]+")
TIKTOK_POST_RE    = re.compile(r"https://(?:www\.)?tiktok\.com/@[^/]+/video/\d+")
X_POST_RE         = re.compile(r"https://(?:www\.)?x\.com/[^/]+/status/\d+")


def classify(results_by_platform: dict[str, list[dict]]) -> dict[str, list[str]]:
    buckets: dict[str, list[str]] = {"instagram": [], "tiktok": [], "x": []}

    for organic in results_by_platform.values():
        for result in organic:
            link = result.get("link", "")
            if INSTAGRAM_POST_RE.match(link):
                buckets["instagram"].append(link)
            elif TIKTOK_POST_RE.match(link):
                buckets["tiktok"].append(link)
            elif X_POST_RE.match(link):
                buckets["x"].append(link)

    for platform in buckets:
        buckets[platform] = list(dict.fromkeys(buckets[platform]))

    return buckets

SOCIAL_DATASETS = {
    "instagram": "gd_lk5ns7kz21pck8jpis",  # Instagram posts
    "tiktok":    "gd_lu702nij2f790tmv9h",  # TikTok posts
    "x":         "gd_lwxkxvnf1cynvib9co",  # X posts
}


def trigger(platform: str, urls: list[str], api_key: str) -> str | None:
    if not urls:
        print(f"  {platform}: no URLs to trigger")
        return None

    response = requests.post(
        TRIGGER_URL,
        params={"dataset_id": SOCIAL_DATASETS[platform], "format": "json"},
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        },
        json=[{"url": url} for url in urls],
        timeout=30,
    )
    response.raise_for_status()
    snapshot_id = response.json().get("snapshot_id")
    print(f"  {platform}: {len(urls)} URLs -> snapshot {snapshot_id}")
    return snapshot_id

def main() -> int:
    api_key = os.environ.get("BRIGHT_DATA_API_KEY")
    zone = os.environ.get("BRIGHT_DATA_SERP_ZONE")

    if not (api_key and zone):
        print("Set BRIGHT_DATA_API_KEY and BRIGHT_DATA_SERP_ZONE before running.")
        return 1

    if len(sys.argv) < 2:
        print("Usage: python listen.py \"your keyword\"")
        return 1

    keyword = sys.argv[1]
    print(f"Discovering mentions of: {keyword}")

    results_by_platform = discover(keyword, api_key, zone)
    buckets = classify(results_by_platform)

    print("Classified URLs:")
    for platform, urls in buckets.items():
        print(f"  {platform}: {len(urls)}")

    print("Triggering scrapers:")
    for platform, urls in buckets.items():
        trigger(platform, urls, api_key)

    return 0


if __name__ == "__main__":
    sys.exit(main())