如何通过 API 在 X 上寻找 Crypto Influencer

https://api.twexapi.io/twitter/search-user/{keyword}/{target_count}

curl --request GET \
  --url 'https://api.twexapi.io/twitter/search-user/crypto%20analyst/50' \
  --header 'Authorization: Bearer <token>'

1{
2  "code": 200,
3  "msg": "success",
4  "data": [
5    {
6      "id": "1717001045992251392",
7      "name": "Example Crypto Analyst",
8      "screen_name": "example_crypto",
9      "description": "Crypto markets, DeFi research, BTC and ETH commentary",
10      "followers_count": 88000,
11      "listed_count": 740,
12      "is_verified": true,
13      "is_blue_verified": true,
14      "verified_type": "blue",
15      "can_dm": true
16    }
17  ]
18}

1import os
2from typing import Any
3from urllib.parse import quote
4
5import requests
6
7TOKEN = os.environ["TWEXAPI_BEARER_TOKEN"]
8BASE_URL = "https://api.twexapi.io/twitter/search-user"
9
10def search_users(keyword: str, *, target_count: int = 50) -> list[dict[str, Any]]:
11    encoded_keyword = quote(keyword)
12    url = f"{BASE_URL}/{encoded_keyword}/{target_count}"
13
14    response = requests.get(
15        url,
16        headers={"Authorization": f"Bearer {TOKEN}"},
17        timeout=30,
18    )
19    response.raise_for_status()
20    return response.json().get("data", [])
21
22if __name__ == "__main__":
23    users = search_users("crypto analyst", target_count=25)
24    for user in users[:10]:
25        print(
26            user.get("screen_name"),
27            user.get("followers_count"),
28            user.get("description"),
29        )

1from typing import Any
2
3CRYPTO_TERMS = [
4    "crypto",
5    "bitcoin",
6    "btc",
7    "ethereum",
8    "eth",
9    "defi",
10    "web3",
11    "nft",
12    "onchain",
13    "token",
14    "dao",
15    "solana",
16]
17
18RISK_TERMS = [
19    "signals",
20    "guaranteed",
21    "100x",
22    "pump",
23    "airdrop farming",
24]
25
26def contains_any(text: str, terms: list[str]) -> bool:
27    lowered = text.lower()
28    return any(term in lowered for term in terms)
29
30def to_kol_row(user: dict[str, Any], keyword: str) -> dict[str, Any]:
31    description = user.get("description") or ""
32
33    return {
34        "keyword": keyword,
35        "user_id": user.get("id"),
36        "screen_name": user.get("screen_name"),
37        "name": user.get("name"),
38        "description": description,
39        "location": user.get("location"),
40        "followers_count": user.get("followers_count") or 0,
41        "listed_count": user.get("listed_count") or 0,
42        "is_verified": user.get("is_verified") is True,
43        "is_blue_verified": user.get("is_blue_verified") is True,
44        "verified_type": user.get("verified_type"),
45        "can_dm": user.get("can_dm"),
46        "crypto_profile_match": contains_any(description, CRYPTO_TERMS),
47        "risk_term_match": contains_any(description, RISK_TERMS),
48        "profile_url": f"https://x.com/{user.get('screen_name')}",
49        "raw": user,
50    }

1import math
2
3def kol_score(row: dict[str, Any]) -> float:
4    score = 0.0
5
6    if row["crypto_profile_match"]:
7        score += 35
8    if row["is_verified"]:
9        score += 8
10    if row["is_blue_verified"]:
11        score += 4
12    if row["verified_type"] == "business":
13        score += 8
14    if row["risk_term_match"]:
15        score -= 20
16
17    followers = row["followers_count"] or 0
18    listed = row["listed_count"] or 0
19    score += min(math.log10(followers + 1) * 10, 45)
20    score += min(math.log10(listed + 1) * 7, 24)
21
22    return round(score, 2)
23
24keywords = [
25    "crypto analyst",
26    "DeFi founder",
27    "Bitcoin trader",
28    "onchain analyst",
29    "Web3 founder",
30]
31
32candidates = []
33seen_user_ids = set()
34
35for keyword in keywords:
36    for user in search_users(keyword, target_count=50):
37        user_id = user.get("id")
38        if not user_id or user_id in seen_user_ids:
39            continue
40        seen_user_ids.add(user_id)
41
42        row = to_kol_row(user, keyword)
43        row["score"] = kol_score(row)
44        row["review_status"] = "manual_review" if row["risk_term_match"] else "candidate"
45        candidates.append(row)
46
47candidates.sort(key=lambda row: row["score"], reverse=True)

1import csv
2import json
3from datetime import datetime, timezone
4from pathlib import Path
5
6timestamp = datetime.now(timezone.utc).isoformat()
7csv_path = Path("crypto-kol-candidates.csv")
8jsonl_path = Path("crypto-kol-candidates.raw.jsonl")
9
10fieldnames = [
11    "score",
12    "review_status",
13    "keyword",
14    "screen_name",
15    "name",
16    "description",
17    "location",
18    "followers_count",
19    "listed_count",
20    "is_verified",
21    "is_blue_verified",
22    "verified_type",
23    "can_dm",
24    "crypto_profile_match",
25    "risk_term_match",
26    "profile_url",
27]
28
29with csv_path.open("w", newline="", encoding="utf-8") as f:
30    writer = csv.DictWriter(f, fieldnames=fieldnames)
31    writer.writeheader()
32    for row in candidates:
33        writer.writerow({field: row.get(field) for field in fieldnames})
34
35with jsonl_path.open("w", encoding="utf-8") as f:
36    for row in candidates:
37        f.write(json.dumps({
38            "fetched_at": timestamp,
39            "candidate": row,
40        }, ensure_ascii=False) + "\n")
41
42print(f"Saved {len(candidates)} crypto KOL candidates")