File size: 2,690 Bytes
1425afc | 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 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 | import os
import uuid
import requests
import subprocess
from urllib.parse import urlparse
from utils.logger import logger
DOWNLOAD_DIR = "jobs"
os.makedirs(DOWNLOAD_DIR, exist_ok=True)
# ---------------------------------------------------
# URL DETECTION
# ---------------------------------------------------
def is_url(value: str):
try:
result = urlparse(value)
return result.scheme in ("http", "https")
except Exception:
return False
def is_social_url(url: str):
domains = [
"youtube.com",
"youtu.be",
"tiktok.com",
"instagram.com",
"facebook.com",
"fb.watch",
"twitter.com",
"x.com"
]
return any(d in url.lower() for d in domains)
# ---------------------------------------------------
# DIRECT FILE DOWNLOAD
# ---------------------------------------------------
def download_direct(url: str) -> str:
filename = f"{uuid.uuid4()}.mp4"
output = os.path.join(DOWNLOAD_DIR, filename)
logger.info(f"[INPUT] Direct download → {url}")
with requests.get(url, stream=True, timeout=120) as r:
r.raise_for_status()
with open(output, "wb") as f:
for chunk in r.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
logger.info(f"[INPUT] Saved → {output}")
return output
# ---------------------------------------------------
# SOCIAL MEDIA DOWNLOAD (yt-dlp)
# ---------------------------------------------------
def download_social(url: str) -> str:
filename = f"{uuid.uuid4()}.mp4"
output = os.path.join(DOWNLOAD_DIR, filename)
logger.info(f"[INPUT] Social download → {url}")
cmd = [
"yt-dlp",
"-f", "bestvideo+bestaudio/best",
"--merge-output-format", "mp4",
"-o", output,
url,
]
subprocess.run(cmd, check=True)
if not os.path.exists(output):
raise Exception("yt-dlp download failed")
logger.info(f"[INPUT] Saved → {output}")
return output
# ---------------------------------------------------
# UNIVERSAL RESOLVER
# ---------------------------------------------------
def resolve_input(input_value):
"""
Accepts:
- Upload path
- Direct URL
- YouTube/TikTok/Instagram/Facebook link
"""
# Already local
if isinstance(input_value, str) and os.path.exists(input_value):
return input_value
# URL input
if isinstance(input_value, str) and is_url(input_value):
if is_social_url(input_value):
return download_social(input_value)
return download_direct(input_value)
raise Exception("Unsupported input type") |