luna/twitter-cli-cookiefile
1
0
Fork 0
Code Issues Pull Requests Actions Releases Activity

feat: add user commands, auto-detect browser, optimize performance

Browse Source 
- Add user/user-posts/followers/following commands
- Add UserProfile model and GraphQL API methods
- Add print_user_profile and print_user_table formatters
- Auto-detect browser for cookies (Chrome → Edge → Firefox → Brave)
- Remove --browser option from all commands
- Remove cookie verification (v1.1 endpoints are gone)
- Use hardcoded fallback query IDs first (skip slow JS bundle scan)
- Update FEATURES from latest twitter-openapi config
- Fix user-posts: add required withVoice variable
- Add tweet URL links in feed output
- Add error handling to all user commands
This commit is contained in:
jackwener
2026-03-05 00:41:26 +08:00
parent 16752c3115
commit 7238b932ab
10 changed files with 770 additions and 353 deletions
Download Patch File Download Diff File Expand all files Collapse all files

302
twitter_cli/client.py
View File

@@ -15,7 +15,7 @@ import ssl
import urllib.request
from typing import Any, Callable, Dict, List, Optional, Tuple
from .models import Author, Metrics, Tweet, TweetMedia
from .models import Author, Metrics, Tweet, TweetMedia, UserProfile
logger = logging.getLogger(__name__)
@@ -27,8 +27,14 @@ BEARER_TOKEN = (
# Last-resort fallback query IDs
FALLBACK_QUERY_IDS = {
"HomeTimeline": "HJFjzBgCs16TqxewQOeLNg",
"HomeTimeline": "c-CzHF1LboFilMpsx4ZCrQ",
"Bookmarks": "VFdMm9iVZxlU6hD86gfW_A",
"CreateTweet": "oB-5XsHNAbjvARJEc8CZFw",
"DeleteTweet": "VaenaVgh5q5ih7kvyVjgtg",
"UserByScreenName": "1VOOyvKkiI3FMmkeDNxM9A",
"UserTweets": "E3opETHurmVJflFsUBVuUQ",
"Followers": "IOh4aS6UdGWGJUYTqliQ7Q",
"Following": "zx6e-TLzRkeDO_a7p4b3JQ",
}
# Community-maintained API definition (auto-updated daily)
@@ -39,14 +45,20 @@ TWITTER_OPENAPI_URL = (
# Default features flags required by the GraphQL endpoint
FEATURES = {
"rweb_video_screen_enabled": False,
"profile_label_improvements_pcf_label_in_post_enabled": True,
"rweb_tipjar_consumption_enabled": True,
"responsive_web_graphql_exclude_directive_enabled": True,
"verified_phone_label_enabled": False,
"creator_subscriptions_tweet_preview_api_enabled": True,
"responsive_web_graphql_timeline_navigation_enabled": True,
"responsive_web_graphql_skip_user_profile_image_extensions_enabled": False,
"premium_content_api_read_enabled": False,
"communities_web_enable_tweet_community_results_fetch": True,
"c9s_tweet_anatomy_moderator_badge_enabled": True,
"responsive_web_grok_analyze_button_fetch_trends_enabled": False,
"responsive_web_grok_analyze_post_followups_enabled": True,
"responsive_web_jetfuel_frame": False,
"responsive_web_grok_share_attachment_enabled": True,
"articles_preview_enabled": True,
"responsive_web_edit_tweet_api_enabled": True,
"graphql_is_translatable_rweb_tweet_is_translatable_enabled": True,
@@ -54,13 +66,15 @@ FEATURES = {
"longform_notetweets_consumption_enabled": True,
"responsive_web_twitter_article_tweet_consumption_enabled": True,
"tweet_awards_web_tipping_enabled": False,
"responsive_web_grok_show_grok_translated_post": False,
"responsive_web_grok_analysis_button_from_backend": False,
"creator_subscriptions_quote_tweet_preview_enabled": False,
"freedom_of_speech_not_reach_fetch_enabled": True,
"standardized_nudges_misinfo": True,
"tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled": True,
"rweb_video_timestamps_enabled": True,
"longform_notetweets_rich_text_read_enabled": True,
"longform_notetweets_inline_media_enabled": True,
"responsive_web_grok_image_annotation_enabled": True,
"responsive_web_enhance_cards_enabled": False,
}
@@ -152,17 +166,18 @@ def _fetch_from_github(operation_name):
def _resolve_query_id(operation_name):
# type: (str) -> str
"""Resolve queryId using three-tier strategy: bundle scan -> GitHub -> fallback."""
"""Resolve queryId using three-tier strategy: fallback -> GitHub -> bundle scan."""
if operation_name in _cached_query_ids:
return _cached_query_ids[operation_name]
logger.info("Auto-detecting %s queryId...", operation_name)
# Tier 1: Hardcoded fallback (instant, no network)
fallback = FALLBACK_QUERY_IDS.get(operation_name)
if fallback:
logger.debug("Using fallback queryId for %s: %s", operation_name, fallback)
_cached_query_ids[operation_name] = fallback
return fallback
# Tier 1: JS bundle scan
_scan_bundles()
if operation_name in _cached_query_ids:
logger.info("Found %s queryId: %s", operation_name, _cached_query_ids[operation_name])
return _cached_query_ids[operation_name]
logger.info("Auto-detecting %s queryId (no fallback available)...", operation_name)
# Tier 2: GitHub
github_id = _fetch_from_github(operation_name)
@@ -170,12 +185,11 @@ def _resolve_query_id(operation_name):
_cached_query_ids[operation_name] = github_id
return github_id
# Tier 3: Hardcoded fallback
fallback = FALLBACK_QUERY_IDS.get(operation_name)
if fallback:
logger.info("Using hardcoded fallback queryId for %s: %s", operation_name, fallback)
_cached_query_ids[operation_name] = fallback
return fallback
# Tier 3: JS bundle scan
_scan_bundles()
if operation_name in _cached_query_ids:
logger.info("Found %s queryId: %s", operation_name, _cached_query_ids[operation_name])
return _cached_query_ids[operation_name]
raise RuntimeError(
'Cannot resolve queryId for "%s" — all detection methods failed' % operation_name
@@ -291,6 +305,260 @@ class TwitterClient:
body = e.read().decode("utf-8", errors="replace")
raise RuntimeError("Twitter API error %d: %s" % (e.code, body[:500]))
def _api_post(self, url, payload):
# type: (str, Dict[str, Any]) -> Any
"""Make authenticated POST request to Twitter API."""
headers = self._build_headers()
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(url, data=data, method="POST")
for k, v in headers.items():
req.add_header(k, v)
ctx = _create_ssl_context()
try:
with urllib.request.urlopen(req, context=ctx, timeout=30) as resp:
body = resp.read().decode("utf-8")
return json.loads(body)
except urllib.error.HTTPError as e:
body = e.read().decode("utf-8", errors="replace")
raise RuntimeError("Twitter API error %d: %s" % (e.code, body[:500]))
def create_tweet(self, text, reply_to=None, quote_tweet_url=None):
# type: (str, Optional[str], Optional[str]) -> Dict[str, Any]
"""Create a tweet, reply, or quote tweet.
Args:
text: Tweet text content.
reply_to: Tweet ID to reply to (optional).
quote_tweet_url: URL of tweet to quote (optional).
Returns:
Dict with tweet_id and text of the created tweet.
"""
query_id = _resolve_query_id("CreateTweet")
url = "https://x.com/i/api/graphql/%s/CreateTweet" % query_id
variables = {
"tweet_text": text,
"dark_request": False,
"media": {"media_entities": [], "possibly_sensitive": False},
"semantic_annotation_ids": [],
} # type: Dict[str, Any]
if reply_to:
variables["reply"] = {
"in_reply_to_tweet_id": reply_to,
"exclude_reply_user_ids": [],
}
if quote_tweet_url:
variables["attachment_url"] = quote_tweet_url
features = {
"communities_web_enable_tweet_community_results_fetch": True,
"c9s_tweet_anatomy_moderator_badge_enabled": True,
"responsive_web_edit_tweet_api_enabled": True,
"graphql_is_translatable_rweb_tweet_is_translatable_enabled": True,
"view_counts_everywhere_api_enabled": True,
"longform_notetweets_consumption_enabled": True,
"responsive_web_twitter_article_tweet_consumption_enabled": True,
"tweet_awards_web_tipping_enabled": False,
"creator_subscriptions_quote_tweet_preview_enabled": False,
"longform_notetweets_rich_text_read_enabled": True,
"longform_notetweets_inline_media_enabled": True,
"articles_preview_enabled": True,
"rweb_video_timestamps_enabled": True,
"rweb_tipjar_consumption_enabled": True,
"responsive_web_graphql_exclude_directive_enabled": True,
"verified_phone_label_enabled": False,
"freedom_of_speech_not_reach_fetch_enabled": True,
"standardized_nudges_misinfo": True,
"tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled": True,
"responsive_web_graphql_skip_user_profile_image_extensions_enabled": False,
"responsive_web_graphql_timeline_navigation_enabled": True,
"responsive_web_enhance_cards_enabled": False,
}
payload = {
"variables": variables,
"features": features,
"queryId": query_id,
}
data = self._api_post(url, payload)
# Parse response
result = _deep_get(data, "data", "create_tweet", "tweet_results", "result")
if not result:
errors = data.get("errors", [])
if errors:
raise RuntimeError("CreateTweet failed: %s" % errors[0].get("message", str(errors)))
raise RuntimeError("CreateTweet failed: unexpected response")
tweet_id = result.get("rest_id", "")
tweet_text = _deep_get(result, "legacy", "full_text") or text
return {"tweet_id": tweet_id, "text": tweet_text}
def delete_tweet(self, tweet_id):
# type: (str) -> bool
"""Delete a tweet by ID.
Returns:
True if deletion was successful.
"""
query_id = _resolve_query_id("DeleteTweet")
url = "https://x.com/i/api/graphql/%s/DeleteTweet" % query_id
payload = {
"variables": {"tweet_id": tweet_id, "dark_request": False},
"queryId": query_id,
}
data = self._api_post(url, payload)
# Check response
result = _deep_get(data, "data", "delete_tweet", "tweet_results")
if result is not None:
return True
errors = data.get("errors", [])
if errors:
raise RuntimeError("DeleteTweet failed: %s" % errors[0].get("message", str(errors)))
# Some successful deletions return empty result
return True
def fetch_user(self, screen_name):
# type: (str) -> UserProfile
"""Fetch user profile by screen name."""
query_id = _resolve_query_id("UserByScreenName")
variables = {
"screen_name": screen_name,
"withSafetyModeUserFields": True,
}
features = {
"hidden_profile_subscriptions_enabled": True,
"rweb_tipjar_consumption_enabled": True,
"responsive_web_graphql_exclude_directive_enabled": True,
"verified_phone_label_enabled": False,
"subscriptions_verification_info_is_identity_verified_enabled": True,
"subscriptions_verification_info_verified_since_enabled": True,
"highlights_tweets_tab_ui_enabled": True,
"responsive_web_twitter_article_notes_tab_enabled": True,
"subscriptions_feature_can_gift_premium": True,
"creator_subscriptions_tweet_preview_api_enabled": True,
"responsive_web_graphql_skip_user_profile_image_extensions_enabled": False,
"responsive_web_graphql_timeline_navigation_enabled": True,
}
url = "https://x.com/i/api/graphql/%s/UserByScreenName?" % query_id
url += "variables=%s&features=%s" % (
urllib.request.quote(json.dumps(variables)),
urllib.request.quote(json.dumps(features)),
)
data = self._api_get(url)
result = _deep_get(data, "data", "user", "result")
if not result:
raise RuntimeError("User @%s not found" % screen_name)
legacy = result.get("legacy", {})
core = result.get("core", {})
return UserProfile(
id=result.get("rest_id", ""),
name=core.get("name") or legacy.get("name", ""),
screen_name=core.get("screen_name") or legacy.get("screen_name", screen_name),
bio=legacy.get("description", ""),
location=legacy.get("location", ""),
url=(
legacy.get("entities", {}).get("url", {}).get("urls", [{}])[0].get("expanded_url", "")
if legacy.get("entities", {}).get("url")
else ""
),
followers_count=legacy.get("followers_count", 0),
following_count=legacy.get("friends_count", 0),
tweets_count=legacy.get("statuses_count", 0),
likes_count=legacy.get("favourites_count", 0),
verified=bool(result.get("is_blue_verified") or legacy.get("verified", False)),
profile_image_url=legacy.get("profile_image_url_https", ""),
created_at=legacy.get("created_at", ""),
)
def fetch_user_tweets(self, user_id, count=20):
# type: (str, int) -> List[Tweet]
"""Fetch tweets posted by a user."""
query_id = _resolve_query_id("UserTweets")
return self._fetch_timeline(
query_id,
"UserTweets",
count,
lambda data: _deep_get(data, "data", "user", "result", "timeline_v2", "timeline", "instructions"),
extra_variables={
"userId": user_id,
"withQuickPromoteEligibilityTweetFields": True,
"withVoice": True,
"withV2Timeline": True,
},
)
def fetch_followers(self, user_id, count=20):
# type: (str, int) -> List[UserProfile]
"""Fetch user's followers."""
query_id = _resolve_query_id("Followers")
return self._fetch_user_list(query_id, "Followers", user_id, count)
def fetch_following(self, user_id, count=20):
# type: (str, int) -> List[UserProfile]
"""Fetch users that this user follows."""
query_id = _resolve_query_id("Following")
return self._fetch_user_list(query_id, "Following", user_id, count)
def _fetch_user_list(self, query_id, operation_name, user_id, count):
# type: (str, str, str, int) -> List[UserProfile]
"""Generic user list fetcher (followers/following)."""
variables = {
"userId": user_id,
"count": min(count, 50),
"includePromotedContent": False,
} # type: Dict[str, Any]
url = "https://x.com/i/api/graphql/%s/%s?" % (query_id, operation_name)
url += "variables=%s&features=%s" % (
urllib.request.quote(json.dumps(variables)),
urllib.request.quote(json.dumps(FEATURES)),
)
data = self._api_get(url)
users = [] # type: List[UserProfile]
instructions = _deep_get(data, "data", "user", "result", "timeline", "timeline", "instructions")
if not isinstance(instructions, list):
return users
for instruction in instructions:
entries = instruction.get("entries", [])
for entry in entries:
content = entry.get("content", {})
item_content = content.get("itemContent", {})
user_results = item_content.get("user_results", {}).get("result")
if not user_results:
continue
legacy = user_results.get("legacy", {})
core = user_results.get("core", {})
if not legacy:
continue
users.append(UserProfile(
id=user_results.get("rest_id", ""),
name=core.get("name") or legacy.get("name", ""),
screen_name=core.get("screen_name") or legacy.get("screen_name", ""),
bio=legacy.get("description", ""),
followers_count=legacy.get("followers_count", 0),
following_count=legacy.get("friends_count", 0),
verified=bool(user_results.get("is_blue_verified") or legacy.get("verified", False)),
profile_image_url=legacy.get("profile_image_url_https", ""),
))
return users[:count]
def _parse_timeline_response(self, data, get_instructions):
# type: (Any, Callable) -> Tuple[List[Tweet], Optional[str]]
"""Parse timeline GraphQL response into tweets + next cursor."""