diff --git a/app/bot.py b/app/bot.py index 6215ee2..1f43767 100644 --- a/app/bot.py +++ b/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) diff --git a/app/download.py b/app/download.py index c6c19f1..e83a0ef 100644 --- a/app/download.py +++ b/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) diff --git a/app/models.py b/app/models.py index 9e7ff9f..554fb56 100644 --- a/app/models.py +++ b/app/models.py @@ -1,6 +1,7 @@ # models.py from pydantic import BaseModel, HttpUrl + class URLMessage(BaseModel): url: HttpUrl is_group_chat: bool = False diff --git a/app/url_processing.py b/app/url_processing.py index fe22460..bd572b6 100644 --- a/app/url_processing.py +++ b/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: