From 7360f6b22e69e58e0f1b1b5473787afab4cd9f9d Mon Sep 17 00:00:00 2001 From: m3taversal Date: Mon, 23 Mar 2026 15:17:11 +0000 Subject: [PATCH] epimetheus: direct tweet lookup via /tweets?tweet_ids= endpoint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Primary path: GET /twitter/tweets?tweet_ids={id} — works for any tweet, any age, returns full content. Replaces the fragile from:username search pagination fallback. Fallback: article endpoint for X long-form articles. Last resort: placeholder with [Could not fetch] message. Pentagon-Agent: Epimetheus <3D35839A-7722-4740-B93D-51157F7D5E70> --- telegram/x_search.py | 65 ++++++++++++++++++++------------------------ 1 file changed, 29 insertions(+), 36 deletions(-) diff --git a/telegram/x_search.py b/telegram/x_search.py index d8a29c7..40ae43c 100644 --- a/telegram/x_search.py +++ b/telegram/x_search.py @@ -179,7 +179,34 @@ async def fetch_tweet_by_url(url: str) -> dict | None: try: async with aiohttp.ClientSession() as session: - # First try article endpoint + # Primary: direct tweet lookup by ID (works for any tweet, any age) + async with session.get( + "https://api.twitterapi.io/twitter/tweets", + params={"tweet_ids": tweet_id}, + headers={"X-API-Key": key}, + timeout=aiohttp.ClientTimeout(total=10), + ) as resp: + if resp.status == 200: + data = await resp.json() + tweets = data.get("tweets", []) + if tweets: + tweet = tweets[0] + author_data = tweet.get("author", {}) + return { + "text": tweet.get("text", ""), + "url": url, + "author": author_data.get("userName", username), + "author_name": author_data.get("name", ""), + "author_followers": author_data.get("followers", 0), + "engagement": (tweet.get("likeCount", 0) or 0) + (tweet.get("retweetCount", 0) or 0), + "likes": tweet.get("likeCount", 0), + "retweets": tweet.get("retweetCount", 0), + "views": tweet.get("viewCount", 0), + "tweet_date": tweet.get("createdAt", ""), + "is_article": False, + } + + # Fallback: try article endpoint (for X long-form articles) async with session.get( "https://api.twitterapi.io/twitter/article", params={"tweet_id": tweet_id}, @@ -202,41 +229,7 @@ async def fetch_tweet_by_url(url: str) -> dict | None: "title": article.get("title", ""), } - # Fallback: search from:username and match by ID - # NOTE: fallback only finds recent tweets (~7 days). Older tweets fail. (Ganymede) - # Try with cursor pagination to search deeper - cursor = "" - for page in range(3): # Up to 3 pages - params = {"query": f"from:{username}", "queryType": "Latest"} - if cursor: - params["cursor"] = cursor - async with session.get( - API_URL, - params=params, - headers={"X-API-Key": key}, - timeout=aiohttp.ClientTimeout(total=10), - ) as resp: - if resp.status >= 400: - break - data = await resp.json() - for tweet in data.get("tweets", []): - if str(tweet.get("id")) == tweet_id: - author = tweet.get("author", {}) - return { - "text": tweet.get("text", ""), - "url": url, - "author": author.get("userName", username), - "author_name": author.get("name", ""), - "author_followers": author.get("followers", 0), - "engagement": (tweet.get("likeCount", 0) or 0) + (tweet.get("retweetCount", 0) or 0), - "tweet_date": tweet.get("createdAt", ""), - "is_article": False, - } - cursor = data.get("next_cursor", "") - if not cursor: - break - - # If still not found, return placeholder (Ganymede: surface failure) + # Both failed — return placeholder (Ganymede: surface failure) return { "text": f"[Could not fetch tweet content from @{username}]", "url": url,