refactor, clear structure

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)

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),

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)

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()

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)

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
 

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