✨ twitter parse support nsfw

defs/fix_twitter_api.py

@@ -0,0 +1,128 @@
+from typing import Optional, List
+
+from pyrogram.enums import ParseMode
+from pyrogram.types import (
+    InlineKeyboardMarkup,
+    InlineKeyboardButton,
+    InputMediaPhoto,
+    InputMediaVideo,
+    InputMediaAnimation,
+)
+
+from init import logs
+from models.apis.fxtwitter.client import fix_twitter_client, FixTwitterError
+from models.apis.fxtwitter.model import User, FixTweet, FixTweetMedia
+
+
+def twitter_link(tweet: FixTweet):
+    origin = tweet.retweet_or_quoted
+    button = [
+        [
+            InlineKeyboardButton(
+                text="Source",
+                url=tweet.url,
+            ),
+            InlineKeyboardButton(text="Author", url=tweet.author.url),
+        ]
+    ]
+    if origin:
+        button[0].insert(1, InlineKeyboardButton(text="RSource", url=origin.url))
+    return InlineKeyboardMarkup(button)
+
+
+def twitter_user_link(user: User):
+    return InlineKeyboardMarkup([[InlineKeyboardButton(text="Author", url=user.url)]])
+
+
+def twitter_medias(tweet: FixTweet):
+    tweet_media_lists = []
+    if tweet.medias:
+        tweet_media_lists.extend(tweet.medias)
+    if tweet.retweet_or_quoted:
+        tweet_media_lists.extend(tweet.retweet_or_quoted.medias)
+    return tweet_media_lists
+
+
+def twitter_media(tweet_media_lists: List[FixTweetMedia], text: str):
+    media_lists = []
+    for idx, media in enumerate(tweet_media_lists):
+        if len(media_lists) > 10:
+            break
+        if media.type == "photo":
+            media_lists.append(
+                InputMediaPhoto(
+                    media.media_url,
+                    caption=text if idx == 0 else None,
+                    parse_mode=ParseMode.HTML,
+                )
+            )
+        elif media.type == "gif":
+            media_lists.append(
+                InputMediaAnimation(
+                    media.media_url,
+                    caption=text if idx == 0 else None,
+                    parse_mode=ParseMode.HTML,
+                )
+            )
+        else:
+            media_lists.append(
+                InputMediaVideo(
+                    media.media_url,
+                    caption=text if idx == 0 else None,
+                    parse_mode=ParseMode.HTML,
+                )
+            )
+    return media_lists
+
+
+def get_twitter_user(user: User):
+    user_name = user.name
+    user_username = user.screen_name
+    text = (
+        f"<b>Twitter User Info</b>\n\n"
+        f"Name: <code>{user_name}</code>\n"
+        f'Username: <a href="https://twitter.com/{user_username}">@{user_username}</a>\n'
+        f"Bio: <code>{user.description}</code>\n"
+        f"Joined: <code>{user.created.strftime('%Y-%m-%d %H:%M:%S')}</code>\n"
+        f"📤 {user.tweets} ❤️{user.likes} "
+        f"粉丝 {user.followers} 关注 {user.following}"
+    )
+    return text
+
+
+def get_twitter_status(tweet: FixTweet):
+    text = tweet.text or "暂 无 内 容"
+    text = f"<code>{text}</code>"
+    final_text = "<b>Twitter Status Info</b>\n\n" f"{text}\n\n"
+    if tweet.retweet_or_quoted:
+        roq = tweet.retweet_or_quoted
+        final_text += (
+            f'<code>RT: {roq.text or "暂 无 内 容"}</code>\n\n'
+            f'{roq.author.one_line} 发表于 {roq.created.strftime("%Y-%m-%d %H:%M:%S")}'
+            f"\n👁 {roq.views}  👍 {roq.likes}   🔁 {roq.retweets}\n"
+            f'{tweet.author.one_line} 转于 {tweet.created.strftime("%Y-%m-%d %H:%M:%S")}\n'
+            f"👁 {tweet.views}  👍 {tweet.likes}   🔁 {tweet.retweets}"
+        )
+    else:
+        final_text += (
+            f'{tweet.author.one_line} 发表于 {tweet.created.strftime("%Y-%m-%d %H:%M:%S")}'
+            f"\n👁 {tweet.views}  👍 {tweet.likes}   🔁 {tweet.retweets}"
+        )
+    return final_text
+
+
+async def fetch_tweet(tweet_id: int) -> Optional[FixTweet]:
+    try:
+        return await fix_twitter_client.tweet_detail(tweet_id)
+    except FixTwitterError as e:
+        logs.error(f"Twitter Error: {e}")
+        return None
+
+
+async def fetch_user(username: str) -> Optional[User]:
+    try:
+        user = await fix_twitter_client.user_by_screen_name(username)
+    except FixTwitterError as e:
+        logs.error(f"Twitter Error: {e}")
+        return None
+    return user

models/apis/fxtwitter/client.py

