app.py

import os
import re
import uuid
import asyncio
import logging
from fastapi import FastAPI, Query, HTTPException
from fastapi.responses import FileResponse, JSONResponse

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

TIKTOK_URL_RE = re.compile(
    r'https?://(?:www\.|vm\.|vt\.)?tiktok\.com/\S+', re.IGNORECASE
)
INSTAGRAM_URL_RE = re.compile(
    r'https?://(?:www\.)?instagram\.com/(?:reel|p)/\S+', re.IGNORECASE
)

TMP_DIR = "/tmp/tiktok"
os.makedirs(TMP_DIR, exist_ok=True)


@app.get("/download")
async def download(url: str = Query(...)):
    if not TIKTOK_URL_RE.match(url):
        raise HTTPException(status_code=400, detail="Invalid TikTok URL")

    job_id = uuid.uuid4().hex
    out_tmpl = f"{TMP_DIR}/{job_id}.%(ext)s"

    cmd = [
        "yt-dlp",
        "--no-warnings",
        "-f", "bestvideo+bestaudio/best",
        "--merge-output-format", "mp4",
        "-o", out_tmpl,
        url,
    ]

    logger.info(f"Running: {' '.join(cmd)}")

    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )

    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=60)
    except asyncio.TimeoutError:
        proc.kill()
        raise HTTPException(status_code=504, detail="Timeout")

    stdout_str = stdout.decode(errors="ignore").strip()
    stderr_str = stderr.decode(errors="ignore").strip()

    logger.info(f"yt-dlp stdout: {stdout_str}")
    logger.error(f"yt-dlp stderr: {stderr_str}")
    logger.info(f"yt-dlp returncode: {proc.returncode}")

    if proc.returncode != 0:
        if "Unsupported URL" in stderr_str and "/photo/" in stdout_str:
            raise HTTPException(status_code=415, detail="Photo post")
        raise HTTPException(status_code=502, detail=stderr_str)

    files = [f for f in os.listdir(TMP_DIR) if f.startswith(job_id)]
    if not files:
        raise HTTPException(status_code=502, detail="File not found after download")

    filepath = os.path.join(TMP_DIR, files[0])

    async def cleanup():
        try:
            os.remove(filepath)
        except Exception:
            pass

    return FileResponse(
        filepath,
        media_type="video/mp4",
        filename="video.mp4",
        background=cleanup,
    )


@app.get("/download_insta")
async def download_insta(url: str = Query(...)):
    if not INSTAGRAM_URL_RE.match(url):
        raise HTTPException(status_code=400, detail="Invalid Instagram URL")

    job_id = uuid.uuid4().hex
    out_dir = f"{TMP_DIR}/{job_id}"
    os.makedirs(out_dir, exist_ok=True)

    cmd = [
        "gallery-dl",
        "--no-mtime",
        "-D", out_dir,
        url,
    ]

    logger.info(f"Running: {' '.join(cmd)}")

    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )

    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=60)
    except asyncio.TimeoutError:
        proc.kill()
import os
import re
import uuid
import shutil
import asyncio
import logging
from fastapi import FastAPI, Query, HTTPException
from fastapi.responses import FileResponse, JSONResponse

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

TIKTOK_URL_RE = re.compile(
    r'https?://(?:www\.|vm\.|vt\.)?tiktok\.com/\S+', re.IGNORECASE
)
INSTAGRAM_URL_RE = re.compile(
    r'https?://(?:www\.)?instagram\.com/\S+', re.IGNORECASE
)

TMP_DIR = "/tmp/tiktok"
os.makedirs(TMP_DIR, exist_ok=True)


@app.get("/download")
async def download(url: str = Query(...)):
    # Если инста — редиректим на логику gallery-dl
    if INSTAGRAM_URL_RE.match(url):
        return await _download_insta(url)

    if not TIKTOK_URL_RE.match(url):
        raise HTTPException(status_code=400, detail="Invalid URL")

    job_id = uuid.uuid4().hex
    out_tmpl = f"{TMP_DIR}/{job_id}.%(ext)s"

    cmd = [
        "yt-dlp",
        "--no-warnings",
        "-f", "bestvideo+bestaudio/best",
        "--merge-output-format", "mp4",
        "-o", out_tmpl,
        url,
    ]

    logger.info(f"Running: {' '.join(cmd)}")

    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )

    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=60)
    except asyncio.TimeoutError:
        proc.kill()
        raise HTTPException(status_code=504, detail="Timeout")

    stdout_str = stdout.decode(errors="ignore").strip()
    stderr_str = stderr.decode(errors="ignore").strip()

    logger.info(f"yt-dlp stdout: {stdout_str}")
    logger.error(f"yt-dlp stderr: {stderr_str}")
    logger.info(f"yt-dlp returncode: {proc.returncode}")

    if proc.returncode != 0:
        if "Unsupported URL" in stderr_str and "/photo/" in stdout_str:
            raise HTTPException(status_code=415, detail="Photo post")
        raise HTTPException(status_code=502, detail=stderr_str)

    files = [f for f in os.listdir(TMP_DIR) if f.startswith(job_id)]
    if not files:
        raise HTTPException(status_code=502, detail="File not found after download")

    filepath = os.path.join(TMP_DIR, files[0])

    async def cleanup():
        try:
            os.remove(filepath)
        except Exception:
            pass

    return FileResponse(
        filepath,
        media_type="video/mp4",
        filename="video.mp4",
        background=cleanup,
    )


