services/whisper/utils/input_resolver.py

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")