API で Twitter の対象オーディエンスを見つける方法

import requests, time

API_KEY = "YOUR_API_KEY"
BASE = "https://api.sorsa.io/v3"
AUTH = {"ApiKey": API_KEY}                                  # GET endpoints
POST_HEADERS = {**AUTH, "Content-Type": "application/json"} # POST endpoints

def find_users_by_bio(query, max_pages=10):
    users, cursor = [], None
    for _ in range(max_pages):
        body = {"query": query}
        if cursor:
            body["next_cursor"] = cursor
        resp = requests.post(f"{BASE}/search-users", headers=POST_HEADERS,
                             json=body, timeout=30)
        resp.raise_for_status()
        data = resp.json()
        users.extend(data.get("users", []))
        cursor = data.get("next_cursor")
        if not cursor:
            break
        time.sleep(0.1)
    return users

users = find_users_by_bio("machine learning engineer")

async function findUsersByBio(query, maxPages = 10) {
  const all = [];
  let cursor = null;
  for (let i = 0; i < maxPages; i++) {
    const body = { query };
    if (cursor) body.next_cursor = cursor;
    const resp = await fetch("https://api.sorsa.io/v3/search-users", {
      method: "POST",
      headers: { ApiKey: "YOUR_API_KEY", "Content-Type": "application/json" },
      body: JSON.stringify(body),
    });
    if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
    const data = await resp.json();
    all.push(...(data.users || []));
    cursor = data.next_cursor;
    if (!cursor) break;
    await new Promise((r) => setTimeout(r, 100));
  }
  return all;
}

qualified = [
    u for u in users
    if u.get("followers_count", 0) >= 1000
    and u.get("tweets_count", 0) >= 100
    and not u.get("protected", False)
]

def get_followers(username, max_pages=20):
    followers, cursor = [], None
    for _ in range(max_pages):
        params = {"username": username}
        if cursor:
            params["next_cursor"] = cursor
        resp = requests.get(f"{BASE}/followers", headers=AUTH,
                            params=params, timeout=30)
        resp.raise_for_status()
        data = resp.json()
        followers.extend(data.get("users", []))
        cursor = data.get("next_cursor")
        if not cursor:
            break
        time.sleep(0.1)
    return followers

followers = get_followers("competitor_handle", max_pages=20)

from collections import Counter

competitors = ["competitor_a", "competitor_b", "competitor_c"]
follower_sets = {h: {f["id"] for f in get_followers(h, max_pages=10)}
                 for h in competitors}

all_ids = [uid for ids in follower_sets.values() for uid in ids]
overlap = {uid for uid, c in Counter(all_ids).items() if c >= 2}

def get_community_members(community_id, max_pages=20):
    members, cursor = [], None
    for _ in range(max_pages):
        body = {"community_link": community_id}
        if cursor:
            body["next_cursor"] = cursor
        resp = requests.post(f"{BASE}/community-members", headers=POST_HEADERS,
                             json=body, timeout=30)
        resp.raise_for_status()
        data = resp.json()
        members.extend(data.get("users", []))
        cursor = data.get("next_cursor")
        if not cursor:
            break
        time.sleep(0.1)
    return members

def find_active_voices(query, min_followers=100, max_pages=10):
    seen, voices, cursor = set(), [], None
    for _ in range(max_pages):
        body = {"query": query, "order": "latest"}
        if cursor:
            body["next_cursor"] = cursor
        resp = requests.post(f"{BASE}/search-tweets", headers=POST_HEADERS,
                             json=body, timeout=30)
        resp.raise_for_status()
        data = resp.json()
        for tw in data.get("tweets", []):
            u = tw.get("user", {})
            uid = u.get("id")
            if uid and uid not in seen and u.get("followers_count", 0) >= min_followers:
                seen.add(uid)
                voices.append({
                    "username": u.get("username"),
                    "followers": u.get("followers_count", 0),
                    "bio": u.get("description", ""),
                    "sample_tweet": (tw.get("full_text") or "")[:160],
                })
        cursor = data.get("next_cursor")
        if not cursor:
            break
        time.sleep(0.1)
    return voices

