リストとコミュニティ

curl "https://api.sorsa.io/v3/list-members?list_id=1234567890" \
  -H "ApiKey: YOUR_API_KEY"

import requests

API_KEY = "YOUR_API_KEY"
BASE = "https://api.sorsa.io/v3"
HEADERS = {"ApiKey": API_KEY}


def get_list_members(list_id, max_pages=50):
    members, cursor = [], None

    for _ in range(max_pages):
        params = {"list_id": list_id}
        if cursor:
            params["next_cursor"] = cursor

        r = requests.get(f"{BASE}/list-members", headers=HEADERS, params=params, timeout=30)
        r.raise_for_status()
        data = r.json()

        members.extend(data.get("users", []))
        cursor = data.get("next_cursor")
        if not cursor:
            break

    return members


members = get_list_members("1234567890")
for u in members[:5]:
    print(f"@{u['username']} ({u['followers_count']:,} followers)")

def get_list_followers(list_link, max_pages=50):
    followers, cursor = [], None

    for _ in range(max_pages):
        params = {"list_link": list_link}
        if cursor:
            params["next_cursor"] = cursor

        r = requests.get(f"{BASE}/list-followers", headers=HEADERS, params=params, timeout=30)
        r.raise_for_status()
        data = r.json()

        followers.extend(data.get("users", []))
        cursor = data.get("next_cursor")
        if not cursor:
            break

    return followers


subs = get_list_followers("https://x.com/i/lists/1234567890")
print(f"{len(subs)} subscribers")

def get_list_tweets(list_id, max_pages=10):
    tweets, cursor = [], None

    for _ in range(max_pages):
        params = {"list_id": list_id}
        if cursor:
            params["next_cursor"] = cursor

        r = requests.get(f"{BASE}/list-tweets", headers=HEADERS, params=params, timeout=30)
        r.raise_for_status()
        data = r.json()

        tweets.extend(data.get("tweets", []))
        cursor = data.get("next_cursor")
        if not cursor:
            break

    return tweets


feed = get_list_tweets("1234567890", max_pages=10)
for t in feed[:5]:
    print(f"@{t['user']['username']}: {t['full_text'][:80]}")

def get_community_members(community_link, max_pages=20):
    members, cursor = [], None

    for _ in range(max_pages):
        body = {"community_link": community_link}
        if cursor:
            body["next_cursor"] = cursor

        r = requests.post(
            f"{BASE}/community-members",
            headers={**HEADERS, "Content-Type": "application/json"},
            json=body,
            timeout=30,
        )
        r.raise_for_status()
        data = r.json()

        members.extend(data.get("users", []))
        cursor = data.get("next_cursor")
        if not cursor:
            break

    return members

def get_community_tweets(community_id, order="latest", max_pages=10):
    tweets, cursor = [], None

    for _ in range(max_pages):
        body = {"community_id": community_id, "order": order}
        if cursor:
            body["next_cursor"] = cursor

        r = requests.post(
            f"{BASE}/community-tweets",
            headers={**HEADERS, "Content-Type": "application/json"},
            json=body,
            timeout=30,
        )
        r.raise_for_status()
        data = r.json()

        tweets.extend(data.get("tweets", []))
        cursor = data.get("next_cursor")
        if not cursor:
            break

    return tweets

def search_community_tweets(community_link, query, order="popular", max_pages=5):
    tweets, cursor = [], None

    for _ in range(max_pages):
        body = {"community_link": community_link, "query": query, "order": order}
        if cursor:
            body["next_cursor"] = cursor

        r = requests.post(
            f"{BASE}/community-search-tweets",
            headers={**HEADERS, "Content-Type": "application/json"},
            json=body,
            timeout=30,
        )
        r.raise_for_status()
        data = r.json()

        tweets.extend(data.get("tweets", []))
        cursor = data.get("next_cursor")
        if not cursor:
            break

    return tweets

import csv

def export_users_to_csv(users, path):
    fields = ["id", "username", "display_name", "description",
              "followers_count", "tweets_count", "verified", "location"]

    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fields)
        writer.writeheader()
        for u in users:
            writer.writerow({
                "id": u.get("id", ""),
                "username": u.get("username", ""),
                "display_name": u.get("display_name", ""),
                "description": (u.get("description") or "").replace("\n", " "),
                "followers_count": u.get("followers_count", 0),
                "tweets_count": u.get("tweets_count", 0),
                "verified": u.get("verified", False),
                "location": u.get("location", ""),
            })


export_users_to_csv(get_list_members("1234567890"), "members.csv")