Merge pull request #23 from jackwener/fix/improvements-v0.8.2
fix: P0 Windows Edge path, time localization, show --output, cleanup tech debt
This commit is contained in:
@@ -13,16 +13,22 @@ import pytest
|
||||
|
||||
|
||||
from twitter_cli.client import (
|
||||
FEATURES,
|
||||
_best_chrome_target,
|
||||
TwitterClient,
|
||||
)
|
||||
from twitter_cli.exceptions import TwitterAPIError
|
||||
from twitter_cli.graphql import (
|
||||
FEATURES,
|
||||
_build_graphql_url,
|
||||
_update_features_from_html,
|
||||
)
|
||||
from twitter_cli.parser import (
|
||||
_deep_get,
|
||||
_extract_cursor,
|
||||
_extract_media,
|
||||
_parse_int,
|
||||
_update_features_from_html,
|
||||
TwitterAPIError,
|
||||
TwitterClient,
|
||||
parse_tweet_result,
|
||||
parse_user_result,
|
||||
)
|
||||
|
||||
|
||||
@@ -339,8 +345,8 @@ class TestPaginationBehavior:
|
||||
return [MagicMock(id="tweet-1")], None
|
||||
|
||||
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)
|
||||
|
||||
assert [tweet.id for tweet in tweets] == ["tweet-1"]
|
||||
@@ -357,8 +363,8 @@ class TestPaginationBehavior:
|
||||
return {"page": len(calls)}
|
||||
|
||||
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)
|
||||
|
||||
assert tweets == []
|
||||
@@ -401,8 +407,8 @@ class TestPaginationBehavior:
|
||||
]
|
||||
|
||||
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)
|
||||
|
||||
assert [user.screen_name for user in users] == ["alice"]
|
||||
@@ -453,7 +459,7 @@ class TestParseTweetResult:
|
||||
client._ct_init_attempted = True
|
||||
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.id == "1234567890"
|
||||
assert tweet.text == "Hello world! This is a test tweet."
|
||||
@@ -475,7 +481,7 @@ class TestParseTweetResult:
|
||||
client._client_transaction = None
|
||||
|
||||
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._gen_ct_headers", return_value={})
|
||||
@@ -491,7 +497,7 @@ class TestParseTweetResult:
|
||||
"__typename": "TweetWithVisibilityResults",
|
||||
"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.id == "1234567890"
|
||||
|
||||
@@ -505,7 +511,7 @@ class TestParseTweetResult:
|
||||
client._ct_init_attempted = True
|
||||
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 ──────────────────────────────────────────────────────
|
||||
@@ -523,7 +529,7 @@ class TestTwitterAPIError:
|
||||
|
||||
class TestParseUserResult:
|
||||
def test_coerces_count_fields_to_int(self):
|
||||
user = TwitterClient._parse_user_result(
|
||||
user = parse_user_result(
|
||||
{
|
||||
"rest_id": "user-1",
|
||||
"legacy": {
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
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:
|
||||
@@ -15,10 +18,9 @@ def _make_client() -> TwitterClient:
|
||||
|
||||
|
||||
def test_parse_home_timeline_fixture(fixture_loader) -> None:
|
||||
client = _make_client()
|
||||
payload = fixture_loader("home_timeline.json")
|
||||
|
||||
tweets, cursor = client._parse_timeline_response(
|
||||
tweets, cursor = parse_timeline_response(
|
||||
payload,
|
||||
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:
|
||||
client = _make_client()
|
||||
payload = fixture_loader("tweet_detail.json")
|
||||
|
||||
tweets, cursor = client._parse_timeline_response(
|
||||
tweets, cursor = parse_timeline_response(
|
||||
payload,
|
||||
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:
|
||||
client = _make_client()
|
||||
payload = fixture_loader("search_timeline.json")
|
||||
|
||||
tweets, cursor = client._parse_timeline_response(
|
||||
tweets, cursor = parse_timeline_response(
|
||||
payload,
|
||||
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:
|
||||
client = _make_client()
|
||||
payload = fixture_loader("list_timeline.json")
|
||||
|
||||
tweets, cursor = client._parse_timeline_response(
|
||||
tweets, cursor = parse_timeline_response(
|
||||
payload,
|
||||
lambda data: _deep_get(data, "data", "list", "tweets_timeline", "timeline", "instructions"),
|
||||
)
|
||||
|
||||
@@ -353,7 +353,13 @@ def iter_cookie_files(browser_name):
|
||||
if sys.platform == "darwin":
|
||||
root = os.path.join(os.path.expanduser("~"), "Library", "Application Support", base_dir)
|
||||
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:
|
||||
root = os.path.join(os.path.expanduser("~"), ".config", base_dir)
|
||||
if not os.path.isdir(root):
|
||||
|
||||
@@ -737,10 +737,11 @@ def _print_show_hint():
|
||||
@click.argument("index", type=click.IntRange(1))
|
||||
@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("--output", "-o", "output_file", type=str, default=None, help="Save tweet detail as JSON to file.")
|
||||
@structured_output_options
|
||||
@click.pass_context
|
||||
def show(ctx, index, max_count, full_text, as_json, as_yaml):
|
||||
# type: (Any, int, Optional[int], bool, bool, bool) -> None
|
||||
def show(ctx, index, max_count, full_text, output_file, as_json, as_yaml):
|
||||
# type: (Any, int, Optional[int], bool, Optional[str], bool, bool) -> None
|
||||
"""View tweet #INDEX from the last feed/search results."""
|
||||
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:
|
||||
_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)
|
||||
|
||||
|
||||
|
||||
@@ -715,7 +715,7 @@ class TwitterClient:
|
||||
variables["cursor"] = cursor
|
||||
|
||||
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:
|
||||
if tweet.id and tweet.id not in seen_ids:
|
||||
@@ -780,7 +780,7 @@ class TwitterClient:
|
||||
item = content.get("itemContent", {})
|
||||
user_results = _deep_get(item, "user_results", "result")
|
||||
if user_results:
|
||||
user = self._parse_user_result(user_results)
|
||||
user = parse_user_result(user_results)
|
||||
if user:
|
||||
new_users.append(user)
|
||||
elif entry_type == "TimelineTimelineCursor":
|
||||
@@ -1081,28 +1081,3 @@ class TwitterClient:
|
||||
except Exception as exc:
|
||||
logger.debug("Failed to generate transaction id: %s", exc)
|
||||
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
|
||||
|
||||
@@ -121,8 +121,3 @@ SEC_CH_UA_ARCH = get_sec_ch_ua_arch()
|
||||
SEC_CH_UA_BITNESS = '"64"'
|
||||
SEC_CH_UA_MODEL = '""'
|
||||
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()
|
||||
|
||||
@@ -10,6 +10,7 @@ from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
|
||||
from .models import Tweet, UserProfile
|
||||
from .timeutil import format_local_time, format_relative_time
|
||||
|
||||
|
||||
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)
|
||||
|
||||
# Stats
|
||||
rel_time = format_relative_time(tweet.created_at)
|
||||
stats = (
|
||||
"❤️ %s 🔄 %s\n💬 %s 👁️ %s"
|
||||
"❤️ %s 🔄 %s\n💬 %s 👁️ %s\n🕐 %s"
|
||||
% (
|
||||
format_number(tweet.metrics.likes),
|
||||
format_number(tweet.metrics.retweets),
|
||||
format_number(tweet.metrics.replies),
|
||||
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),
|
||||
)
|
||||
)
|
||||
local_time = format_local_time(tweet.created_at)
|
||||
rel_time = format_relative_time(tweet.created_at)
|
||||
body_parts.append(
|
||||
"🕐 %s · https://x.com/%s/status/%s"
|
||||
% (tweet.created_at, tweet.author.screen_name, tweet.id)
|
||||
"🕐 %s (%s) · https://x.com/%s/status/%s"
|
||||
% (local_time, rel_time, tweet.author.screen_name, tweet.id)
|
||||
)
|
||||
|
||||
console.print(Panel(
|
||||
|
||||
@@ -6,6 +6,7 @@ import json
|
||||
from typing import Any, Dict, Iterable, List, Optional
|
||||
|
||||
from .models import Author, Metrics, Tweet, TweetMedia, UserProfile
|
||||
from .timeutil import format_local_time
|
||||
|
||||
|
||||
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,
|
||||
},
|
||||
"createdAt": tweet.created_at,
|
||||
"createdAtLocal": format_local_time(tweet.created_at),
|
||||
"media": [
|
||||
{
|
||||
"type": media.type,
|
||||
|
||||
71
twitter_cli/timeutil.py
Normal file
71
twitter_cli/timeutil.py
Normal 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
|
||||
Reference in New Issue
Block a user