Commit 46ee63e

@@ -27,7 +27,7 @@ async def preview_twitter(
     message: Message,
     url: str = "",
     db_key: str = "",
-    platform: str = "",
+    platform: str = "x",
     twitter_provider: str = PROVIDER.TWITTER,
     twitter_comments_provider: str = PROVIDER.TWITTER_COMMENTS,
     *,
@@ -41,16 +41,11 @@ async def preview_twitter(
         message (Message): The trigger message object.
         url (str, optional): The twitter link.
         db_key (str, optional): The cache key.
-        platform (str): The domain of the link: twitter, x, fxtwitter, fixupx
+        platform (str): The social media platform.
         twitter_provider (str): The extractor to use: fxtwitter or tikhub.
         twitter_comments_provider (str, optional): The twitter comments extractor: "tikhub" or "false".
         fallback (bool, optional): Fallback to other bots. Defaults to True.
-
-    If skip_fxtwitter is set to True, and the domain is fxtwitter or fixupx, this function is skipped.
     """
-    if true(kwargs.get("skip_fxtwitter")) and platform in ["fxtwitter", "fixupx"]:
-        return
-    platform = "x"  # set to x for all twitter links
     if kwargs.get("show_progress") and "progress" not in kwargs:
         res = await send2tg(client, message, texts=f"🔗正在解析推特链接\n{url}", **kwargs)
         kwargs["progress"] = res[0]
@@ -91,12 +86,25 @@ async def preview_twitter(
             succ = True
         except Exception as e:
             logger.warning(f"Twitter API [fxtwitter] failed: {e}")
+
+    if not succ and "vxtwitter" in twitter_provider:  # try vxtwitter
+        try:
+            this_info = await get_tweet_info_via_vxtwitter(url=url)
+            if not this_info:
+                error = "❌[VxTwitter]推特解析失败"
+                await modify_progress(text=error, **kwargs)
+                raise APIError(error)  # noqa: TRY301
+            master_info = await get_tweet_info_via_vxtwitter(handle=this_info["replying_to_user"], post_id=this_info["replying_post_id"]) if this_info["has_master"] else {}
+            quote_info = await get_tweet_info_via_vxtwitter(quote_info=this_info["quote_info"]) if this_info["has_quote"] else {}
+            succ = True
+        except Exception as e:
+            logger.warning(f"Twitter API [vxtwitter] failed: {e}")
+
     if not succ:
         if fallback:
             await modify_progress(text="❌推特解析失败, 尝试第三方Bot...", **kwargs)
             await send_to_social_media_bridge(client, message, url, platform, **kwargs)
         return
-
     media = []
     media_ids = set()  # deduplicate media
     master_media = []
@@ -214,7 +222,7 @@ async def preview_twitter(
 
 
 @cache.memoize(ttl=30)
-async def get_tweet_info_via_tikhub(url: str = "", post_id: str = "", quote_info: dict | None = None, **kwargs) -> dict:  # type: ignore
+async def get_tweet_info_via_tikhub(url: str = "", post_id: str = "", quote_info: dict | None = None, **kwargs) -> dict:
     """Get a single tweet info.
 
     url: https://x.com/{handle}/status/{post_id}
@@ -306,7 +314,7 @@ async def get_tweet_info_via_tikhub(url: str = "", post_id: str = "", quote_info
 
 
 @cache.memoize(ttl=30)
-async def get_tweet_info_via_fxtwitter(url: str = "", handle: str = "", post_id: str = "", quote_info: dict | None = None) -> dict:  # type: ignore
+async def get_tweet_info_via_fxtwitter(url: str = "", handle: str = "", post_id: str = "", quote_info: dict | None = None) -> dict:
     """Get a single tweet info.
 
     url: https://x.com/{handle}/status/{post_id}
@@ -317,20 +325,20 @@ async def get_tweet_info_via_fxtwitter(url: str = "", handle: str = "", post_id:
     data = {}
     if quote_info:
         data = copy.deepcopy(quote_info)
-        handle = data.get("author", {}).get("name", "")
+        handle = glom(data, "author.name", default="")
         post_id = data.get("id", "")
     else:
         api_url = f"{API.FXTWITTER}/{handle}/status/{post_id}"
         logger.info(f"Twitter preview via fxtwitter: {api_url}")
         headers = {"user-agent": TELEGRAM_UA}
-        resp = await hx_req(api_url, headers=headers)
-        if resp.get("hx_error") or str(resp.get("tweet", {}).get("id")) != str(post_id):
+        resp = await hx_req(api_url, headers=headers, proxy=PROXY.TWITTER)
+        if resp.get("hx_error") or str(glom(resp, "tweet.id", default="")) != str(post_id):
             logger.error("Failed to get tweet info via fxtwitter")
             return {}
         data: dict = resp["tweet"]
 
-    info = {"handle": data.get("author", {}).get("screen_name", handle), "post_id": data.get("id", post_id)}
-    media = data.get("media", {}).get("all", [])
+    info = {"handle": glom(data, "author.screen_name", default=handle), "post_id": data.get("id", post_id)}
+    media = glom(data, "media.all", default=[])
     for x in media:
         if x.get("type", "") == "video" and "mp4" not in x.get("format", ""):  # this is a m3u8 url, choose mp4 instead
             m3u8_url = x.get("url", "")
@@ -341,7 +349,7 @@ async def get_tweet_info_via_fxtwitter(url: str = "", handle: str = "", post_id:
         x["id"] = x["url"]  # record media "id" for de-duplication
 
     info["media"] = media
-    info["author"] = data.get("author", {}).get("name", "")
+    info["author"] = glom(data, "author.name", default="")
     if ts := data.get("created_timestamp", ""):
         dt = datetime.fromtimestamp(round(float(ts)), tz=UTC).astimezone(ZoneInfo(TZ))
         info["time"] = f"{dt:%Y-%m-%d %H:%M:%S}"
@@ -355,6 +363,51 @@ async def get_tweet_info_via_fxtwitter(url: str = "", handle: str = "", post_id:
     return info
 
 
+@cache.memoize(ttl=30)
+async def get_tweet_info_via_vxtwitter(url: str = "", handle: str = "", post_id: str = "", quote_info: dict | None = None) -> dict:
+    """Get a single tweet info.
+
+    url: https://x.com/{handle}/status/{post_id}
+    """
+    if not handle or not post_id:
+        handle = url.split("/")[-3]
+        post_id = url.split("/")[-1]
+    data = {}
+    if quote_info:
+        data = copy.deepcopy(quote_info)
+        handle = data.get("user_screen_name", "")
+        post_id = data.get("tweetID", "")
+    else:
+        api_url = f"{API.VXTWITTER}/Twitter/status/{post_id}"
+        logger.info(f"Twitter preview via vxtwitter: {api_url}")
+        headers = {"user-agent": TELEGRAM_UA}
+        data = await hx_req(api_url, headers=headers, proxy=PROXY.TWITTER, check_kv={"user_screen_name": handle})
+        if data.get("hx_error"):
+            logger.error("Failed to get tweet info via vxtwitter")
+            return {}
+        if data.get("retweet"):
+            data = data["retweet"]
+    info = {"handle": glom(data, "screen_name", default=handle), "post_id": data.get("tweetID", post_id)}
+    media = data.get("media_extended", [])
+    for x in media:
+        x["id"] = x.get("url", "")  # record media "id" for de-duplication
+        if x.get("type", "") == "image":  # change `image` -> `photo`
+            x["type"] = "photo"
+    info["media"] = media
+    info["author"] = data.get("user_name", f"@{info['handle']}")
+    if ts := data.get("date_epoch", 0):
+        dt = datetime.fromtimestamp(round(float(ts)), tz=UTC).astimezone(ZoneInfo(TZ))
+        info["time"] = f"{dt:%Y-%m-%d %H:%M:%S}"
+    info["texts"] = data.get("text", "")
+    info["device"] = data.get("source", "").removeprefix("Twitter for").removeprefix("Twitter").removesuffix("App").strip().removesuffix("Web")
+    info["replying_to_user"] = data.get("replyingTo", "")
+    info["replying_post_id"] = data.get("replyingToID", "")
+    info["quote_info"] = data.get("qrt", {})
+    info["has_master"] = bool(data.get("replyingTo"))
+    info["has_quote"] = bool(data.get("qrt"))
+    return info
+
+
 def remove_twitter_suffix(text: str, post_id: str = "", *, same_id_only: bool = True) -> str:
     """Remove twitter link suffix.
 

@@ -99,6 +99,7 @@ class PREFIX:
 
 class API:
     FXTWITTER = os.getenv("FXTWITTER_API", "https://api.fxtwitter.com")
+    VXTWITTER = os.getenv("VXTWITTER_API", "https://api.vxtwitter.com")
     DDINSTAGRAM = os.getenv("DDINSTAGRAM_API", "https://www.ddinstagram.com")
     TIKHUB = os.getenv("TIKHUB", "https://api.tikhub.io")
     TIKHUB_FREE = os.getenv("TIKHUB_FREE", "https://api.douyin.wtf")
@@ -133,7 +134,7 @@ class DANMU:
 class PROVIDER:  # default API provider
     DOUYIN = os.getenv("DOUYIN_PROVIDER", "free-tikhub-bridge").lower()  # free or tikhub
     DOUYIN_COMMENTS = os.getenv("DOUYIN_COMMENTS_PROVIDER", "free-tikhub").lower()  # free or tikhub or a false value (0, false, none, null, etc.)
-    TWITTER = os.getenv("TWITTER_PROVIDER", "tikhub-fxtwitter").lower()  # tikhub or fxtwitter
+    TWITTER = os.getenv("TWITTER_PROVIDER", "tikhub-vxtwitter-fxtwitter").lower()
     TWITTER_COMMENTS = os.getenv("TWITTER_COMMENTS_PROVIDER", "tikhub").lower()  # tikhub or a false value (0, false, none, null, etc.)
     INSTAGRAM = os.getenv("INSTAGRAM_PROVIDER", "tikhub-ddinstagram-bridge").lower()  # tikhub, ddinstagram, bridge
     INSTAGRAM_COMMENTS = os.getenv("INSTAGRAM_COMMENTS_PROVIDER", "tikhub").lower()  # tikhub or a false value (0, false, none, null, etc.)

@@ -300,12 +300,11 @@ async def match_social_media_link(text: str, *, flatten_first: bool = True) -> d
     # https://twitter.com/taylorswift13/status/1794805688696275131
     # https://fixupx.com/taylorswift13/status/1794805688696275131
     # https://fxtwitter.com/taylorswift13/status/1794805688696275131
-    if matched := re.search(r"(https?://)?(:?twitter|x|fxtwitter|fixupx)\.com\/(\w+)\/status/(\d+)", text):
-        platform = matched.group(2)
+    if matched := re.search(r"(https?://)?(:?twitter|x|fxtwitter|fixupx|vxtwitter)\.com\/(\w+)\/status/(\d+)", text):
         handle = matched.group(3)
         post_id = matched.group(4)
         url = f"https://x.com/{handle}/status/{post_id}"
-        return {"platform": platform, "handle": handle, "post_id": post_id, "url": url, "db_key": bare_url(url)}
+        return {"platform": "x", "handle": handle, "post_id": post_id, "url": url, "db_key": bare_url(url)}
 
     # weibo video first, then weibo post
     # https://video.weibo.com/show?fid=1034:5123779299311660