Create terabox.py

utils/terabox.py

@@ -0,0 +1,72 @@
+# filename: utils/terabox.py
+
+import re
+import asyncio
+import logging
+from functools import partial
+import requests
+
+import config
+
+logger = logging.getLogger(__name__)
+
+# A set of regex patterns to extract the short ID from various Terabox URL formats
+TERABOX_URL_PATTERNS = [
+    r'terabox\.com/s/([a-zA-Z0-9_-]+)',
+    r'teraboxapp\.com/s/([a-zA-Z0-9_-]+)',
+    r'1024tera\.com/s/([a-zA-Z0-9_-]+)',
+    r'freeterabox\.com/s/([a-zA-Z0-9_-]+)',
+    r'terabox\.com/sharing/link\?surl=([a-zA-Z0-9_-]+)',
+    r'terasharelink\.com/s/([a-zA-Z0-9_-]+)',
+    r'4funbox\.com/s/([a-zA-Z0-9_-]+)',
+    r'box-links\.com/s/([a-zA-Z0-9_-]+)'
+]
+
+async def extract_terabox_short_id(full_url: str) -> str | None:
+    """Extracts the unique short ID from a Terabox URL."""
+    for pattern in TERABOX_URL_PATTERNS:
+        if match := re.search(pattern, full_url, re.IGNORECASE):
+            return match.group(1)
+    return None
+
+async def get_final_url_and_metadata(original_link: str) -> dict:
+    """
+    Contacts the worker API to get the direct download link and file metadata.
+    Returns a dictionary with details or an error.
+    """
+    payload = {"link": original_link}
+    headers = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36"
+    }
+    
+    try:
+        # We run the synchronous requests call in a separate thread to avoid blocking our async bot
+        loop = asyncio.get_event_loop()
+        response = await loop.run_in_executor(
+            None,
+            partial(requests.post, config.TERABOX_WORKER_URL, headers=headers, json=payload, timeout=30)
+        )
+        response.raise_for_status()
+        data = response.json()
+
+        # Check for expected keys in the worker's response
+        if data.get("ok") and all(k in data for k in ["file_name", "file_size_bytes", "download_link"]):
+            return {
+                "success": True,
+                "file_name": data["file_name"],
+                "file_size": data["file_size_bytes"],
+                "url": data["download_link"],
+                "error": None
+            }
+        else:
+            error_message = data.get("message", "Worker returned incomplete or invalid data.")
+            logger.warning(f"Worker API returned an error for link {original_link}: {error_message}")
+            return {"success": False, "error": error_message}
+
+    except requests.exceptions.RequestException as e:
+        logger.error(f"Network error while contacting worker for link {original_link}: {e}")
+        return {"success": False, "error": "A network error occurred while contacting the processing service."}
+    except Exception as e:
+        logger.error(f"An unexpected error occurred in get_final_url_and_metadata for link {original_link}: {e}", exc_info=True)
+        return {"success": False, "error": "An unexpected server error occurred."}
+