refactor: dynamic UA matching, session reuse, score Optional, --output on all commands

- constants.py: sync_chrome_version() aligns UA/sec-ch-ua with impersonate target
- auth.py: reuse shared cffi session instead of creating duplicate
- filter.py: eliminate double weight building in score_tweet
- models.py: Tweet.score → Optional[float] for accurate display
- cli.py: add --output to search/likes/user-posts for consistency
This commit is contained in:
jackwener
2026-03-09 21:15:28 +08:00
parent fda9b1c3dc
commit 8313a7012f
9 changed files with 79 additions and 41 deletions

View File

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "twitter-cli" name = "twitter-cli"
version = "0.4.3" version = "0.4.4"
description = "A CLI for Twitter/X — feed, bookmarks, and user timeline in terminal" description = "A CLI for Twitter/X — feed, bookmarks, and user timeline in terminal"
readme = "README.md" readme = "README.md"
license = "Apache-2.0" license = "Apache-2.0"

View File

@@ -17,9 +17,7 @@ import subprocess
import sys import sys
from typing import Dict, Optional from typing import Dict, Optional
from curl_cffi import requests as _cffi_requests from .constants import BEARER_TOKEN, get_user_agent
from .constants import BEARER_TOKEN, USER_AGENT
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -49,7 +47,7 @@ def verify_cookies(auth_token, ct0, cookie_string=None):
Tries multiple endpoints. Only raises on clear auth failures (401/403). Tries multiple endpoints. Only raises on clear auth failures (401/403).
For other errors (404, network), returns empty dict (proceed without verification). For other errors (404, network), returns empty dict (proceed without verification).
""" """
from .client import _best_chrome_target from .client import _get_cffi_session
urls = [ urls = [
"https://api.x.com/1.1/account/verify_credentials.json", "https://api.x.com/1.1/account/verify_credentials.json",
@@ -65,14 +63,11 @@ def verify_cookies(auth_token, ct0, cookie_string=None):
"X-Csrf-Token": ct0, "X-Csrf-Token": ct0,
"X-Twitter-Active-User": "yes", "X-Twitter-Active-User": "yes",
"X-Twitter-Auth-Type": "OAuth2Session", "X-Twitter-Auth-Type": "OAuth2Session",
"User-Agent": USER_AGENT, "User-Agent": get_user_agent(),
} }
proxy = os.environ.get("TWITTER_PROXY", "") # Reuse the shared curl_cffi session for consistent TLS fingerprint
session = _cffi_requests.Session( session = _get_cffi_session()
impersonate=_best_chrome_target(),
proxies={"https": proxy, "http": proxy} if proxy else None,
)
for url in urls: for url in urls:
try: try:

View File

@@ -244,8 +244,9 @@ def user(screen_name):
@click.argument("screen_name") @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=20, help="Max number of tweets to fetch.")
@click.option("--json", "as_json", is_flag=True, help="Output as JSON.") @click.option("--json", "as_json", is_flag=True, help="Output as JSON.")
def user_posts(screen_name, max_count, as_json): @click.option("--output", "-o", "output_file", type=str, default=None, help="Save tweets to JSON file.")
# type: (str, int, bool) -> None def user_posts(screen_name, max_count, as_json, output_file):
# type: (str, int, bool, Optional[str]) -> None
"""List a user's tweets. SCREEN_NAME is the @handle (without @).""" """List a user's tweets. SCREEN_NAME is the @handle (without @)."""
screen_name = screen_name.lstrip("@") screen_name = screen_name.lstrip("@")
config = load_config() config = load_config()
@@ -258,7 +259,7 @@ def user_posts(screen_name, max_count, as_json):
sys.exit(1) sys.exit(1)
_fetch_and_display( _fetch_and_display(
lambda count: client.fetch_user_tweets(profile.id, count), lambda count: client.fetch_user_tweets(profile.id, count),
"@%s tweets" % screen_name, "📝", max_count, as_json, None, False, config, "@%s tweets" % screen_name, "📝", max_count, as_json, output_file, False, config,
) )
@@ -277,15 +278,16 @@ SEARCH_PRODUCTS = ["Top", "Latest", "Photos", "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=20, help="Max number of tweets to fetch.")
@click.option("--json", "as_json", is_flag=True, help="Output as JSON.") @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.") @click.option("--filter", "do_filter", is_flag=True, help="Enable score-based filtering.")
def search(query, product, max_count, as_json, do_filter): def search(query, product, max_count, as_json, output_file, do_filter):
# type: (str, str, int, bool, bool) -> None # type: (str, str, int, bool, Optional[str], bool) -> None
"""Search tweets by QUERY string.""" """Search tweets by QUERY string."""
config = load_config() config = load_config()
client = _get_client(config) client = _get_client(config)
_fetch_and_display( _fetch_and_display(
lambda count: client.fetch_search(query, count, product), lambda count: client.fetch_search(query, count, product),
"'%s' (%s)" % (query, product), "🔍", max_count, as_json, None, do_filter, config, "'%s' (%s)" % (query, product), "🔍", max_count, as_json, output_file, do_filter, config,
) )
@@ -293,9 +295,10 @@ def search(query, product, max_count, as_json, do_filter):
@click.argument("screen_name") @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=20, help="Max number of tweets to fetch.")
@click.option("--json", "as_json", is_flag=True, help="Output as JSON.") @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.") @click.option("--filter", "do_filter", is_flag=True, help="Enable score-based filtering.")
def likes(screen_name, max_count, as_json, do_filter): def likes(screen_name, max_count, as_json, output_file, do_filter):
# type: (str, int, bool, bool) -> None # type: (str, int, bool, Optional[str], bool) -> None
"""Show tweets liked by a user. SCREEN_NAME is the @handle (without @).""" """Show tweets liked by a user. SCREEN_NAME is the @handle (without @)."""
screen_name = screen_name.lstrip("@") screen_name = screen_name.lstrip("@")
config = load_config() config = load_config()
@@ -308,7 +311,7 @@ def likes(screen_name, max_count, as_json, do_filter):
sys.exit(1) sys.exit(1)
_fetch_and_display( _fetch_and_display(
lambda count: client.fetch_user_likes(profile.id, count), lambda count: client.fetch_user_likes(profile.id, count),
"@%s likes" % screen_name, "❤️", max_count, as_json, None, do_filter, config, "@%s likes" % screen_name, "❤️", max_count, as_json, output_file, do_filter, config,
) )

