🐛 修复 twitter status info

defs/bilibili.py

@@ -64,7 +64,7 @@ def cut_text(old_str, cut):
             next_str = next_str[1:]
         elif s == "\n":
             str_list.append(next_str[: i - 1])
-            next_str = next_str[i - 1:]
+            next_str = next_str[i - 1 :]
             si = 0
             i = 0
             continue
@@ -173,9 +173,7 @@ async def binfo_image_create(video_info: dict):
     bg_y += title_bg_y
 
     # 简介
-    dynamic = (
-        "该视频没有简介" if video_info["desc"] == "" else video_info["desc"]
-    )
+    dynamic = "该视频没有简介" if video_info["desc"] == "" else video_info["desc"]
     dynamic_font = ImageFont.truetype(
         f"resources{sep}font{sep}sarasa-mono-sc-semibold.ttf", 18
     )

defs/splash.py

@@ -59,6 +59,7 @@ def retry(func):
             logger.warning(f"Sleeping for {e.value}s")
             await asyncio.sleep(e.value + 1)
             return await func(*args, **kwargs)
+
     return wrapper
 
 

defs/twitter_api.py

@@ -0,0 +1,136 @@
+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.twitter.client import twitter_client, TwitterError
+from models.apis.twitter.model import User, Tweet, MediaItem
+
+
+def twitter_link(tweet: Tweet):
+    origin = tweet.retweet_or_quoted
+    button = [
+        [
+            InlineKeyboardButton(
+                text="Source",
+                url=tweet.url,
+            ),
+            InlineKeyboardButton(text="Author", url=tweet.user.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: Tweet):
+    tweet_media_lists = []
+    if tweet.extended_entities:
+        tweet_media_lists.extend(tweet.extended_entities.media)
+    if tweet.retweet and tweet.retweet.extended_entities:
+        tweet_media_lists.extend(tweet.retweet.extended_entities.media)
+    if tweet.quoted and tweet.quoted.extended_entities:
+        tweet_media_lists.extend(tweet.quoted.extended_entities.media)
+    return tweet_media_lists
+
+
+def twitter_media(tweet_media_lists: List[MediaItem], 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
+    verified = "💎" if user.verified else ""
+    protected = "🔒" if user.protected else ""
+    text = (
+        f"<b>Twitter User Info</b>\n\n"
+        f"Name: {verified}{protected}<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.statuses_count} ❤️{user.favourites_count} "
+        f"粉丝 {user.followers_count} 关注 {user.friends_count}"
+    )
+    return text
+
+
+def get_twitter_status(tweet: Tweet):
+    text = tweet.full_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.full_text or "暂 无 内 容"}</code>\n\n'
+            f'{roq.user.one_line} 发表于 {roq.created.strftime("%Y-%m-%d %H:%M:%S")}'
+            f"\n👍 {roq.favorite_count}   🔁 {roq.retweet_count}\n"
+            f'{tweet.user.one_line} 转于 {tweet.created.strftime("%Y-%m-%d %H:%M:%S")}\n'
+            f"👍 {tweet.favorite_count}   🔁 {tweet.retweet_count}"
+        )
+    else:
+        final_text += (
+            f'{tweet.user.one_line} 发表于 {tweet.created.strftime("%Y-%m-%d %H:%M:%S")}'
+            f"\n👍 {tweet.favorite_count}   🔁 {tweet.retweet_count}"
+        )
+    return final_text
+
+
+async def fetch_tweet(tweet_id: int) -> Optional[Tweet]:
+    try:
+        tweet = await twitter_client.tweet_detail(tweet_id)
+        for t in tweet:
+            if t.id_str == str(tweet_id):
+                return t
+        return tweet[0]
+    except TwitterError as e:
+        logs.error(f"Twitter Error: {e}")
+        return None
+
+
+async def fetch_user(username: str) -> Optional[User]:
+    try:
+        user = await twitter_client.user_by_screen_name(username)
+    except TwitterError as e:
+        logs.error(f"Twitter Error: {e}")
+        return None
+    return user

models/apis/splash.py

@@ -53,14 +53,18 @@ class Splash(BaseModel):
             return ""
         if not self.game_short_name:
             return ""
-        return f"https://www.miyoushe.com/{self.game_short_name}/article/{self.article_id}"
+        return (
+            f"https://www.miyoushe.com/{self.game_short_name}/article/{self.article_id}"
+        )
 
     @property
     def text(self) -> str:
-        return f"#id{self.id} \n" \
-               f"ID：<code>{self.id}</code>\n" \
-               f"所属分区：<code>{self.game_id} - {self.game_short_name}</code>\n" \
-               f"开始时间：<code>{self.online_time_str}</code>\n" \
-               f"结束时间：<code>{self.offline_time_str}</code>\n" \
-               f"链接： {self.splash_image}\n" \
-               f"文章链接： {self.article_url}"
+        return (
+            f"#id{self.id} \n"
+            f"ID：<code>{self.id}</code>\n"
+            f"所属分区：<code>{self.game_id} - {self.game_short_name}</code>\n"
+            f"开始时间：<code>{self.online_time_str}</code>\n"
+            f"结束时间：<code>{self.offline_time_str}</code>\n"
+            f"链接： {self.splash_image}\n"
+            f"文章链接： {self.article_url}"
+        )

models/apis/twitter/client.py

@@ -0,0 +1,259 @@
+import base64
+import json
+import random
+from typing import Dict, Any, List
+
+from httpx import AsyncClient, Cookies
+
+from .model import Tweet, User
+
+
+class TwitterError(Exception):
+    def __init__(self, status_code: int, message: str):
+        self.status_code = status_code
+        self.message = message
+
+
+class TwitterClient:
+    headers = {
+        "authorization": "",
+        "x-guest-token": "",
+        "x-twitter-auth-type": "",
+        "x-twitter-client-language": "en",
+        "x-twitter-active-user": "yes",
+        "x-csrf-token": "",
+        "Referer": "https://twitter.com/",
+    }
+    tokens = ["CjulERsDeqhhjSme66ECg:IQWdVyqFxghAtURHGeGiWAsmCAGmdW3WmbEx6Hck"]
+
+    _variables = {
+        "count": 20,
+        "includePromotedContent": False,
+        "withSuperFollowsUserFields": True,
+        "withBirdwatchPivots": False,
+        "withDownvotePerspective": False,
+        "withReactionsMetadata": False,
+        "withReactionsPerspective": False,
+        "withSuperFollowsTweetFields": True,
+        "withClientEventToken": False,
+        "withBirdwatchNotes": False,
+        "withVoice": True,
+        "withV2Timeline": False,
+        "__fs_interactive_text": False,
+        "__fs_dont_mention_me_view_api_enabled": False,
+    }
+    _features = {
+        "tweets": {
+            "rweb_lists_timeline_redesign_enabled": True,
+            "responsive_web_graphql_exclude_directive_enabled": True,
+            "verified_phone_label_enabled": False,
+            "creator_subscriptions_tweet_preview_api_enabled": True,
+            "responsive_web_graphql_timeline_navigation_enabled": True,
+            "responsive_web_graphql_skip_user_profile_image_extensions_enabled": False,
+            "tweetypie_unmention_optimization_enabled": True,
+            "responsive_web_edit_tweet_api_enabled": True,
+            "graphql_is_translatable_rweb_tweet_is_translatable_enabled": True,
+            "view_counts_everywhere_api_enabled": True,
+            "longform_notetweets_consumption_enabled": True,
+            "responsive_web_twitter_article_tweet_consumption_enabled": False,
+            "tweet_awards_web_tipping_enabled": False,
+            "freedom_of_speech_not_reach_fetch_enabled": True,
+            "standardized_nudges_misinfo": True,
+            "tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled": True,
+            "longform_notetweets_rich_text_read_enabled": True,
+            "longform_notetweets_inline_media_enabled": True,
+            "responsive_web_media_download_video_enabled": False,
+            "responsive_web_enhance_cards_enabled": False,
+        },
+        "user_by_screen_name": {
+            "hidden_profile_likes_enabled": False,
+            "responsive_web_graphql_exclude_directive_enabled": True,
+            "verified_phone_label_enabled": False,
+            "subscriptions_verification_info_verified_since_enabled": True,
+            "highlights_tweets_tab_ui_enabled": True,
+            "creator_subscriptions_tweet_preview_api_enabled": True,
+            "responsive_web_graphql_skip_user_profile_image_extensions_enabled": False,
+            "responsive_web_graphql_timeline_navigation_enabled": True,
+        },
+    }
+
+    def __init__(self):
+        self.cookie = Cookies()
+        self.client = AsyncClient(cookies=self.cookie, timeout=60.0)
+        self.inited = False
+
+    async def request(
+        self, url: str, method: str = "GET", headers: Dict[str, str] = None, **kwargs
+    ):
+        headers = headers or self.headers
+        headers = {key: value for key, value in headers.items() if value is not None}
+        return await self.client.request(method, url, headers=headers, **kwargs)
+
+    async def reset_session(self):
+        self.headers[
+            "authorization"
+        ] = f"Basic {base64.b64encode(self.tokens[0].encode()).decode()}"
+        response = await self.request(
+            "https://api.twitter.com/oauth2/token",
+            method="POST",
+            params={"grant_type": "client_credentials"},
+        )
+        access_token = response.json()["access_token"]
+        self.headers["authorization"] = f"Bearer {access_token}"
+        # 生成csrf - token
+        # 32个随机十六进制字符
+        csrf_token = "".join([random.choice("0123456789abcdef") for _ in range(32)])
+        self.cookie.set("ct0", csrf_token, domain=".twitter.com")
+        self.headers["x-csrf-token"] = csrf_token
+        self.headers["x-guest-token"] = ""
+        # 发起初始化请求
+        response = await self.request(
+            "https://api.twitter.com/1.1/guest/activate.json",
+            method="POST",
+        )
+        # 获取guest - token
+        guest_token = response.json()["guest_token"]
+        self.headers["x-guest-token"] = guest_token
+        self.cookie.set("gt", guest_token, domain=".twitter.com")
+        # 发起第二个初始化请求, 获取_twitter_sess
+        await self.request(
+            "https://twitter.com/i/js_inst",
+            method="GET",
+            params={"c_name": "ui_metrics"},
+        )
+
+    async def func(self, url: str, method: str = "GET", **kwargs):
+        if not self.inited:
+            await self.reset_session()
+            self.inited = True
+        response = await self.request(
+            url,
+            method=method,
+            **kwargs,
+        )
+        if response.status_code == 403:
+            await self.reset_session()
+            response = await self.request(
+                url,
+                method=method,
+                **kwargs,
+            )
+        if response.status_code != 200:
+            raise TwitterError(response.status_code, response.text)
+        csrf_token = response.cookies.get("ct0")
+        if csrf_token:
+            self.headers["x-csrf-token"] = csrf_token
+        return response
+
+    async def pagination_tweets(
+        self,
+        endpoint: str,
+        variables: Dict[str, Any],
+        path: List[str],
+    ):
+        _variables = self._variables.copy()
+        _variables.update(variables)
+        data = await self.func(
+            f"https://twitter.com/i/api{endpoint}",
+            params={
+                "variables": json.dumps(_variables),
+                "features": json.dumps(self._features["tweets"]),
+            },
+        )
+        if not path:
+            instructions = data.json()["data"]["user"]["result"]["timeline"][
+                "timeline"
+            ]["instructions"]
+        else:
+            instructions = data.json()["data"]
+            for key in path:
+                _instructions = instructions.get(key)
+                if _instructions is not None:
+                    instructions = _instructions
+                    break
+            instructions = instructions["instructions"]
+        for i in instructions:
+            if i["type"] == "TimelineAddEntries":
+                return i["entries"]
+
+    @staticmethod
+    def gather_legacy_from_data(entries: List[Dict], filters: str = "tweet-"):
+        tweets: List[Tweet] = []
+        filter_entries = []
+        for entry in entries:
+            entry_id: str = entry.get("entryId")
+            if entry_id:
+                if filters == "none":
+                    if entry_id.startswith("tweet-"):
+                        filter_entries.append(entry)
+                    elif entry_id.startswith(
+                        "homeConversation-"
+                    ) or entry_id.startswith("conversationthread-"):
+                        filter_entries.extend(entry["content"]["items"])
+                else:
+                    if entry_id.startswith(filters):
+                        filter_entries.append(entry)
+        for entry in filter_entries:
+            retweet = None
+            quoted = None
+            content = entry.get("content") or entry.get("item")
+            tweet = (
+                content.get("itemContent", {})
+                .get("tweet_results", {})
+                .get("result", {})
+            )
+            if tweet and tweet.get("tweet"):
+                tweet = tweet["tweet"]
+            if tweet:
+                retweet = tweet.get("retweeted_status_result", {}).get("result")
+                quoted = tweet.get("quoted_status_result", {}).get("result")
+            if (not tweet) or (not tweet.get("legacy")):
+                continue
+            if retweet and not retweet.get("legacy"):
+                retweet = None
+            if quoted and not quoted.get("legacy"):
+                quoted = None
+            tweet = Tweet(
+                **tweet["legacy"],
+                **{"user": tweet["core"]["user_results"]["result"]["legacy"]},
+            )
+            if retweet:
+                tweet.retweet = Tweet(
+                    **retweet["legacy"],
+                    **{"user": retweet["core"]["user_results"]["result"]["legacy"]},
+                )
+            if quoted:
+                tweet.quoted = Tweet(
+                    **quoted["legacy"],
+                    **{"user": quoted["core"]["user_results"]["result"]["legacy"]},
+                )
+            tweets.append(tweet)
+        return tweets
+
+    async def tweet_detail(self, tid: int):
+        data = await self.pagination_tweets(
+            "/graphql/3XDB26fBve-MmjHaWTUZxA/TweetDetail",
+            variables={
+                "focalTweetId": tid,
+                "with_rux_injections": False,
+                "withCommunity": True,
+                "withQuickPromoteEligibilityTweetFields": True,
+                "withBirdwatchNotes": False,
+            },
+            path=["threaded_conversation_with_injections"],
+        )
+        return self.gather_legacy_from_data(data, filters="none")
+
+    async def user_by_screen_name(self, username: str):
+        _variables = {"screen_name": username, "withHighlightedLabel": True}
+        data = await self.func(
+            "https://twitter.com/i/api/graphql/oUZZZ8Oddwxs8Cd3iW3UEA/UserByScreenName",
+            params={
+                "variables": json.dumps(_variables),
+                "features": json.dumps(self._features["user_by_screen_name"]),
+            },
+        )
+        return User(**data.json()["data"]["user"]["result"]["legacy"])
+
+
+twitter_client = TwitterClient()

models/apis/twitter/model.py

@@ -0,0 +1,127 @@
+from datetime import datetime, timedelta
+from typing import List, Optional
+
+from pydantic import BaseModel
+
+
+class VideoInfoVariant(BaseModel):
+    bitrate: Optional[int]
+    content_type: str
+    url: str
+
+
+class VideoInfo(BaseModel):
+    variants: List[VideoInfoVariant]
+
+    @property
+    def best_variant(self) -> VideoInfoVariant:
+        variants = [
+            i for i in self.variants if i.content_type.startswith("video/mp4")
+        ] or self.variants
+        return max(variants, key=lambda x: x.bitrate or 0)
+
+
+class MediaItem(BaseModel):
+    display_url: str
+    expanded_url: str
+    id_str: str
+    media_url_https: str
+    """ 真实链接 """
+    type: str
+    """ 类型 """
+    url: str
+    """ 短链接 """
+    video_info: Optional[VideoInfo]
+    """ 视频信息 """
+
+    @property
+    def media_url(self):
+        if self.type == "photo":
+            ext = self.media_url_https.split(".")[-1]
+            return f"{self.media_url_https[:-(len(ext) + 1)]}?format={ext}&name=orig"
+        elif self.type == "video":
+            return self.video_info.best_variant.url
+        return self.media_url_https
+
+
+class ExtendedEntities(BaseModel):
+    media: Optional[List[MediaItem]]
+
+
+class User(BaseModel):
+    created_at: str
+    description: str
+    statuses_count: int = 0
+    favourites_count: int = 0
+    followers_count: int = 0
+    """ 关注者 """
+    friends_count: int = 0
+    """ 正在关注 """
+    location: str
+    name: str
+    profile_banner_url: Optional[str]
+    profile_image_url_https: str
+    screen_name: str
+    verified: bool = False
+    protected: bool = False
+
+    @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 icon(self) -> str:
+        """头像"""
+        return self.profile_image_url_https.replace("_normal.jpg", ".jpg")
+
+    @property
+    def url(self) -> str:
+        """链接"""
+        return f"https://twitter.com/{self.screen_name}"
+
+    @property
+    def one_line(self) -> str:
+        verified = "" if not self.verified else "💎"
+        protected = "" if not self.protected else "🔒"
+        return f'{verified}{protected}<a href="{self.url}">{self.screen_name}</a>'
+
+
+class Tweet(BaseModel):
+    bookmark_count: int
+    """ 书签次数 """
+    created_at: str
+    conversation_id_str: str
+    extended_entities: Optional[ExtendedEntities]
+    favorite_count: int
+    """ 喜欢次数 """
+    full_text: str
+    quote_count: int
+    """ 引用 """
+    reply_count: int
+    """ 回复 """
+    retweet_count: int
+    """ 转推 """
+    id_str: str
+    """ tweet id """
+    user: User
+    retweet: "Tweet" = None
+    quoted: "Tweet" = 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 url(self) -> str:
+        return f"https://twitter.com/{self.user.screen_name}/status/{self.id_str}"
+
+    @property
+    def retweet_or_quoted(self) -> "Tweet":
+        return self.retweet or self.quoted

modules/banme.py

@@ -5,18 +5,20 @@ import random
 from datetime import datetime, timedelta
 from pyrogram import Client, filters
 from pyrogram.enums import ChatMemberStatus
-from pyrogram.types import Message, ChatPermissions, CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton
+from pyrogram.types import (
+    Message,
+    ChatPermissions,
+    CallbackQuery,
+    InlineKeyboardMarkup,
+    InlineKeyboardButton,
+)
 from init import bot
 from scheduler import reply_message
 
 
 def gen_cancel_button(uid: int):
     return InlineKeyboardMarkup(
-        [
-            [
-                InlineKeyboardButton(text="别口球我！", callback_data=f"banme_cancel_{uid}")
-            ]
-        ]
+        [[InlineKeyboardButton(text="别口球我！", callback_data=f"banme_cancel_{uid}")]]
     )
 
 
@@ -37,7 +39,7 @@ async def ban_me_command(client: Client, message: Message):
 
     # 检查bot和用户身份
     if (
-            await client.get_chat_member(message.chat.id, "self")
+        await client.get_chat_member(message.chat.id, "self")
     ).status != ChatMemberStatus.ADMINISTRATOR:
         await message.reply("Bot非群管理员, 无法执行禁言操作QAQ")
         return
@@ -64,12 +66,12 @@ async def ban_me_command(client: Client, message: Message):
         ChatPermissions(),
         datetime.now() + timedelta(seconds=act_time),
     )
-    await reply_message(message, msg, reply_markup=gen_cancel_button(message.from_user.id))
+    await reply_message(
+        message, msg, reply_markup=gen_cancel_button(message.from_user.id)
+    )
 
 
-@bot.on_callback_query(
-    filters.regex(r"^banme_cancel_(\d+)$")
-)
+@bot.on_callback_query(filters.regex(r"^banme_cancel_(\d+)$"))
 async def ban_me_cancel(client: Client, callback_query: CallbackQuery):
     if not callback_query.from_user:
         return

modules/geo.py

@@ -10,9 +10,7 @@ from init import request, bot
 REQUEST_URL = f"https://restapi.amap.com/v3/geocode/geo?key={amap_key}&"
 
 
-@bot.on_message(
-    filters.incoming & filters.command(["geo", f"geo@{bot.me.username}"])
-)
+@bot.on_message(filters.incoming & filters.command(["geo", f"geo@{bot.me.username}"]))
 async def geo_command(_: Client, message: Message):
     if len(message.command) <= 1:
         await message.reply("没有找到要查询的中国 经纬度/地址 ...")

modules/start.py

@@ -1,5 +1,10 @@
 from pyrogram import Client, filters
-from pyrogram.types import Message, InlineQuery, InlineQueryResultArticle, InputTextMessageContent
+from pyrogram.types import (
+    Message,
+    InlineQuery,
+    InlineQueryResultArticle,
+    InputTextMessageContent,
+)
 from defs.button import gen_button, Button
 from init import bot
 

