fix: P0 Windows Edge path, add time localization, show --output, cleanup tech debt

- Fix auth.py subprocess script Windows Edge cookie path inconsistency
- Add timeutil.py for UTC→local time and relative time conversion
- Integrate time localization into formatter.py and serialization.py
- Add --output/-o option to show command for saving tweet detail as JSON
- Remove constants.py legacy aliases (USER_AGENT, SEC_CH_UA)
- Remove client.py backward-compat delegation methods and re-exports
- Update test imports to use parser module directly
This commit is contained in:
jackwener
2026-03-14 13:26:36 +08:00
parent 80e5a62890
commit ec4589c2d1
9 changed files with 128 additions and 63 deletions

View File

@@ -13,16 +13,22 @@ import pytest
from twitter_cli.client import ( from twitter_cli.client import (
FEATURES,
_best_chrome_target, _best_chrome_target,
TwitterClient,
)
from twitter_cli.exceptions import TwitterAPIError
from twitter_cli.graphql import (
FEATURES,
_build_graphql_url, _build_graphql_url,
_update_features_from_html,
)
from twitter_cli.parser import (
_deep_get, _deep_get,
_extract_cursor, _extract_cursor,
_extract_media, _extract_media,
_parse_int, _parse_int,
_update_features_from_html, parse_tweet_result,
TwitterAPIError, parse_user_result,
TwitterClient,
) )
@@ -339,8 +345,8 @@ class TestPaginationBehavior:
return [MagicMock(id="tweet-1")], None return [MagicMock(id="tweet-1")], None
client._graphql_get = _graphql_get client._graphql_get = _graphql_get
client._parse_timeline_response = _parse_timeline_response
with patch('twitter_cli.client.parse_timeline_response', side_effect=_parse_timeline_response):
tweets = client._fetch_timeline("HomeTimeline", 1, lambda data: data) tweets = client._fetch_timeline("HomeTimeline", 1, lambda data: data)
assert [tweet.id for tweet in tweets] == ["tweet-1"] assert [tweet.id for tweet in tweets] == ["tweet-1"]
@@ -357,8 +363,8 @@ class TestPaginationBehavior:
return {"page": len(calls)} return {"page": len(calls)}
client._graphql_get = _graphql_get client._graphql_get = _graphql_get
client._parse_timeline_response = lambda data, get_instructions: ([], "cursor-same")
with patch('twitter_cli.client.parse_timeline_response', return_value=([], "cursor-same")):
tweets = client._fetch_timeline("HomeTimeline", 1, lambda data: data) tweets = client._fetch_timeline("HomeTimeline", 1, lambda data: data)
assert tweets == [] assert tweets == []
@@ -401,8 +407,8 @@ class TestPaginationBehavior:
] ]
client._graphql_get = _graphql_get client._graphql_get = _graphql_get
client._parse_user_result = _parse_user_result
with patch('twitter_cli.client.parse_user_result', side_effect=_parse_user_result):
users = client._fetch_user_list("Followers", "1", 1, _get_instructions) users = client._fetch_user_list("Followers", "1", 1, _get_instructions)
assert [user.screen_name for user in users] == ["alice"] assert [user.screen_name for user in users] == ["alice"]
@@ -453,7 +459,7 @@ class TestParseTweetResult:
client._ct_init_attempted = True client._ct_init_attempted = True
client._client_transaction = None client._client_transaction = None
tweet = client._parse_tweet_result(copy.deepcopy(self.SAMPLE_TWEET_RESULT)) tweet = parse_tweet_result(copy.deepcopy(self.SAMPLE_TWEET_RESULT))
assert tweet is not None assert tweet is not None
assert tweet.id == "1234567890" assert tweet.id == "1234567890"
assert tweet.text == "Hello world! This is a test tweet." assert tweet.text == "Hello world! This is a test tweet."
@@ -475,7 +481,7 @@ class TestParseTweetResult:
client._client_transaction = None client._client_transaction = None
result = {"__typename": "TweetTombstone"} result = {"__typename": "TweetTombstone"}
assert client._parse_tweet_result(result) is None assert parse_tweet_result(result) is None
@patch("twitter_cli.client._get_cffi_session") @patch("twitter_cli.client._get_cffi_session")
@patch("twitter_cli.client._gen_ct_headers", return_value={}) @patch("twitter_cli.client._gen_ct_headers", return_value={})
@@ -491,7 +497,7 @@ class TestParseTweetResult:
"__typename": "TweetWithVisibilityResults", "__typename": "TweetWithVisibilityResults",
"tweet": copy.deepcopy(self.SAMPLE_TWEET_RESULT), "tweet": copy.deepcopy(self.SAMPLE_TWEET_RESULT),
} }
tweet = client._parse_tweet_result(wrapped) tweet = parse_tweet_result(wrapped)
assert tweet is not None assert tweet is not None
assert tweet.id == "1234567890" assert tweet.id == "1234567890"
@@ -505,7 +511,7 @@ class TestParseTweetResult:
client._ct_init_attempted = True client._ct_init_attempted = True
client._client_transaction = None client._client_transaction = None
assert client._parse_tweet_result(self.SAMPLE_TWEET_RESULT, depth=3) is None assert parse_tweet_result(self.SAMPLE_TWEET_RESULT, depth=3) is None
# ── TwitterAPIError ────────────────────────────────────────────────────── # ── TwitterAPIError ──────────────────────────────────────────────────────
@@ -523,7 +529,7 @@ class TestTwitterAPIError:
class TestParseUserResult: class TestParseUserResult:
def test_coerces_count_fields_to_int(self): def test_coerces_count_fields_to_int(self):
user = TwitterClient._parse_user_result( user = parse_user_result(
{ {
"rest_id": "user-1", "rest_id": "user-1",
"legacy": { "legacy": {

View File

@@ -1,6 +1,9 @@
from __future__ import annotations from __future__ import annotations
from twitter_cli.client import TwitterClient, _deep_get from unittest.mock import patch
from twitter_cli.client import TwitterClient
from twitter_cli.parser import _deep_get, parse_timeline_response
def _make_client() -> TwitterClient: def _make_client() -> TwitterClient:
@@ -15,10 +18,9 @@ def _make_client() -> TwitterClient:
def test_parse_home_timeline_fixture(fixture_loader) -> None: def test_parse_home_timeline_fixture(fixture_loader) -> None:
client = _make_client()
payload = fixture_loader("home_timeline.json") payload = fixture_loader("home_timeline.json")
tweets, cursor = client._parse_timeline_response( tweets, cursor = parse_timeline_response(
payload, payload,
lambda data: _deep_get(data, "data", "home", "home_timeline_urt", "instructions"), lambda data: _deep_get(data, "data", "home", "home_timeline_urt", "instructions"),
) )
@@ -34,10 +36,9 @@ def test_parse_home_timeline_fixture(fixture_loader) -> None:
def test_parse_tweet_detail_fixture_with_nested_items(fixture_loader) -> None: def test_parse_tweet_detail_fixture_with_nested_items(fixture_loader) -> None:
client = _make_client()
payload = fixture_loader("tweet_detail.json") payload = fixture_loader("tweet_detail.json")
tweets, cursor = client._parse_timeline_response( tweets, cursor = parse_timeline_response(
payload, payload,
lambda data: _deep_get(data, "data", "threaded_conversation_with_injections_v2", "instructions"), lambda data: _deep_get(data, "data", "threaded_conversation_with_injections_v2", "instructions"),
) )
@@ -47,10 +48,9 @@ def test_parse_tweet_detail_fixture_with_nested_items(fixture_loader) -> None:
def test_parse_search_timeline_fixture_with_module_items(fixture_loader) -> None: def test_parse_search_timeline_fixture_with_module_items(fixture_loader) -> None:
client = _make_client()
payload = fixture_loader("search_timeline.json") payload = fixture_loader("search_timeline.json")
tweets, cursor = client._parse_timeline_response( tweets, cursor = parse_timeline_response(
payload, payload,
lambda data: _deep_get(data, "data", "search_by_raw_query", "search_timeline", "timeline", "instructions"), lambda data: _deep_get(data, "data", "search_by_raw_query", "search_timeline", "timeline", "instructions"),
) )
@@ -62,10 +62,9 @@ def test_parse_search_timeline_fixture_with_module_items(fixture_loader) -> None
def test_parse_list_timeline_fixture_with_visibility_wrapper(fixture_loader) -> None: def test_parse_list_timeline_fixture_with_visibility_wrapper(fixture_loader) -> None:
client = _make_client()
payload = fixture_loader("list_timeline.json") payload = fixture_loader("list_timeline.json")
tweets, cursor = client._parse_timeline_response( tweets, cursor = parse_timeline_response(
payload, payload,
lambda data: _deep_get(data, "data", "list", "tweets_timeline", "timeline", "instructions"), lambda data: _deep_get(data, "data", "list", "tweets_timeline", "timeline", "instructions"),
) )

View File

@@ -353,7 +353,13 @@ def iter_cookie_files(browser_name):
if sys.platform == "darwin": if sys.platform == "darwin":
root = os.path.join(os.path.expanduser("~"), "Library", "Application Support", base_dir) root = os.path.join(os.path.expanduser("~"), "Library", "Application Support", base_dir)
elif sys.platform == "win32": elif sys.platform == "win32":
root = os.path.join(os.environ.get("LOCALAPPDATA", ""), base_dir, "User Data") if browser_name == "edge":
root = os.path.join(os.environ.get("LOCALAPPDATA", ""), "Microsoft", "Edge", "User Data")
else:
root = os.path.join(os.environ.get("LOCALAPPDATA", ""), base_dir)
else:
if browser_name == "edge":
root = os.path.join(os.path.expanduser("~"), ".config", "microsoft-edge")
else: else:
root = os.path.join(os.path.expanduser("~"), ".config", base_dir) root = os.path.join(os.path.expanduser("~"), ".config", base_dir)
if not os.path.isdir(root): if not os.path.isdir(root):

View File

@@ -737,10 +737,11 @@ def _print_show_hint():
@click.argument("index", type=click.IntRange(1)) @click.argument("index", type=click.IntRange(1))
@click.option("--max", "-n", "max_count", type=int, default=None, help="Max replies to fetch.") @click.option("--max", "-n", "max_count", type=int, default=None, help="Max replies to fetch.")
@click.option("--full-text", is_flag=True, help="Show full reply text in table output.") @click.option("--full-text", is_flag=True, help="Show full reply text in table output.")
@click.option("--output", "-o", "output_file", type=str, default=None, help="Save tweet detail as JSON to file.")
@structured_output_options @structured_output_options
@click.pass_context @click.pass_context
def show(ctx, index, max_count, full_text, as_json, as_yaml): def show(ctx, index, max_count, full_text, output_file, as_json, as_yaml):
# type: (Any, int, Optional[int], bool, bool, bool) -> None # type: (Any, int, Optional[int], bool, Optional[str], bool, bool) -> None
"""View tweet #INDEX from the last feed/search results.""" """View tweet #INDEX from the last feed/search results."""
compact = ctx.obj.get("compact", False) compact = ctx.obj.get("compact", False)
@@ -769,6 +770,11 @@ def show(ctx, index, max_count, full_text, as_json, as_yaml):
except RuntimeError as exc: except RuntimeError as exc:
_exit_with_error(exc) _exit_with_error(exc)
if output_file:
Path(output_file).write_text(tweets_to_json(tweets), encoding="utf-8")
if rich_output:
console.print("💾 Saved to %s\n" % output_file)
_emit_tweet_detail(tweets, compact=compact, as_json=as_json, as_yaml=as_yaml, full_text=full_text) _emit_tweet_detail(tweets, compact=compact, as_json=as_json, as_yaml=as_yaml, full_text=full_text)

View File

@@ -715,7 +715,7 @@ class TwitterClient:
variables["cursor"] = cursor variables["cursor"] = cursor
data = self._graphql_get(operation_name, variables, FEATURES, field_toggles=field_toggles) data = self._graphql_get(operation_name, variables, FEATURES, field_toggles=field_toggles)
new_tweets, next_cursor = self._parse_timeline_response(data, get_instructions) new_tweets, next_cursor = parse_timeline_response(data, get_instructions)
for tweet in new_tweets: for tweet in new_tweets:
if tweet.id and tweet.id not in seen_ids: if tweet.id and tweet.id not in seen_ids:
@@ -780,7 +780,7 @@ class TwitterClient:
item = content.get("itemContent", {}) item = content.get("itemContent", {})
user_results = _deep_get(item, "user_results", "result") user_results = _deep_get(item, "user_results", "result")
if user_results: if user_results:
user = self._parse_user_result(user_results) user = parse_user_result(user_results)
if user: if user:
new_users.append(user) new_users.append(user)
elif entry_type == "TimelineTimelineCursor": elif entry_type == "TimelineTimelineCursor":
@@ -1081,28 +1081,3 @@ class TwitterClient:
except Exception as exc: except Exception as exc:
logger.debug("Failed to generate transaction id: %s", exc) logger.debug("Failed to generate transaction id: %s", exc)
return headers return headers
# ── Backward-compatible delegation to parser module ──────────────
@staticmethod
def _parse_user_result(user_data):
# type: (Dict[str, Any]) -> Optional[UserProfile]
"""Parse a user result object into UserProfile."""
return parse_user_result(user_data)
def _parse_tweet_result(self, result, depth=0):
# type: (Dict[str, Any], int) -> Optional[Tweet]
"""Parse a single TweetResult into a Tweet dataclass."""
return parse_tweet_result(result, depth)
def _parse_timeline_response(self, data, get_instructions):
# type: (Any, Callable[[Any], Any]) -> Tuple[List[Tweet], Optional[str]]
"""Parse timeline GraphQL response into tweets and next cursor."""
return parse_timeline_response(data, get_instructions)
# ── Backward compatibility re-exports ────────────────────────────────────
# These keep existing test imports working without modification.
from .graphql import FALLBACK_QUERY_IDS # noqa: E402, F401
from .parser import _extract_cursor, _extract_media # noqa: E402, F401

View File

@@ -121,8 +121,3 @@ SEC_CH_UA_ARCH = get_sec_ch_ua_arch()
SEC_CH_UA_BITNESS = '"64"' SEC_CH_UA_BITNESS = '"64"'
SEC_CH_UA_MODEL = '""' SEC_CH_UA_MODEL = '""'
SEC_CH_UA_PLATFORM_VERSION = get_sec_ch_ua_platform_version() SEC_CH_UA_PLATFORM_VERSION = get_sec_ch_ua_platform_version()
# Legacy aliases — modules that import these get the default value.
# _build_headers() should use get_user_agent() / get_sec_ch_ua() instead.
USER_AGENT = get_user_agent()
SEC_CH_UA = get_sec_ch_ua()

View File

@@ -10,6 +10,7 @@ from rich.panel import Panel
from rich.table import Table from rich.table import Table
from .models import Tweet, UserProfile from .models import Tweet, UserProfile
from .timeutil import format_local_time, format_relative_time
def format_number(n: int) -> str: def format_number(n: int) -> str:
@@ -77,13 +78,15 @@ def print_tweet_table(
text += "\n🔗 x.com/%s/status/%s" % (tweet.author.screen_name, tweet.id) text += "\n🔗 x.com/%s/status/%s" % (tweet.author.screen_name, tweet.id)
# Stats # Stats
rel_time = format_relative_time(tweet.created_at)
stats = ( stats = (
"❤️ %s 🔄 %s\n💬 %s 👁️ %s" "❤️ %s 🔄 %s\n💬 %s 👁️ %s\n🕐 %s"
% ( % (
format_number(tweet.metrics.likes), format_number(tweet.metrics.likes),
format_number(tweet.metrics.retweets), format_number(tweet.metrics.retweets),
format_number(tweet.metrics.replies), format_number(tweet.metrics.replies),
format_number(tweet.metrics.views), format_number(tweet.metrics.views),
rel_time,
) )
) )
@@ -138,9 +141,11 @@ def print_tweet_detail(tweet: Tweet, console: Optional[Console] = None) -> None:
format_number(tweet.metrics.views), format_number(tweet.metrics.views),
) )
) )
local_time = format_local_time(tweet.created_at)
rel_time = format_relative_time(tweet.created_at)
body_parts.append( body_parts.append(
"🕐 %s · https://x.com/%s/status/%s" "🕐 %s (%s) · https://x.com/%s/status/%s"
% (tweet.created_at, tweet.author.screen_name, tweet.id) % (local_time, rel_time, tweet.author.screen_name, tweet.id)
) )
console.print(Panel( console.print(Panel(

View File

@@ -6,6 +6,7 @@ import json
from typing import Any, Dict, Iterable, List, Optional from typing import Any, Dict, Iterable, List, Optional
from .models import Author, Metrics, Tweet, TweetMedia, UserProfile from .models import Author, Metrics, Tweet, TweetMedia, UserProfile
from .timeutil import format_local_time
def tweet_to_dict(tweet: Tweet) -> Dict[str, Any]: def tweet_to_dict(tweet: Tweet) -> Dict[str, Any]:
@@ -29,6 +30,7 @@ def tweet_to_dict(tweet: Tweet) -> Dict[str, Any]:
"bookmarks": tweet.metrics.bookmarks, "bookmarks": tweet.metrics.bookmarks,
}, },
"createdAt": tweet.created_at, "createdAt": tweet.created_at,
"createdAtLocal": format_local_time(tweet.created_at),
"media": [ "media": [
{ {
"type": media.type, "type": media.type,

71
twitter_cli/timeutil.py Normal file
View File

@@ -0,0 +1,71 @@
"""Time formatting utilities for twitter-cli.
Converts Twitter API timestamps (e.g. "Sat Mar 08 12:00:00 +0000 2026")
into human-friendly local time and relative time strings.
"""
from __future__ import annotations
import logging
from datetime import datetime, timezone
from typing import Optional
logger = logging.getLogger(__name__)
# Twitter API timestamp format: "Sat Mar 08 12:00:00 +0000 2026"
_TWITTER_TIME_FORMAT = "%a %b %d %H:%M:%S %z %Y"
def _parse_twitter_time(created_at: str) -> Optional[datetime]:
"""Parse a Twitter API timestamp into a timezone-aware datetime."""
if not created_at:
return None
try:
return datetime.strptime(created_at, _TWITTER_TIME_FORMAT)
except (ValueError, TypeError):
logger.debug("Failed to parse Twitter timestamp: %s", created_at)
return None
def format_local_time(created_at: str) -> str:
"""Convert Twitter timestamp to local time string.
Returns "2026-03-14 21:08" or the original string on parse failure.
"""
dt = _parse_twitter_time(created_at)
if dt is None:
return created_at
local_dt = dt.astimezone()
return local_dt.strftime("%Y-%m-%d %H:%M")
def format_relative_time(created_at: str) -> str:
"""Convert Twitter timestamp to a relative time string.
Returns e.g. "2m ago", "3h ago", "5d ago", or the original string on failure.
"""
dt = _parse_twitter_time(created_at)
if dt is None:
return created_at
now = datetime.now(timezone.utc)
delta = now - dt
seconds = int(delta.total_seconds())
if seconds < 0:
return "just now"
if seconds < 60:
return "%ds ago" % seconds
minutes = seconds // 60
if minutes < 60:
return "%dm ago" % minutes
hours = minutes // 60
if hours < 24:
return "%dh ago" % hours
days = hours // 24
if days < 30:
return "%dd ago" % days
months = days // 30
if months < 12:
return "%dmo ago" % months
years = days // 365
return "%dy ago" % years