voices = find_active_voices(
    '("need a CRM" OR "looking for a CRM") lang:en -filter:retweets')

def get_verified_followers(username, max_pages=10):
    verified, cursor = [], None
    for _ in range(max_pages):
        params = {"username": username}
        if cursor:
            params["next_cursor"] = cursor
        resp = requests.get(f"{BASE}/verified-followers", headers=AUTH,
                            params=params, timeout=30)
        resp.raise_for_status()
        data = resp.json()
        verified.extend(data.get("users", []))
        cursor = data.get("next_cursor")
        if not cursor:
            break
        time.sleep(0.1)
    return verified

verified = get_verified_followers("openai")
verified.sort(key=lambda u: u.get("followers_count", 0), reverse=True)

def get_retweeters(tweet_link, max_pages=10):
    users, cursor = [], None
    for _ in range(max_pages):
        body = {"tweet_link": tweet_link}
        if cursor:
            body["next_cursor"] = cursor
        resp = requests.post(f"{BASE}/retweeters", headers=POST_HEADERS,
                             json=body, timeout=30)
        resp.raise_for_status()
        data = resp.json()
        users.extend(data.get("users", []))
        cursor = data.get("next_cursor")
        if not cursor:
            break
        time.sleep(0.1)
    return users

def get_quoters(tweet_link, max_pages=10):
    quotes, cursor = [], None
    for _ in range(max_pages):
        body = {"tweet_link": tweet_link}
        if cursor:
            body["next_cursor"] = cursor
        resp = requests.post(f"{BASE}/quotes", headers=POST_HEADERS,
                             json=body, timeout=30)
        resp.raise_for_status()
        data = resp.json()
        quotes.extend(data.get("tweets", []))
        cursor = data.get("next_cursor")
        if not cursor:
            break
        time.sleep(0.1)
    return quotes

def score_by_source(by_source):
    """by_source: {source_name: [user objects]} -> deduped users with source_count."""
    index = {}
    for source, users in by_source.items():
        for u in users:
            uid = u["id"]
            index.setdefault(uid, {"user": u, "sources": set()})["sources"].add(source)

    result = []
    for entry in index.values():
        u = entry["user"].copy()
        u["source_count"] = len(entry["sources"])
        u["sources"] = sorted(entry["sources"])
        result.append(u)

    result.sort(key=lambda r: (-r["source_count"], -r.get("followers_count", 0)))
    return result

combined = score_by_source({
    "followers_competitor_a": followers,
    "bio_ml_engineer": users,
    "intent_crm": voices,
    "community_indie_hackers": get_community_members("1966045657589813686"),
})

from datetime import datetime, timezone, timedelta

def is_quality_account(user, min_followers=500, min_tweets=100, max_following_ratio=10):
    if user.get("protected", False):
        return False
    if user.get("followers_count", 0) < min_followers:
        return False
    if user.get("tweets_count", 0) < min_tweets:
        return False
    followers = user.get("followers_count", 1)
    if user.get("followings_count", 0) > followers * max_following_ratio:
        return False
    created = user.get("created_at")
    if created:
        try:
            dt = datetime.fromisoformat(created.replace("Z", "+00:00"))
            if dt > datetime.now(timezone.utc) - timedelta(days=30):
                return False
        except ValueError:
            pass
    if not (user.get("description") or "").strip():
        return False
    return True

qualified = [u for u in combined if is_quality_account(u)]

import csv

def export_users_to_csv(users, output_file="audience.csv"):
    fields = ["user_id", "username", "display_name", "description",
              "followers_count", "followings_count", "tweets_count",
              "location", "verified", "created_at"]
    with open(output_file, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=fields)
        w.writeheader()
        for u in users:
            w.writerow({
                "user_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),
                "followings_count": u.get("followings_count", 0),
                "tweets_count": u.get("tweets_count", 0),
                "location": u.get("location", ""),
                "verified": u.get("verified", False),
                "created_at": u.get("created_at", ""),
            })