From 4e0a694696d343bfb578f8dbb3483e3f095da675 Mon Sep 17 00:00:00 2001 From: shenlei Date: Fri, 31 Jul 2026 18:33:14 +0900 Subject: [PATCH] Add scheduled X post fetcher deployment --- .env.example | 2 + .gitignore | 3 + README.md | 80 ++++++++++++++++++ deploy/fetch-x.service | 21 +++++ deploy/fetch-x.timer | 10 +++ deploy/install-linux.sh | 33 ++++++++ fetch_x_posts.py | 178 ++++++++++++++++++++++++++++++++++++++++ 7 files changed, 327 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 deploy/fetch-x.service create mode 100644 deploy/fetch-x.timer create mode 100755 deploy/install-linux.sh create mode 100644 fetch_x_posts.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..5a96fc2 --- /dev/null +++ b/.env.example @@ -0,0 +1,2 @@ +# Create .env from this file locally, then replace with the Bearer Token from X Developer Console. +X_BEARER_TOKEN=replace_me diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..065cade --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +.env +*.db +__pycache__/ diff --git a/README.md b/README.md index e69de29..d6d7b5e 100644 --- a/README.md +++ b/README.md @@ -0,0 +1,80 @@ +# 每天抓取指定 X 账号推文 + +这个项目提供一个零第三方依赖的 Python 脚本，使用 **X 官方 API v2** 查询指定账号的近期原创推文，并去重存入本地 SQLite 数据库。不会抓取私密账号，也默认排除转推。 + +## 配置并手动运行 + +1. 在 [X Developer Console](https://developer.x.com/) 创建有 API 访问权限的项目，并取得 Bearer Token。 +2. 复制环境变量示例并填入 token（不要把 `.env` 提交到 Git）： + + ```sh + cp .env.example .env + ``` + +3. 运行： + + ```sh + set -a; source .env; set +a + python3 fetch_x_posts.py @OpenAI @XDevelopers + ``` + +首次运行默认取过去 24 小时；可用 `--initial-lookback-hours 168` 取最多 7 天。之后的运行会从上次成功时间向前回退 5 分钟查询，SQLite 的主键去重保证不会重复保存。 + +数据保存在 `x_posts.db`：`posts` 表包含正文、发布时间、公开互动指标、原始 API JSON 和可直接打开的链接。查看最近记录： + +```sh +sqlite3 x_posts.db 'SELECT username, created_at, text, url FROM posts ORDER BY created_at DESC LIMIT 20;' +``` + +## macOS 每天定时（推荐） + +将下面内容存为 `~/Library/LaunchAgents/com.local.fetch-x.plist`，把其中的项目路径和账号改成自己的；token 请直接填入 `EnvironmentVariables`，或改为调用一个仅自己可读的 wrapper 脚本。 + +```xml + + + + Labelcom.local.fetch-x + ProgramArguments + /usr/bin/python3 + /Users/YOU/Work/fetch_x/fetch_x_posts.py + @OpenAI@XDevelopers + + WorkingDirectory/Users/YOU/Work/fetch_x + EnvironmentVariables + X_BEARER_TOKENYOUR_BEARER_TOKEN + + StartCalendarIntervalHour9Minute0 + StandardOutPath/Users/YOU/Work/fetch_x/fetch-x.log + StandardErrorPath/Users/YOU/Work/fetch_x/fetch-x-error.log + +``` + +加载并立即验证一次： + +```sh +launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/com.local.fetch-x.plist +launchctl kickstart -k gui/$(id -u)/com.local.fetch-x +``` + +X 的“recent search”接口只能检索最近 7 天的推文，且实际 API 权限、额度和速率限制取决于你的 X 开发者套餐。官方接口文档见 [Search recent Posts](https://docs.x.com/x-api/posts/search-recent-posts)。 + +## 部署到 Linux 服务器（systemd） + +以下步骤适用于 Ubuntu/Debian 等使用 systemd 的服务器。把整个项目上传到服务器后，在项目目录执行： + +```sh +sudo bash deploy/install-linux.sh +sudoedit /etc/fetch-x/fetch-x.env +``` + +在环境文件中写入一行 `X_BEARER_TOKEN=你的token`，然后把 [deploy/fetch-x.service](deploy/fetch-x.service) 的账号参数 `@OpenAI @XDevelopers` 换成目标账号并重载： + +```sh +sudo systemctl daemon-reload +sudo systemctl start fetch-x.service +sudo journalctl -u fetch-x.service -n 100 --no-pager +systemctl list-timers fetch-x.timer +``` + +默认每天服务器本地时间 09:00 运行，修改 [deploy/fetch-x.timer](deploy/fetch-x.timer) 的 `OnCalendar` 后执行 `sudo systemctl daemon-reload && sudo systemctl restart fetch-x.timer`。定时器设有 `Persistent=true`，服务器关机错过的执行会在下次开机后补跑。 diff --git a/deploy/fetch-x.service b/deploy/fetch-x.service new file mode 100644 index 0000000..824b2fc --- /dev/null +++ b/deploy/fetch-x.service @@ -0,0 +1,21 @@ +[Unit] +Description=Fetch selected X posts into SQLite +After=network-online.target +Wants=network-online.target + +[Service] +Type=oneshot +User=fetchx +WorkingDirectory=/opt/fetch-x +EnvironmentFile=/etc/fetch-x/fetch-x.env +ExecStart=/usr/bin/python3 /opt/fetch-x/fetch_x_posts.py @OpenAI @XDevelopers +StandardOutput=journal +StandardError=journal + +# Avoid a hung API request blocking a subsequent scheduled run indefinitely. +TimeoutStartSec=10min +NoNewPrivileges=true +PrivateTmp=true +ProtectSystem=strict +ProtectHome=true +ReadWritePaths=/opt/fetch-x diff --git a/deploy/fetch-x.timer b/deploy/fetch-x.timer new file mode 100644 index 0000000..8b03cd9 --- /dev/null +++ b/deploy/fetch-x.timer @@ -0,0 +1,10 @@ +[Unit] +Description=Run X post fetcher every day + +[Timer] +OnCalendar=*-*-* 09:00:00 +Persistent=true +Unit=fetch-x.service + +[Install] +WantedBy=timers.target diff --git a/deploy/install-linux.sh b/deploy/install-linux.sh new file mode 100755 index 0000000..03686d4 --- /dev/null +++ b/deploy/install-linux.sh @@ -0,0 +1,33 @@ +#!/usr/bin/env bash +# Run as root from the project directory on Debian/Ubuntu-like Linux hosts. +set -euo pipefail + +APP_DIR=/opt/fetch-x +CONFIG_DIR=/etc/fetch-x +SERVICE_USER=fetchx + +if [[ ${EUID} -ne 0 ]]; then + echo "Run with sudo: sudo bash deploy/install-linux.sh" >&2 + exit 1 +fi + +if [[ ! -f fetch_x_posts.py || ! -f deploy/fetch-x.service ]]; then + echo "Run this from the fetch_x project directory." >&2 + exit 1 +fi + +id -u "$SERVICE_USER" >/dev/null 2>&1 || useradd --system --create-home --shell /usr/sbin/nologin "$SERVICE_USER" +install -d -o "$SERVICE_USER" -g "$SERVICE_USER" -m 0750 "$APP_DIR" +install -o "$SERVICE_USER" -g "$SERVICE_USER" -m 0640 fetch_x_posts.py "$APP_DIR/fetch_x_posts.py" +install -d -m 0750 "$CONFIG_DIR" + +if [[ ! -f "$CONFIG_DIR/fetch-x.env" ]]; then + install -m 0600 /dev/null "$CONFIG_DIR/fetch-x.env" + echo "Created $CONFIG_DIR/fetch-x.env. Add X_BEARER_TOKEN=... before starting the service." +fi + +install -m 0644 deploy/fetch-x.service /etc/systemd/system/fetch-x.service +install -m 0644 deploy/fetch-x.timer /etc/systemd/system/fetch-x.timer +systemctl daemon-reload +systemctl enable --now fetch-x.timer +systemctl list-timers fetch-x.timer --no-pager diff --git a/fetch_x_posts.py b/fetch_x_posts.py new file mode 100644 index 0000000..7d1e200 --- /dev/null +++ b/fetch_x_posts.py @@ -0,0 +1,178 @@ +#!/usr/bin/env python3 +"""Fetch recent posts from selected X accounts into a local SQLite database. + +Requires an X API bearer token. See README.md for setup and scheduling. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sqlite3 +import sys +import time +from datetime import UTC, datetime, timedelta +from pathlib import Path +from urllib.error import HTTPError, URLError +from urllib.parse import urlencode +from urllib.request import Request, urlopen + + +API_URL = "https://api.x.com/2/tweets/search/recent" +DEFAULT_DB = Path(__file__).with_name("x_posts.db") +OVERLAP = timedelta(minutes=5) # avoids gaps around a scheduled run boundary + + +def utc_now() -> datetime: + return datetime.now(UTC).replace(microsecond=0) + + +def iso_time(value: datetime) -> str: + return value.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def init_db(connection: sqlite3.Connection) -> None: + connection.executescript( + """ + CREATE TABLE IF NOT EXISTS posts ( + id TEXT PRIMARY KEY, + username TEXT NOT NULL, + created_at TEXT NOT NULL, + text TEXT NOT NULL, + url TEXT NOT NULL, + public_metrics TEXT, + raw_json TEXT NOT NULL, + fetched_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_posts_username_created + ON posts(username, created_at DESC); + CREATE TABLE IF NOT EXISTS state ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ); + """ + ) + + +def get_state(connection: sqlite3.Connection, key: str) -> str | None: + row = connection.execute("SELECT value FROM state WHERE key = ?", (key,)).fetchone() + return row[0] if row else None + + +def set_state(connection: sqlite3.Connection, key: str, value: str) -> None: + connection.execute( + "INSERT INTO state(key, value) VALUES(?, ?) " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value", + (key, value), + ) + + +def request_json(token: str, params: dict[str, str]) -> dict: + request = Request( + f"{API_URL}?{urlencode(params)}", + headers={"Authorization": f"Bearer {token}", "User-Agent": "x-post-fetcher/1.0"}, + ) + try: + with urlopen(request, timeout=30) as response: + return json.load(response) + except HTTPError as error: + body = error.read().decode("utf-8", errors="replace") + raise RuntimeError(f"X API returned HTTP {error.code}: {body}") from error + except URLError as error: + raise RuntimeError(f"Unable to reach X API: {error.reason}") from error + + +def fetch_account(token: str, username: str, start_time: str) -> list[dict]: + """Fetch every matching post in the requested period, following pagination.""" + posts: list[dict] = [] + next_token: str | None = None + while True: + params = { + "query": f"from:{username} -is:retweet", + "start_time": start_time, + "max_results": "100", + "sort_order": "recency", + "tweet.fields": "created_at,public_metrics,referenced_tweets,lang,entities", + } + if next_token: + params["next_token"] = next_token + payload = request_json(token, params) + posts.extend(payload.get("data", [])) + next_token = payload.get("meta", {}).get("next_token") + if not next_token: + return posts + time.sleep(1) # be conservative with API rate limits + + +def save_posts(connection: sqlite3.Connection, username: str, posts: list[dict], fetched_at: str) -> int: + saved = 0 + for post in posts: + cursor = connection.execute( + """ + INSERT OR IGNORE INTO posts + (id, username, created_at, text, url, public_metrics, raw_json, fetched_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + post["id"], username, + post.get("created_at", ""), post.get("text", ""), + f"https://x.com/{username}/status/{post['id']}", + json.dumps(post.get("public_metrics"), ensure_ascii=False), + json.dumps(post, ensure_ascii=False), fetched_at, + ), + ) + saved += cursor.rowcount + return saved + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Fetch selected X accounts into SQLite.") + parser.add_argument("usernames", nargs="+", help="X usernames, with or without @") + parser.add_argument("--db", type=Path, default=DEFAULT_DB, help=f"SQLite database (default: {DEFAULT_DB})") + parser.add_argument("--initial-lookback-hours", type=int, default=24, + help="How much history to fetch on the first run (max 168; default: 24)") + return parser.parse_args() + + +def main() -> int: + args = parse_args() + token = os.environ.get("X_BEARER_TOKEN") + if not token: + print("Missing X_BEARER_TOKEN environment variable.", file=sys.stderr) + return 2 + if not 1 <= args.initial_lookback_hours <= 168: + print("--initial-lookback-hours must be between 1 and 168.", file=sys.stderr) + return 2 + + usernames = list(dict.fromkeys(name.lstrip("@").strip() for name in args.usernames)) + if any(not name for name in usernames): + print("Usernames cannot be empty.", file=sys.stderr) + return 2 + args.db.parent.mkdir(parents=True, exist_ok=True) + now = utc_now() + fetched_at = iso_time(now) + with sqlite3.connect(args.db) as connection: + init_db(connection) + last_success = get_state(connection, "last_success_at") + if last_success: + start = datetime.fromisoformat(last_success.replace("Z", "+00:00")) - OVERLAP + else: + start = now - timedelta(hours=args.initial_lookback_hours) + # The recent-search endpoint cannot look back further than seven days. + start = max(start, now - timedelta(days=7)) + start_time = iso_time(start) + total = 0 + for username in usernames: + posts = fetch_account(token, username, start_time) + saved = save_posts(connection, username, posts, fetched_at) + total += saved + print(f"@{username}: fetched {len(posts)}, saved {saved} new post(s)") + # Only advance after all accounts completed successfully. + set_state(connection, "last_success_at", fetched_at) + print(f"Done. {total} new post(s) saved to {args.db}.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())