tebrox / terabox_utils.py
understanding's picture
Create terabox_utils.py
3ec2de0 verified
raw
history blame
1.49 kB
# 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)