Spaces:
Sleeping
Sleeping
# terabox_utils.py | |
import aiohttp | |
import os | |
import uuid | |
import config | |
async def extract_terabox_short_id(link: str) -> str: | |
import re | |
match = re.search(r'/s/([a-zA-Z0-9]+)', link) | |
return match.group(1) if match else None | |
async def get_final_url_and_filename(original_link: str): | |
try: | |
async with aiohttp.ClientSession() as session: | |
async with session.post(config.TERABOX_WORKER_URL, json={"url": original_link}) as resp: | |
if resp.status != 200: | |
return None, None, f"Worker returned status {resp.status}" | |
data = await resp.json() | |
return data.get("url"), data.get("filename"), None | |
except Exception as e: | |
return None, None, str(e) | |
async def download_terabox_file(bot, chat_id, message_id, download_url, filename): | |
local_path = f"downloads/{uuid.uuid4()}_{filename}" | |
try: | |
async with aiohttp.ClientSession() as session: | |
async with session.get(download_url) as resp: | |
if resp.status != 200: | |
return None, None, f"Download failed with status {resp.status}" | |
with open(local_path, "wb") as f: | |
while True: | |
chunk = await resp.content.read(1024 * 1024) | |
if not chunk: | |
break | |
f.write(chunk) | |
return local_path, None, None | |
except Exception as e: | |
return None, None, str(e) | |