From 19ab11d6a4e046d0d54ca85f65f8d8df2c13e704 Mon Sep 17 00:00:00 2001 From: jackwener Date: Tue, 10 Mar 2026 11:02:32 +0800 Subject: [PATCH] fix: harden auth flow and sync browser support docs --- README.md | 34 +++--- SKILL.md | 18 +-- tests/conftest.py | 12 ++ tests/test_auth.py | 166 ++++++++++++++++++++++++++ tests/test_cli.py | 62 ++++++++++ tests/test_parser_fixtures.py | 93 +++++++++++++++ twitter_cli/auth.py | 145 ++++++++++++----------- twitter_cli/cli.py | 212 ++++++++++++++++++++++------------ 8 files changed, 572 insertions(+), 170 deletions(-) create mode 100644 tests/test_auth.py create mode 100644 tests/test_parser_fixtures.py diff --git a/README.md b/README.md index 24f11a2..df51ebe 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,7 @@ A terminal-first CLI for Twitter/X: read timelines, bookmarks, and user profiles - Delete: remove your own tweets - Like / Unlike: manage tweet likes - Retweet / Unretweet: manage retweets -- Bookmark: favorite/unfavorite +- Bookmark: bookmark/unbookmark (`favorite/unfavorite` kept as compatibility aliases) **Auth & Anti-Detection:** - Cookie auth: use browser cookies or environment variables @@ -83,8 +83,8 @@ twitter feed --json > tweets.json twitter feed --input tweets.json # Bookmarks -twitter favorites -twitter favorites --max 30 --json +twitter bookmarks +twitter bookmarks --max 30 --json # Search twitter search "Claude Code" @@ -117,8 +117,8 @@ twitter like 1234567890 twitter unlike 1234567890 twitter retweet 1234567890 twitter unretweet 1234567890 -twitter favorite 1234567890 -twitter unfavorite 1234567890 +twitter bookmark 1234567890 +twitter unbookmark 1234567890 ``` ### Authentication @@ -126,7 +126,7 @@ twitter unfavorite 1234567890 twitter-cli uses this auth priority: 1. **Environment variables**: `TWITTER_AUTH_TOKEN` + `TWITTER_CT0` -2. **Browser cookies** (recommended): auto-extract from Chrome/Edge/Firefox/Brave +2. **Browser cookies** (recommended): auto-extract from Arc/Chrome/Edge/Firefox/Brave Browser extraction is recommended — it forwards ALL Twitter cookies (not just `auth_token` + `ct0`), making requests indistinguishable from real browser traffic. @@ -174,6 +174,10 @@ rateLimit: maxCount: 200 # hard cap on fetched items ``` +Fetch behavior: + +- `fetch.count` is the default item count for read commands when `--max` is omitted + Filter behavior: - Default behavior: no ranking filter unless `--filter` is passed @@ -206,8 +210,9 @@ Mode behavior: ### Troubleshooting - `No Twitter cookies found` - - Ensure you are logged in to `x.com` in a supported browser. + - Ensure you are logged in to `x.com` in a supported browser (Arc/Chrome/Edge/Firefox/Brave). - Or set `TWITTER_AUTH_TOKEN` and `TWITTER_CT0` manually. + - Run with `-v` to see browser extraction diagnostics. - `Cookie expired or invalid (HTTP 401/403)` - Re-login to `x.com` and retry. @@ -290,7 +295,7 @@ After installation, OpenClaw can call `twitter-cli` commands directly. - 删除：删除自己的推文 - 点赞 / 取消点赞 - 转推 / 取消转推 -- 收藏 / 取消收藏：favorite/unfavorite +- 书签 / 取消书签：bookmark/unbookmark（保留 `favorite/unfavorite` 兼容别名） **认证与反风控:** - Cookie 认证：支持环境变量和浏览器自动提取 @@ -317,7 +322,7 @@ twitter feed -t following twitter feed --filter # 收藏 -twitter favorites +twitter bookmarks # 搜索 twitter search "Claude Code" @@ -348,8 +353,8 @@ twitter like 1234567890 twitter unlike 1234567890 twitter retweet 1234567890 twitter unretweet 1234567890 -twitter favorite 1234567890 -twitter unfavorite 1234567890 +twitter bookmark 1234567890 +twitter unbookmark 1234567890 ``` ### 认证说明 @@ -357,7 +362,7 @@ twitter unfavorite 1234567890 认证优先级： 1. **环境变量**：`TWITTER_AUTH_TOKEN` + `TWITTER_CT0` -2. **浏览器提取**（推荐）：Chrome/Edge/Firefox/Brave 全量 Cookie 提取 +2. **浏览器提取**（推荐）：Arc/Chrome/Edge/Firefox/Brave 全量 Cookie 提取 推荐使用浏览器提取方式，会转发所有 Twitter Cookie，让请求和真实浏览器完全一致。 @@ -375,6 +380,8 @@ export TWITTER_PROXY=socks5://127.0.0.1:1080 ### 筛选算法 +未传 `--max` 时，所有读取命令默认使用 `config.yaml` 里的 `fetch.count`。 + 只有在传入 `--filter` 时才会启用筛选评分；默认不筛选。 评分公式： @@ -395,7 +402,8 @@ score = likes_w * likes ### 常见问题 -- 报错 `No Twitter cookies found`：请先登录 `x.com` 或手动设置环境变量。 +- 报错 `No Twitter cookies found`：请先登录 `x.com`，并确认浏览器为 Arc/Chrome/Edge/Firefox/Brave 之一，或手动设置环境变量。 +- 如需查看浏览器提取细节，可加 `-v` 打开诊断日志。 - 报错 `Cookie expired or invalid`：Cookie 过期，重新登录后重试。 - 报错 `Twitter API error 404`：通常是 queryId 轮换，重试即可。 diff --git a/SKILL.md b/SKILL.md index 42514ef..d14938d 100644 --- a/SKILL.md +++ b/SKILL.md @@ -25,7 +25,7 @@ uv tool install twitter-cli ## Authentication -- Auto-extracts browser cookies from Chrome/Edge/Firefox/Brave. +- Auto-extracts browser cookies from Arc/Chrome/Edge/Firefox/Brave. - Or set environment variables: `TWITTER_AUTH_TOKEN` + `TWITTER_CT0`. ## Command Reference @@ -44,9 +44,9 @@ twitter feed --input tweets.json # Read from local JSON file ### Bookmarks ```bash -twitter favorites # List bookmarked tweets -twitter favorites --max 30 --json -twitter favorites --filter # Apply ranking filter +twitter bookmarks # List bookmarked tweets +twitter bookmarks --max 30 --json +twitter bookmarks --filter # Apply ranking filter ``` ### Search @@ -94,8 +94,8 @@ twitter like 1234567890 # Like twitter unlike 1234567890 # Unlike twitter retweet 1234567890 # Retweet twitter unretweet 1234567890 # Unretweet -twitter favorite 1234567890 # Bookmark -twitter unfavorite 1234567890 # Unbookmark +twitter bookmark 1234567890 # Bookmark +twitter unbookmark 1234567890 # Unbookmark ``` ## JSON / Scripting @@ -115,7 +115,7 @@ Filtering is opt-in (disabled by default). Enable with `--filter`. ```bash twitter feed --filter -twitter favorites --filter +twitter bookmarks --filter ``` The scoring formula: @@ -144,12 +144,12 @@ twitter user elonmusk --json # Daily reading workflow twitter feed -t following --filter -twitter favorites --filter +twitter bookmarks --filter ``` ## Error Handling -- `No Twitter cookies found` — login to `x.com` in a supported browser, or set env vars. +- `No Twitter cookies found` — login to `x.com` in Arc/Chrome/Edge/Firefox/Brave, or set env vars. - `Cookie expired or invalid (HTTP 401/403)` — re-login to `x.com` and retry. - `Twitter API error 404` — queryId rotation, retry the command (client has live fallback). diff --git a/tests/conftest.py b/tests/conftest.py index ed8a42f..cdb2594 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,7 @@ from __future__ import annotations +import json +from pathlib import Path from typing import Any import pytest @@ -34,3 +36,13 @@ def tweet_factory(): ) return _make_tweet + + +@pytest.fixture() +def fixture_loader(): + fixture_dir = Path(__file__).parent / "fixtures" + + def _load(name: str) -> Any: + return json.loads((fixture_dir / name).read_text(encoding="utf-8")) + + return _load diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..af49302 --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +import json +import sys +from types import SimpleNamespace + +import pytest + +from twitter_cli import auth + + +def test_get_cookies_prefers_env(monkeypatch) -> None: + monkeypatch.setattr(auth, "load_from_env", lambda: {"auth_token": "env-token", "ct0": "env-csrf"}) + monkeypatch.setattr(auth, "extract_from_browser", lambda: pytest.fail("should not extract from browser")) + seen = [] + monkeypatch.setattr( + auth, + "verify_cookies", + lambda auth_token, ct0, cookie_string=None: seen.append((auth_token, ct0, cookie_string)) or {}, + ) + + cookies = auth.get_cookies() + + assert cookies == {"auth_token": "env-token", "ct0": "env-csrf"} + assert seen == [("env-token", "env-csrf", None)] + + +def test_get_cookies_reextracts_after_verify_failure(monkeypatch) -> None: + monkeypatch.setattr(auth, "load_from_env", lambda: None) + extracted = iter( + [ + {"auth_token": "stale-token", "ct0": "stale-csrf", "cookie_string": "stale=1"}, + {"auth_token": "fresh-token", "ct0": "fresh-csrf", "cookie_string": "fresh=1"}, + ] + ) + monkeypatch.setattr(auth, "extract_from_browser", lambda: next(extracted)) + + calls = [] + + def _verify(auth_token, ct0, cookie_string=None): + calls.append((auth_token, ct0, cookie_string)) + if auth_token == "stale-token": + raise RuntimeError("expired") + return {} + + monkeypatch.setattr(auth, "verify_cookies", _verify) + + cookies = auth.get_cookies() + + assert cookies["auth_token"] == "fresh-token" + assert calls == [ + ("stale-token", "stale-csrf", "stale=1"), + ("fresh-token", "fresh-csrf", "fresh=1"), + ] + + +def test_load_from_env_logs_incomplete_env(monkeypatch, caplog) -> None: + monkeypatch.setenv("TWITTER_AUTH_TOKEN", "token") + monkeypatch.delenv("TWITTER_CT0", raising=False) + + with caplog.at_level("DEBUG"): + cookies = auth.load_from_env() + + assert cookies is None + assert "Environment cookies incomplete" in caplog.text + + +def test_extract_cookies_from_jar_logs_missing_required_cookies(caplog) -> None: + class Cookie: + def __init__(self, domain: str, name: str, value: str) -> None: + self.domain = domain + self.name = name + self.value = value + + jar = [Cookie(".x.com", "auth_token", "token")] + + with caplog.at_level("DEBUG"): + cookies = auth._extract_cookies_from_jar(jar, source="test-jar") + + assert cookies is None + assert "test-jar" in caplog.text + assert "ct0=False" in caplog.text + + +def test_extract_from_browser_logs_warning_when_all_methods_fail(monkeypatch, caplog) -> None: + monkeypatch.setattr(auth, "_extract_in_process", lambda: None) + monkeypatch.setattr(auth, "_extract_via_subprocess", lambda: None) + + with caplog.at_level("WARNING"): + cookies = auth.extract_from_browser() + + assert cookies is None + assert "Twitter cookie extraction failed in both in-process and subprocess modes" in caplog.text + + +def test_extract_in_process_supports_arc(monkeypatch) -> None: + class Cookie: + def __init__(self, domain: str, name: str, value: str) -> None: + self.domain = domain + self.name = name + self.value = value + + fake_module = SimpleNamespace( + arc=lambda: [Cookie(".x.com", "auth_token", "token"), Cookie(".x.com", "ct0", "csrf")], + chrome=lambda: pytest.fail("chrome should not be used when arc succeeds"), + edge=lambda: pytest.fail("edge should not be used when arc succeeds"), + firefox=lambda: pytest.fail("firefox should not be used when arc succeeds"), + brave=lambda: pytest.fail("brave should not be used when arc succeeds"), + ) + monkeypatch.setitem(sys.modules, "browser_cookie3", fake_module) + + cookies = auth._extract_in_process() + + assert cookies is not None + assert cookies["auth_token"] == "token" + assert cookies["ct0"] == "csrf" + + +def test_extract_via_subprocess_script_includes_arc(monkeypatch) -> None: + class Completed: + def __init__(self, stdout: str, stderr: str = "") -> None: + self.stdout = stdout + self.stderr = stderr + + seen = {} + + def _run(cmd, capture_output=True, text=True, timeout=15): + script = cmd[2] + seen["script"] = script + return Completed(json.dumps({"error": "No Twitter cookies found", "attempts": []})) + + monkeypatch.setattr(auth.subprocess, "run", _run) + + cookies = auth._extract_via_subprocess() + + assert cookies is None + assert '("arc", browser_cookie3.arc)' in seen["script"] + + +def test_verify_cookies_logs_attempt_summary_on_non_auth_failures(monkeypatch, caplog) -> None: + class Response: + def __init__(self, status_code: int, payload=None) -> None: + self.status_code = status_code + self._payload = payload or {} + + def json(self): + return self._payload + + class Session: + def __init__(self) -> None: + self.calls = 0 + + def get(self, url, headers=None, timeout=5): + self.calls += 1 + if self.calls == 1: + return Response(404) + raise Exception("network") + + monkeypatch.setattr("twitter_cli.client._get_cffi_session", lambda: Session()) + + with caplog.at_level("INFO"): + result = auth.verify_cookies("token", "csrf") + + assert result == {} + assert "verify_credentials.json=404" in caplog.text + assert "settings.json=Exception" in caplog.text diff --git a/tests/test_cli.py b/tests/test_cli.py index baf87d9..2eb0fd4 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,6 +1,7 @@ from __future__ import annotations from click.testing import CliRunner +import pytest from twitter_cli.cli import cli from twitter_cli.models import UserProfile @@ -26,3 +27,64 @@ def test_cli_feed_json_input_path(tmp_path, tweet_factory) -> None: result = runner.invoke(cli, ["feed", "--input", str(json_path), "--json"]) assert result.exit_code == 0 assert '"id": "1"' in result.output + + +@pytest.mark.parametrize( + "args", + [ + ["favorites"], + ["bookmarks"], + ["search", "x"], + ["user-posts", "alice"], + ["likes", "alice"], + ["list", "123"], + ], +) +def test_cli_commands_wrap_client_creation_errors(monkeypatch, args) -> None: + monkeypatch.setattr( + "twitter_cli.cli._get_client", + lambda config=None: (_ for _ in ()).throw(RuntimeError("boom")), + ) + runner = CliRunner() + + result = runner.invoke(cli, args) + + assert result.exit_code == 1 + assert "boom" in result.output + assert type(result.exception).__name__ == "SystemExit" + + +def test_cli_tweet_accepts_shared_url_with_query(monkeypatch) -> None: + class FakeClient: + def fetch_tweet_detail(self, tweet_id: str, max_count: int): + assert tweet_id == "12345" + assert max_count == 50 + return [] + + monkeypatch.setattr("twitter_cli.cli._get_client", lambda config=None: FakeClient()) + monkeypatch.setattr( + "twitter_cli.cli.load_config", + lambda: {"fetch": {"count": 50}, "filter": {}, "rateLimit": {}}, + ) + runner = CliRunner() + + result = runner.invoke(cli, ["tweet", "https://x.com/user/status/12345?s=20"]) + + assert result.exit_code == 0 + + +def test_cli_bookmark_alias_works(monkeypatch) -> None: + calls = [] + + class FakeClient: + def bookmark_tweet(self, tweet_id: str) -> bool: + calls.append(tweet_id) + return True + + monkeypatch.setattr("twitter_cli.cli._get_client", lambda config=None: FakeClient()) + runner = CliRunner() + + result = runner.invoke(cli, ["bookmark", "123"]) + + assert result.exit_code == 0 + assert calls == ["123"] diff --git a/tests/test_parser_fixtures.py b/tests/test_parser_fixtures.py new file mode 100644 index 0000000..4d42d49 --- /dev/null +++ b/tests/test_parser_fixtures.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +from twitter_cli.client import TwitterClient, _deep_get + + +def _make_client() -> TwitterClient: + client = TwitterClient.__new__(TwitterClient) + client._ct_init_attempted = True + client._client_transaction = None + client._request_delay = 0.0 + client._max_retries = 0 + client._retry_base_delay = 0.0 + client._max_count = 200 + return client + + +def test_parse_home_timeline_fixture(fixture_loader) -> None: + client = _make_client() + payload = fixture_loader("home_timeline.json") + + tweets, cursor = client._parse_timeline_response( + payload, + lambda data: _deep_get(data, "data", "home", "home_timeline_urt", "instructions"), + ) + + assert [tweet.id for tweet in tweets] == ["1", "20"] + assert cursor == "cursor-bottom-1" + assert tweets[0].media[0].type == "photo" + assert tweets[0].urls == ["https://example.com/post"] + assert tweets[1].is_retweet is True + assert tweets[1].retweeted_by == "bob" + assert tweets[1].quoted_tweet is not None + assert tweets[1].quoted_tweet.id == "30" + + +def test_parse_tweet_detail_fixture_with_nested_items(fixture_loader) -> None: + client = _make_client() + payload = fixture_loader("tweet_detail.json") + + tweets, cursor = client._parse_timeline_response( + payload, + lambda data: _deep_get(data, "data", "threaded_conversation_with_injections_v2", "instructions"), + ) + + assert [tweet.id for tweet in tweets] == ["100", "101"] + assert cursor == "conversation-cursor" + + +def test_parse_search_timeline_fixture_with_module_items(fixture_loader) -> None: + client = _make_client() + payload = fixture_loader("search_timeline.json") + + tweets, cursor = client._parse_timeline_response( + payload, + lambda data: _deep_get(data, "data", "search_by_raw_query", "search_timeline", "timeline", "instructions"), + ) + + assert [tweet.id for tweet in tweets] == ["500"] + assert cursor == "search-cursor" + assert tweets[0].media[0].type == "video" + assert tweets[0].media[0].url == "https://video-high.mp4" + + +def test_parse_list_timeline_fixture_with_visibility_wrapper(fixture_loader) -> None: + client = _make_client() + payload = fixture_loader("list_timeline.json") + + tweets, cursor = client._parse_timeline_response( + payload, + lambda data: _deep_get(data, "data", "list", "tweets_timeline", "timeline", "instructions"), + ) + + assert [tweet.id for tweet in tweets] == ["700"] + assert cursor == "list-cursor" + assert tweets[0].author.verified is True + assert tweets[0].lang == "zh" + + +def test_fetch_user_list_with_fixture(monkeypatch, fixture_loader) -> None: + client = _make_client() + payload = fixture_loader("followers_page.json") + monkeypatch.setattr(client, "_graphql_get", lambda operation_name, variables, features: payload) + + users = client._fetch_user_list( + "Followers", + "user-id", + 20, + lambda data: _deep_get(data, "data", "user", "result", "timeline", "timeline", "instructions"), + ) + + assert len(users) == 1 + assert users[0].screen_name == "follower1" + assert users[0].verified is True diff --git a/twitter_cli/auth.py b/twitter_cli/auth.py index c403398..f8d50e3 100644 --- a/twitter_cli/auth.py +++ b/twitter_cli/auth.py @@ -36,6 +36,12 @@ def load_from_env() -> Optional[Dict[str, str]]: ct0 = os.environ.get("TWITTER_CT0", "") if auth_token and ct0: return {"auth_token": auth_token, "ct0": ct0} + if auth_token or ct0: + logger.debug( + "Environment cookies incomplete: auth_token=%s ct0=%s", + bool(auth_token), + bool(ct0), + ) return None @@ -68,8 +74,15 @@ def verify_cookies(auth_token, ct0, cookie_string=None): # Reuse the shared curl_cffi session for consistent TLS fingerprint session = _get_cffi_session() + attempts = [] + + logger.debug( + "Verifying Twitter cookies with %s cookie header", + "full forwarded" if cookie_string else "minimal", + ) for url in urls: + endpoint = url.split("/")[-1] try: resp = session.get(url, headers=headers, timeout=5) if resp.status_code in (401, 403): @@ -78,28 +91,37 @@ def verify_cookies(auth_token, ct0, cookie_string=None): ) if resp.status_code == 200: data = resp.json() + attempts.append("%s=200" % endpoint) + logger.debug("Cookie verification succeeded via %s", endpoint) return {"screen_name": data.get("screen_name", "")} + attempts.append("%s=%d" % (endpoint, resp.status_code)) logger.debug("Verification endpoint %s returned HTTP %d, trying next...", url, resp.status_code) continue except RuntimeError: raise except Exception as e: + attempts.append("%s=%s" % (endpoint, type(e).__name__)) 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") + logger.info( + "Cookie verification skipped (attempts: %s), will verify on first API call", + ", ".join(attempts) if attempts else "none", + ) return {} -def _extract_cookies_from_jar(jar): - # type: (Any) -> Optional[Dict[str, str]] +def _extract_cookies_from_jar(jar, source="unknown"): + # type: (Any, str) -> Optional[Dict[str, str]] """Extract Twitter cookies from a cookie jar.""" result = {} # type: Dict[str, str] all_cookies = {} # type: Dict[str, str] + twitter_cookie_count = 0 for cookie in jar: domain = cookie.domain or "" if _is_twitter_domain(domain): + twitter_cookie_count += 1 if cookie.name == "auth_token": result["auth_token"] = cookie.value elif cookie.name == "ct0": @@ -112,6 +134,13 @@ def _extract_cookies_from_jar(jar): cookies["cookie_string"] = "; ".join("%s=%s" % (k, v) for k, v in all_cookies.items()) logger.info("Extracted %d total cookies for full browser fingerprint", len(all_cookies)) return cookies + logger.debug( + "Cookie jar %s did not contain usable Twitter auth cookies (twitter_cookies=%d, auth_token=%s, ct0=%s)", + source, + twitter_cookie_count, + "auth_token" in result, + "ct0" in result, + ) return None @@ -136,17 +165,22 @@ def _extract_in_process(): ("firefox", browser_cookie3.firefox), ("brave", browser_cookie3.brave), ] + attempts = [] for name, fn in browsers: try: jar = fn() except Exception as e: logger.debug("%s in-process extraction failed: %s", name, e) + attempts.append("%s=%s" % (name, type(e).__name__)) continue - cookies = _extract_cookies_from_jar(jar) + cookies = _extract_cookies_from_jar(jar, source="%s(in-process)" % name) if cookies: logger.info("Found cookies in %s (in-process)", name) return cookies + attempts.append("%s=no-cookies" % name) + if attempts: + logger.debug("In-process extraction attempts: %s", ", ".join(attempts)) return None @@ -168,11 +202,13 @@ browsers = [ ("firefox", browser_cookie3.firefox), ("brave", browser_cookie3.brave), ] +attempts = [] for name, fn in browsers: try: jar = fn() - except Exception: + except Exception as exc: + attempts.append(f"{name}={type(exc).__name__}") continue result = {} all_cookies = {} @@ -190,8 +226,14 @@ for name, fn in browsers: result["all_cookies"] = all_cookies print(json.dumps(result)) sys.exit(0) + attempts.append( + f"{name}=no-cookies(auth_token={'auth_token' in result},ct0={'ct0' in result})" + ) -print(json.dumps({"error": "No Twitter cookies found in any browser. Make sure you are logged into x.com."})) +print(json.dumps({ + "error": "No Twitter cookies found in any browser. Make sure you are logged into x.com.", + "attempts": attempts, +})) sys.exit(1) ''' @@ -221,6 +263,9 @@ sys.exit(1) data = json.loads(output) if "error" in data: + attempts = data.get("attempts") or [] + if attempts: + logger.debug("Subprocess extraction attempts: %s", ", ".join(str(item) for item in attempts)) return None logger.info("Found cookies in %s (subprocess)", data.get("browser", "unknown")) @@ -232,7 +277,17 @@ sys.exit(1) cookies["cookie_string"] = cookie_str logger.info("Extracted %d total cookies for full browser fingerprint", len(all_cookies)) return cookies - except (subprocess.TimeoutExpired, json.JSONDecodeError, KeyError, FileNotFoundError): + except subprocess.TimeoutExpired: + logger.debug("Cookie extraction subprocess timed out") + return None + except json.JSONDecodeError as exc: + logger.debug("Cookie extraction subprocess returned invalid JSON: %s", exc) + return None + except KeyError as exc: + logger.debug("Cookie extraction subprocess returned incomplete payload: %s", exc) + return None + except FileNotFoundError as exc: + logger.debug("Cookie extraction subprocess launcher missing: %s", exc) return None @@ -250,11 +305,14 @@ def extract_from_browser() -> Optional[Dict[str, str]]: # 2. Subprocess fallback (handles SQLite lock, but fails on macOS Keychain) logger.debug("In-process extraction failed, trying subprocess fallback") - return _extract_via_subprocess() + cookies = _extract_via_subprocess() + if not cookies: + logger.warning("Twitter cookie extraction failed in both in-process and subprocess modes") + return cookies def get_cookies() -> Dict[str, str]: - """Get Twitter cookies. Priority: env vars -> cache file -> browser extraction. + """Get Twitter cookies. Priority: env vars -> browser extraction. Raises RuntimeError if no cookies found. """ @@ -265,17 +323,10 @@ def get_cookies() -> Dict[str, str]: if cookies: logger.info("Loaded cookies from environment variables") - # 2. Try cached cookies (file cache with TTL) - if not cookies: - cookies = _load_cookie_cache() - if cookies: - logger.info("Loaded cookies from cache") - - # 3. Try browser extraction (auto-detect) + # 2. Try browser extraction (auto-detect) if not cookies: + logger.debug("Attempting browser cookie extraction") cookies = extract_from_browser() - if cookies: - _save_cookie_cache(cookies) if not cookies: raise RuntimeError( @@ -288,66 +339,12 @@ def get_cookies() -> Dict[str, str]: try: verify_cookies(cookies["auth_token"], cookies["ct0"], cookies.get("cookie_string")) except RuntimeError: - # Auth failure — invalidate cache and re-extract from browser - logger.info("Cookie verification failed, invalidating cache and re-extracting") - invalidate_cookie_cache() + # Auth failure — re-extract from browser and retry verification + logger.info("Cookie verification failed, re-extracting from browser") fresh_cookies = extract_from_browser() if fresh_cookies: - _save_cookie_cache(fresh_cookies) # Verify fresh cookies — if this also fails, let it raise verify_cookies(fresh_cookies["auth_token"], fresh_cookies["ct0"], fresh_cookies.get("cookie_string")) return fresh_cookies raise return cookies - - -# ── Cookie file cache ─────────────────────────────────────────────────── - -_CACHE_DIR = os.path.join(os.path.expanduser("~"), ".cache", "twitter-cli") -_CACHE_FILE = os.path.join(_CACHE_DIR, "cookies.json") -_CACHE_TTL_SECONDS = 24 * 3600 # 24 hours - - -def _load_cookie_cache(): - # type: () -> Optional[Dict[str, str]] - """Load cookies from file cache if within TTL.""" - try: - if not os.path.exists(_CACHE_FILE): - return None - import time as _time - mtime = os.path.getmtime(_CACHE_FILE) - if _time.time() - mtime > _CACHE_TTL_SECONDS: - logger.debug("Cookie cache expired (>%ds)", _CACHE_TTL_SECONDS) - return None - with open(_CACHE_FILE, "r", encoding="utf-8") as f: - data = json.load(f) - if isinstance(data, dict) and "auth_token" in data and "ct0" in data: - return data - except Exception as exc: - logger.debug("Failed to load cookie cache: %s", exc) - return None - - -def _save_cookie_cache(cookies): - # type: (Dict[str, str]) -> None - """Save cookies to file cache.""" - try: - os.makedirs(_CACHE_DIR, exist_ok=True) - with open(_CACHE_FILE, "w", encoding="utf-8") as f: - json.dump(cookies, f, ensure_ascii=False) - # Restrict permissions — cookies are sensitive - os.chmod(_CACHE_FILE, 0o600) - logger.info("Saved cookies to cache (%s)", _CACHE_FILE) - except Exception as exc: - logger.debug("Failed to save cookie cache: %s", exc) - - -def invalidate_cookie_cache(): - # type: () -> None - """Delete the cookie cache file.""" - try: - if os.path.exists(_CACHE_FILE): - os.remove(_CACHE_FILE) - logger.info("Cookie cache invalidated") - except Exception as exc: - logger.debug("Failed to invalidate cookie cache: %s", exc) diff --git a/twitter_cli/cli.py b/twitter_cli/cli.py index 532d358..b8b4614 100644 --- a/twitter_cli/cli.py +++ b/twitter_cli/cli.py @@ -3,7 +3,7 @@ Read commands: twitter feed # home timeline (For You) twitter feed -t following # following feed - twitter favorites # bookmarks + twitter bookmarks # bookmarks twitter search "query" # search tweets twitter user elonmusk # user profile twitter user-posts elonmusk # user tweets @@ -17,19 +17,20 @@ Write commands: twitter post "text" # post a tweet twitter delete # delete a tweet twitter like/unlike # like/unlike - twitter favorite/unfavorite # bookmark/unbookmark + twitter bookmark/unbookmark # bookmark/unbookmark twitter retweet/unretweet # retweet/unretweet """ from __future__ import annotations +import json import logging +import re import sys import time +import urllib.parse from pathlib import Path -import json - import click from rich.console import Console @@ -50,6 +51,7 @@ from .serialization import tweets_from_json, tweets_to_json, user_profile_to_dic console = Console(stderr=True) FEED_TYPES = ["for-you", "following"] +SEARCH_PRODUCTS = ["Top", "Latest", "Photos", "Videos"] def _setup_logging(verbose): @@ -90,6 +92,20 @@ def _get_client(config=None): ) +def _exit_with_error(exc): + # type: (RuntimeError) -> None + console.print("[red]❌ %s[/red]" % exc) + sys.exit(1) + + +def _run_guarded(action): + # type: (Callable[[], Any]) -> Any + try: + return action() + except RuntimeError as exc: + _exit_with_error(exc) + + def _resolve_fetch_count(max_count, configured): # type: (Optional[int], int) -> int """Resolve fetch count with bounds checks.""" @@ -100,6 +116,35 @@ def _resolve_fetch_count(max_count, configured): return max(configured, 1) +def _resolve_configured_count(config, max_count): + # type: (dict, Optional[int]) -> int + return _resolve_fetch_count(max_count, config.get("fetch", {}).get("count", 50)) + + +def _normalize_tweet_id(value): + # type: (str) -> str + """Extract a numeric tweet ID from raw input or a full X/Twitter URL.""" + raw = value.strip() + if not raw: + raise RuntimeError("Tweet ID or URL is required") + + parsed = urllib.parse.urlparse(raw) + candidate = raw + if parsed.scheme and parsed.netloc: + path = parsed.path.rstrip("/") + match = re.search(r"/status/(\d+)$", path) + if not match: + raise RuntimeError("Invalid tweet URL: %s" % value) + candidate = match.group(1) + else: + candidate = raw.rstrip("/").split("/")[-1] + candidate = candidate.split("?", 1)[0].split("#", 1)[0] + + if not candidate.isdigit(): + raise RuntimeError("Invalid tweet ID: %s" % value) + return candidate + + def _apply_filter(tweets, do_filter, config): # type: (List[Tweet], bool, dict) -> List[Tweet] """Optionally apply tweet filtering.""" @@ -128,15 +173,14 @@ def _fetch_and_display(fetch_fn, label, emoji, max_count, as_json, output_file, if config is None: config = load_config() try: - fetch_count = _resolve_fetch_count(max_count, config.get("fetch", {}).get("count", 50)) + fetch_count = _resolve_configured_count(config, max_count) console.print("%s Fetching %s (%d tweets)...\n" % (emoji, label, fetch_count)) start = time.time() tweets = fetch_fn(fetch_count) elapsed = time.time() - start console.print("✅ Fetched %d %s in %.1fs\n" % (len(tweets), label, elapsed)) except RuntimeError as exc: - console.print("[red]❌ %s[/red]" % exc) - sys.exit(1) + _exit_with_error(exc) filtered = _apply_filter(tweets, do_filter, config) @@ -176,7 +220,7 @@ def feed(feed_type, max_count, as_json, input_file, output_file, do_filter): tweets = _load_tweets_from_json(input_file) console.print(" Loaded %d tweets" % len(tweets)) else: - fetch_count = _resolve_fetch_count(max_count, config.get("fetch", {}).get("count", 50)) + fetch_count = _resolve_configured_count(config, max_count) client = _get_client(config) label = "following feed" if feed_type == "following" else "home timeline" console.print("📡 Fetching %s (%d tweets)...\n" % (label, fetch_count)) @@ -188,8 +232,7 @@ def feed(feed_type, max_count, as_json, input_file, output_file, do_filter): elapsed = time.time() - start console.print("✅ Fetched %d tweets in %.1fs\n" % (len(tweets), elapsed)) except RuntimeError as exc: - console.print("[red]❌ %s[/red]" % exc) - sys.exit(1) + _exit_with_error(exc) filtered = _apply_filter(tweets, do_filter, config) @@ -216,11 +259,24 @@ def favorites(max_count, as_json, output_file, do_filter): # type: (Optional[int], bool, Optional[str], bool) -> None """Fetch bookmarked (favorite) tweets.""" config = load_config() - client = _get_client(config) - _fetch_and_display( - lambda count: client.fetch_bookmarks(count), - "favorites", "🔖", max_count, as_json, output_file, do_filter, config, - ) + def _run(): + client = _get_client(config) + _fetch_and_display( + lambda count: client.fetch_bookmarks(count), + "bookmarks", "🔖", max_count, as_json, output_file, do_filter, config, + ) + _run_guarded(_run) + + +@cli.command(name="bookmarks") +@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("--output", "-o", "output_file", type=str, default=None, help="Save tweets to JSON file.") +@click.option("--filter", "do_filter", is_flag=True, help="Enable score-based filtering.") +def bookmarks(max_count, as_json, output_file, do_filter): + # type: (Optional[int], bool, Optional[str], bool) -> None + """Fetch bookmarked tweets.""" + favorites.callback(max_count=max_count, as_json=as_json, output_file=output_file, do_filter=do_filter) @cli.command() @@ -236,8 +292,7 @@ def user(screen_name, as_json): console.print("👤 Fetching user @%s..." % screen_name) profile = client.fetch_user(screen_name) except RuntimeError as exc: - console.print("[red]❌ %s[/red]" % exc) - sys.exit(1) + _exit_with_error(exc) if as_json: click.echo(json.dumps(user_profile_to_dict(profile), ensure_ascii=False, indent=2)) @@ -248,7 +303,7 @@ def user(screen_name, as_json): @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("--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("--output", "-o", "output_file", type=str, default=None, help="Save tweets to JSON file.") def user_posts(screen_name, max_count, as_json, output_file): @@ -256,20 +311,15 @@ def user_posts(screen_name, max_count, as_json, output_file): """List a user's tweets. SCREEN_NAME is the @handle (without @).""" screen_name = screen_name.lstrip("@") config = load_config() - client = _get_client(config) - console.print("👤 Fetching @%s's profile..." % screen_name) - try: + def _run(): + client = _get_client(config) + console.print("👤 Fetching @%s's profile..." % screen_name) profile = client.fetch_user(screen_name) - except RuntimeError as exc: - console.print("[red]❌ %s[/red]" % exc) - sys.exit(1) - _fetch_and_display( - lambda count: client.fetch_user_tweets(profile.id, count), - "@%s tweets" % screen_name, "📝", max_count, as_json, output_file, False, config, - ) - - -SEARCH_PRODUCTS = ["Top", "Latest", "Photos", "Videos"] + _fetch_and_display( + lambda count: client.fetch_user_tweets(profile.id, count), + "@%s tweets" % screen_name, "📝", max_count, as_json, output_file, False, config, + ) + _run_guarded(_run) @cli.command() @@ -282,7 +332,7 @@ SEARCH_PRODUCTS = ["Top", "Latest", "Photos", "Videos"] default="Top", help="Search tab: Top, Latest, Photos, or Videos.", ) -@click.option("--max", "-n", "max_count", type=int, default=20, help="Max 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("--output", "-o", "output_file", type=str, default=None, help="Save tweets to JSON file.") @click.option("--filter", "do_filter", is_flag=True, help="Enable score-based filtering.") @@ -290,16 +340,18 @@ def search(query, product, max_count, as_json, output_file, do_filter): # type: (str, str, int, bool, Optional[str], bool) -> None """Search tweets by QUERY string.""" config = load_config() - client = _get_client(config) - _fetch_and_display( - lambda count: client.fetch_search(query, count, product), - "'%s' (%s)" % (query, product), "🔍", max_count, as_json, output_file, do_filter, config, - ) + def _run(): + client = _get_client(config) + _fetch_and_display( + lambda count: client.fetch_search(query, count, product), + "'%s' (%s)" % (query, product), "🔍", max_count, as_json, output_file, do_filter, config, + ) + _run_guarded(_run) @cli.command() @click.argument("screen_name") -@click.option("--max", "-n", "max_count", type=int, default=20, help="Max 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("--output", "-o", "output_file", type=str, default=None, help="Save tweets to JSON file.") @click.option("--filter", "do_filter", is_flag=True, help="Enable score-based filtering.") @@ -308,39 +360,35 @@ def likes(screen_name, max_count, as_json, output_file, do_filter): """Show tweets liked by a user. SCREEN_NAME is the @handle (without @).""" screen_name = screen_name.lstrip("@") config = load_config() - client = _get_client(config) - console.print("👤 Fetching @%s's profile..." % screen_name) - try: + def _run(): + client = _get_client(config) + console.print("👤 Fetching @%s's profile..." % screen_name) profile = client.fetch_user(screen_name) - except RuntimeError as exc: - console.print("[red]❌ %s[/red]" % exc) - sys.exit(1) - _fetch_and_display( - lambda count: client.fetch_user_likes(profile.id, count), - "@%s likes" % screen_name, "❤️", max_count, as_json, output_file, do_filter, config, - ) + _fetch_and_display( + lambda count: client.fetch_user_likes(profile.id, count), + "@%s likes" % screen_name, "❤️", max_count, as_json, output_file, do_filter, config, + ) + _run_guarded(_run) @cli.command() @click.argument("tweet_id") -@click.option("--max", "-n", "max_count", type=int, default=20, help="Max replies to fetch.") +@click.option("--max", "-n", "max_count", type=int, default=None, help="Max replies to fetch.") @click.option("--json", "as_json", is_flag=True, help="Output as JSON.") def tweet(tweet_id, max_count, as_json): # type: (str, int, bool) -> None """View a tweet and its replies. TWEET_ID is the numeric tweet ID or full URL.""" - # Extract tweet ID from URL if given - tweet_id = tweet_id.strip().rstrip("/").split("/")[-1] + tweet_id = _normalize_tweet_id(tweet_id) config = load_config() try: client = _get_client(config) console.print("🐦 Fetching tweet %s...\n" % tweet_id) start = time.time() - tweets = client.fetch_tweet_detail(tweet_id, max_count) + tweets = client.fetch_tweet_detail(tweet_id, _resolve_configured_count(config, max_count)) elapsed = time.time() - start console.print("✅ Fetched %d tweets in %.1fs\n" % (len(tweets), elapsed)) except RuntimeError as exc: - console.print("[red]❌ %s[/red]" % exc) - sys.exit(1) + _exit_with_error(exc) if as_json: click.echo(tweets_to_json(tweets)) @@ -356,23 +404,25 @@ def tweet(tweet_id, max_count, as_json): @cli.command(name="list") @click.argument("list_id") -@click.option("--max", "-n", "max_count", type=int, default=20, help="Max tweets to fetch.") +@click.option("--max", "-n", "max_count", type=int, default=None, help="Max tweets to fetch.") @click.option("--json", "as_json", is_flag=True, help="Output as JSON.") @click.option("--filter", "do_filter", is_flag=True, help="Enable score-based filtering.") def list_timeline(list_id, max_count, as_json, do_filter): # type: (str, int, bool, bool) -> None """Fetch tweets from a Twitter List. LIST_ID is the numeric list ID.""" config = load_config() - client = _get_client(config) - _fetch_and_display( - lambda count: client.fetch_list_timeline(list_id, count), - "list %s" % list_id, "📋", max_count, as_json, None, do_filter, config, - ) + def _run(): + client = _get_client(config) + _fetch_and_display( + lambda count: client.fetch_list_timeline(list_id, count), + "list %s" % list_id, "📋", max_count, as_json, None, do_filter, config, + ) + _run_guarded(_run) @cli.command() @click.argument("screen_name") -@click.option("--max", "-n", "max_count", type=int, default=20, help="Max users to fetch.") +@click.option("--max", "-n", "max_count", type=int, default=None, help="Max users to fetch.") @click.option("--json", "as_json", is_flag=True, help="Output as JSON.") def followers(screen_name, max_count, as_json): # type: (str, int, bool) -> None @@ -383,14 +433,14 @@ def followers(screen_name, max_count, as_json): client = _get_client(config) console.print("👤 Fetching @%s's profile..." % screen_name) profile = client.fetch_user(screen_name) - console.print("👥 Fetching followers (%d)...\n" % max_count) + fetch_count = _resolve_configured_count(config, max_count) + console.print("👥 Fetching followers (%d)...\n" % fetch_count) start = time.time() - users = client.fetch_followers(profile.id, max_count) + users = client.fetch_followers(profile.id, fetch_count) elapsed = time.time() - start console.print("✅ Fetched %d followers in %.1fs\n" % (len(users), elapsed)) except RuntimeError as exc: - console.print("[red]❌ %s[/red]" % exc) - sys.exit(1) + _exit_with_error(exc) if as_json: click.echo(users_to_json(users)) @@ -402,7 +452,7 @@ def followers(screen_name, max_count, as_json): @cli.command() @click.argument("screen_name") -@click.option("--max", "-n", "max_count", type=int, default=20, help="Max users to fetch.") +@click.option("--max", "-n", "max_count", type=int, default=None, help="Max users to fetch.") @click.option("--json", "as_json", is_flag=True, help="Output as JSON.") def following(screen_name, max_count, as_json): # type: (str, int, bool) -> None @@ -413,14 +463,14 @@ def following(screen_name, max_count, as_json): client = _get_client(config) console.print("👤 Fetching @%s's profile..." % screen_name) profile = client.fetch_user(screen_name) - console.print("👥 Fetching following (%d)...\n" % max_count) + fetch_count = _resolve_configured_count(config, max_count) + console.print("👥 Fetching following (%d)...\n" % fetch_count) start = time.time() - users = client.fetch_following(profile.id, max_count) + users = client.fetch_following(profile.id, fetch_count) elapsed = time.time() - start console.print("✅ Fetched %d following in %.1fs\n" % (len(users), elapsed)) except RuntimeError as exc: - console.print("[red]❌ %s[/red]" % exc) - sys.exit(1) + _exit_with_error(exc) if as_json: click.echo(users_to_json(users)) @@ -442,8 +492,7 @@ def _write_action(emoji, action_desc, client_method, tweet_id): getattr(client, client_method)(tweet_id) console.print("[green]✅ Done.[/green]") except RuntimeError as exc: - console.print("[red]❌ %s[/red]" % exc) - sys.exit(1) + _exit_with_error(exc) @cli.command() @@ -461,8 +510,7 @@ def post(text, reply_to): console.print("[green]✅ Tweet posted![/green]") console.print("🔗 https://x.com/i/status/%s" % tweet_id) except RuntimeError as exc: - console.print("[red]❌ %s[/red]" % exc) - sys.exit(1) + _exit_with_error(exc) @cli.command(name="delete") @@ -514,6 +562,14 @@ def favorite(tweet_id): _write_action("🔖", "Bookmarking tweet", "bookmark_tweet", tweet_id) +@cli.command() +@click.argument("tweet_id") +def bookmark(tweet_id): + # type: (str,) -> None + """Bookmark a tweet. TWEET_ID is the numeric tweet ID.""" + _write_action("🔖", "Bookmarking tweet", "bookmark_tweet", tweet_id) + + @cli.command() @click.argument("tweet_id") def unfavorite(tweet_id): @@ -522,5 +578,13 @@ def unfavorite(tweet_id): _write_action("🔖", "Removing bookmark", "unbookmark_tweet", tweet_id) +@cli.command() +@click.argument("tweet_id") +def unbookmark(tweet_id): + # type: (str,) -> None + """Remove a tweet from bookmarks. TWEET_ID is the numeric tweet ID.""" + _write_action("🔖", "Removing bookmark", "unbookmark_tweet", tweet_id) + + if __name__ == "__main__": cli()