@@ -0,0 +1,52 @@
+from typing import Dict
+
+from httpx import AsyncClient
+
+from .model import FixTweet, FixTweetMedia, User
+
+
+class FixTwitterError(Exception):
+    def __init__(self, status_code: int, message: str):
+        self.status_code = status_code
+        self.message = message
+
+
+class FixTwitterClient:
+    def __init__(self):
+        self.client = AsyncClient(timeout=60.0)
+        self.url = "https://api.fxtwitter.com/"
+
+    @staticmethod
+    def gen_tweet(data: Dict) -> FixTweet:
+        tweet = FixTweet(**data)
+        if medias := data.get("media"):
+            if all_media := medias.get("all"):
+                tweet.medias = [FixTweetMedia(**i) for i in all_media]
+        if quote := data.get("quote"):
+            tweet.quoted = FixTwitterClient.gen_tweet(quote)
+        if retweet := data.get("retweet"):
+            tweet.retweeted = FixTwitterClient.gen_tweet(retweet)
+        return tweet
+
+    async def tweet_detail(self, tid: int) -> FixTweet:
+        url = f"{self.url}X/status/{tid}"
+        response = await self.client.get(url)
+        if response.status_code != 200:
+            raise FixTwitterError(response.status_code, response.text)
+        data = response.json()
+        if data.get("code", 200) != 200:
+            raise FixTwitterError(data.get("code"), data.get("message"))
+        return self.gen_tweet(data["tweet"])
+
+    async def user_by_screen_name(self, username: str) -> User:
+        url = f"{self.url}{username}"
+        response = await self.client.get(url)
+        if response.status_code != 200:
+            raise FixTwitterError(response.status_code, response.text)
+        data = response.json()
+        if data.get("code", 200) != 200:
+            raise FixTwitterError(data.get("code"), data.get("message"))
+        return User(**data["user"])
+
+
+fix_twitter_client = FixTwitterClient()

models/apis/fxtwitter/model.py

@@ -0,0 +1,76 @@
+from datetime import datetime, timedelta
+from typing import List, Optional
+
+from pydantic import BaseModel
+
+
+class User(BaseModel):
+    id: int
+    name: str
+    screen_name: str
+    avatar_url: str = ""
+    banner_url: str = ""
+    description: str = ""
+    location: str = ""
+    url: str
+    followers: int = 0
+    following: int = 0
+    joined: str
+    tweets: int = 0
+    likes: int = 0
+
+    @property
+    def created(self) -> datetime:
+        """入推时间"""
+        # 'Fri Jun 30 12:08:56 +0000 2023'
+        return datetime.strptime(self.joined, "%a %b %d %H:%M:%S %z %Y") + timedelta(
+            hours=8
+        )
+
+    @property
+    def icon(self) -> str:
+        """头像"""
+        return self.avatar_url.replace("_normal.jpg", ".jpg")
+
+    @property
+    def one_line(self) -> str:
+        return f'<a href="{self.url}">{self.screen_name}</a>'
+
+
+class FixTweetMedia(BaseModel):
+    type: str
+    url: str
+    width: int = 0
+    height: int = 0
+    altText: str = ""
+
+
+class FixTweet(BaseModel):
+    url: str
+    id: int
+    text: str
+    replies: int
+    """ 回复 """
+    retweets: int
+    """ 转推 """
+    likes: int
+    """ 喜欢 """
+    created_at: str
+    views: int
+    """ 阅读次数 """
+    source: str
+    medias: Optional[List[FixTweetMedia]] = None
+    author: User
+    retweeted: "FixTweet" = None
+    quoted: "FixTweet" = None
+
+    @property
+    def created(self) -> datetime:
+        # 'Fri Jun 30 12:08:56 +0000 2023'
+        return datetime.strptime(
+            self.created_at, "%a %b %d %H:%M:%S %z %Y"
+        ) + timedelta(hours=8)
+
+    @property
+    def retweet_or_quoted(self) -> "FixTweet":
+        return self.retweeted or self.quoted

modules/twitter_api.py

@@ -4,7 +4,7 @@ from pyrogram import Client, filters, ContinuePropagation
 from pyrogram.enums import MessageEntityType
 from pyrogram.types import Message
 
-from defs.twitter_api import (
+from defs.fix_twitter_api import (
     fetch_tweet,
     get_twitter_status,
     twitter_link,
@@ -85,18 +85,7 @@ async def process_user(message: Message, username: str):
     )
 
 
-@bot.on_message(filters.incoming & filters.text & filters.regex(r"twitter.com/"))
-async def twitter_share(_: Client, message: Message):
-    if not message.text:
-        return
-    for num in range(len(message.entities)):
-        entity = message.entities[num]
-        if entity.type == MessageEntityType.URL:
-            url = message.text[entity.offset : entity.offset + entity.length]
-        elif entity.type == MessageEntityType.TEXT_LINK:
-            url = entity.url
-        else:
-            continue
+async def process_url(url: str, message: Message):
     url = urlparse(url)
     if url.hostname and url.hostname in ["twitter.com", "vxtwitter.com"]:
         if url.path.find("status") >= 0:
@@ -116,4 +105,19 @@ async def twitter_share(_: Client, message: Message):
                 await process_user(message, uid)
             except Exception as e:
                 print(e)
+
+
+@bot.on_message(filters.incoming & filters.text & filters.regex(r"twitter.com/"))
+async def twitter_share(_: Client, message: Message):
+    if not message.text:
+        return
+    for num in range(len(message.entities)):
+        entity = message.entities[num]
+        if entity.type == MessageEntityType.URL:
+            url = message.text[entity.offset : entity.offset + entity.length]
+        elif entity.type == MessageEntityType.TEXT_LINK:
+            url = entity.url
+        else:
+            continue
+        await process_url(url, message)
     raise ContinuePropagation