modules/twitter_api.py

@@ -0,0 +1,118 @@
+from urllib.parse import urlparse
+
+from pyrogram import Client, filters, ContinuePropagation
+from pyrogram.enums import MessageEntityType
+from pyrogram.types import Message
+
+from defs.twitter_api import (
+    fetch_tweet,
+    get_twitter_status,
+    twitter_link,
+    twitter_media,
+    fetch_user,
+    get_twitter_user,
+    twitter_user_link,
+    twitter_medias,
+)
+from models.apis.twitter.model import MediaItem
+
+
+async def send_single_tweet(message: Message, media: MediaItem, text: str, button):
+    if media.type == "photo":
+        await message.reply_photo(
+            media.media_url,
+            quote=True,
+            caption=text,
+            reply_markup=button,
+        )
+    elif media.type == "video":
+        await message.reply_video(
+            media.media_url,
+            quote=True,
+            caption=text,
+            reply_markup=button,
+        )
+    elif media.type == "gif":
+        await message.reply_animation(
+            media.media_url,
+            quote=True,
+            caption=text,
+            reply_markup=button,
+        )
+    else:
+        await message.reply_document(
+            media.media_url,
+            quote=True,
+            caption=text,
+            reply_markup=button,
+        )
+
+
+async def process_status(message: Message, status: str):
+    try:
+        status = int(status)
+    except ValueError:
+        return
+    tweet = await fetch_tweet(status)
+    if not tweet:
+        return
+    text = get_twitter_status(tweet)
+    button = twitter_link(tweet)
+    medias = twitter_medias(tweet)
+    if len(medias) == 1:
+        media = medias[0]
+        await send_single_tweet(message, media, text, button)
+        return
+    media_lists = twitter_media(medias, text)
+    if media_lists:
+        await message.reply_media_group(media_lists, quote=True)
+    else:
+        await message.reply(text, quote=True, reply_markup=button)
+
+
+async def process_user(message: Message, username: str):
+    user = await fetch_user(username)
+    if not user:
+        return
+    text = get_twitter_user(user)
+    button = twitter_user_link(user)
+    await message.reply_photo(
+        user.icon,
+        caption=text,
+        quote=True,
+        reply_markup=button,
+    )
+
+
+@Client.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
+        url = urlparse(url)
+        if url.hostname and url.hostname in ["twitter.com", "vxtwitter.com"]:
+            if url.path.find("status") >= 0:
+                status_id = str(
+                    url.path[url.path.find("status") + 7 :].split("/")[0]
+                ).split("?")[0]
+                try:
+                    await process_status(message, status_id)
+                except Exception as e:
+                    print(e)
+            elif url.path == "/":
+                return
+            else:
+                # 解析用户
+                uid = url.path.replace("/", "")
+                try:
+                    await process_user(message, uid)
+                except Exception as e:
+                    print(e)
+    raise ContinuePropagation

scheduler.py

@@ -30,7 +30,7 @@ def add_delete_message_job(message: Message, delete_seconds: int = 60):
         name=f"{message.chat.id}|{message.id}|delete_message",
         args=[message],
         run_date=datetime.datetime.now(pytz.timezone("Asia/Shanghai"))
-                 + datetime.timedelta(seconds=delete_seconds),
+        + datetime.timedelta(seconds=delete_seconds),
         replace_existing=True,
     )
 
@@ -43,12 +43,14 @@ def add_delete_file_job(path: str, delete_seconds: int = 3600):
         name=f"{hash(path)}|delete_file",
         args=[path],
         run_date=datetime.datetime.now(pytz.timezone("Asia/Shanghai"))
-                 + datetime.timedelta(seconds=delete_seconds),
+        + datetime.timedelta(seconds=delete_seconds),
         replace_existing=True,
     )
 
 
-async def reply_message(msg: Message, text: str, delete_origin: bool = True, *args, **kwargs):
+async def reply_message(
+    msg: Message, text: str, delete_origin: bool = True, *args, **kwargs
+):
     reply_msg = await msg.reply(text, *args, **kwargs)
     add_delete_message_job(reply_msg)
     if delete_origin: