fix: add fieldToggles support for TweetDetail

- TweetDetail requires fieldToggles (withArticleRichContentState: true)
  to populate tweet_results in entries — without it, server returns {}
- Add fieldToggles parameter throughout: _build_graphql_url, _graphql_get,
  _fetch_timeline
- Tested: tweet detail, followers, following, like — all working
This commit is contained in:
jackwener
2026-03-07 20:20:59 +08:00
parent 80499384c9
commit 6c73a9f0b6

View File

@@ -136,15 +136,20 @@ def _url_fetch(url, headers=None):
return response.read().decode("utf-8") return response.read().decode("utf-8")
def _build_graphql_url(query_id, operation_name, variables, features): def _build_graphql_url(query_id, operation_name, variables, features, field_toggles=None):
# type: (str, str, Dict[str, Any], Dict[str, Any]) -> str # type: (str, str, Dict[str, Any], Dict[str, Any], Optional[Dict[str, Any]]) -> str
"""Build GraphQL GET URL with encoded variables/features.""" """Build GraphQL GET URL with encoded variables/features/fieldToggles."""
return "https://x.com/i/api/graphql/%s/%s?variables=%s&features=%s" % ( url = "https://x.com/i/api/graphql/%s/%s?variables=%s&features=%s" % (
query_id, query_id,
operation_name, operation_name,
urllib.parse.quote(json.dumps(variables, separators=(",", ":"))), urllib.parse.quote(json.dumps(variables, separators=(",", ":"))),
urllib.parse.quote(json.dumps(features, separators=(",", ":"))), urllib.parse.quote(json.dumps(features, separators=(",", ":"))),
) )
if field_toggles:
url += "&fieldToggles=%s" % urllib.parse.quote(
json.dumps(field_toggles, separators=(",", ":"))
)
return url
def _scan_bundles(): def _scan_bundles():
@@ -396,6 +401,7 @@ class TwitterClient:
"focalTweetId": tweet_id, "focalTweetId": tweet_id,
"referrer": "tweet", "referrer": "tweet",
"with_rux_injections": False, "with_rux_injections": False,
"includePromotedContent": True,
"rankingMode": "Relevance", "rankingMode": "Relevance",
"withCommunity": True, "withCommunity": True,
"withQuickPromoteEligibilityTweetFields": True, "withQuickPromoteEligibilityTweetFields": True,
@@ -403,6 +409,12 @@ class TwitterClient:
"withVoice": True, "withVoice": True,
}, },
override_base_variables=True, override_base_variables=True,
field_toggles={
"withArticleRichContentState": True,
"withArticlePlainText": False,
"withGrokAnalyze": False,
"withDisallowedReplyControls": False,
},
) )
def fetch_list_timeline(self, list_id, count=20): def fetch_list_timeline(self, list_id, count=20):
@@ -497,8 +509,8 @@ class TwitterClient:
self._graphql_post("DeleteBookmark", {"tweet_id": tweet_id}) self._graphql_post("DeleteBookmark", {"tweet_id": tweet_id})
return True return True
def _fetch_timeline(self, operation_name, count, get_instructions, extra_variables=None, override_base_variables=False): def _fetch_timeline(self, operation_name, count, get_instructions, extra_variables=None, override_base_variables=False, field_toggles=None):
# type: (str, int, Callable[[Any], Any], Optional[Dict[str, Any]], bool) -> List[Tweet] # type: (str, int, Callable[[Any], Any], Optional[Dict[str, Any]], bool, Optional[Dict[str, Any]]) -> List[Tweet]
"""Generic timeline fetcher with pagination and deduplication. """Generic timeline fetcher with pagination and deduplication.
Args: Args:
@@ -534,7 +546,7 @@ class TwitterClient:
if cursor: if cursor:
variables["cursor"] = cursor variables["cursor"] = cursor
data = self._graphql_get(operation_name, variables, FEATURES) 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 = self._parse_timeline_response(data, get_instructions)
for tweet in new_tweets: for tweet in new_tweets:
@@ -553,12 +565,12 @@ class TwitterClient:
return tweets[:count] return tweets[:count]
def _graphql_get(self, operation_name, variables, features): def _graphql_get(self, operation_name, variables, features, field_toggles=None):
# type: (str, Dict[str, Any], Dict[str, Any]) -> Dict[str, Any] # type: (str, Dict[str, Any], Dict[str, Any], Optional[Dict[str, Any]]) -> Dict[str, Any]
"""Issue GraphQL GET request with automatic stale-fallback retry.""" """Issue GraphQL GET request with automatic stale-fallback retry."""
query_id = _resolve_query_id(operation_name, prefer_fallback=True) query_id = _resolve_query_id(operation_name, prefer_fallback=True)
using_fallback = query_id == FALLBACK_QUERY_IDS.get(operation_name) using_fallback = query_id == FALLBACK_QUERY_IDS.get(operation_name)
url = _build_graphql_url(query_id, operation_name, variables, features) url = _build_graphql_url(query_id, operation_name, variables, features, field_toggles)
try: try:
return self._api_get(url) return self._api_get(url)
@@ -568,7 +580,7 @@ class TwitterClient:
logger.info("Retrying %s with live queryId after 404", operation_name) logger.info("Retrying %s with live queryId after 404", operation_name)
_invalidate_query_id(operation_name) _invalidate_query_id(operation_name)
refreshed_query_id = _resolve_query_id(operation_name, prefer_fallback=False) refreshed_query_id = _resolve_query_id(operation_name, prefer_fallback=False)
retry_url = _build_graphql_url(refreshed_query_id, operation_name, variables, features) retry_url = _build_graphql_url(refreshed_query_id, operation_name, variables, features, field_toggles)
return self._api_get(retry_url) return self._api_get(retry_url)
raise RuntimeError(str(exc)) raise RuntimeError(str(exc))