如何通过 API 检测 X 付费推广

1import requests
2from datetime import datetime, timezone
3
4API_URL = "https://api.twexapi.io/twitter/tweets/lookup"
5TOKEN = "YOUR_TWEXAPI_BEARER_TOKEN"
6
7def clean_tweet_ids(tweet_ids):
8    cleaned = []
9    seen = set()
10
11    for value in tweet_ids:
12        tweet_id = str(value).strip()
13        if not tweet_id or tweet_id in seen:
14            continue
15        cleaned.append(tweet_id)
16        seen.add(tweet_id)
17
18    return cleaned
19
20def lookup_tweets(tweet_ids):
21    cleaned_ids = clean_tweet_ids(tweet_ids)
22    if not cleaned_ids:
23        return []
24
25    response = requests.post(
26        API_URL,
27        headers={
28            "Authorization": f"Bearer {TOKEN}",
29            "Content-Type": "application/json",
30            "Accept": "application/json",
31        },
32        json=cleaned_ids,
33        timeout=30,
34    )
35    response.raise_for_status()
36    payload = response.json()
37
38    if payload.get("code", 200) >= 400:
39        raise RuntimeError(payload.get("msg", "TwexAPI returned an error"))
40
41    return payload.get("data") or []
42
43def classify_paid_promotions(tweet_ids, candidate_source):
44    fetched_at = datetime.now(timezone.utc).isoformat()
45    rows = []
46
47    for tweet in lookup_tweets(tweet_ids):
48        if not tweet:
49            continue
50
51        text = tweet.get("full_text") or tweet.get("text") or ""
52        user = tweet.get("user") or {}
53        is_paid = tweet.get("is_paid_promotion") is True
54
55        rows.append({
56            "tweet_id": tweet.get("tweet_id") or tweet.get("id"),
57            "is_paid_promotion": is_paid,
58            "promotion_label": "paid" if is_paid else "not_marked_paid",
59            "candidate_source": candidate_source,
60            "screen_name": user.get("screen_name"),
61            "created_at_datetime": tweet.get("created_at_datetime"),
62            "favorite_count": tweet.get("favorite_count"),
63            "retweet_count": tweet.get("retweet_count"),
64            "reply_count": tweet.get("reply_count"),
65            "view_count": tweet.get("view_count"),
66            "text_preview": text[:160],
67            "fetched_at": fetched_at,
68            "raw": tweet,
69        })
70
71    return rows
72
73candidate_ids = ["123456789", "987654321", "555666777"]
74results = classify_paid_promotions(candidate_ids, candidate_source="competitor_launch_search")
75
76paid_rows = [row for row in results if row["is_paid_promotion"]]
77print(f"Checked {len(results)} tweets; paid promotion marked: {len(paid_rows)}")