siftlode/backend/app/downloads/upload.py

"""Local file ingest for the download center (E7 S5) — the non-yt-dlp path into the media library.

A user-uploaded file joins the SAME library as a downloaded one, so it inherits the whole function
set already built (metadata chips, share links, and — for media — in-app play/edit). Unlike a
download it bypasses yt-dlp AND the worker entirely: the file is already on the host, so ingest
probes it, places it on disk, and creates an already-`ready` MediaAsset + `done` DownloadJob in one
step — the same "the file exists, mark it done now" shortcut `service.enqueue` takes on a cache hit.

Kinds (the invariant lives in `mediakind`): a probe with a real video stream -> video, audio-only
-> audio, neither -> a generic 'file' (pdf/zip/docx/…) that is download-only and never rendered
inline (see the force-download serving in the file/watch routes). This module is the ingest seam a
future sub-function (e.g. convert) plugs a new adapter into, mirroring how it sidesteps the worker.

An upload does NOT get an `expires_at`: it's user-provided and may be irreplaceable (there's no
source to re-download), so it doesn't age out on the retention clock the way a re-fetchable download
does — it lives until the user deletes it.
"""

import secrets
from datetime import datetime, timezone
from pathlib import Path

from sqlalchemy.orm import Session

from app.config import settings
from app.downloads import edit as editmod
from app.downloads import mediakind, quota, service, storage
from app.models import DownloadJob, MediaAsset

# ffprobe container formats that mean "this is a still image / image sequence", NOT playable media.
# A standalone jpg/png/gif/webp/bmp reports a real video stream (codec mjpeg/png/…) with no
# `attached_pic` flag, so without this guard it would classify as media_kind='video' and get a Play
# button + a broken in-app <video> player (and inline serving on the public share). Keyed on the
# container format so an animated gif/webp also lands as a generic 'file' (no browser plays those in
# a <video> either). SVG and anything else ffprobe can't decode already fails the probe → 'file'.
_IMAGE_FORMATS = {
    "image2",
    "image2pipe",
    "png_pipe",
    "mjpeg",
    "mjpeg_pipe",
    "jpeg_pipe",
    "gif",
    "webp_pipe",
    "bmp_pipe",
    "tiff_pipe",
    "ico",
    "svg_pipe",
}


def _probe(path: Path) -> tuple[str | None, str | None, int | None, int | None, int | None]:
    """(vcodec, acodec, width, height, duration_s) from one ffprobe call; all None if ffprobe fails
    or the file isn't playable media (an unprobeable file — or a still image — is treated as a generic
    non-media upload, so it becomes a download-only 'file' rather than broken 'video').

    A video stream flagged `attached_pic` (embedded cover art in an mp3/m4a) is skipped — it's a
    thumbnail, not a real video track, so an audio file with album art still classifies as audio."""
    data = editmod.run_ffprobe_json(path, show_streams=True, show_format=True)
    if data is None:
        return None, None, None, None, None

    fmt = data.get("format") or {}
    fmt_names = {n.strip() for n in (fmt.get("format_name") or "").split(",")}
    if fmt_names & _IMAGE_FORMATS:
        # A still image / image sequence is a generic file, not video — no vcodec means media_kind='file'.
        return None, None, None, None, None

    vcodec = acodec = None
    width = height = None
    for stream in data.get("streams", []):
        kind = stream.get("codec_type")
        if kind == "video" and vcodec is None:
            if (stream.get("disposition") or {}).get("attached_pic"):
                continue  # cover art, not a video track
            vcodec = stream.get("codec_name")
            width = stream.get("width")
            height = stream.get("height")
        elif kind == "audio" and acodec is None:
            acodec = stream.get("codec_name")

    duration = None
    try:
        raw = fmt.get("duration")
        if raw is not None:
            duration = int(float(raw))
    except (TypeError, ValueError):
        duration = None
    return vcodec, acodec, width, height, duration


def _title_from_filename(filename: str) -> str:
    """A human display title from the original upload name: the stem with separators tidied to
    spaces (kept as-is otherwise — this is display metadata, not the shell-safe on-disk name)."""
    stem = Path(filename).stem.replace("_", " ").replace(".", " ").strip()
    return (stem or "Upload")[:255]


def ingest_upload(
    db: Session,
    user_id: int,
    *,
    filename: str,
    src_path: Path,
    size: int,
    display_name: str | None = None,
) -> DownloadJob:
    """Ingest an already-uploaded file at `src_path` into `user_id`'s library and return the `done`
    job. `src_path` is MOVED into place on success (the caller stages it to a temp file first).

    Raises quota.QuotaExceeded (holdings/footprint cap) — checked BEFORE any disk write, so a
    rejected upload leaves nothing behind and the caller just cleans up its temp file."""
    quota.check_upload(db, user_id, size)  # the file's size is known → charge it up front

    vcodec, acodec, width, height, duration = _probe(src_path)
    kind = mediakind.derive(vcodec, acodec)

    # An upload never dedups (each file is its own asset): a random ref makes the cache key unique,
    # so `format_sig` is a constant marker rather than a format hash.
    ref = f"{user_id}:{secrets.token_hex(8)}"[:512]
    asset = MediaAsset(
        source_kind="upload", source_ref=ref, format_sig="upload", status="pending", ref_count=0
    )
    db.add(asset)
    db.flush()  # need asset.id for the on-disk path

    rel = storage.upload_rel_path(user_id, asset.id, filename)
    storage.place_file(src_path, settings.download_root, rel)
    # From here the file is on disk under `.uploads/`. If the transaction fails before it commits, the
    # asset/job rows roll back but the placed file would be ORPHANED — nothing references it, the
    # per-user delete path keys off asset.rel_path (which is gone), and the GC orphan sweep skips the
    # dot-prefixed `.uploads/` tree. So roll back the on-disk side too on any failure.
    try:
        final = storage.abs_path(settings.download_root, rel)
        actual_size = final.stat().st_size if final.exists() else size

        asset.status = "ready"
        asset.rel_path = rel
        asset.size_bytes = actual_size
        asset.container = storage.upload_ext(filename)
        asset.vcodec = vcodec
        asset.acodec = acodec
        asset.width = width
        asset.height = height
        asset.duration_s = duration
        asset.media_kind = kind
        asset.title = (display_name or "").strip() or _title_from_filename(filename)
        asset.last_access_at = datetime.now(timezone.utc)
        # No expires_at — see the module docstring (an upload is irreplaceable, so it doesn't age out).
        if kind == mediakind.VIDEO:
            # A frame poster gives the library row a thumbnail + the share link an og:image. Audio has
            # no frame and a generic file no player, so neither gets one. Best-effort (ffmpeg decodes
            # server side even for a codec the browser can't play).
            asset.poster_path = editmod.ensure_poster(asset, settings.download_root)

        job = DownloadJob(
            user_id=user_id,
            asset_id=asset.id,
            job_kind="upload",
            source_kind="upload",
            source_ref=ref,
            profile_snapshot={},
            display_name=(display_name or "").strip() or None,
            status="done",
            progress=100,
            queue_pos=service.next_queue_pos(db),
        )
        db.add(job)
        service.incr_ref_count(db, asset.id)
        db.commit()
    except Exception:
        db.rollback()
        # The rows are gone; remove the now-unreferenced file so no bytes leak. Clean the poster by its
        # DETERMINISTIC path (rel with a .jpg suffix, the same path ensure_poster writes to) rather than
        # ensure_poster's return value: it writes the frame directly to that path and returns None on an
        # ffmpeg failure, so a partial poster can be on disk while the return is None — gating on the
        # return would skip that orphan. unlink is missing_ok, so the poster path is a harmless no-op
        # when none was generated (audio/file, or a video whose poster never started).
        for leaked in (rel, str(Path(rel).with_suffix(".jpg"))):
            storage.abs_path(settings.download_root, leaked).unlink(missing_ok=True)
        raise
    db.refresh(job)
    return job