View File

@@ -15,7 +15,10 @@ from curl_cffi import requests as _cffi_requests
from x_client_transaction import ClientTransaction from x_client_transaction import ClientTransaction
from x_client_transaction.utils import generate_headers as _gen_ct_headers, get_ondemand_file_url from x_client_transaction.utils import generate_headers as _gen_ct_headers, get_ondemand_file_url
from .constants import BEARER_TOKEN, USER_AGENT, SEC_CH_UA, SEC_CH_UA_MOBILE, SEC_CH_UA_PLATFORM from .constants import (
BEARER_TOKEN, SEC_CH_UA_MOBILE, SEC_CH_UA_PLATFORM,
get_sec_ch_ua, get_user_agent, sync_chrome_version,
)
from .models import Author, Metrics, Tweet, TweetMedia, UserProfile from .models import Author, Metrics, Tweet, TweetMedia, UserProfile
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -124,6 +127,7 @@ def _get_cffi_session():
import os import os
proxy = os.environ.get("TWITTER_PROXY", "") proxy = os.environ.get("TWITTER_PROXY", "")
target = _best_chrome_target() target = _best_chrome_target()
sync_chrome_version(target) # align UA/sec-ch-ua with impersonate target
_cffi_session = _cffi_requests.Session( _cffi_session = _cffi_requests.Session(
impersonate=target, impersonate=target,
proxies={"https": proxy, "http": proxy} if proxy else None, proxies={"https": proxy, "http": proxy} if proxy else None,
@@ -174,7 +178,7 @@ def _scan_bundles():
_bundles_scanned = True _bundles_scanned = True
try: try:
html = _url_fetch("https://x.com", {"user-agent": USER_AGENT}) html = _url_fetch("https://x.com", {"user-agent": get_user_agent()})
script_pattern = re.compile( script_pattern = re.compile(
r'(?:src|href)=["\']' r'(?:src|href)=["\']'
r'(https://abs\.twimg\.com/responsive-web/client-web[^"\']+\.js)' r'(https://abs\.twimg\.com/responsive-web/client-web[^"\']+\.js)'
@@ -685,12 +689,12 @@ class TwitterClient:
"X-Twitter-Active-User": "yes", "X-Twitter-Active-User": "yes",
"X-Twitter-Auth-Type": "OAuth2Session", "X-Twitter-Auth-Type": "OAuth2Session",
"X-Twitter-Client-Language": "en", "X-Twitter-Client-Language": "en",
"User-Agent": USER_AGENT, "User-Agent": get_user_agent(),
"Origin": "https://x.com", "Origin": "https://x.com",
"Referer": "https://x.com", "Referer": "https://x.com",
"Accept": "*/*", "Accept": "*/*",
"Accept-Language": "en-US,en;q=0.9", "Accept-Language": "en-US,en;q=0.9",
"sec-ch-ua": SEC_CH_UA, "sec-ch-ua": get_sec_ch_ua(),
"sec-ch-ua-mobile": SEC_CH_UA_MOBILE, "sec-ch-ua-mobile": SEC_CH_UA_MOBILE,
"sec-ch-ua-platform": SEC_CH_UA_PLATFORM, "sec-ch-ua-platform": SEC_CH_UA_PLATFORM,
"Sec-Fetch-Dest": "empty", "Sec-Fetch-Dest": "empty",

View File

@@ -1,17 +1,50 @@
"""Shared constants for twitter-cli.""" """Shared constants for twitter-cli."""
import re
BEARER_TOKEN = ( BEARER_TOKEN = (
"AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs" "AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs"
"%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA" "%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA"
) )
USER_AGENT = ( # Default Chrome version — updated by _best_chrome_target() at runtime
_DEFAULT_CHROME_VERSION = "133"
_chrome_version = _DEFAULT_CHROME_VERSION # mutable, set by sync_chrome_version()
def sync_chrome_version(impersonate_target):
# type: (str) -> None
"""Sync USER_AGENT / SEC_CH_UA with the actual impersonate target.
Called once when _get_cffi_session() picks a target (e.g. "chrome136").
"""
global _chrome_version
match = re.search(r"(\d+)", impersonate_target)
if match:
_chrome_version = match.group(1)
def get_user_agent():
# type: () -> str
return (
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) " "AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/133.0.0.0 Safari/537.36" "Chrome/%s.0.0.0 Safari/537.36" % _chrome_version
) )
# Chrome Client Hints — sent by modern Chrome on every request
SEC_CH_UA = '"Chromium";v="133", "Not(A:Brand";v="99", "Google Chrome";v="133"' def get_sec_ch_ua():
# type: () -> str
return '"Chromium";v="%s", "Not(A:Brand";v="99", "Google Chrome";v="%s"' % (
_chrome_version, _chrome_version,
)
# Static Client Hints
SEC_CH_UA_MOBILE = "?0" SEC_CH_UA_MOBILE = "?0"
SEC_CH_UA_PLATFORM = '"macOS"' SEC_CH_UA_PLATFORM = '"macOS"'
# 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

@@ -21,7 +21,7 @@ DEFAULT_WEIGHTS = {
def score_tweet(tweet, weights=None): def score_tweet(tweet, weights=None):
# type: (Tweet, Optional[FilterWeights]) -> float # type: (Tweet, Optional[Dict[str, float]]) -> float
"""Calculate engagement score for a single tweet. """Calculate engagement score for a single tweet.
Formula: Formula:
@@ -30,15 +30,18 @@ def score_tweet(tweet, weights=None):
+ w_replies × replies + w_replies × replies
+ w_bookmarks × bookmarks + w_bookmarks × bookmarks
+ w_views_log × log10(views) + w_views_log × log10(views)
Args:
weights: Pre-built weight dict. If None, uses DEFAULT_WEIGHTS.
""" """
weight_map = _build_weights(weights or {}) w = weights or DEFAULT_WEIGHTS
m = tweet.metrics m = tweet.metrics
return ( return (
weight_map["likes"] * m.likes w.get("likes", 1.0) * m.likes
+ weight_map["retweets"] * m.retweets + w.get("retweets", 3.0) * m.retweets
+ weight_map["replies"] * m.replies + w.get("replies", 2.0) * m.replies
+ weight_map["bookmarks"] * m.bookmarks + w.get("bookmarks", 5.0) * m.bookmarks
+ weight_map["views_log"] * math.log10(max(m.views, 1)) + w.get("views_log", 0.5) * math.log10(max(m.views, 1))
) )

View File

@@ -49,7 +49,7 @@ class Tweet:
lang: str = "" lang: str = ""
retweeted_by: Optional[str] = None retweeted_by: Optional[str] = None
quoted_tweet: Optional[Tweet] = None quoted_tweet: Optional[Tweet] = None
score: float = 0.0 score: Optional[float] = None
@dataclass @dataclass

View File

@@ -112,7 +112,7 @@ def tweet_from_dict(data: Dict[str, Any]) -> Tweet:
lang=str(data.get("lang") or ""), lang=str(data.get("lang") or ""),
retweeted_by=_optional_str(data.get("retweetedBy")), retweeted_by=_optional_str(data.get("retweetedBy")),
quoted_tweet=quoted_tweet, quoted_tweet=quoted_tweet,
score=float(data.get("score") or 0.0), score=float(data["score"]) if data.get("score") is not None else None,
) )

2
uv.lock generated
View File

@@ -950,7 +950,7 @@ wheels = [
[[package]] [[package]]
name = "twitter-cli" name = "twitter-cli"
version = "0.4.2" version = "0.4.3"
source = { editable = "." } source = { editable = "." }
dependencies = [ dependencies = [
{ name = "beautifulsoup4" }, { name = "beautifulsoup4" },