如何通过 API 搜索 X 社区内的帖子

curl --request POST \
  --url https://api.twexapi.io/twitter/community/search-tweets \
  --header 'Authorization: Bearer <token>' \
  --header 'Content-Type: application/json' \
  --data '{
    "community_id": "1234567890123456789",
    "target_count": 50,
    "query": "customer support"
  }'

1{
2  "code": 200,
3  "msg": "success",
4  "data": [
5    {
6      "tweet_id": "1803006263529541838",
7      "text": "Example community post about customer support",
8      "created_at_datetime": "2024-06-17T03:51:48.000Z",
9      "favorite_count": 18,
10      "retweet_count": 3,
11      "reply_count": 6,
12      "hashtags": ["support"]
13    }
14  ]
15}

1import os
2from typing import Any
3
4import requests
5
6API_URL = "https://api.twexapi.io/twitter/community/search-tweets"
7TOKEN = os.environ["TWEXAPI_BEARER_TOKEN"]
8
9def search_community_tweets(
10    community_id: str,
11    query: str,
12    *,
13    target_count: int = 50,
14) -> list[dict[str, Any]]:
15    payload = {
16        "community_id": community_id,
17        "target_count": target_count,
18        "query": query,
19    }
20
21    response = requests.post(
22        API_URL,
23        headers={
24            "Authorization": f"Bearer {TOKEN}",
25            "Content-Type": "application/json",
26        },
27        json=payload,
28        timeout=30,
29    )
30    response.raise_for_status()
31    return response.json().get("data", [])
32
33if __name__ == "__main__":
34    tweets = search_community_tweets(
35        "1234567890123456789",
36        "customer support",
37        target_count=25,
38    )
39    for tweet in tweets[:5]:
40        text = tweet.get("full_text") or tweet.get("text") or ""
41        print(tweet.get("tweet_id"), text[:140])

1def to_review_row(tweet: dict[str, Any]) -> dict[str, Any]:
2    favorite_count = tweet.get("favorite_count") or 0
3    retweet_count = tweet.get("retweet_count") or 0
4    reply_count = tweet.get("reply_count") or 0
5    quote_count = tweet.get("quote_count") or 0
6    engagement = favorite_count + retweet_count + reply_count + quote_count
7
8    return {
9        "tweet_id": tweet.get("tweet_id"),
10        "created_at": tweet.get("created_at_datetime") or tweet.get("created_at"),
11        "engagement": engagement,
12        "hashtags": ", ".join(tweet.get("hashtags") or []),
13        "text": tweet.get("full_text") or tweet.get("text") or "",
14        "url": f"https://x.com/i/status/{tweet.get('tweet_id')}",
15    }
16
17tweets = search_community_tweets(
18    "1234567890123456789",
19    "pricing",
20    target_count=100,
21)
22rows = [to_review_row(tweet) for tweet in tweets]
23rows.sort(key=lambda row: row["engagement"], reverse=True)
24
25for row in rows[:10]:
26    print(row["engagement"], row["url"], row["text"][:100])

1communities = {
2    "1234567890123456789": "Product builders",
3    "2345678901234567890": "Support leaders",
4    "3456789012345678901": "AI operators",
5}
6
7query = "onboarding"
8summary = []
9
10for community_id, name in communities.items():
11    tweets = search_community_tweets(community_id, query, target_count=50)
12    rows = [to_review_row(tweet) for tweet in tweets]
13    total_engagement = sum(row["engagement"] for row in rows)
14    summary.append(
15        {
16            "community_id": community_id,
17            "name": name,
18            "tweet_count": len(rows),
19            "total_engagement": total_engagement,
20        }
21    )
22
23for item in sorted(summary, key=lambda row: row["total_engagement"], reverse=True):
24    print(item["name"], item["tweet_count"], item["total_engagement"])