优化 API 使用

import requests

API_KEY = "YOUR_API_KEY"

{
  "tweets": [
    {
      "id": "2029914600217473314",
      "full_text": "Great thread on API design patterns...",
      "likes_count": 142,
      "user": {
        "id": "1422280682240450563",
        "username": "dev_sarah",
        "display_name": "Sarah Chen",
        "description": "Staff engineer @stripe. APIs, distributed systems.",
        "followers_count": 12400,
        "followings_count": 890,
        "tweets_count": 4521,
        "verified": true,
        "location": "San Francisco",
        "created_at": "2021-02-01T09:15:22Z"
      }
    }
  ]
}

# Collect unique users from a tweet search - zero extra API calls
seen_ids = set()
unique_users = []

for tweet in search_results:
    user = tweet["user"]
    if user["id"] not in seen_ids:
        seen_ids.add(user["id"])
        unique_users.append(user)

print(f"Found {len(unique_users)} unique users from {len(search_results)} tweets")

# A separate /info call for each account would use 10 requests.

# Efficient: 10 accounts = 1 request
resp = requests.get(
    "https://api.sorsa.io/v3/info-batch",
    headers={"ApiKey": API_KEY},
    params={"usernames": ["NASA", "SpaceX", "Tesla", "OpenAI", "stripe",
                           "shopify", "vercel", "github", "notion", "linear"]},
)
resp.raise_for_status()
profiles = resp.json().get("users", [])

# A separate /tweet-info call for each tweet would use up to 100 requests.

# Efficient: 100 tweets = 1 request
resp = requests.post(
    "https://api.sorsa.io/v3/tweet-info-bulk",
    headers={"ApiKey": API_KEY, "Content-Type": "application/json"},
    json={
        "tweet_links": [
            "https://x.com/user/status/111111",
            "https://x.com/user/status/222222",
            # ... up to 100 links
        ]
    },
)
resp.raise_for_status()
tweets = resp.json().get("tweets", [])

# Separate /user-tweets calls would use 30 requests per polling cycle.
LIST_ID = "YOUR_LIST_ID"

# Efficient: 1 request covers all 30 accounts
resp = requests.get(
    f"https://api.sorsa.io/v3/list-tweets?list_id={LIST_ID}",
    headers={"ApiKey": API_KEY},
    timeout=30,
)
resp.raise_for_status()

# Resolving an ID and then fetching its profile would use two requests.

# Efficient: 1 request per account
response = requests.get(
    "https://api.sorsa.io/v3/info",
    headers={"ApiKey": API_KEY},
    params={"username": "stripe"},       # accepts username, user_id, or user_link
    timeout=30,
)
response.raise_for_status()
profile = response.json()
# profile already contains the user ID, plus everything else

import sqlite3

db = sqlite3.connect("audience.db")
db.execute("""
    CREATE TABLE IF NOT EXISTS users (
        user_id TEXT PRIMARY KEY, username TEXT, display_name TEXT,
        description TEXT, followers_count INTEGER, tweets_count INTEGER,
        verified INTEGER, updated_at TEXT
    )
""")

def upsert_user(db, user):
    """Insert or update a user record keyed by permanent User ID."""
    db.execute("""
        INSERT INTO users (user_id, username, display_name, description,
                          followers_count, tweets_count, verified, updated_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, datetime('now'))
        ON CONFLICT(user_id) DO UPDATE SET
            username = excluded.username,
            display_name = excluded.display_name,
            description = excluded.description,
            followers_count = excluded.followers_count,
            tweets_count = excluded.tweets_count,
            verified = excluded.verified,
            updated_at = datetime('now')
    """, (
        user["id"], user["username"], user.get("display_name", ""),
        user.get("description", ""), user.get("followers_count", 0),
        user.get("tweets_count", 0), user.get("verified", False),
    ))
    db.commit()


# Every time you encounter a user in any API response, upsert:
for tweet in search_results:
    upsert_user(db, tweet["user"])
    # The user's profile data stays fresh without separate /info calls