ターゲット層の発見

import os
import time
import requests

API_KEY = os.environ["SORSA_API_KEY"]
BASE_URL = "https://api.sorsa.io/v3"

def fetch_pages(method, endpoint, payload, result_key, max_pages=10):
    """Fetch a bounded number of pages; raise on HTTP errors."""
    items = []
    cursor = None
    seen_cursors = set()

    for _ in range(max_pages):
        values = dict(payload)
        if cursor:
            values["next_cursor"] = cursor
        options = {"params": values} if method == "GET" else {"json": values}
        response = requests.request(
            method, f"{BASE_URL}{endpoint}",
            headers={"ApiKey": API_KEY}, timeout=30, **options,
        )
        response.raise_for_status()
        data = response.json()
        items.extend(data.get(result_key) or [])
        cursor = data.get("next_cursor")
        if not cursor:
            break
        if cursor in seen_cursors:
            raise RuntimeError("Pagination returned a repeated cursor")
        seen_cursors.add(cursor)
        time.sleep(0.1)

    return items

{
  "query": "Product Manager"
}

def find_users_by_bio(query, max_pages=10):
    return fetch_pages("POST", "/search-users", {"query": query}, "users", max_pages)

bio_results = find_users_by_bio("machine learning engineer")
qualified = [
    u for u in bio_results
    if (u.get("followers_count") or 0) >= 1000
    and (u.get("tweets_count") or 0) >= 100
    and not u.get("protected", False)
]

const API_KEY = process.env.SORSA_API_KEY;
if (!API_KEY) throw new Error("Set SORSA_API_KEY before running this example");

async function findUsersByBio(query, maxPages = 10) {
  const users = [];
  const seenCursors = new Set();
  let cursor = null;

  for (let i = 0; i < maxPages; i++) {
    const body = { query };
    if (cursor) body.next_cursor = cursor;
    const response = await fetch("https://api.sorsa.io/v3/search-users", {
      method: "POST",
      headers: { ApiKey: API_KEY, "Content-Type": "application/json" },
      body: JSON.stringify(body),
      signal: AbortSignal.timeout(30000),
    });
    if (!response.ok) throw new Error(`HTTP ${response.status}`);
    const data = await response.json();
    users.push(...(data.users || []));
    cursor = data.next_cursor;
    if (!cursor) break;
    if (seenCursors.has(cursor)) throw new Error("Repeated pagination cursor");
    seenCursors.add(cursor);
    await new Promise((resolve) => setTimeout(resolve, 100));
  }
  return users;
}

GET https://api.sorsa.io/v3/followers?username=competitor_handle

def get_followers(username, max_pages=10):
    return fetch_pages("GET", "/followers", {"username": username}, "users", max_pages)

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

from collections import Counter

competitors = ["competitor_a", "competitor_b", "competitor_c"]
follower_sets = {
    handle: {u["id"] for u in get_followers(handle, max_pages=10)}
    for handle in competitors
}
counts = Counter(uid for ids in follower_sets.values() for uid in ids)
overlap = {uid for uid, count in counts.items() if count >= 2}

{
  "community_link": "1966045657589813686"
}

def get_community_members(community_id, max_pages=20):
    return fetch_pages(
        "POST", "/community-members",
        {"community_link": community_id}, "users", max_pages,
    )

def find_active_voices(query, min_followers=100, max_pages=10):
    tweets = fetch_pages(
        "POST", "/search-tweets",
        {"query": query, "order": "latest"}, "tweets", max_pages,
    )
    voices = {}
    for tweet in tweets:
        user = tweet.get("user")
        if not user or not user.get("id"):
            continue
        if (user.get("followers_count") or 0) < min_followers:
            continue
        if user["id"] not in voices:
            voices[user["id"]] = {
                **user,
                "sample_tweet": (tweet.get("full_text") or "")[:160],
            }
    return list(voices.values())

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

def get_verified_followers(username, max_pages=10):
    return fetch_pages(
        "GET", "/verified-followers", {"username": username}, "users", max_pages,
    )

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

def get_retweeters(tweet_link, max_pages=10):
    return fetch_pages(
        "POST", "/retweeters", {"tweet_link": tweet_link}, "users", max_pages,
    )

def get_quoters(tweet_link, max_pages=10):
    quote_tweets = fetch_pages(
        "POST", "/quotes", {"tweet_link": tweet_link}, "tweets", max_pages,
    )
    return list({
        tweet["user"]["id"]: tweet["user"]
        for tweet in quote_tweets if tweet.get("user")
    }.values())

def score_by_source(by_source):
    index = {}
    for source, users in by_source.items():
        for user in users:
            uid = user["id"]
            if uid not in index:
                index[uid] = {"user": dict(user), "sources": set()}
            else:
                # Fill gaps when one source returns a compact profile.
                for field, value in user.items():
                    if index[uid]["user"].get(field) is None and value is not None:
                        index[uid]["user"][field] = value
            index[uid]["sources"].add(source)

    result = [
        {**entry["user"], "source_count": len(entry["sources"]),
         "sources": sorted(entry["sources"])}
        for entry in index.values()
    ]
    return sorted(result, key=lambda u: (-u["source_count"], -(u.get("followers_count") or 0)))

# Uses the results from Techniques 1, 2, and 4 above.
combined = score_by_source({
    "profile_search": bio_results,
    "competitor_followers": followers,
    "topic_discussion": intent_voices,
})

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
    followers = user.get("followers_count") or 0
    if followers < min_followers or (user.get("tweets_count") or 0) < min_tweets:
        return False
    if (user.get("followings_count") or 0) > followers * max_following_ratio:
        return False
    if not (user.get("description") or "").strip():
        return False

    created = user.get("created_at")
    if not created:
        return False
    try:
        dt = datetime.fromisoformat(created.replace("Z", "+00:00"))
        if dt.tzinfo is None:
            return False
    except (TypeError, ValueError):
        return False
    return dt <= datetime.now(timezone.utc) - timedelta(days=30)

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

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 file:
        writer = csv.DictWriter(file, fieldnames=fields)
        writer.writeheader()
        for user in users:
            row = {field: user.get(field, "") for field in fields}
            row["user_id"] = user["id"]
            row["description"] = (user.get("description") or "").replace("\n", " ")
            writer.writerow(row)

export_users_to_csv(qualified)