File size: 2,458 Bytes
9d382e5 5592c90 9d382e5 5592c90 9d382e5 5592c90 91f7053 5592c90 91f7053 5592c90 9d382e5 5592c90 9d382e5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | import os
import re
import httpx
import uuid
from config import DOWNLOAD_DIR, MAX_FILE_SIZE
os.makedirs(DOWNLOAD_DIR, exist_ok=True)
async def download_file(url: str):
async with httpx.AsyncClient(timeout=300, follow_redirects=True) as client:
r = await client.head(url)
content_length = int(r.headers.get("content-length", 0))
content_type = r.headers.get("content-type", "application/octet-stream")
r = await client.get(url)
if r.status_code != 200:
return None, f"Failed to download (HTTP {r.status_code})", None, None
filename = _extract_filename(r.headers, url, content_type)
if not filename:
ext = _ext_from_url(url, content_type)
filename = f"{uuid.uuid4()}{ext}"
filepath = os.path.join(DOWNLOAD_DIR, filename)
with open(filepath, "wb") as f:
f.write(r.content)
return filepath, None, content_type, len(r.content)
def _extract_filename(headers, url: str, content_type: str) -> str:
cd = headers.get("Content-Disposition", "")
match = re.search(r'filename\*?="?([^";]+)"?', cd)
if match:
name = match.group(1).strip()
if name:
return name
from urllib.parse import urlparse, unquote
path = urlparse(url).path
if "/" in path:
name = path.rsplit("/", 1)[-1]
if name and "." in name:
return unquote(name)
ext = _ext_from_url(url, content_type)
if ext:
return f"download{ext}"
return ""
def _ext_from_url(url: str, content_type: str) -> str:
match = re.search(r"\.([a-zA-Z0-9]+)(?:\?|$)", url)
if match:
return f".{match.group(1)}"
mapping = {
"image/jpeg": ".jpg",
"image/png": ".png",
"image/gif": ".gif",
"application/pdf": ".pdf",
"application/zip": ".zip",
"text/plain": ".txt",
"application/x-msdownload": ".exe",
"application/vnd.ms-excel": ".xls",
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx",
"application/msword": ".doc",
"application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx",
"video/mp4": ".mp4",
"audio/mpeg": ".mp3",
}
return mapping.get(content_type, "")
def cleanup_file(filepath: str):
try:
if os.path.exists(filepath):
os.remove(filepath)
except Exception:
pass
|