Clip Downloader

import json
import os
import re
import time
import uuid
import subprocess
import tempfile
import shutil

from flask import Flask, request, jsonify
from google.cloud import storage

app = Flask(__name__)

# --- Config (set as environment variables on Cloud Run) ---
AUTH_TOKEN = os.environ.get("AUTH_TOKEN")            # shared secret, required
BUCKET_NAME = os.environ.get("BUCKET_NAME")          # GCS bucket name, required
# Videos longer than this (seconds) skip the slow best-quality conversion
# so the iPhone Shortcut never runs out of time.
MAX_BEST_QUALITY_SECONDS = float(os.environ.get("MAX_BEST_QUALITY_SECONDS", "95"))

storage_client = storage.Client()


def check_auth(req):
    """Reject anything that doesn't present the shared secret."""
    header = req.headers.get("Authorization", "")
    expected = f"Bearer {AUTH_TOKEN}"
    return AUTH_TOKEN and header == expected


@app.route("/download", methods=["POST"])
def download():
    if not check_auth(request):
        return jsonify({"error": "unauthorized"}), 401

    # Be forgiving about how the iPhone Shortcut sends the link: it may
    # arrive as {"url": "..."}, a list, or buried in other text (e.g. a
    # TikTok "check out this video https://..." share). Fall back to
    # pulling the first http(s) link out of the raw body.
    raw = request.get_data(as_text=True) or ""
    data = request.get_json(silent=True)
    url = ""
    if isinstance(data, dict):
        value = data.get("url", "")
        if isinstance(value, list):
            value = value[0] if value else ""
        url = str(value).strip() if value is not None else ""
    if not url:
        match = re.search(r"https?://[^\s\"'<>\\]+", raw.replace("\\/", "/"))
        if match:
            url = match.group(0)

    # Echo what we actually received on failure, so a misconfigured
    # Shortcut shows its own mistake directly in "Show Result".
    received = {
        "content_type": request.content_type,
        "body": raw[:300],
    }
    print(f"/download received: {received}", flush=True)

    if not url:
        return jsonify({
            "error": "missing 'url' in request body",
            "received": received,
        }), 400

    # Basic allow-list: only these platforms, since that's all this is for.
    allowed_domains = ["instagram.com", "tiktok.com", "youtube.com", "youtu.be"]
    if not any(d in url for d in allowed_domains):
        return jsonify({
            "error": "url domain not allowed",
            "url": url[:200],
            "received": received,
        }), 400

    work_dir = tempfile.mkdtemp(prefix="dl_")

    try:
        # Each site only gets its own cookies; other sites (e.g. TikTok)
        # get none, since another site's cookies just confuse them. Both
        # files are baked into the container image (see Dockerfile).
        if "instagram.com" in url:
            cookies_path = "/app/cookies_instagram.txt"
        elif "youtube.com" in url or "youtu.be" in url:
            cookies_path = "/app/cookies.txt"
        else:
            cookies_path = None

        base = [
            "yt-dlp",
            "--no-playlist",
            # Lets yt-dlp fetch its JS challenge-solver (needed for YouTube).
            "--remote-components", "ejs:github",
        ]
        if cookies_path and os.path.exists(cookies_path):
            base += ["--cookies", cookies_path]

        # Cloud Run's filesystem is read-only outside /tmp, but yt-dlp's
        # JS solver and Deno need somewhere writable for their caches.
        env = os.environ.copy()
        env["XDG_CACHE_HOME"] = "/tmp/xdg-cache"
        env["DENO_DIR"] = "/tmp/deno-cache"
        os.makedirs(env["XDG_CACHE_HOME"], exist_ok=True)
        os.makedirs(env["DENO_DIR"], exist_ok=True)

        # Step 1: look the video up (no download yet) to learn its length.
        t_start = time.time()
        info = subprocess.run(
            base + ["-J", url],
            capture_output=True, text=True, timeout=120, env=env,
        )
        if info.returncode != 0:
            return jsonify({
                "error": "yt-dlp failed",
                "detail": info.stderr[-2000:],
            }), 502
        info_path = os.path.join(work_dir, "info.json")
        with open(info_path, "w") as f:
            f.write(info.stdout)
        try:
            info_data = json.loads(info.stdout)
            duration = float(info_data.get("duration") or 0)
            # Log what versions the site offers (helps explain choices).
            offered = [
                f"{f.get('format_id')}:{f.get('vcodec') or '?'}:{f.get('height') or '-'}p"
                for f in info_data.get("formats", [])
                if f.get("vcodec") != "none"
            ]
            print(f"formats offered: {' '.join(offered)[:900]}", flush=True)
        except (ValueError, AttributeError):
            duration = 0

        # Step 2: pick the version. The iPhone Shortcut gives up after
        # about a minute, and converting Instagram's best (VP9) version
        # takes longer the longer the video is. So: best quality for short
        # videos; for long ones, the site's ready-made H.264 version, which
        # needs no conversion and finishes in seconds at any length.
        if duration > MAX_BEST_QUALITY_SECONDS:
            mode = "fast (ready-made H.264)"
            choice = [
                # Anything that ISN'T VP9/AV1 (Instagram's ready-made
                # versions carry no codec label, hence the "?" which also
                # accepts unlabelled formats). Falls back to anything.
                "-f", "b[vcodec!*=?vp][vcodec!*=?av01]"
                      "/bv*[vcodec!*=?vp][vcodec!*=?av01]+ba/bv*+ba/b",
                "-S", "res,vcodec:h264,acodec:aac",
            ]
        else:
            mode = "best quality"
            # Highest resolution; among equal resolutions prefer H.264/AAC.
            choice = ["-S", "res,vcodec:h264,acodec:aac"]
        print(f"duration: {duration:.0f}s -> {mode}", flush=True)

        out_dir = os.path.join(work_dir, "out")
        os.makedirs(out_dir, exist_ok=True)
        result = subprocess.run(
            base + choice + [
                "-N", "4",  # download several fragments in parallel
                "--merge-output-format", "mp4",
                "-o", os.path.join(out_dir, "%(id)s.%(ext)s"),
                "--load-info-json", info_path,
            ],
            capture_output=True,
            text=True,
            timeout=280,  # keep comfortably under Cloud Run's request timeout
            env=env,
        )
        print(f"timing: yt-dlp {time.time() - t_start:.1f}s", flush=True)

        if result.returncode != 0:
            return jsonify({
                "error": "yt-dlp failed",
                "detail": result.stderr[-2000:],  # tail, in case it's long
            }), 502

        # Find whatever file yt-dlp produced
        files = os.listdir(out_dir)
        if not files:
            return jsonify({"error": "no output file produced"}), 500
        local_path = os.path.join(out_dir, files[0])

        # Guarantee Photos-compatible output. yt-dlp's own postprocessors
        # (Merger, --recode-video) only run in specific circumstances —
        # Merger only fires if two separate streams were downloaded, and
        # --recode-video skips silently if the container already says
        # mp4 even when the codec inside is wrong. So instead we check
        # the actual codecs with ffprobe and re-encode with ffmpeg
        # ourselves whenever needed, regardless of what yt-dlp did.
        probe = subprocess.run(
            [
                "ffprobe", "-v", "error",
                "-show_entries", "stream=codec_type,codec_name",
                "-of", "json",
                local_path,
            ],
            capture_output=True,
            text=True,
            timeout=30,
        )
        # JSON output is unambiguous (the old CSV output put the fields
        # in the opposite order from what was assumed, so the codecs were
        # never recognized and every file got needlessly re-encoded).
        codecs = {}
        try:
            for stream in json.loads(probe.stdout).get("streams", []):
                codecs.setdefault(stream.get("codec_type"), stream.get("codec_name", ""))
        except (ValueError, AttributeError):
            pass
        video_codec = codecs.get("video", "")
        audio_codec = codecs.get("audio", "")


        print(f"codecs: video={video_codec} audio={audio_codec}", flush=True)

        # Always do a final ffmpeg pass, because downloaded files carry an
        # empty creation date (1 Jan 1904). Photos files videos by that
        # date, so they got buried at the very start of the library
        # instead of showing up in Recents. Stamping "now" fixes that.
        # When the codecs are already fine this is a stream copy (~1s);
        # only an incompatible stream (e.g. Instagram's VP9) is re-encoded.
        converted_path = os.path.join(work_dir, "video.mp4")
        now_iso = time.strftime("%Y-%m-%dT%H:%M:%S.000000Z", time.gmtime())
        # Only re-encode the stream that actually needs it. Copying
        # an already-compatible stream is nearly instant, while
        # re-encoding video is by far the slowest step.
        video_args = (
            ["-c:v", "copy"] if video_codec == "h264"
            else [
                "-c:v", "libx264", "-preset", "ultrafast", "-crf", "26",
                "-pix_fmt", "yuv420p", "-threads", "0",
            ]
        )
        audio_args = (
            ["-c:a", "copy"] if audio_codec == "aac"
            else ["-c:a", "aac", "-b:a", "192k"]
        )
        t_convert = time.time()
        convert = subprocess.run(
            ["ffmpeg", "-y", "-i", local_path]
            + video_args + audio_args
            + ["-metadata", f"creation_time={now_iso}",
               "-movflags", "+faststart", converted_path],
            capture_output=True,
            text=True,
            timeout=280,
        )
        print(f"timing: ffmpeg {time.time() - t_convert:.1f}s", flush=True)
        if convert.returncode != 0:
            return jsonify({
                "error": "conversion failed",
                "detail": convert.stderr[-2000:],
            }), 502
        local_path = converted_path

        # Upload to GCS
        upload_name = os.path.basename(local_path)
        blob_name = f"downloads/{uuid.uuid4().hex}_{upload_name}"
        bucket = storage_client.bucket(BUCKET_NAME)
        blob = bucket.blob(blob_name)
        blob.upload_from_filename(local_path)

        # Cloud Run's default credentials can't sign URLs (no private key),
        # so instead we make just this one file briefly public. A daily
        # lifecycle rule on the bucket cleans these up automatically.
        blob.make_public()

        return jsonify({
            "status": "ok",
            "download_url": blob.public_url,
        })

    except subprocess.TimeoutExpired:
        return jsonify({"error": "download timed out"}), 504
    finally:
        shutil.rmtree(work_dir, ignore_errors=True)


