refactor: unify exception handling, add ISO 8601 time, dedup commands, expand tests

- Replace _error_code_for_message() string matching with error_code attribute on exception classes - Add error_code to all TwitterError subclasses (AuthenticationError, RateLimitError, etc.) - Add InvalidInputError exception class - TwitterAPIError derives error_code from HTTP status code automatically - auth.py: use AuthenticationError instead of RuntimeError - cli.py: catch (TwitterError, RuntimeError) for backward compat - Extract _fetch_and_display_users() to deduplicate followers/following commands - Add format_iso8601() to timeutil.py - Add createdAtISO field to tweet and user profile serialization - New test files: test_output.py, test_cache.py, test_timeutil.py - Expand test_filter.py (topN, score mode, custom weights, empty input) - Tests: 152 → 194 unit tests, all passing
2026-03-16 18:24:35 +08:00
parent 0b91e66998
commit e496d8f870
11 changed files with 482 additions and 73 deletions
--- a/twitter_cli/auth.py
+++ b/twitter_cli/auth.py
@@ -19,6 +19,7 @@ import sys
 from typing import Any, Dict, List, Optional, Tuple

 from .constants import BEARER_TOKEN, get_user_agent
+from .exceptions import AuthenticationError

 logger = logging.getLogger(__name__)

