Spaces:
Running
Running
File size: 2,690 Bytes
a1ece72 0c3b94e a1ece72 0c3b94e a1ece72 0c3b94e a1ece72 0c3b94e a1ece72 0c3b94e a1ece72 0c3b94e a1ece72 0c3b94e a1ece72 0c3b94e a1ece72 0c3b94e a1ece72 0c3b94e a1ece72 0c3b94e a1ece72 0c3b94e a1ece72 0c3b94e a1ece72 0c3b94e a1ece72 0c3b94e a1ece72 0c3b94e a1ece72 | 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") |