From c1c1d42892da58966e580c473abcdc18daa863d3 Mon Sep 17 00:00:00 2001 From: YoursFunny Date: Sun, 23 Jun 2024 22:03:12 +0800 Subject: [PATCH] refactor, clear structure --- common.py | 19 ++----------------- main.py | 10 ++++++---- utils/logger.py | 11 +++++++++++ utils/net.py | 5 ----- utils/regex.py | 6 ++++++ utils/telegram.py | 4 ++-- utils/tweet.py | 9 +++++---- 7 files changed, 32 insertions(+), 32 deletions(-) create mode 100644 utils/logger.py create mode 100644 utils/regex.py diff --git a/common.py b/common.py index 92fa417..22fd606 100644 --- a/common.py +++ b/common.py @@ -1,9 +1,8 @@ -import logging import os -import re try: - import uvloop, asyncio + import uvloop + import asyncio asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) except ImportError: @@ -20,17 +19,3 @@ if WEBHOOK: WEBHOOK_KEY = os.getenv("WEBHOOK_KEY", "cert/private.key") WEBHOOK_CERT = os.getenv("WEBHOOK_CERT", "cert/cert.pem") WEBHOOK_SECRET_TOKEN = os.getenv("WEBHOOK_SECRET_TOKEN") - -x_url_regex = re.compile(r"^(?:https?://)(?:www\.|mobile\.|)(?:x|twitter|fixvx|vxtwitter)\.com/(.+)/status/(\d+)") -x_media_regex = re.compile(r"^(?:https?://)(pbs|video)\.twimg\.com/(.*)") -x_tco_regex = re.compile(r"(?:https?://)t\.co/.+$", re.M) -message_url_regex = re.compile(r"\[.+]", re.S) - -logging.basicConfig( - level=os.getenv("LOG_LEVEL", "WARNING"), - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' -) - - -def get_logger(name: str) -> logging.Logger: - return logging.getLogger(name) diff --git a/main.py b/main.py index ad8ddfe..1fc5e2c 100644 --- a/main.py +++ b/main.py @@ -10,13 +10,15 @@ from telegram.ext import (ApplicationBuilder, CallbackQueryHandler, CommandHandl InlineQueryHandler, MessageHandler, PicklePersistence, filters) import common +import utils.regex as regex +from utils.logger import get_logger from utils.telegram import Telegram if TYPE_CHECKING: from telegram import Chat, Message, Update from telegram.ext import Application, ContextTypes -logger = common.get_logger(__name__) +logger = get_logger(__name__) def send_action(action): @@ -105,7 +107,7 @@ async def edit_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No )) else: update_text = html.escape(update.message.text) - match = common.message_url_regex.search(update_text) + match = regex.message_url.search(update_text) if match: match = match.span() update_text = update_text[:match[0]] + message_url.format( @@ -234,8 +236,8 @@ def main(): user_filter.add_user_ids(common.ADMIN) handlers = [ - InlineQueryHandler(inline_query, common.x_url_regex), - MessageHandler(filters.Regex(common.x_url_regex) & filters.ChatType.PRIVATE, url_media), + InlineQueryHandler(inline_query, regex.x_url), + MessageHandler(filters.Regex(regex.x_url) & filters.ChatType.PRIVATE, url_media), CommandHandler("set_forward_channel", cmd_set_forward_channel), CommandHandler("remove_forward_channel", cmd_remove_forward_channel), CommandHandler("edit_before_forward", cmd_edit_before_forward), diff --git a/utils/logger.py b/utils/logger.py new file mode 100644 index 0000000..6c60327 --- /dev/null +++ b/utils/logger.py @@ -0,0 +1,11 @@ +import logging +import os + +logging.basicConfig( + level=os.getenv("LOG_LEVEL", "WARNING"), + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) + + +def get_logger(name: str) -> logging.Logger: + return logging.getLogger(name) diff --git a/utils/net.py b/utils/net.py index 84ab0b3..623c6ed 100644 --- a/utils/net.py +++ b/utils/net.py @@ -2,10 +2,6 @@ from __future__ import annotations from httpx import AsyncClient -from common import get_logger - -logger = get_logger(__name__) - def create_client() -> AsyncClient: return AsyncClient(http2=True) @@ -16,7 +12,6 @@ async def close_client(_client: AsyncClient) -> None: async def fetch_json(_client: AsyncClient, url: str) -> dict: - logger.info(f"Fetching {url}") response = await _client.get(url) assert response.is_success, f"Failed to fetch {url}, status code {response.status_code}" return response.json() diff --git a/utils/regex.py b/utils/regex.py new file mode 100644 index 0000000..a5b3144 --- /dev/null +++ b/utils/regex.py @@ -0,0 +1,6 @@ +import re + +x_url = re.compile(r"^(?:https?://)(?:www\.|mobile\.|)(?:x|twitter|fixvx|vxtwitter)\.com/(.+)/status/(\d+)") +x_media_url = re.compile(r"^(?:https?://)(pbs|video)\.twimg\.com/(.*)") +x_tco_url = re.compile(r"(?:https?://)t\.co/.+$", re.M) +message_url = re.compile(r"\[.+]", re.S) diff --git a/utils/telegram.py b/utils/telegram.py index 76a201f..e00ec54 100644 --- a/utils/telegram.py +++ b/utils/telegram.py @@ -1,7 +1,7 @@ from __future__ import annotations -from common import x_url_regex from .net import NetClient +from .regex import x_url from .tweet import TelegramTweet @@ -10,7 +10,7 @@ class Telegram(NetClient): self._url = url async def __aenter__(self): - if x_url_regex.match(self._url): + if x_url.match(self._url): async with TelegramTweet(self._url) as tweet: return tweet diff --git a/utils/tweet.py b/utils/tweet.py index 325381d..c41551e 100644 --- a/utils/tweet.py +++ b/utils/tweet.py @@ -8,8 +8,9 @@ from uuid import uuid4 from telegram import (InlineQueryResultMpeg4Gif, InlineQueryResultPhoto, InlineQueryResultVideo, InputMediaPhoto, InputMediaVideo) -from common import get_logger, x_media_regex, x_tco_regex, x_url_regex +from .logger import get_logger from .net import NetClient +from .regex import x_media_url, x_tco_url, x_url if TYPE_CHECKING: from .types import TweetInfo, TypeInlineQueryResult, TypeMessageMediaResult @@ -38,7 +39,7 @@ class TweetMedia: @cached_property def _uri(self) -> str | None: - if match := x_media_regex.match(self._url): + if match := x_media_url.match(self._url): return match.group(2).removesuffix('.jpg').removesuffix('.png') return None @@ -140,14 +141,14 @@ class ProcessTweet: pass async def _fetch_tweet(self) -> TweetInfo: - match = x_url_regex.match(self._url) + match = x_url.match(self._url) assert match, f"Invalid URL: {self._url}" auther_id, tweet_id = match.groups() return await NetClient.fetch_json(vx_api_url.format(auther_id, tweet_id)) @property def _tweet_text(self) -> str: - match = x_tco_regex.search(self._tweet['text']) + match = x_tco_url.search(self._tweet['text']) return self._tweet['text'][:match.start()].strip(" ") if match else self._tweet['text'] @property