From 7238b932ab9d5baddaa74ce83669aad6236c8955 Mon Sep 17 00:00:00 2001 From: jackwener Date: Thu, 5 Mar 2026 00:41:26 +0800 Subject: [PATCH] feat: add user commands, auto-detect browser, optimize performance MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add user/user-posts/followers/following commands - Add UserProfile model and GraphQL API methods - Add print_user_profile and print_user_table formatters - Auto-detect browser for cookies (Chrome → Edge → Firefox → Brave) - Remove --browser option from all commands - Remove cookie verification (v1.1 endpoints are gone) - Use hardcoded fallback query IDs first (skip slow JS bundle scan) - Update FEATURES from latest twitter-openapi config - Fix user-posts: add required withVoice variable - Add tweet URL links in feed output - Add error handling to all user commands --- README.md | 84 +++++++---- config.yaml | 7 - pyproject.toml | 2 +- twitter_cli/auth.py | 161 +++++++++++++------- twitter_cli/cli.py | 288 +++++++++++++++++++++++++++--------- twitter_cli/client.py | 302 +++++++++++++++++++++++++++++++++++--- twitter_cli/config.py | 14 -- twitter_cli/formatter.py | 84 ++++++++++- twitter_cli/models.py | 17 +++ twitter_cli/summarizer.py | 164 --------------------- 10 files changed, 770 insertions(+), 353 deletions(-) delete mode 100644 twitter_cli/summarizer.py diff --git a/README.md b/README.md index 56303ea..427c9e2 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Twitter CLI -从你的 Twitter/X 首页抓取推文，智能筛选高价值内容，AI 自动生成摘要。 +Twitter/X 命令行工具 — 读取 Timeline、管理推文。 **零 API Key** — 使用浏览器 Cookie 认证，免费访问 Twitter。 @@ -19,38 +19,70 @@ twitter feed ## 使用方式 +### 读取 + ```bash -# 完整 pipeline：抓取 50 条 → 筛选 top 20 → AI 总结 +# 抓取首页 timeline twitter feed # 自定义抓取条数 -twitter feed --count 50 - -# 只抓取 + 筛选，跳过 AI 总结 -twitter feed --no-summary - -# JSON 输出（可重定向到文件） -twitter feed --json > tweets.json - -# 对已有数据做筛选 + 总结 -twitter feed --input tweets.json +twitter feed --max 50 # 跳过筛选 twitter feed --no-filter -# 指定浏览器 -twitter feed --browser firefox +# JSON 输出 +twitter feed --json > tweets.json + +# 从已有数据加载 +twitter feed --input tweets.json + # 抓取收藏 -twitter bookmarks -twitter bookmarks --count 30 --json +twitter favorite +twitter favorite --max 30 --json +``` + +### 用户 + +```bash +# 查看用户资料 +twitter user elonmusk + +# 列出用户推文 +twitter user-posts elonmusk --max 20 + +# 查看粉丝 +twitter followers elonmusk --max 30 + +# 查看关注 +twitter following elonmusk --max 30 +``` + +### 发推 + +```bash +# 发新推文 +twitter post "Hello World" + +# 回复推文 +twitter reply "这是回复内容" + +# 引用转推（传 URL 或 ID 均可） +twitter quote "这是引用内容" + +# 删除推文（会有确认提示） +twitter delete + +# 跳过删除确认 +twitter delete --yes ``` ## Pipeline ``` -抓取 (GraphQL API) → 筛选 (Engagement Score) → AI 总结 - 50 条 top 20 按主题分组 +抓取 (GraphQL API) → 筛选 (Engagement Score) + 50 条 top 20 ``` ### 筛选算法 @@ -62,10 +94,6 @@ score = 1.0 × likes + 3.0 × retweets + 2.0 × replies + 5.0 × bookmarks + 0.5 × log10(views) ``` -### AI 总结 - -支持 **OpenAI-compatible**（doubao / deepseek / openai）和 **Anthropic**（Claude）两种 API 格式。 - ## 配置 编辑 `config.yaml`： @@ -83,18 +111,11 @@ filter: replies: 2.0 bookmarks: 5.0 views_log: 0.5 - -ai: - provider: "openai" # "openai" or "anthropic" - api_key: "" # 或设置环境变量 AI_API_KEY - model: "doubao-seed-2.0-code" - base_url: "https://ark.cn-beijing.volces.com/api/coding" - language: "zh-CN" ``` ### Cookie 配置 -**方式 1：自动提取**（推荐） — 确保 Chrome 已登录 x.com，程序自动通过 `browser-cookie3` 读取。 +**方式 1：自动提取**（推荐） — 确保浏览器已登录 x.com，程序自动通过 `browser-cookie3` 按 Chrome → Edge → Firefox → Brave 顺序尝试读取。 **方式 2：环境变量** — 设置： @@ -111,10 +132,9 @@ export TWITTER_CT0=your_ct0 twitter_cli/ ├── __init__.py # 版本信息 ├── cli.py # CLI 入口 (click) -├── client.py # Twitter GraphQL API Client +├── client.py # Twitter GraphQL API Client (GET + POST) ├── auth.py # Cookie 提取 (env / browser-cookie3) ├── filter.py # Engagement scoring + 筛选 -├── summarizer.py # AI 总结 (OpenAI + Anthropic) ├── formatter.py # Rich 终端输出 + JSON ├── config.py # YAML 配置加载 └── models.py # 数据模型 (dataclass) diff --git a/config.yaml b/config.yaml index ad4ee75..2649084 100644 --- a/config.yaml +++ b/config.yaml @@ -13,10 +13,3 @@ filter: replies: 2.0 bookmarks: 5.0 views_log: 0.5 - -ai: - provider: "openai" - api_key: "" - model: "doubao-seed-2.0-code" - base_url: "https://ark.cn-beijing.volces.com/api/coding" - language: "zh-CN" diff --git a/pyproject.toml b/pyproject.toml index 5066be2..b4ce9b7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "hatchling.build" [project] name = "twitter-cli" version = "0.1.0" -description = "A CLI for Twitter/X — feed, bookmarks, filtering, AI summary" +description = "A CLI for Twitter/X — feed, bookmarks, tweet, filtering" readme = "README.md" license = "Apache-2.0" requires-python = ">=3.8" diff --git a/twitter_cli/auth.py b/twitter_cli/auth.py index a525f50..e8adac9 100644 --- a/twitter_cli/auth.py +++ b/twitter_cli/auth.py @@ -8,10 +8,21 @@ Supports: from __future__ import annotations import json +import logging import os +import ssl import subprocess import sys -from typing import Dict, Optional +import urllib.request +from typing import Any, Dict, Optional + +logger = logging.getLogger(__name__) + +# Public bearer token (same as in client.py) +_BEARER_TOKEN = ( + "AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs" + "%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA" +) def load_from_env() -> Optional[Dict[str, str]]: @@ -23,9 +34,63 @@ def load_from_env() -> Optional[Dict[str, str]]: return None -def extract_from_browser(browser: str = "chrome") -> Optional[Dict[str, str]]: +def verify_cookies(auth_token, ct0): + # type: (str, str) -> Dict[str, Any] + """Verify cookies by calling a Twitter API endpoint. + + Tries multiple endpoints. Only raises on clear auth failures (401/403). + For other errors (404, network), returns empty dict (proceed without verification). + """ + # Endpoints to try, in order of preference + urls = [ + "https://api.x.com/1.1/account/verify_credentials.json", + "https://x.com/i/api/1.1/account/settings.json", + ] + + headers = { + "Authorization": "Bearer %s" % _BEARER_TOKEN, + "Cookie": "auth_token=%s; ct0=%s" % (auth_token, ct0), + "X-Csrf-Token": ct0, + "X-Twitter-Active-User": "yes", + "X-Twitter-Auth-Type": "OAuth2Session", + "User-Agent": ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/131.0.0.0 Safari/537.36" + ), + } + + for url in urls: + req = urllib.request.Request(url) + for k, v in headers.items(): + req.add_header(k, v) + + ctx = ssl.create_default_context() + try: + with urllib.request.urlopen(req, context=ctx, timeout=3) as resp: + data = json.loads(resp.read().decode("utf-8")) + return {"screen_name": data.get("screen_name", "")} + except urllib.error.HTTPError as e: + if e.code in (401, 403): + raise RuntimeError( + "Cookie expired or invalid (HTTP %d). Please re-login to x.com in your browser." % e.code + ) + # 404 or other — try next endpoint + logger.debug("Verification endpoint %s returned HTTP %d, trying next...", url, e.code) + continue + except Exception as e: + logger.debug("Verification endpoint %s failed: %s", url, e) + continue + + # All endpoints failed with non-auth errors — proceed without verification + logger.info("Cookie verification skipped (no working endpoint), will verify on first API call") + return {} + + +def extract_from_browser() -> Optional[Dict[str, str]]: """Auto-extract cookies from local browser using browser-cookie3. + Tries browsers in order: Chrome -> Edge -> Firefox -> Brave. Runs in a subprocess to avoid SQLite database lock issues when the browser is running. """ @@ -37,40 +102,34 @@ except ImportError: print(json.dumps({"error": "browser-cookie3 not installed"})) sys.exit(1) -browser_funcs = { - "chrome": browser_cookie3.chrome, - "firefox": browser_cookie3.firefox, - "edge": browser_cookie3.edge, - "brave": browser_cookie3.brave, -} +browsers = [ + ("chrome", browser_cookie3.chrome), + ("edge", browser_cookie3.edge), + ("firefox", browser_cookie3.firefox), + ("brave", browser_cookie3.brave), +] -browser_name = "%s" -fn = browser_funcs.get(browser_name) -if not fn: - print(json.dumps({"error": "Unsupported browser: " + browser_name})) - sys.exit(1) +for name, fn in browsers: + try: + jar = fn() + except Exception: + continue + result = {} + for cookie in jar: + domain = cookie.domain or "" + if domain.endswith(".x.com") or domain.endswith(".twitter.com") or domain in ("x.com", "twitter.com", ".x.com", ".twitter.com"): + if cookie.name == "auth_token": + result["auth_token"] = cookie.value + elif cookie.name == "ct0": + result["ct0"] = cookie.value + if "auth_token" in result and "ct0" in result: + result["browser"] = name + print(json.dumps(result)) + sys.exit(0) -try: - jar = fn() -except Exception as e: - print(json.dumps({"error": str(e)})) - sys.exit(1) - -result = {} -for cookie in jar: - domain = cookie.domain or "" - if domain.endswith(".x.com") or domain.endswith(".twitter.com") or domain in ("x.com", "twitter.com", ".x.com", ".twitter.com"): - if cookie.name == "auth_token": - result["auth_token"] = cookie.value - elif cookie.name == "ct0": - result["ct0"] = cookie.value - -if "auth_token" in result and "ct0" in result: - print(json.dumps(result)) -else: - print(json.dumps({"error": "Could not find auth_token and ct0 cookies. Make sure you are logged into x.com in " + browser_name + "."})) - sys.exit(1) -''' % browser +print(json.dumps({"error": "No Twitter cookies found in any browser. Make sure you are logged into x.com."})) +sys.exit(1) +''' try: result = subprocess.run( @@ -97,29 +156,33 @@ else: data = json.loads(output) if "error" in data: return None + logger.info("Found cookies in %s", data.get("browser", "unknown")) return {"auth_token": data["auth_token"], "ct0": data["ct0"]} except (subprocess.TimeoutExpired, json.JSONDecodeError, KeyError, FileNotFoundError): return None -def get_cookies(browser: str = "chrome") -> Dict[str, str]: - """Get Twitter cookies. Priority: env vars -> browser extraction. +def get_cookies() -> Dict[str, str]: + """Get Twitter cookies. Priority: env vars -> browser extraction (Chrome/Edge/Firefox/Brave). - Returns dict with 'auth_token' and 'ct0' keys. Raises RuntimeError if no cookies found. """ + cookies = None # type: Optional[Dict[str, str]] + # 1. Try environment variables - env_cookies = load_from_env() - if env_cookies: - return env_cookies + cookies = load_from_env() + if cookies: + logger.info("Loaded cookies from environment variables") - # 2. Try browser extraction - browser_cookies = extract_from_browser(browser) - if browser_cookies: - return browser_cookies + # 2. Try browser extraction (auto-detect) + if not cookies: + cookies = extract_from_browser() - raise RuntimeError( - "No Twitter cookies found.\n" - "Option 1: Set TWITTER_AUTH_TOKEN and TWITTER_CT0 environment variables\n" - "Option 2: Make sure you are logged into x.com in your browser" - ) + if not cookies: + raise RuntimeError( + "No Twitter cookies found.\n" + "Option 1: Set TWITTER_AUTH_TOKEN and TWITTER_CT0 environment variables\n" + "Option 2: Make sure you are logged into x.com in your browser (Chrome/Edge/Firefox/Brave)" + ) + + return cookies diff --git a/twitter_cli/cli.py b/twitter_cli/cli.py index 38d388f..ea13dbc 100644 --- a/twitter_cli/cli.py +++ b/twitter_cli/cli.py @@ -1,16 +1,18 @@ """CLI entry point for twitter-cli. Usage: - twitter feed # full pipeline: fetch → filter → AI summarize - twitter feed --count 50 # custom fetch count - twitter feed --no-summary # skip AI summary + twitter feed # fetch home timeline → filter + twitter feed --max 50 # custom fetch count twitter feed --no-filter # skip filtering twitter feed --json # JSON output - twitter feed --browser firefox # specify browser for cookie extraction - twitter bookmarks # fetch bookmarks - twitter bookmarks --count 30 - twitter feed --input tweets.json # summarize existing data + twitter favorite # fetch bookmarks + twitter favorite --max 30 + twitter feed --input tweets.json # load existing data twitter feed --output out.json # save filtered tweets + twitter post "Hello" # post a tweet + twitter reply ID "text" # reply to a tweet + twitter quote ID "text" # quote a tweet + twitter delete ID # delete a tweet """ from __future__ import annotations @@ -33,10 +35,12 @@ from .filter import filter_tweets from .formatter import ( print_filter_stats, print_tweet_table, + print_user_profile, + print_user_table, tweets_to_json, ) from .models import Author, Metrics, Tweet, TweetMedia -from .summarizer import summarize + console = Console() @@ -132,16 +136,14 @@ def cli(verbose): # ===== Feed ===== @cli.command() -@click.option("--count", "-n", type=int, default=None, help="Number of tweets to fetch.") +@click.option("--max", "-n", "max_count", type=int, default=None, help="Max number of tweets to fetch.") @click.option("--json", "as_json", is_flag=True, help="Output as JSON.") -@click.option("--browser", "-b", default="chrome", help="Browser to extract cookies from.") @click.option("--input", "-i", "input_file", type=str, default=None, help="Load tweets from JSON file.") @click.option("--output", "-o", "output_file", type=str, default=None, help="Save filtered tweets to JSON file.") @click.option("--no-filter", is_flag=True, help="Skip filtering.") -@click.option("--no-summary", is_flag=True, help="Skip AI summary.") -def feed(count, as_json, browser, input_file, output_file, no_filter, no_summary): - # type: (int, bool, str, str, str, bool, bool) -> None - """Fetch home timeline — full pipeline: fetch → filter → AI summarize.""" +def feed(max_count, as_json, input_file, output_file, no_filter): + # type: (int, bool, str, str, bool) -> None + """Fetch home timeline with filtering.""" config = load_config() # Step 1: Get tweets @@ -150,10 +152,10 @@ def feed(count, as_json, browser, input_file, output_file, no_filter, no_summary tweets = _load_tweets_from_json(input_file) console.print(" Loaded %d tweets" % len(tweets)) else: - fetch_count = count or config.get("fetch", {}).get("count", 50) + fetch_count = max_count or config.get("fetch", {}).get("count", 50) console.print("\n🔐 Getting Twitter cookies...") try: - cookies = get_cookies(browser) + cookies = get_cookies() except RuntimeError as e: console.print("[red]❌ %s[/red]" % e) sys.exit(1) @@ -188,58 +190,33 @@ def feed(count, as_json, browser, input_file, output_file, no_filter, no_summary print_tweet_table(filtered, console) console.print() - # Step 3: AI Summary - if no_summary: - return - ai_config = config.get("ai", {}) - if not ai_config.get("api_key"): - console.print( - "[yellow]⚠️ AI summary skipped: no API key configured.[/yellow]\n" - " Set ai.api_key in config.yaml or export AI_API_KEY=your_key" - ) - return - - try: - console.print("🤖 Calling AI (%s/%s)..." % (ai_config.get("provider", "openai"), ai_config.get("model", ""))) - summary = summarize(filtered, ai_config) - console.print("\n" + "═" * 50) - console.print("📝 AI Summary") - console.print("═" * 50 + "\n") - console.print(summary) - console.print() - except Exception as e: - console.print("[red]❌ AI summary failed: %s[/red]" % e) - - -# ===== Bookmarks ===== +# ===== Favorite ===== @cli.command() -@click.option("--count", "-n", type=int, default=None, help="Number of tweets to fetch.") +@click.option("--max", "-n", "max_count", type=int, default=None, help="Max number of tweets to fetch.") @click.option("--json", "as_json", is_flag=True, help="Output as JSON.") -@click.option("--browser", "-b", default="chrome", help="Browser to extract cookies from.") @click.option("--output", "-o", "output_file", type=str, default=None, help="Save tweets to JSON file.") @click.option("--no-filter", is_flag=True, help="Skip filtering.") -@click.option("--no-summary", is_flag=True, help="Skip AI summary.") -def bookmarks(count, as_json, browser, output_file, no_filter, no_summary): - # type: (int, bool, str, str, bool, bool) -> None - """Fetch bookmarked tweets.""" +def favorite(max_count, as_json, output_file, no_filter): + # type: (int, bool, str, bool) -> None + """Fetch bookmarked (favorite) tweets.""" config = load_config() - fetch_count = count or 50 + fetch_count = max_count or 50 console.print("\n🔐 Getting Twitter cookies...") try: - cookies = get_cookies(browser) + cookies = get_cookies() except RuntimeError as e: console.print("[red]❌ %s[/red]" % e) sys.exit(1) client = TwitterClient(cookies["auth_token"], cookies["ct0"]) - console.print("🔖 Fetching bookmarks (%d tweets)...\n" % fetch_count) + console.print("🔖 Fetching favorites (%d tweets)...\n" % fetch_count) start = time.time() tweets = client.fetch_bookmarks(fetch_count) elapsed = time.time() - start - console.print("✅ Fetched %d bookmarks in %.1fs\n" % (len(tweets), elapsed)) + console.print("✅ Fetched %d favorites in %.1fs\n" % (len(tweets), elapsed)) # Filter if no_filter: @@ -261,29 +238,204 @@ def bookmarks(count, as_json, browser, output_file, no_filter, no_summary): click.echo(tweets_to_json(filtered)) return - print_tweet_table(filtered, console, title="🔖 Bookmarks — %d tweets" % len(filtered)) + print_tweet_table(filtered, console, title="🔖 Favorites — %d tweets" % len(filtered)) console.print() - # AI Summary - if no_summary: - return - ai_config = config.get("ai", {}) - if not ai_config.get("api_key"): - console.print( - "[yellow]⚠️ AI summary skipped: no API key configured.[/yellow]" - ) - return +# ===== User ===== +@cli.command() +@click.argument("screen_name") +def user(screen_name): + # type: (str,) -> None + """View a user's profile. SCREEN_NAME is the @handle (without @).""" + screen_name = screen_name.lstrip("@") + client = _get_client() + console.print("👤 Fetching user @%s..." % screen_name) try: - console.print("🤖 Calling AI...") - summary = summarize(filtered, ai_config) - console.print("\n" + "═" * 50) - console.print("📝 AI Summary") - console.print("═" * 50 + "\n") - console.print(summary) - except Exception as e: - console.print("[red]❌ AI summary failed: %s[/red]" % e) + profile = client.fetch_user(screen_name) + console.print() + print_user_profile(profile, console) + except RuntimeError as e: + console.print("[red]❌ %s[/red]" % e) + sys.exit(1) + + +@cli.command("user-posts") +@click.argument("screen_name") +@click.option("--max", "-n", "max_count", type=int, default=20, help="Max number of tweets to fetch.") +@click.option("--json", "as_json", is_flag=True, help="Output as JSON.") +def user_posts(screen_name, max_count, as_json): + # type: (str, int, bool) -> None + """List a user's tweets. SCREEN_NAME is the @handle (without @).""" + screen_name = screen_name.lstrip("@") + client = _get_client() + console.print("👤 Fetching @%s's profile..." % screen_name) + try: + profile = client.fetch_user(screen_name) + except RuntimeError as e: + console.print("[red]❌ %s[/red]" % e) + sys.exit(1) + + console.print("📝 Fetching tweets (%d)...\n" % max_count) + start = time.time() + try: + tweets = client.fetch_user_tweets(profile.id, max_count) + except RuntimeError as e: + console.print("[red]❌ %s[/red]" % e) + sys.exit(1) + elapsed = time.time() - start + console.print("✅ Fetched %d tweets in %.1fs\n" % (len(tweets), elapsed)) + + if as_json: + click.echo(tweets_to_json(tweets)) + return + + print_tweet_table(tweets, console, title="📝 @%s — %d tweets" % (screen_name, len(tweets))) + console.print() + + +@cli.command() +@click.argument("screen_name") +@click.option("--max", "-n", "max_count", type=int, default=20, help="Max number of users to show.") +def followers(screen_name, max_count): + # type: (str, int) -> None + """List a user's followers. SCREEN_NAME is the @handle (without @).""" + screen_name = screen_name.lstrip("@") + client = _get_client() + console.print("👤 Fetching @%s's profile..." % screen_name) + try: + profile = client.fetch_user(screen_name) + except RuntimeError as e: + console.print("[red]❌ %s[/red]" % e) + sys.exit(1) + + console.print("👥 Fetching followers...\n") + try: + users = client.fetch_followers(profile.id, max_count) + except RuntimeError as e: + console.print("[red]❌ %s[/red]" % e) + sys.exit(1) + print_user_table(users, console, title="👥 @%s's followers — %d" % (screen_name, len(users))) + console.print() + + +@cli.command() +@click.argument("screen_name") +@click.option("--max", "-n", "max_count", type=int, default=20, help="Max number of users to show.") +def following(screen_name, max_count): + # type: (str, int) -> None + """List users that someone follows. SCREEN_NAME is the @handle (without @).""" + screen_name = screen_name.lstrip("@") + client = _get_client() + console.print("👤 Fetching @%s's profile..." % screen_name) + try: + profile = client.fetch_user(screen_name) + except RuntimeError as e: + console.print("[red]❌ %s[/red]" % e) + sys.exit(1) + + console.print("👥 Fetching following...\n") + try: + users = client.fetch_following(profile.id, max_count) + except RuntimeError as e: + console.print("[red]❌ %s[/red]" % e) + sys.exit(1) + print_user_table(users, console, title="👥 @%s follows — %d" % (screen_name, len(users))) + console.print() + + +# ===== Post / Reply / Quote / Delete ===== + +def _get_client(): + # type: () -> TwitterClient + """Helper to authenticate and create a TwitterClient.""" + console.print("\n🔐 Getting Twitter cookies...") + try: + cookies = get_cookies() + except RuntimeError as e: + console.print("[red]❌ %s[/red]" % e) + sys.exit(1) + return TwitterClient(cookies["auth_token"], cookies["ct0"]) + + +@cli.command() +@click.argument("text") +def post(text): + # type: (str,) -> None + """Post a new tweet.""" + client = _get_client() + console.print("✏️ Posting tweet...") + try: + result = client.create_tweet(text) + tweet_id = result["tweet_id"] + console.print("\n[green]✅ Tweet posted![/green]") + console.print(" ID: %s" % tweet_id) + console.print(" URL: https://x.com/i/status/%s" % tweet_id) + console.print(' Text: "%s"' % result["text"][:100]) + except RuntimeError as e: + console.print("[red]❌ %s[/red]" % e) + sys.exit(1) + + +@cli.command() +@click.argument("tweet_id") +@click.argument("text") +def reply(tweet_id, text): + # type: (str, str) -> None + """Reply to a tweet.""" + client = _get_client() + console.print("💬 Replying to %s..." % tweet_id) + try: + result = client.create_tweet(text, reply_to=tweet_id) + new_id = result["tweet_id"] + console.print("\n[green]✅ Reply posted![/green]") + console.print(" ID: %s" % new_id) + console.print(" URL: https://x.com/i/status/%s" % new_id) + console.print(' Text: "%s"' % result["text"][:100]) + except RuntimeError as e: + console.print("[red]❌ %s[/red]" % e) + sys.exit(1) + + +@cli.command() +@click.argument("tweet_url") +@click.argument("text") +def quote(tweet_url, text): + # type: (str, str) -> None + """Quote a tweet. TWEET_URL can be a full URL or tweet ID.""" + # If user passes just an ID, convert to URL + if not tweet_url.startswith("http"): + tweet_url = "https://x.com/i/status/%s" % tweet_url + client = _get_client() + console.print("🔄 Quoting %s..." % tweet_url) + try: + result = client.create_tweet(text, quote_tweet_url=tweet_url) + new_id = result["tweet_id"] + console.print("\n[green]✅ Quote tweet posted![/green]") + console.print(" ID: %s" % new_id) + console.print(" URL: https://x.com/i/status/%s" % new_id) + console.print(' Text: "%s"' % result["text"][:100]) + except RuntimeError as e: + console.print("[red]❌ %s[/red]" % e) + sys.exit(1) + + +@cli.command() +@click.argument("tweet_id") +@click.confirmation_option(prompt="Are you sure you want to delete this tweet?") +def delete(tweet_id): + # type: (str,) -> None + """Delete a tweet by ID.""" + client = _get_client() + console.print("🗑️ Deleting tweet %s..." % tweet_id) + try: + client.delete_tweet(tweet_id) + console.print("\n[green]✅ Tweet deleted![/green]") + console.print(" ID: %s" % tweet_id) + except RuntimeError as e: + console.print("[red]❌ %s[/red]" % e) + sys.exit(1) if __name__ == "__main__": diff --git a/twitter_cli/client.py b/twitter_cli/client.py index 306b651..76a4ed7 100644 --- a/twitter_cli/client.py +++ b/twitter_cli/client.py @@ -15,7 +15,7 @@ import ssl import urllib.request from typing import Any, Callable, Dict, List, Optional, Tuple -from .models import Author, Metrics, Tweet, TweetMedia +from .models import Author, Metrics, Tweet, TweetMedia, UserProfile logger = logging.getLogger(__name__) @@ -27,8 +27,14 @@ BEARER_TOKEN = ( # Last-resort fallback query IDs FALLBACK_QUERY_IDS = { - "HomeTimeline": "HJFjzBgCs16TqxewQOeLNg", + "HomeTimeline": "c-CzHF1LboFilMpsx4ZCrQ", "Bookmarks": "VFdMm9iVZxlU6hD86gfW_A", + "CreateTweet": "oB-5XsHNAbjvARJEc8CZFw", + "DeleteTweet": "VaenaVgh5q5ih7kvyVjgtg", + "UserByScreenName": "1VOOyvKkiI3FMmkeDNxM9A", + "UserTweets": "E3opETHurmVJflFsUBVuUQ", + "Followers": "IOh4aS6UdGWGJUYTqliQ7Q", + "Following": "zx6e-TLzRkeDO_a7p4b3JQ", } # Community-maintained API definition (auto-updated daily) @@ -39,14 +45,20 @@ TWITTER_OPENAPI_URL = ( # Default features flags required by the GraphQL endpoint FEATURES = { + "rweb_video_screen_enabled": False, + "profile_label_improvements_pcf_label_in_post_enabled": True, "rweb_tipjar_consumption_enabled": True, - "responsive_web_graphql_exclude_directive_enabled": True, "verified_phone_label_enabled": False, "creator_subscriptions_tweet_preview_api_enabled": True, "responsive_web_graphql_timeline_navigation_enabled": True, "responsive_web_graphql_skip_user_profile_image_extensions_enabled": False, + "premium_content_api_read_enabled": False, "communities_web_enable_tweet_community_results_fetch": True, "c9s_tweet_anatomy_moderator_badge_enabled": True, + "responsive_web_grok_analyze_button_fetch_trends_enabled": False, + "responsive_web_grok_analyze_post_followups_enabled": True, + "responsive_web_jetfuel_frame": False, + "responsive_web_grok_share_attachment_enabled": True, "articles_preview_enabled": True, "responsive_web_edit_tweet_api_enabled": True, "graphql_is_translatable_rweb_tweet_is_translatable_enabled": True, @@ -54,13 +66,15 @@ FEATURES = { "longform_notetweets_consumption_enabled": True, "responsive_web_twitter_article_tweet_consumption_enabled": True, "tweet_awards_web_tipping_enabled": False, + "responsive_web_grok_show_grok_translated_post": False, + "responsive_web_grok_analysis_button_from_backend": False, "creator_subscriptions_quote_tweet_preview_enabled": False, "freedom_of_speech_not_reach_fetch_enabled": True, "standardized_nudges_misinfo": True, "tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled": True, - "rweb_video_timestamps_enabled": True, "longform_notetweets_rich_text_read_enabled": True, "longform_notetweets_inline_media_enabled": True, + "responsive_web_grok_image_annotation_enabled": True, "responsive_web_enhance_cards_enabled": False, } @@ -152,17 +166,18 @@ def _fetch_from_github(operation_name): def _resolve_query_id(operation_name): # type: (str) -> str - """Resolve queryId using three-tier strategy: bundle scan -> GitHub -> fallback.""" + """Resolve queryId using three-tier strategy: fallback -> GitHub -> bundle scan.""" if operation_name in _cached_query_ids: return _cached_query_ids[operation_name] - logger.info("Auto-detecting %s queryId...", operation_name) + # Tier 1: Hardcoded fallback (instant, no network) + fallback = FALLBACK_QUERY_IDS.get(operation_name) + if fallback: + logger.debug("Using fallback queryId for %s: %s", operation_name, fallback) + _cached_query_ids[operation_name] = fallback + return fallback - # Tier 1: JS bundle scan - _scan_bundles() - if operation_name in _cached_query_ids: - logger.info("Found %s queryId: %s", operation_name, _cached_query_ids[operation_name]) - return _cached_query_ids[operation_name] + logger.info("Auto-detecting %s queryId (no fallback available)...", operation_name) # Tier 2: GitHub github_id = _fetch_from_github(operation_name) @@ -170,12 +185,11 @@ def _resolve_query_id(operation_name): _cached_query_ids[operation_name] = github_id return github_id - # Tier 3: Hardcoded fallback - fallback = FALLBACK_QUERY_IDS.get(operation_name) - if fallback: - logger.info("Using hardcoded fallback queryId for %s: %s", operation_name, fallback) - _cached_query_ids[operation_name] = fallback - return fallback + # Tier 3: JS bundle scan + _scan_bundles() + if operation_name in _cached_query_ids: + logger.info("Found %s queryId: %s", operation_name, _cached_query_ids[operation_name]) + return _cached_query_ids[operation_name] raise RuntimeError( 'Cannot resolve queryId for "%s" — all detection methods failed' % operation_name @@ -291,6 +305,260 @@ class TwitterClient: body = e.read().decode("utf-8", errors="replace") raise RuntimeError("Twitter API error %d: %s" % (e.code, body[:500])) + def _api_post(self, url, payload): + # type: (str, Dict[str, Any]) -> Any + """Make authenticated POST request to Twitter API.""" + headers = self._build_headers() + data = json.dumps(payload).encode("utf-8") + req = urllib.request.Request(url, data=data, method="POST") + for k, v in headers.items(): + req.add_header(k, v) + + ctx = _create_ssl_context() + try: + with urllib.request.urlopen(req, context=ctx, timeout=30) as resp: + body = resp.read().decode("utf-8") + return json.loads(body) + except urllib.error.HTTPError as e: + body = e.read().decode("utf-8", errors="replace") + raise RuntimeError("Twitter API error %d: %s" % (e.code, body[:500])) + + def create_tweet(self, text, reply_to=None, quote_tweet_url=None): + # type: (str, Optional[str], Optional[str]) -> Dict[str, Any] + """Create a tweet, reply, or quote tweet. + + Args: + text: Tweet text content. + reply_to: Tweet ID to reply to (optional). + quote_tweet_url: URL of tweet to quote (optional). + + Returns: + Dict with tweet_id and text of the created tweet. + """ + query_id = _resolve_query_id("CreateTweet") + url = "https://x.com/i/api/graphql/%s/CreateTweet" % query_id + + variables = { + "tweet_text": text, + "dark_request": False, + "media": {"media_entities": [], "possibly_sensitive": False}, + "semantic_annotation_ids": [], + } # type: Dict[str, Any] + + if reply_to: + variables["reply"] = { + "in_reply_to_tweet_id": reply_to, + "exclude_reply_user_ids": [], + } + + if quote_tweet_url: + variables["attachment_url"] = quote_tweet_url + + features = { + "communities_web_enable_tweet_community_results_fetch": True, + "c9s_tweet_anatomy_moderator_badge_enabled": True, + "responsive_web_edit_tweet_api_enabled": True, + "graphql_is_translatable_rweb_tweet_is_translatable_enabled": True, + "view_counts_everywhere_api_enabled": True, + "longform_notetweets_consumption_enabled": True, + "responsive_web_twitter_article_tweet_consumption_enabled": True, + "tweet_awards_web_tipping_enabled": False, + "creator_subscriptions_quote_tweet_preview_enabled": False, + "longform_notetweets_rich_text_read_enabled": True, + "longform_notetweets_inline_media_enabled": True, + "articles_preview_enabled": True, + "rweb_video_timestamps_enabled": True, + "rweb_tipjar_consumption_enabled": True, + "responsive_web_graphql_exclude_directive_enabled": True, + "verified_phone_label_enabled": False, + "freedom_of_speech_not_reach_fetch_enabled": True, + "standardized_nudges_misinfo": True, + "tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled": True, + "responsive_web_graphql_skip_user_profile_image_extensions_enabled": False, + "responsive_web_graphql_timeline_navigation_enabled": True, + "responsive_web_enhance_cards_enabled": False, + } + + payload = { + "variables": variables, + "features": features, + "queryId": query_id, + } + + data = self._api_post(url, payload) + + # Parse response + result = _deep_get(data, "data", "create_tweet", "tweet_results", "result") + if not result: + errors = data.get("errors", []) + if errors: + raise RuntimeError("CreateTweet failed: %s" % errors[0].get("message", str(errors))) + raise RuntimeError("CreateTweet failed: unexpected response") + + tweet_id = result.get("rest_id", "") + tweet_text = _deep_get(result, "legacy", "full_text") or text + return {"tweet_id": tweet_id, "text": tweet_text} + + def delete_tweet(self, tweet_id): + # type: (str) -> bool + """Delete a tweet by ID. + + Returns: + True if deletion was successful. + """ + query_id = _resolve_query_id("DeleteTweet") + url = "https://x.com/i/api/graphql/%s/DeleteTweet" % query_id + + payload = { + "variables": {"tweet_id": tweet_id, "dark_request": False}, + "queryId": query_id, + } + + data = self._api_post(url, payload) + + # Check response + result = _deep_get(data, "data", "delete_tweet", "tweet_results") + if result is not None: + return True + + errors = data.get("errors", []) + if errors: + raise RuntimeError("DeleteTweet failed: %s" % errors[0].get("message", str(errors))) + # Some successful deletions return empty result + return True + + def fetch_user(self, screen_name): + # type: (str) -> UserProfile + """Fetch user profile by screen name.""" + query_id = _resolve_query_id("UserByScreenName") + variables = { + "screen_name": screen_name, + "withSafetyModeUserFields": True, + } + features = { + "hidden_profile_subscriptions_enabled": True, + "rweb_tipjar_consumption_enabled": True, + "responsive_web_graphql_exclude_directive_enabled": True, + "verified_phone_label_enabled": False, + "subscriptions_verification_info_is_identity_verified_enabled": True, + "subscriptions_verification_info_verified_since_enabled": True, + "highlights_tweets_tab_ui_enabled": True, + "responsive_web_twitter_article_notes_tab_enabled": True, + "subscriptions_feature_can_gift_premium": True, + "creator_subscriptions_tweet_preview_api_enabled": True, + "responsive_web_graphql_skip_user_profile_image_extensions_enabled": False, + "responsive_web_graphql_timeline_navigation_enabled": True, + } + + url = "https://x.com/i/api/graphql/%s/UserByScreenName?" % query_id + url += "variables=%s&features=%s" % ( + urllib.request.quote(json.dumps(variables)), + urllib.request.quote(json.dumps(features)), + ) + + data = self._api_get(url) + result = _deep_get(data, "data", "user", "result") + if not result: + raise RuntimeError("User @%s not found" % screen_name) + + legacy = result.get("legacy", {}) + core = result.get("core", {}) + return UserProfile( + id=result.get("rest_id", ""), + name=core.get("name") or legacy.get("name", ""), + screen_name=core.get("screen_name") or legacy.get("screen_name", screen_name), + bio=legacy.get("description", ""), + location=legacy.get("location", ""), + url=( + legacy.get("entities", {}).get("url", {}).get("urls", [{}])[0].get("expanded_url", "") + if legacy.get("entities", {}).get("url") + else "" + ), + followers_count=legacy.get("followers_count", 0), + following_count=legacy.get("friends_count", 0), + tweets_count=legacy.get("statuses_count", 0), + likes_count=legacy.get("favourites_count", 0), + verified=bool(result.get("is_blue_verified") or legacy.get("verified", False)), + profile_image_url=legacy.get("profile_image_url_https", ""), + created_at=legacy.get("created_at", ""), + ) + + def fetch_user_tweets(self, user_id, count=20): + # type: (str, int) -> List[Tweet] + """Fetch tweets posted by a user.""" + query_id = _resolve_query_id("UserTweets") + return self._fetch_timeline( + query_id, + "UserTweets", + count, + lambda data: _deep_get(data, "data", "user", "result", "timeline_v2", "timeline", "instructions"), + extra_variables={ + "userId": user_id, + "withQuickPromoteEligibilityTweetFields": True, + "withVoice": True, + "withV2Timeline": True, + }, + ) + + def fetch_followers(self, user_id, count=20): + # type: (str, int) -> List[UserProfile] + """Fetch user's followers.""" + query_id = _resolve_query_id("Followers") + return self._fetch_user_list(query_id, "Followers", user_id, count) + + def fetch_following(self, user_id, count=20): + # type: (str, int) -> List[UserProfile] + """Fetch users that this user follows.""" + query_id = _resolve_query_id("Following") + return self._fetch_user_list(query_id, "Following", user_id, count) + + def _fetch_user_list(self, query_id, operation_name, user_id, count): + # type: (str, str, str, int) -> List[UserProfile] + """Generic user list fetcher (followers/following).""" + variables = { + "userId": user_id, + "count": min(count, 50), + "includePromotedContent": False, + } # type: Dict[str, Any] + + url = "https://x.com/i/api/graphql/%s/%s?" % (query_id, operation_name) + url += "variables=%s&features=%s" % ( + urllib.request.quote(json.dumps(variables)), + urllib.request.quote(json.dumps(FEATURES)), + ) + + data = self._api_get(url) + users = [] # type: List[UserProfile] + + instructions = _deep_get(data, "data", "user", "result", "timeline", "timeline", "instructions") + if not isinstance(instructions, list): + return users + + for instruction in instructions: + entries = instruction.get("entries", []) + for entry in entries: + content = entry.get("content", {}) + item_content = content.get("itemContent", {}) + user_results = item_content.get("user_results", {}).get("result") + if not user_results: + continue + legacy = user_results.get("legacy", {}) + core = user_results.get("core", {}) + if not legacy: + continue + users.append(UserProfile( + id=user_results.get("rest_id", ""), + name=core.get("name") or legacy.get("name", ""), + screen_name=core.get("screen_name") or legacy.get("screen_name", ""), + bio=legacy.get("description", ""), + followers_count=legacy.get("followers_count", 0), + following_count=legacy.get("friends_count", 0), + verified=bool(user_results.get("is_blue_verified") or legacy.get("verified", False)), + profile_image_url=legacy.get("profile_image_url_https", ""), + )) + + return users[:count] + def _parse_timeline_response(self, data, get_instructions): # type: (Any, Callable) -> Tuple[List[Tweet], Optional[str]] """Parse timeline GraphQL response into tweets + next cursor.""" diff --git a/twitter_cli/config.py b/twitter_cli/config.py index 22ce784..c24261f 100644 --- a/twitter_cli/config.py +++ b/twitter_cli/config.py @@ -5,7 +5,6 @@ Uses a simple built-in YAML parser to avoid adding PyYAML as a dependency. from __future__ import annotations -import os import re from pathlib import Path from typing import Any, Dict, List, Union @@ -29,13 +28,6 @@ DEFAULT_CONFIG = { "views_log": 0.5, }, }, - "ai": { - "provider": "openai", - "api_key": "", - "model": "doubao-seed-2.0-code", - "base_url": "https://ark.cn-beijing.volces.com/api/coding", - "language": "zh-CN", - }, } # type: Dict[str, Any] @@ -161,15 +153,9 @@ def load_config(config_path=None): # Ensure nested dicts exist config.setdefault("fetch", DEFAULT_CONFIG["fetch"]) config.setdefault("filter", DEFAULT_CONFIG["filter"]) - config.setdefault("ai", DEFAULT_CONFIG["ai"]) # Deep-copy filter weights if needed if "filter" in config and "weights" not in config["filter"]: config["filter"]["weights"] = dict(DEFAULT_CONFIG["filter"]["weights"]) - # AI API key fallback to env var - ai = config.get("ai", {}) - if not ai.get("api_key"): - ai["api_key"] = os.environ.get("AI_API_KEY", "") - return config diff --git a/twitter_cli/formatter.py b/twitter_cli/formatter.py index 1d52aae..62ec741 100644 --- a/twitter_cli/formatter.py +++ b/twitter_cli/formatter.py @@ -10,7 +10,7 @@ from rich.panel import Panel from rich.table import Table from rich.text import Text -from .models import Tweet +from .models import Tweet, UserProfile def format_number(n): @@ -69,6 +69,9 @@ def print_tweet_table(tweets, console=None, title=None): qt_text = qt.text.replace("\n", " ")[:60] text += "\n┌ @%s: %s" % (qt.author.screen_name, qt_text) + # Tweet link + text += "\n🔗 x.com/%s/status/%s" % (tweet.author.screen_name, tweet.id) + # Stats stats = ( "❤️ %s 🔄 %s\n💬 %s 👁️ %s" @@ -205,3 +208,82 @@ def tweets_to_json(tweets): } result.append(d) return json.dumps(result, ensure_ascii=False, indent=2) + + +def print_user_profile(user, console=None): + # type: (UserProfile, Optional[Console]) -> None + """Print user profile as a rich panel.""" + if console is None: + console = Console() + + verified = " ✓" if user.verified else "" + header = "@%s%s (%s)" % (user.screen_name, verified, user.name) + + lines = [] + if user.bio: + lines.append(user.bio) + lines.append("") + + if user.location: + lines.append("📍 %s" % user.location) + if user.url: + lines.append("🔗 %s" % user.url) + if user.location or user.url: + lines.append("") + + lines.append( + "👥 %s followers · %s following · %s tweets · %s likes" + % ( + format_number(user.followers_count), + format_number(user.following_count), + format_number(user.tweets_count), + format_number(user.likes_count), + ) + ) + + if user.created_at: + lines.append("📅 Joined %s" % user.created_at) + lines.append("🔗 x.com/%s" % user.screen_name) + + console.print(Panel( + "\n".join(lines), + title=header, + border_style="cyan", + expand=True, + )) + + +def print_user_table(users, console=None, title=None): + # type: (List[UserProfile], Optional[Console], Optional[str]) -> None + """Print a list of users as a rich table.""" + if console is None: + console = Console() + + if not title: + title = "👥 Users — %d" % len(users) + + table = Table(title=title, show_lines=True, expand=True) + table.add_column("#", style="dim", width=3, justify="right") + table.add_column("User", style="cyan", width=20, no_wrap=True) + table.add_column("Bio", ratio=3) + table.add_column("Stats", style="green", width=22, no_wrap=True) + + for i, user in enumerate(users): + verified = " ✓" if user.verified else "" + user_text = "@%s%s\n%s" % (user.screen_name, verified, user.name) + + bio = (user.bio or "").replace("\n", " ").strip() + if len(bio) > 100: + bio = bio[:97] + "..." + + stats = ( + "👥 %s followers\n📝 %s following" + % ( + format_number(user.followers_count), + format_number(user.following_count), + ) + ) + + table.add_row(str(i + 1), user_text, bio, stats) + + console.print(table) diff --git a/twitter_cli/models.py b/twitter_cli/models.py index e8f9faa..d07880a 100644 --- a/twitter_cli/models.py +++ b/twitter_cli/models.py @@ -50,3 +50,20 @@ class Tweet: retweeted_by: Optional[str] = None quoted_tweet: Optional[Tweet] = None score: float = 0.0 + + +@dataclass +class UserProfile: + id: str + name: str + screen_name: str + bio: str = "" + location: str = "" + url: str = "" + followers_count: int = 0 + following_count: int = 0 + tweets_count: int = 0 + likes_count: int = 0 + verified: bool = False + profile_image_url: str = "" + created_at: str = "" diff --git a/twitter_cli/summarizer.py b/twitter_cli/summarizer.py deleted file mode 100644 index 1c0ef4d..0000000 --- a/twitter_cli/summarizer.py +++ /dev/null @@ -1,164 +0,0 @@ -"""AI summarization module. - -Supports OpenAI-compatible (doubao, deepseek, openai) and Anthropic APIs. -Uses urllib.request for zero extra dependencies. -""" - -from __future__ import annotations - -import json -import logging -import ssl -import urllib.request -from typing import Any, Dict, List - -from .models import Tweet - -logger = logging.getLogger(__name__) - -SYSTEM_MESSAGE = "你是一个专业的 Twitter/X 信息流分析师，擅长提炼关键信息和发现趋势。" - - -def _build_prompt(tweets, language="zh-CN"): - # type: (List[Tweet], str) -> str - """Build the summarization prompt.""" - lines = [] - for i, t in enumerate(tweets): - score_str = " [score: %.1f]" % t.score if t.score else "" - rt = " (RT by @%s)" % t.retweeted_by if t.is_retweet and t.retweeted_by else "" - media_str = "" - if t.media: - media_str = " [%s]" % ", ".join(m.type for m in t.media) - url_str = "" - if t.urls: - url_str = "\n Links: %s" % ", ".join(t.urls) - quoted = "" - if t.quoted_tweet: - qt = t.quoted_tweet - quoted = "\n Quoting @%s: %s..." % (qt.author.screen_name, qt.text[:100].replace("\n", " ")) - - text_preview = t.text.replace("\n", " ")[:300] - lines.append( - '%d. @%s (%s)%s%s\n' - ' "%s"\n' - ' ❤️%d 🔄%d 💬%d 🔖%d 👁️%d%s%s%s' - % ( - i + 1, t.author.screen_name, t.author.name, rt, score_str, - text_preview, - t.metrics.likes, t.metrics.retweets, t.metrics.replies, - t.metrics.bookmarks, t.metrics.views, - media_str, url_str, quoted, - ) - ) - - tweet_summaries = "\n\n".join(lines) - - if language.startswith("zh"): - lang_inst = "请用中文输出。" - else: - lang_inst = "Please output in %s." % language - - return ( - "你是一个 Twitter/X 信息流分析师。请对以下 %d 条推文进行摘要总结。\n\n" - "要求：\n" - "1. 按主题分组（如：AI & 编程、Crypto、工具推荐、生活观点等）\n" - "2. 每组列出关键推文和核心观点，标注作者 @handle\n" - "3. 标注数据亮点（高赞/高收藏推文用 🔥 标记）\n" - "4. 最后用 2-3 句话总结今天 timeline 的整体趋势\n" - "5. %s\n\n" - "推文数据：\n\n%s" - ) % (len(tweets), lang_inst, tweet_summaries) - - -def _call_openai(prompt, config): - # type: (str, Dict[str, Any]) -> str - """Call OpenAI-compatible API.""" - url = config.get("base_url", "").rstrip("/") - if not url.endswith("/chat/completions"): - if not url.endswith("/v1"): - url += "/v1" - url += "/chat/completions" - - payload = json.dumps({ - "model": config.get("model", ""), - "messages": [ - {"role": "system", "content": SYSTEM_MESSAGE}, - {"role": "user", "content": prompt}, - ], - "temperature": 0.3, - "max_tokens": 4096, - }).encode("utf-8") - - req = urllib.request.Request(url, data=payload) - req.add_header("Content-Type", "application/json") - req.add_header("Authorization", "Bearer %s" % config.get("api_key", "")) - - ctx = ssl.create_default_context() - with urllib.request.urlopen(req, context=ctx, timeout=120) as resp: - data = json.loads(resp.read().decode("utf-8")) - - choices = data.get("choices", []) - if choices: - return choices[0].get("message", {}).get("content", "") - return "" - - -def _call_anthropic(prompt, config): - # type: (str, Dict[str, Any]) -> str - """Call Anthropic Messages API.""" - url = config.get("base_url", "").rstrip("/") - if not url.endswith("/messages"): - if not url.endswith("/v1"): - url += "/v1" - url += "/messages" - - payload = json.dumps({ - "model": config.get("model", ""), - "system": SYSTEM_MESSAGE, - "messages": [{"role": "user", "content": prompt}], - "temperature": 0.3, - "max_tokens": 4096, - }).encode("utf-8") - - req = urllib.request.Request(url, data=payload) - req.add_header("Content-Type", "application/json") - req.add_header("x-api-key", config.get("api_key", "")) - req.add_header("anthropic-version", "2023-06-01") - - ctx = ssl.create_default_context() - with urllib.request.urlopen(req, context=ctx, timeout=120) as resp: - data = json.loads(resp.read().decode("utf-8")) - - content_blocks = data.get("content", []) - for block in content_blocks: - if block.get("type") == "text": - return block.get("text", "") - return "" - - -def summarize(tweets, config): - # type: (List[Tweet], Dict[str, Any]) -> str - """Summarize tweets using the configured AI provider. - - Config keys: provider, api_key, model, base_url, language - """ - api_key = config.get("api_key", "") - if not api_key: - raise RuntimeError( - "AI API key not configured.\n" - "Set ai.api_key in config.yaml or export AI_API_KEY=your_key" - ) - - if not tweets: - return "No tweets to summarize." - - language = config.get("language", "zh-CN") - prompt = _build_prompt(tweets, language) - provider = config.get("provider", "openai") - - logger.info("Calling AI (%s/%s)...", provider, config.get("model", "")) - - if provider == "anthropic": - return _call_anthropic(prompt, config) - else: - return _call_openai(prompt, config)