public social platform 感情分析チュートリアル：Python と API で構築するリアルタイム感情分析パイプライン

1import os
2from datetime import datetime, timezone
3
4import requests
5
6API_URL = "https://api.twexapi.io/twitter/advanced_search/page"
7TOKEN = os.environ["TWEXAPI_BEARER_TOKEN"]
8
9def search_page(query, next_cursor=None):
10    payload = {
11        "searchTerms": [query],
12        "sortBy": "Latest"
13    }
14
15    if next_cursor:
16        payload["next_cursor"] = next_cursor
17
18    response = requests.post(
19        API_URL,
20        headers={
21            "Authorization": f"Bearer {TOKEN}",
22            "Content-Type": "application/json",
23            "Accept": "application/json",
24        },
25        json=payload,
26        timeout=30,
27    )
28    response.raise_for_status()
29    data = response.json()
30
31    if data.get("code", 200) >= 400:
32        raise RuntimeError(data.get("msg", "TwexAPI returned an error"))
33
34    return data
35
36def collect_recent_tweets(query, seen_ids, max_pages=3):
37    next_cursor = None
38    collected = []
39
40    for page_number in range(max_pages):
41        page = search_page(query, next_cursor)
42
43        for tweet in page.get("data", []):
44            if not tweet:
45                continue
46
47            tweet_id = tweet.get("tweet_id") or tweet.get("id")
48            if not tweet_id or tweet_id in seen_ids:
49                continue
50
51            seen_ids.add(tweet_id)
52            collected.append({
53                "query": query,
54                "fetched_at": datetime.now(timezone.utc).isoformat(),
55                "raw": tweet,
56            })
57
58        if not page.get("has_next_page"):
59            break
60
61        next_cursor = page.get("next_cursor")
62        if not next_cursor:
63            break
64
65    return collected
66
67seen_ids = set()
68tweets = collect_recent_tweets(
69    query='"crypto regulation" lang:en -filter:retweets',
70    seen_ids=seen_ids,
71    max_pages=2,
72)
73
74print(f"New tweets ready for scoring: {len(tweets)}")

1from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
2
3analyzer = SentimentIntensityAnalyzer()
4MODEL_VERSION = "vader-3.3.2-en-baseline"
5
6def sentiment_label(score):
7    if score >= 0.2:
8        return "positive"
9    if score <= -0.2:
10        return "negative"
11    return "neutral"
12
13def text_for_model(tweet):
14    return tweet.get("full_text") or tweet.get("text") or ""
15
16def score_tweets(records):
17    scored = []
18
19    for record in records:
20        tweet = record["raw"]
21        text = text_for_model(tweet).strip()
22        if not text:
23            continue
24
25        compound = analyzer.polarity_scores(text)["compound"]
26        scored.append({
27            "tweet_id": tweet.get("tweet_id") or tweet.get("id"),
28            "query": record["query"],
29            "fetched_at": record["fetched_at"],
30            "created_at_datetime": tweet.get("created_at_datetime"),
31            "lang": tweet.get("lang"),
32            "text": text,
33            "sentiment_score": compound,
34            "sentiment_label": sentiment_label(compound),
35            "model_version": MODEL_VERSION,
36            "favorite_count": tweet.get("favorite_count"),
37            "retweet_count": tweet.get("retweet_count"),
38            "reply_count": tweet.get("reply_count"),
39            "raw": tweet,
40        })
41
42    return scored
43
44scored_tweets = score_tweets(tweets)
45print(f"Scored tweets: {len(scored_tweets)}")

1def summarize_window(scored_rows):
2    if not scored_rows:
3        return {
4            "tweet_count": 0,
5            "avg_sentiment": None,
6            "positive_ratio": 0,
7            "negative_ratio": 0,
8            "top_negative_examples": [],
9        }
10
11    tweet_count = len(scored_rows)
12    avg_sentiment = sum(row["sentiment_score"] for row in scored_rows) / tweet_count
13    positive_count = sum(1 for row in scored_rows if row["sentiment_label"] == "positive")
14    negative_count = sum(1 for row in scored_rows if row["sentiment_label"] == "negative")
15
16    top_negative_examples = sorted(
17        scored_rows,
18        key=lambda row: (row["sentiment_score"], -int(row.get("reply_count") or 0)),
19    )[:5]
20
21    return {
22        "tweet_count": tweet_count,
23        "avg_sentiment": round(avg_sentiment, 4),
24        "positive_ratio": round(positive_count / tweet_count, 4),
25        "negative_ratio": round(negative_count / tweet_count, 4),
26        "top_negative_examples": [
27            {
28                "tweet_id": row["tweet_id"],
29                "score": row["sentiment_score"],
30                "text": row["text"][:180],
31            }
32            for row in top_negative_examples
33        ],
34    }
35
36summary = summarize_window(scored_tweets)
37print(summary)