@@ -136,7 +137,7 @@ def verify_cookies(auth_token: str, ct0: str, cookie_string: Optional[str] = Non
        try:
            resp = session.get(url, headers=headers, timeout=5)
            if resp.status_code in (401, 403):
-                raise RuntimeError(
+                raise AuthenticationError(
                    "Cookie expired or invalid (HTTP %d). Please re-login to x.com in your browser." % resp.status_code
                )
            if resp.status_code == 200:
@@ -616,7 +617,7 @@ def get_cookies() -> Dict[str, str]:
        lines.append("Option 2: Make sure you are logged into x.com in your browser (Arc/Chrome/Edge/Firefox/Brave)")
        lines.append("")
        lines.append("Run 'twitter -v <command>' for debug diagnostics.")
-        raise RuntimeError("\n".join(lines))
+        raise AuthenticationError("\n".join(lines))

    # Verify only for explicit auth failures; transient endpoint issues are tolerated.
    try:
--- a/twitter_cli/cli.py
+++ b/twitter_cli/cli.py
@@ -44,6 +44,7 @@ from rich.console import Console
 from . import __version__
 from .auth import get_cookies
 from .cache import resolve_cached_tweet, save_tweet_cache
+from .exceptions import TwitterError
 from .client import TwitterClient
 from .config import load_config
 from .filter import filter_tweets
@@ -156,9 +157,13 @@ def _get_client(config=None, quiet=False):



-def _exit_with_error(exc):
-    # type: (RuntimeError) -> None
-    if emit_error(_error_code_for_message(str(exc)), str(exc)):
+def _error_code_from_exc(exc: Exception) -> str:
+    """Extract structured error code from an exception."""
+    return getattr(exc, "error_code", "api_error")
+
+
+def _exit_with_error(exc: Exception) -> None:
+    if emit_error(_error_code_from_exc(exc), str(exc)):
        sys.exit(1)
    console.print("[red]❌ %s[/red]" % exc)
    sys.exit(1)
@@ -168,24 +173,10 @@ def _run_guarded(action):
    # type: (Callable[[], Any]) -> Any
    try:
        return action()
-    except RuntimeError as exc:
+    except (TwitterError, RuntimeError) as exc:
        _exit_with_error(exc)


-def _error_code_for_message(message):
-    # type: (str) -> str
-    lowered = message.lower()
-    if "cookie expired" in lowered or "no twitter cookies found" in lowered or "invalid cookie" in lowered:
-        return "not_authenticated"
-    if "rate limited" in lowered or "http 429" in lowered:
-        return "rate_limited"
-    if "invalid tweet" in lowered or "required" in lowered or "--max must" in lowered:
-        return "invalid_input"
-    if "not found" in lowered:
-        return "not_found"
-    return "api_error"
-
-
 def _resolve_fetch_count(max_count, configured):
    # type: (Optional[int], int) -> int
    """Resolve fetch count with bounds checks."""
@@ -258,13 +249,13 @@ def _print_lines(lines: List[str], mode: StructuredMode) -> None:


 def _handle_structured_runtime_error(
-    exc: RuntimeError,
+    exc: Exception,
    *,
    mode: StructuredMode,
    details: Optional[Dict[str, Any]] = None,
 ) -> None:
    if _emit_mode_payload(
-        error_payload(_error_code_for_message(str(exc)), str(exc), details=details),
+        error_payload(_error_code_from_exc(exc), str(exc), details=details),
        mode,
    ):
        raise SystemExit(1) from None
@@ -285,7 +276,7 @@ def _run_write_command(
        client = _get_client(load_config())
        _print_lines(progress_lines or [], mode)
        payload = operation(client)
-    except RuntimeError as exc:
+    except (TwitterError, RuntimeError) as exc:
        _handle_structured_runtime_error(exc, mode=mode, details=error_details)
        return None

@@ -325,7 +316,7 @@ def _fetch_and_display(fetch_fn, label, emoji, max_count, as_json, as_yaml, outp
        elapsed = time.time() - start
        if rich_output:
            console.print("✅ Fetched %d %s in %.1fs\n" % (len(tweets), label, elapsed))
-    except RuntimeError as exc:
+    except (TwitterError, RuntimeError) as exc:
        _exit_with_error(exc)

    filtered = _apply_filter(tweets, do_filter, config, rich_output=rich_output)
@@ -420,7 +411,7 @@ def feed(ctx, feed_type, max_count, as_json, as_yaml, input_file, output_file, d
            elapsed = time.time() - start
            if rich_output:
                console.print("✅ Fetched %d tweets in %.1fs\n" % (len(tweets), elapsed))
-    except RuntimeError as exc:
+    except (TwitterError, RuntimeError) as exc:
        _exit_with_error(exc)

    filtered = _apply_filter(tweets, do_filter, config, rich_output=rich_output)
@@ -502,7 +493,7 @@ def user(screen_name, as_json, as_yaml):
        if rich_output:
            console.print("👤 Fetching user @%s..." % screen_name)
        profile = client.fetch_user(screen_name)
-    except RuntimeError as exc:
+    except (TwitterError, RuntimeError) as exc:
        _exit_with_error(exc)

    if not emit_structured(user_profile_to_dict(profile), as_json=as_json, as_yaml=as_yaml):
@@ -693,7 +684,7 @@ def tweet(ctx, tweet_id, max_count, full_text, as_json, as_yaml):
        elapsed = time.time() - start
        if rich_output:
            console.print("✅ Fetched %d tweets in %.1fs\n" % (len(tweets), elapsed))
-    except RuntimeError as exc:
+    except (TwitterError, RuntimeError) as exc:
        _exit_with_error(exc)

    _emit_tweet_detail(tweets, compact=compact, as_json=as_json, as_yaml=as_yaml, full_text=full_text)
@@ -757,7 +748,7 @@ def show(ctx, index, max_count, full_text, output_file, as_json, as_yaml):
        elapsed = time.time() - start
        if rich_output:
            console.print("✅ Fetched %d tweets in %.1fs\n" % (len(tweets), elapsed))
-    except RuntimeError as exc:
+    except (TwitterError, RuntimeError) as exc:
        _exit_with_error(exc)

    if output_file:
@@ -795,7 +786,7 @@ def article(ctx, tweet_id, as_json, as_yaml, as_markdown, output_file):
        elapsed = time.time() - start
        if rich_output:
            console.print("✅ Fetched article in %.1fs\n" % elapsed)
-    except RuntimeError as exc:
+    except (TwitterError, RuntimeError) as exc:
        _exit_with_error(exc)

    markdown = article_to_markdown(article_tweet)
@@ -836,13 +827,15 @@ def list_timeline(ctx, list_id, max_count, as_json, as_yaml, do_filter, full_tex
    _run_guarded(_run)


-@cli.command()
-@click.argument("screen_name")
-@click.option("--max", "-n", "max_count", type=int, default=None, help="Max users to fetch.")
-@structured_output_options
-def followers(screen_name, max_count, as_json, as_yaml):
-    # type: (str, int, bool, bool) -> None
-    """List followers of a user. SCREEN_NAME is the @handle (without @)."""
+def _fetch_and_display_users(
+    screen_name: str,
+    fetch_fn_name: str,
+    label: str,
+    max_count: Optional[int],
+    as_json: bool,
+    as_yaml: bool,
+) -> None:
+    """Shared fetch-and-display logic for followers/following commands."""
    screen_name = screen_name.lstrip("@")
    config = load_config()
    try:
@@ -853,22 +846,32 @@ def followers(screen_name, max_count, as_json, as_yaml):
        profile = client.fetch_user(screen_name)
        fetch_count = _resolve_configured_count(config, max_count)
        if rich_output:
-            console.print("👥 Fetching followers (%d)...\n" % fetch_count)
+            console.print("👥 Fetching %s (%d)...\n" % (label, fetch_count))
        start = time.time()
-        users = client.fetch_followers(profile.id, fetch_count)
+        users = getattr(client, fetch_fn_name)(profile.id, fetch_count)
        elapsed = time.time() - start
        if rich_output:
-            console.print("✅ Fetched %d followers in %.1fs\n" % (len(users), elapsed))
-    except RuntimeError as exc:
+            console.print("✅ Fetched %d %s in %.1fs\n" % (len(users), label, elapsed))
+    except (TwitterError, RuntimeError) as exc:
        _exit_with_error(exc)

    if emit_structured(users_to_data(users), as_json=as_json, as_yaml=as_yaml):
        return

-    print_user_table(users, console, title="👥 @%s followers — %d" % (screen_name, len(users)))
+    print_user_table(users, console, title="👥 @%s %s — %d" % (screen_name, label, len(users)))
    console.print()


+@cli.command()
+@click.argument("screen_name")
+@click.option("--max", "-n", "max_count", type=int, default=None, help="Max users to fetch.")
+@structured_output_options
+def followers(screen_name, max_count, as_json, as_yaml):
+    # type: (str, int, bool, bool) -> None
+    """List followers of a user. SCREEN_NAME is the @handle (without @)."""
+    _fetch_and_display_users(screen_name, "fetch_followers", "followers", max_count, as_json, as_yaml)
+
+
@cli.command()
@click.argument("screen_name")
@click.option("--max", "-n", "max_count", type=int, default=None, help="Max users to fetch.")
@@ -876,30 +879,7 @@ def followers(screen_name, max_count, as_json, as_yaml):
 def following(screen_name, max_count, as_json, as_yaml):
    # type: (str, int, bool, bool) -> None
    """List accounts a user is following. SCREEN_NAME is the @handle (without @)."""
-    screen_name = screen_name.lstrip("@")
-    config = load_config()
-    try:
-        rich_output = use_rich_output(as_json=as_json, as_yaml=as_yaml)
-        client = _get_client(config, quiet=not rich_output)
-        if rich_output:
-            console.print("👤 Fetching @%s's profile..." % screen_name)
-        profile = client.fetch_user(screen_name)
-        fetch_count = _resolve_configured_count(config, max_count)
-        if rich_output:
-            console.print("👥 Fetching following (%d)...\n" % fetch_count)
-        start = time.time()
-        users = client.fetch_following(profile.id, fetch_count)
-        elapsed = time.time() - start
-        if rich_output:
-            console.print("✅ Fetched %d following in %.1fs\n" % (len(users), elapsed))
-    except RuntimeError as exc:
-        _exit_with_error(exc)
-
-    if emit_structured(users_to_data(users), as_json=as_json, as_yaml=as_yaml):
-        return
-
-    print_user_table(users, console, title="👥 @%s following — %d" % (screen_name, len(users)))
-    console.print()
+    _fetch_and_display_users(screen_name, "fetch_following", "following", max_count, as_json, as_yaml)


 # ── Write commands ──────────────────────────────────────────────────────
@@ -1055,8 +1035,8 @@ def status(as_json, as_yaml):
        rich_output = use_rich_output(as_json=as_json, as_yaml=as_yaml)
        client = _get_client(config, quiet=not rich_output)
        profile = client.fetch_me()
-    except RuntimeError as exc:
-        payload = error_payload("not_authenticated", str(exc))
+    except (TwitterError, RuntimeError) as exc:
+        payload = error_payload(_error_code_from_exc(exc), str(exc))
        if emit_structured(payload, as_json=as_json, as_yaml=as_yaml):
            sys.exit(1)
        _exit_with_error(exc)
@@ -1082,8 +1062,8 @@ def whoami(as_json, as_yaml):
        if rich_output:
            console.print("👤 Fetching current user...")
        profile = client.fetch_me()
-    except RuntimeError as exc:
-        if emit_structured(error_payload("not_authenticated", str(exc)), as_json=as_json, as_yaml=as_yaml):
+    except (TwitterError, RuntimeError) as exc:
+        if emit_structured(error_payload(_error_code_from_exc(exc), str(exc)), as_json=as_json, as_yaml=as_yaml):
            raise SystemExit(1) from None
        _exit_with_error(exc)

--- a/twitter_cli/commands/init.py
+++ b/twitter_cli/commands/init.py
@@ -0,0 +1,7 @@
+"""CLI command sub-modules for twitter-cli.
+
+Commands are split into three groups:
+  - read: feed, bookmarks, search, tweet, article, show, list, favorites
+  - write: post, reply, quote, delete, like/unlike, retweet/unretweet, bookmark/unbookmark
+  - user: user, user-posts, likes, followers, following, whoami, status, follow/unfollow
+"""
--- a/twitter_cli/exceptions.py
+++ b/twitter_cli/exceptions.py
@@ -6,7 +6,7 @@ Provides a structured exception hierarchy for categorized error handling:
 - Network errors
 - Query ID resolution failures

-Modeled after bilibili-cli/xiaohongshu-cli exception patterns.
+Each exception carries an `error_code` attribute for structured output.
 """

 from __future__ import annotations
@@ -15,30 +15,50 @@ from __future__ import annotations
 class TwitterError(RuntimeError):
    """Base exception for twitter-cli errors."""

+    error_code: str = "api_error"
+

 class AuthenticationError(TwitterError):
    """Raised when cookies are missing, expired, or invalid."""

+    error_code = "not_authenticated"
+

 class RateLimitError(TwitterError):
    """Raised when Twitter rate limits the request (HTTP 429)."""

+    error_code = "rate_limited"
+

 class NotFoundError(TwitterError):
    """Raised when a user or tweet is not found."""

+    error_code = "not_found"
+

 class NetworkError(TwitterError):
    """Raised when upstream network requests fail."""

+    error_code = "network_error"
+

 class QueryIdError(TwitterError):
    """Raised when a GraphQL queryId cannot be resolved."""

+    error_code = "query_id_error"
+

 class MediaUploadError(TwitterError):
    """Raised when media upload fails (file not found, too large, unsupported format, API error)."""

+    error_code = "media_upload_error"
+
+
+class InvalidInputError(TwitterError):
+    """Raised when user input is invalid (bad tweet ID, invalid options, etc.)."""
+
+    error_code = "invalid_input"
+

 class TwitterAPIError(TwitterError):
    """Raised on non-OK Twitter API responses with HTTP status + message."""
@@ -46,4 +66,13 @@ class TwitterAPIError(TwitterError):
    def __init__(self, status_code: int, message: str):
        self.status_code = status_code
        self.message = message
+        # Derive error_code from HTTP status
+        if status_code in (401, 403):
+            self.error_code = "not_authenticated"
+        elif status_code == 429:
+            self.error_code = "rate_limited"
+        elif status_code == 404:
+            self.error_code = "not_found"
+        else:
+            self.error_code = "api_error"
        super().__init__("Twitter API error (HTTP %d): %s" % (status_code, message))
--- a/twitter_cli/serialization.py
+++ b/twitter_cli/serialization.py
@@ -6,7 +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
+from .timeutil import format_iso8601, format_local_time


 def tweet_to_dict(tweet: Tweet) -> Dict[str, Any]:
@@ -31,6 +31,7 @@ def tweet_to_dict(tweet: Tweet) -> Dict[str, Any]:
        },
        "createdAt": tweet.created_at,
        "createdAtLocal": format_local_time(tweet.created_at),
+        "createdAtISO": format_iso8601(tweet.created_at),
        "media": [
            {
                "type": media.type,
@@ -190,6 +191,7 @@ def user_profile_to_dict(user: UserProfile) -> Dict[str, Any]:
        "verified": user.verified,
        "profileImageUrl": user.profile_image_url,
        "createdAt": user.created_at,
+        "createdAtISO": format_iso8601(user.created_at),
    }


--- a/twitter_cli/timeutil.py
+++ b/twitter_cli/timeutil.py
@@ -69,3 +69,14 @@ def format_relative_time(created_at: str) -> str:
        return "%dmo ago" % months
    years = days // 365
    return "%dy ago" % years
+
+
+def format_iso8601(created_at: str) -> str:
+    """Convert Twitter timestamp to ISO 8601 format.
+
+    Returns e.g. "2026-03-08T12:00:00+00:00" or the original string on failure.
+    """
+    dt = _parse_twitter_time(created_at)
+    if dt is None:
+        return created_at
+    return dt.isoformat()