2024-10-28 09:02:25+04:00

app/bot.py

@@ -3,14 +3,14 @@
 
 import logging
 import re
-from pydantic import ValidationError  # For handling validation errors
-from .models import URLMessage  # Import the URLMessage model
 
 from aiogram import Bot, Dispatcher, types
 from aiogram.utils import executor
+from pydantic import ValidationError  # For handling validation errors
 
 from app.config import settings
 
+from .models import URLMessage  # Import the URLMessage model
 from .url_processing import process_url_request
 
 bot = Bot(token=settings.BOT_TOKEN)
@@ -59,10 +59,14 @@ async def handle_message(message: types.Message):
     for url in urls:
         try:
             # Create a URLMessage instance to validate the URL and chat type
-            url_message = URLMessage(url=url, is_group_chat=message.chat.type in ["group", "supergroup"])
+            url_message = URLMessage(
+                url=url, is_group_chat=message.chat.type in ["group", "supergroup"]
+            )
 
             # Pass validated URL and chat type to the process_url_request function
-            result = await process_url_request(url_message.url, url_message.is_group_chat)
+            result = await process_url_request(
+                url_message.url, url_message.is_group_chat
+            )
 
             # If result is None, do not reply
             if result is not None:
@@ -72,6 +76,7 @@ async def handle_message(message: types.Message):
             # Reply with validation error if URL is invalid
             await message.reply(f"Invalid URL provided: {e}")
 
+
 def start_bot():
     dp.register_message_handler(start, commands="start")
     executor.start_polling(dp, skip_updates=False)

app/download.py

@@ -3,15 +3,20 @@
 
 import logging
 import os
+
 import yt_dlp
+
 from app.config import settings
 
 logger = logging.getLogger(__name__)
 
+
 class UnsupportedUrlError(Exception):
     """Custom exception for unsupported URLs"""
+
     pass
 
+
 async def yt_dlp_download(url: str) -> str:
     video_path = os.path.join(settings.CACHE_DIR, f"{sanitize_subfolder_name(url)}.mp4")
 
@@ -33,15 +38,22 @@ async def yt_dlp_download(url: str) -> str:
             raise UnsupportedUrlError(f"Unsupported URL: {url}")
         else:
             logger.error(f"DownloadError for URL: {url} - {str(e)}")
-            raise RuntimeError(f"Failed to download video from URL: {url}. Check if the URL is correct and accessible.") from e
+            raise RuntimeError(
+                f"Failed to download video from URL: {url}. Check if the URL is correct and accessible."
+            ) from e
 
     except yt_dlp.PostProcessingError as e:
         logger.error(f"PostProcessingError for URL: {url} - {str(e)}")
-        raise RuntimeError(f"An error occurred while processing the video file for URL: {url}.") from e
+        raise RuntimeError(
+            f"An error occurred while processing the video file for URL: {url}."
+        ) from e
 
     except Exception as e:
         logger.error(f"Unexpected error for URL: {url} - {str(e)}")
-        raise RuntimeError(f"An unexpected error occurred while processing the URL: {url}. Please try again later.") from e
+        raise RuntimeError(
+            f"An unexpected error occurred while processing the URL: {url}. Please try again later."
+        ) from e
+
 
 def sanitize_subfolder_name(url: str) -> str:
     return "".join(c if c.isalnum() else "_" for c in url)

app/models.py

@@ -1,6 +1,7 @@
 # models.py
 from pydantic import BaseModel, HttpUrl
 
+
 class URLMessage(BaseModel):
     url: HttpUrl
     is_group_chat: bool = False

app/url_processing.py

@@ -1,16 +1,19 @@
 # url_processing.py
 
-import re
 import logging
 import os
-import requests
+import re
 from urllib.parse import urlparse, urlunparse
 
+import requests
+
 from app.config import settings
-from .download import yt_dlp_download, UnsupportedUrlError
+
+from .download import UnsupportedUrlError, yt_dlp_download
 
 logger = logging.getLogger(__name__)
 
+
 def follow_redirects(url: str, timeout=10) -> str:
     try:
         response = requests.head(url, allow_redirects=True, timeout=timeout)
@@ -25,6 +28,7 @@ def follow_redirects(url: str, timeout=10) -> str:
         logger.warning(f"Timeout for URL: {url}")
         return url
 
+
 rewrite_map = {
     r"^https://(www\.)?tiktok\.com": "https://tfxktok.com",
     r"^https://(www\.)?twitter\.com": "https://www.fxtwitter.com",
@@ -33,24 +37,28 @@ def follow_redirects(url: str, timeout=10) -> str:
     r"^https://(www\.)?instagram\.com/reel/": "https://www.ddinstagram.com/reel/",
 }
 
+
 async def process_url_request(url: str, is_group_chat: bool = False) -> str:
     # Ensure url is a string
     url = str(url)
+    final_url = url  # Initialize final_url with url for error handling
 
     # Check if the URL is a YouTube URL and apply the custom transformation
     youtube_patterns = [
-        (r"^https://www\.youtube\.com/watch\?v=([a-zA-Z0-9_-]+)", r"https://www.yfxtube.com/watch?v=\1"),
+        (
+            r"^https://www\.youtube\.com/watch\?v=([a-zA-Z0-9_-]+)",
+            r"https://www.yfxtube.com/watch?v=\1",
+        ),
         (r"^https://youtu\.be/([a-zA-Z0-9_-]+)", r"https://fxyoutu.be/\1"),
     ]
 
     for pattern, replacement in youtube_patterns:
         if re.match(pattern, url):
-            # Replace the URL according to the custom rule and return without downloading
             modified_url = re.sub(pattern, replacement, url)
             return (
                 "YouTube video cannot be downloaded, but here’s an alternative link:"
                 + f"\n\n[📎 Modified URL]({modified_url})"
-                + f"\n\n[📎 Original]({url})"  # Using `url` directly here
+                + f"\n\n[📎 Original]({url})"
             )
 
     try: