如何监控 Serenity 在 X 上发布的新公开帖

from:aleabitoreddit -filter:retweets -filter:replies

POST https://api.twexapi.io/twitter/advanced_search/page

Authorization: Bearer <your_token>
Content-Type: application/json

curl --request POST \
  --url https://api.twexapi.io/twitter/advanced_search/page \
  --header 'Authorization: Bearer <your_token>' \
  --header 'Content-Type: application/json' \
  --data '{
    "searchTerms": ["from:aleabitoreddit -filter:retweets -filter:replies"],
    "sortBy": "Latest"
  }'

1import json
2import time
3from pathlib import Path
4
5import requests
6
7TOKEN = "<your_token>"
8URL = "https://api.twexapi.io/twitter/advanced_search/page"
9QUERY = "from:aleabitoreddit -filter:retweets -filter:replies"
10HEADERS = {"Authorization": f"Bearer {TOKEN}"}
11STATE_FILE = Path("serenity-monitor-state.json")
12POLL_SECONDS = 300
13
14def load_seen_ids():
15    if not STATE_FILE.exists():
16        return set(), False
17    payload = json.loads(STATE_FILE.read_text(encoding="utf-8"))
18    return set(payload.get("seen_ids", [])), payload.get("initialized", False)
19
20def save_state(seen_ids, initialized=True):
21    STATE_FILE.write_text(
22        json.dumps(
23            {"initialized": initialized, "seen_ids": sorted(seen_ids)},
24            ensure_ascii=False,
25            indent=2,
26        ),
27        encoding="utf-8",
28    )
29
30def fetch_latest():
31    response = requests.post(
32        URL,
33        headers=HEADERS,
34        json={"searchTerms": [QUERY], "sortBy": "Latest"},
35        timeout=30,
36    )
37    response.raise_for_status()
38    return [tweet for tweet in response.json().get("data", []) if tweet]
39
40def notify(tweet):
41    print("NEW", tweet.get("tweet_id"), tweet.get("text", "")[:160])
42
43seen_ids, initialized = load_seen_ids()
44
45while True:
46    latest = fetch_latest()
47    latest_ids = {tweet["tweet_id"] for tweet in latest if tweet.get("tweet_id")}
48
49    if not initialized:
50        seen_ids.update(latest_ids)
51        initialized = True
52        save_state(seen_ids, initialized=True)
53        print(f"已初始化 {len(seen_ids)} 条已有帖子")
54    else:
55        new_tweets = []
56        for tweet in reversed(latest):
57            tweet_id = tweet.get("tweet_id")
58            if tweet_id and tweet_id not in seen_ids:
59                new_tweets.append(tweet)
60                seen_ids.add(tweet_id)
61
62        if new_tweets:
63            save_state(seen_ids, initialized=True)
64            for tweet in new_tweets:
65                notify(tweet)
66
67    time.sleep(POLL_SECONDS)