feat: detect subscriber-only tweets via tweetInterstitial (#33)
Extract visibility metadata from TweetWithVisibilityResults wrapper before unwrapping. Adds is_subscriber_only field to Tweet model, with full serialization roundtrip and test coverage. Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -38,6 +38,7 @@ def tweet_factory():
|
|||||||
score=overrides.pop("score", 0.0),
|
score=overrides.pop("score", 0.0),
|
||||||
article_title=overrides.pop("article_title", None),
|
article_title=overrides.pop("article_title", None),
|
||||||
article_text=overrides.pop("article_text", None),
|
article_text=overrides.pop("article_text", None),
|
||||||
|
is_subscriber_only=overrides.pop("is_subscriber_only", False),
|
||||||
)
|
)
|
||||||
|
|
||||||
return _make_tweet
|
return _make_tweet
|
||||||
|
|||||||
4
tests/fixtures/list_timeline.json
vendored
4
tests/fixtures/list_timeline.json
vendored
@@ -15,6 +15,10 @@
|
|||||||
"tweet_results": {
|
"tweet_results": {
|
||||||
"result": {
|
"result": {
|
||||||
"__typename": "TweetWithVisibilityResults",
|
"__typename": "TweetWithVisibilityResults",
|
||||||
|
"tweetInterstitial": {
|
||||||
|
"__typename": "TweetInterstitial",
|
||||||
|
"text": { "rtl": false, "text": "Subscribe to @lister to see this post" }
|
||||||
|
},
|
||||||
"tweet": {
|
"tweet": {
|
||||||
"__typename": "Tweet",
|
"__typename": "Tweet",
|
||||||
"rest_id": "700",
|
"rest_id": "700",
|
||||||
|
|||||||
@@ -641,6 +641,7 @@ class TestParseTweetResult:
|
|||||||
tweet = parse_tweet_result(wrapped)
|
tweet = parse_tweet_result(wrapped)
|
||||||
assert tweet is not None
|
assert tweet is not None
|
||||||
assert tweet.id == "1234567890"
|
assert tweet.id == "1234567890"
|
||||||
|
assert tweet.is_subscriber_only is False
|
||||||
|
|
||||||
@patch("twitter_cli.client._get_cffi_session")
|
@patch("twitter_cli.client._get_cffi_session")
|
||||||
@patch("twitter_cli.client._gen_ct_headers", return_value={})
|
@patch("twitter_cli.client._gen_ct_headers", return_value={})
|
||||||
|
|||||||
@@ -74,6 +74,7 @@ def test_parse_list_timeline_fixture_with_visibility_wrapper(fixture_loader) ->
|
|||||||
assert cursor == "list-cursor"
|
assert cursor == "list-cursor"
|
||||||
assert tweets[0].author.verified is True
|
assert tweets[0].author.verified is True
|
||||||
assert tweets[0].lang == "zh"
|
assert tweets[0].lang == "zh"
|
||||||
|
assert tweets[0].is_subscriber_only is True
|
||||||
|
|
||||||
|
|
||||||
def test_fetch_user_list_with_fixture(monkeypatch, fixture_loader) -> None:
|
def test_fetch_user_list_with_fixture(monkeypatch, fixture_loader) -> None:
|
||||||
|
|||||||
@@ -77,3 +77,11 @@ def test_tweet_roundtrip_preserves_article_fields(tweet_factory) -> None:
|
|||||||
|
|
||||||
assert restored.article_title == "Long-form title"
|
assert restored.article_title == "Long-form title"
|
||||||
assert restored.article_text == "Intro\n\n## Details"
|
assert restored.article_text == "Intro\n\n## Details"
|
||||||
|
|
||||||
|
|
||||||
|
def test_tweet_roundtrip_preserves_subscriber_only(tweet_factory) -> None:
|
||||||
|
tweet = tweet_factory("99", is_subscriber_only=True)
|
||||||
|
payload = tweet_to_dict(tweet)
|
||||||
|
assert payload["isSubscriberOnly"] is True
|
||||||
|
restored = tweet_from_dict(payload)
|
||||||
|
assert restored.is_subscriber_only is True
|
||||||
|
|||||||
@@ -52,6 +52,7 @@ class Tweet:
|
|||||||
score: Optional[float] = None
|
score: Optional[float] = None
|
||||||
article_title: Optional[str] = None
|
article_title: Optional[str] = None
|
||||||
article_text: Optional[str] = None
|
article_text: Optional[str] = None
|
||||||
|
is_subscriber_only: bool = False
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
|
|||||||
@@ -241,15 +241,21 @@ def parse_user_result(user_data):
|
|||||||
# ── Tweet parsing ────────────────────────────────────────────────────────
|
# ── Tweet parsing ────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|
||||||
|
def _unwrap_visibility(result):
|
||||||
|
# type: (Dict[str, Any]) -> Tuple[Dict[str, Any], bool]
|
||||||
|
"""Unwrap TweetWithVisibilityResults, returning (inner_data, is_subscriber_only)."""
|
||||||
|
if result.get("__typename") == "TweetWithVisibilityResults" and result.get("tweet"):
|
||||||
|
return result["tweet"], bool(result.get("tweetInterstitial"))
|
||||||
|
return result, False
|
||||||
|
|
||||||
|
|
||||||
def parse_tweet_result(result, depth=0):
|
def parse_tweet_result(result, depth=0):
|
||||||
# type: (Dict[str, Any], int) -> Optional[Tweet]
|
# type: (Dict[str, Any], int) -> Optional[Tweet]
|
||||||
"""Parse a single TweetResult into a Tweet dataclass."""
|
"""Parse a single TweetResult into a Tweet dataclass."""
|
||||||
if depth > 2:
|
if depth > 2:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
tweet_data = result
|
tweet_data, is_subscriber_only = _unwrap_visibility(result)
|
||||||
if result.get("__typename") == "TweetWithVisibilityResults" and result.get("tweet"):
|
|
||||||
tweet_data = result["tweet"]
|
|
||||||
if tweet_data.get("__typename") == "TweetTombstone":
|
if tweet_data.get("__typename") == "TweetTombstone":
|
||||||
return None
|
return None
|
||||||
|
|
||||||
@@ -270,8 +276,7 @@ def parse_tweet_result(result, depth=0):
|
|||||||
|
|
||||||
if is_retweet:
|
if is_retweet:
|
||||||
retweet_result = _deep_get(legacy, "retweeted_status_result", "result") or {}
|
retweet_result = _deep_get(legacy, "retweeted_status_result", "result") or {}
|
||||||
if retweet_result.get("__typename") == "TweetWithVisibilityResults" and retweet_result.get("tweet"):
|
retweet_result, retweet_subscriber_only = _unwrap_visibility(retweet_result)
|
||||||
retweet_result = retweet_result["tweet"]
|
|
||||||
rt_legacy = retweet_result.get("legacy")
|
rt_legacy = retweet_result.get("legacy")
|
||||||
rt_core = retweet_result.get("core")
|
rt_core = retweet_result.get("core")
|
||||||
if isinstance(rt_legacy, dict) and isinstance(rt_core, dict):
|
if isinstance(rt_legacy, dict) and isinstance(rt_core, dict):
|
||||||
@@ -312,6 +317,7 @@ def parse_tweet_result(result, depth=0):
|
|||||||
retweeted_by=retweeted_by,
|
retweeted_by=retweeted_by,
|
||||||
quoted_tweet=quoted_tweet,
|
quoted_tweet=quoted_tweet,
|
||||||
lang=actual_legacy.get("lang", ""),
|
lang=actual_legacy.get("lang", ""),
|
||||||
|
is_subscriber_only=retweet_subscriber_only if is_retweet else is_subscriber_only,
|
||||||
**_parse_article(actual_data),
|
**_parse_article(actual_data),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -46,6 +46,7 @@ def tweet_to_dict(tweet: Tweet) -> Dict[str, Any]:
|
|||||||
"retweetedBy": tweet.retweeted_by,
|
"retweetedBy": tweet.retweeted_by,
|
||||||
"lang": tweet.lang,
|
"lang": tweet.lang,
|
||||||
"score": tweet.score,
|
"score": tweet.score,
|
||||||
|
"isSubscriberOnly": tweet.is_subscriber_only,
|
||||||
}
|
}
|
||||||
if tweet.article_title is not None:
|
if tweet.article_title is not None:
|
||||||
data["articleTitle"] = tweet.article_title
|
data["articleTitle"] = tweet.article_title
|
||||||
@@ -122,6 +123,7 @@ def tweet_from_dict(data: Dict[str, Any]) -> Tweet:
|
|||||||
score=float(data["score"]) if data.get("score") is not None else None,
|
score=float(data["score"]) if data.get("score") is not None else None,
|
||||||
article_title=_optional_str(data.get("articleTitle")),
|
article_title=_optional_str(data.get("articleTitle")),
|
||||||
article_text=_optional_str(data.get("articleText")),
|
article_text=_optional_str(data.get("articleText")),
|
||||||
|
is_subscriber_only=bool(data.get("isSubscriberOnly", False)),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user