@app.get("/download_insta")
async def download_insta(url: str = Query(...)):
    if not INSTAGRAM_URL_RE.match(url):
        raise HTTPException(status_code=400, detail="Invalid Instagram URL")
    return await _download_insta(url)


async def _download_insta(url: str):
    job_id = uuid.uuid4().hex
    out_dir = f"{TMP_DIR}/{job_id}"
    os.makedirs(out_dir, exist_ok=True)

    cmd = [
        "gallery-dl",
        "--no-mtime",
        "-D", out_dir,
        url,
    ]

    logger.info(f"Running: {' '.join(cmd)}")

    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )

import os
import re
import uuid
import shutil
import asyncio
import logging
from fastapi import FastAPI, Query, HTTPException
from fastapi.responses import FileResponse, JSONResponse

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

TIKTOK_URL_RE = re.compile(
    r'https?://(?:www\.|vm\.|vt\.)?tiktok\.com/\S+', re.IGNORECASE
)
INSTAGRAM_URL_RE = re.compile(
    r'https?://(?:www\.)?instagram\.com/\S+', re.IGNORECASE
)

TMP_DIR = "/tmp/tiktok"
os.makedirs(TMP_DIR, exist_ok=True)


@app.get("/download")
async def download(url: str = Query(...)):
    if INSTAGRAM_URL_RE.match(url):
        return await _download_insta(url)

    if not TIKTOK_URL_RE.match(url):
        raise HTTPException(status_code=400, detail="Invalid URL")

    job_id = uuid.uuid4().hex
    out_tmpl = f"{TMP_DIR}/{job_id}.%(ext)s"

    cmd = [
        "yt-dlp",
        "--no-warnings",
        "-f", "bestvideo+bestaudio/best",
        "--merge-output-format", "mp4",
        "-o", out_tmpl,
        url,
    ]

    logger.info(f"Running: {' '.join(cmd)}")

    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )

    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=60)
    except asyncio.TimeoutError:
        proc.kill()
        raise HTTPException(status_code=504, detail="Timeout")

    stdout_str = stdout.decode(errors="ignore").strip()
    stderr_str = stderr.decode(errors="ignore").strip()

    logger.info(f"yt-dlp stdout: {stdout_str}")
    logger.error(f"yt-dlp stderr: {stderr_str}")
    logger.info(f"yt-dlp returncode: {proc.returncode}")

    if proc.returncode != 0:
        if "Unsupported URL" in stderr_str and "/photo/" in stdout_str:
            raise HTTPException(status_code=415, detail="Photo post")
        raise HTTPException(status_code=502, detail=stderr_str)

    files = [f for f in os.listdir(TMP_DIR) if f.startswith(job_id)]
    if not files:
        raise HTTPException(status_code=502, detail="File not found after download")

    filepath = os.path.join(TMP_DIR, files[0])

    async def cleanup():
        try:
            os.remove(filepath)
        except Exception:
            pass

    return FileResponse(
        filepath,
        media_type="video/mp4",
        filename="video.mp4",
        background=cleanup,
    )


@app.get("/download_insta")
async def download_insta(url: str = Query(...)):
    if not INSTAGRAM_URL_RE.match(url):
        raise HTTPException(status_code=400, detail="Invalid Instagram URL")
    return await _download_insta(url)


async def _download_insta(url: str):
    job_id = uuid.uuid4().hex
    out_tmpl = f"{TMP_DIR}/{job_id}.%(ext)s"

    cmd = [
        "yt-dlp",
        "--no-warnings",
        "--no-check-certificate",
        "-f", "bestvideo+bestaudio/best",
        "--merge-output-format", "mp4",
        "--add-header", "User-Agent:Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1",
        "--add-header", "Accept-Language:en-US,en;q=0.9",
        "-o", out_tmpl,
        url,
    ]

    logger.info(f"Running insta: {' '.join(cmd)}")

    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )

    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=60)
    except asyncio.TimeoutError:
        proc.kill()
        raise HTTPException(status_code=504, detail="Timeout")

    stdout_str = stdout.decode(errors="ignore").strip()
    stderr_str = stderr.decode(errors="ignore").strip()

    logger.info(f"yt-dlp insta stdout: {stdout_str}")
    logger.error(f"yt-dlp insta stderr: {stderr_str}")
    logger.info(f"yt-dlp insta returncode: {proc.returncode}")

    if proc.returncode != 0:
        raise HTTPException(status_code=502, detail=stderr_str or "yt-dlp failed")

    files = [f for f in os.listdir(TMP_DIR) if f.startswith(job_id)]
    if not files:
        raise HTTPException(status_code=502, detail="File not found after download")

    filepath = os.path.join(TMP_DIR, files[0])

    async def cleanup():
        try:
            os.remove(filepath)
        except Exception:
            pass

    return FileResponse(
        filepath,
        media_type="video/mp4",
        filename="video.mp4",
        background=cleanup,
    )


@app.get("/")
async def root():
    return JSONResponse({"status": "ok", "usage": "GET /download?url=<tiktok_or_instagram_url>"})