@app.route("/health", methods=["GET"])
def health():
    return jsonify({"status": "ok"})


if __name__ == "__main__":
    port = int(os.environ.get("PORT", 8080))
    app.run(host="0.0.0.0", port=port)

FROM python:3.12-slim

# ffmpeg: needed by yt-dlp to merge/convert audio+video streams
# curl/unzip: needed to install Deno
RUN apt-get update && apt-get install -y --no-install-recommends \
    ffmpeg \
    curl \
    unzip \
    ca-certificates \
    && rm -rf /var/lib/apt/lists/*

# Deno: required by yt-dlp (since 2025.11.12) to run YouTube's
# obfuscated JS and extract playable video URLs.
RUN curl -fsSL https://deno.land/install.sh | sh
ENV DENO_INSTALL="/root/.deno"
ENV PATH="$DENO_INSTALL/bin:$PATH"

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -U -r requirements.txt

# Optional: cookies.txt (YouTube) and/or cookies_instagram.txt (Instagram),
# both Netscape-format cookie files exported from a browser logged into
# that site, placed alongside main.py before deploying. These let yt-dlp
# authenticate and avoid each platform's cloud-IP bot detection. They're
# kept as separate files because they're separate accounts/domains.
# main.py is guaranteed to exist, so this line never fails even if one or
# both cookie files are absent (Docker's documented pattern for optional
# files).
COPY main.py cookies.txt* cookies_instagram.txt* /app/

# Cloud Run sets $PORT; gunicorn binds to it
ENV PORT=8080
EXPOSE 8080

CMD exec gunicorn --bind :$PORT --workers 1 --threads 4 --timeout 290 main:app

flask==3.0.3
gunicorn==22.0.0
yt-dlp
google-cloud-storage==2.18.2

{
  "rule": [
    {
      "action": { "type": "Delete" },
      "condition": { "age": 1 }
    }
  ]
}

#!/bin/bash
# Deploys your clip downloader with the right settings. Run: ./deploy.sh
# Works from anywhere: it always deploys the folder it's in.
#
# Fill in your own two values first. Keep this file private once it holds
# your secret: never share it or put it on GitHub.
SECRET="YOUR_SECRET"
BUCKET="YOUR_BUCKET"

cd "$(dirname "$0")" || exit 1
gcloud run deploy clip-downloader \
  --source . \
  --region us-central1 \
  --allow-unauthenticated \
  --memory 4Gi \
  --cpu 4 \
  --cpu-boost \
  --concurrency 1 \
  --timeout 300 \
  --max-instances 2 \
  --set-env-vars "AUTH_TOKEN=$SECRET,BUCKET_NAME=$BUCKET"