Verified each deferred per-subsystem item against current source, then fixed the
real ones (tradeoffs left as-is, see siftlode-ops/CODE-HYGIENE.md closeout):
- search BUG-3: don't finalise shorts_probed on un-enriched stubs (enrichment
failure/omitted rows) — restores the scheduler's enriched_at guard so a real
Short can't leak into the feed and never get reclassified.
- downloads GC: 4th pass reaps orphaned errored, fileless, unreferenced assets
(they carry no TTL, so the expiry passes never cleared them).
- downloads B6: quota/edit errors now return a structured {code,reason,limit}
the trilingual client localises, instead of hardcoded English + dot-decimal GB.
- channels BB1: reset-backfill re-arms deep sync on every subscriber (bulk update)
instead of 404ing when the admin isn't personally subscribed.
- channels BB2: enrich the stub on BOTH the normal and "already exists" desync
path (nest the insert try inside the client `with`).
- channels CB3: drop the redundant second _discovery_rows read (identity-mapped
rows are already enriched; re-sort in Python for the title tie-break).
- playlists PC1: read the live playlist once in push() and share it with plan_push
+ push_playlist (halves read-quota, closes the second TOCTOU window).
- playlists PC2/PC5: batch the cover thumbnails in one window query (was N+1);
rename combines count+duration into one aggregate + a single cover lookup.
- messages MB2: get_thread only resolves a messageable partner OR one we already
share a thread with (closes the user-enumeration oracle).
- messages MB3: push a live unread-count to the user's other tabs after mark-read.
- messages MC4: list_conversations uses DISTINCT ON + grouped COUNT instead of
loading the whole message history into memory.
- config AB3: per-spec allow_empty so smtp_user/youtube_api_proxy can be blanked
to disable them rather than snapping back to the env default.
918 lines
33 KiB
Python
918 lines
33 KiB
Python
"""Download center HTTP API — the per-user surface over the worker/cache/quota layers.
|
|
|
|
All routes are behind `require_human` (the shared demo account can't spend server disk / needs a
|
|
real identity). Admin routes add `admin_user`. Live job state is polled by the frontend (the
|
|
worker is a separate process); this module never downloads — it enqueues, manages job state,
|
|
serves finished files (range-aware, with the user's custom display name), and shares them.
|
|
"""
|
|
import re
|
|
from datetime import datetime, timedelta, timezone
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from fastapi.responses import FileResponse
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import sysconfig
|
|
from app.auth import admin_user, require_human
|
|
from app.config import settings
|
|
from app.db import get_db
|
|
from app.downloads import edit as editmod
|
|
from app.downloads import links as linksmod
|
|
from app.downloads import quota, service, storage
|
|
from app.models import (
|
|
DownloadJob,
|
|
DownloadLink,
|
|
DownloadProfile,
|
|
DownloadQuota,
|
|
DownloadShare,
|
|
MediaAsset,
|
|
User,
|
|
)
|
|
from app.security import hash_password
|
|
|
|
router = APIRouter(prefix="/api/downloads", tags=["downloads"])
|
|
admin_router = APIRouter(prefix="/api/admin/downloads", tags=["admin-downloads"])
|
|
|
|
_VIDEO_ID = re.compile(r"^[A-Za-z0-9_-]{11}$")
|
|
|
|
|
|
# --- source + serialization ----------------------------------------------------------------
|
|
|
|
def resolve_source(raw: str) -> tuple[str, str]:
|
|
"""Map a user-supplied id/URL to (source_kind, source_ref). YouTube is normalized to its
|
|
11-char id (so the cache dedups across watch/youtu.be/shorts URLs); anything else is kept
|
|
as a raw URL for yt-dlp's generic extractor."""
|
|
raw = (raw or "").strip()
|
|
if not raw:
|
|
raise HTTPException(status_code=400, detail="A video link or id is required.")
|
|
if _VIDEO_ID.match(raw):
|
|
return "youtube", raw
|
|
if raw.startswith(("http://", "https://")):
|
|
u = urlparse(raw)
|
|
host = u.netloc.lower()
|
|
if "youtube.com" in host:
|
|
if u.path.startswith(("/shorts/", "/embed/", "/live/")):
|
|
vid = u.path.split("/")[2] if len(u.path.split("/")) > 2 else ""
|
|
if _VIDEO_ID.match(vid):
|
|
return "youtube", vid
|
|
qs = parse_qs(u.query).get("v", [])
|
|
if qs and _VIDEO_ID.match(qs[0]):
|
|
return "youtube", qs[0]
|
|
if "youtu.be" in host:
|
|
vid = u.path.lstrip("/").split("/")[0]
|
|
if _VIDEO_ID.match(vid):
|
|
return "youtube", vid
|
|
return "url", raw
|
|
raise HTTPException(status_code=400, detail="That doesn't look like a valid video link.")
|
|
|
|
|
|
def _serialize(job: DownloadJob, asset: MediaAsset | None) -> dict:
|
|
ready = asset is not None and asset.status == "ready" and bool(asset.rel_path)
|
|
return {
|
|
"id": job.id,
|
|
"status": job.status,
|
|
"progress": job.progress,
|
|
"speed_bps": job.speed_bps,
|
|
"eta_s": job.eta_s,
|
|
"phase": job.phase,
|
|
"error": job.error,
|
|
"queue_pos": job.queue_pos,
|
|
"created_at": job.created_at.isoformat() if job.created_at else None,
|
|
"source_kind": job.source_kind,
|
|
"source_ref": job.source_ref,
|
|
"source_url": service.reference_url(job, asset),
|
|
# A locally-generated poster to fall back to when the source had no thumbnail.
|
|
"poster_url": f"/api/downloads/{job.id}/poster.jpg"
|
|
if (asset is not None and asset.poster_path)
|
|
else None,
|
|
"profile_id": job.profile_id,
|
|
"spec": job.profile_snapshot,
|
|
"display_name": job.display_name,
|
|
"display_uploader": job.display_uploader,
|
|
"display_uploader_url": job.display_uploader_url,
|
|
"extra_links": job.extra_links or [],
|
|
"job_kind": job.job_kind,
|
|
"source_job_id": job.source_job_id,
|
|
"edit_spec": job.edit_spec,
|
|
"can_download": ready and job.status == "done",
|
|
"expired": job.asset_id is None and job.status == "done",
|
|
"asset": None
|
|
if asset is None
|
|
else {
|
|
"id": asset.id,
|
|
"status": asset.status,
|
|
"title": asset.title,
|
|
"uploader": asset.uploader,
|
|
"upload_date": asset.upload_date.isoformat() if asset.upload_date else None,
|
|
"duration_s": asset.duration_s,
|
|
"size_bytes": asset.size_bytes,
|
|
"width": asset.width,
|
|
"height": asset.height,
|
|
"container": asset.container,
|
|
"uploader_url": asset.uploader_url,
|
|
"thumbnail_url": asset.thumbnail_url,
|
|
"expires_at": asset.expires_at.isoformat() if asset.expires_at else None,
|
|
},
|
|
}
|
|
|
|
|
|
def _assets_for(db: Session, jobs: list[DownloadJob]) -> dict[int, MediaAsset]:
|
|
ids = {j.asset_id for j in jobs if j.asset_id}
|
|
if not ids:
|
|
return {}
|
|
rows = db.execute(select(MediaAsset).where(MediaAsset.id.in_(ids))).scalars()
|
|
return {a.id: a for a in rows}
|
|
|
|
|
|
def _serialize_job(db: Session, job: DownloadJob) -> dict:
|
|
"""Resolve a job's cache asset (if any) and serialize the pair — the resolve-then-serialize step
|
|
every single-job handler shares."""
|
|
asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
|
|
return _serialize(job, asset)
|
|
|
|
|
|
def _own_job(db: Session, user: User, job_id: int) -> DownloadJob:
|
|
job = db.get(DownloadJob, job_id)
|
|
if job is None or job.user_id != user.id:
|
|
raise HTTPException(status_code=404, detail="Unknown download")
|
|
return job
|
|
|
|
|
|
# --- profiles ------------------------------------------------------------------------------
|
|
|
|
def _serialize_profile(p: DownloadProfile) -> dict:
|
|
return {
|
|
"id": p.id,
|
|
"name": p.name,
|
|
"spec": p.spec,
|
|
"is_builtin": p.is_builtin,
|
|
"sort_order": p.sort_order,
|
|
}
|
|
|
|
|
|
@router.get("/profiles")
|
|
def list_profiles(
|
|
user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> list[dict]:
|
|
rows = (
|
|
db.execute(
|
|
select(DownloadProfile)
|
|
.where(
|
|
(DownloadProfile.is_builtin.is_(True))
|
|
| (DownloadProfile.user_id == user.id)
|
|
)
|
|
.order_by(DownloadProfile.is_builtin.desc(), DownloadProfile.sort_order, DownloadProfile.id)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
return [_serialize_profile(p) for p in rows]
|
|
|
|
|
|
@router.post("/profiles")
|
|
def create_profile(
|
|
payload: dict, user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
name = (payload.get("name") or "").strip()
|
|
spec = payload.get("spec")
|
|
if not name:
|
|
raise HTTPException(status_code=400, detail="A profile name is required.")
|
|
if not isinstance(spec, dict):
|
|
raise HTTPException(status_code=400, detail="spec must be an object")
|
|
maxpos = db.execute(
|
|
select(func.coalesce(func.max(DownloadProfile.sort_order), 100)).where(
|
|
DownloadProfile.user_id == user.id
|
|
)
|
|
).scalar_one()
|
|
p = DownloadProfile(
|
|
user_id=user.id, name=name[:80], spec=spec, is_builtin=False, sort_order=maxpos + 1
|
|
)
|
|
db.add(p)
|
|
db.commit()
|
|
return _serialize_profile(p)
|
|
|
|
|
|
@router.patch("/profiles/{profile_id}")
|
|
def update_profile(
|
|
profile_id: int,
|
|
payload: dict,
|
|
user: User = Depends(require_human),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
p = db.get(DownloadProfile, profile_id)
|
|
if p is None or p.is_builtin or p.user_id != user.id:
|
|
raise HTTPException(status_code=404, detail="Unknown profile")
|
|
if "name" in payload:
|
|
name = (payload.get("name") or "").strip()
|
|
if not name:
|
|
raise HTTPException(status_code=400, detail="A profile name is required.")
|
|
p.name = name[:80]
|
|
if "spec" in payload:
|
|
if not isinstance(payload["spec"], dict):
|
|
raise HTTPException(status_code=400, detail="spec must be an object")
|
|
p.spec = payload["spec"]
|
|
db.commit()
|
|
return _serialize_profile(p)
|
|
|
|
|
|
@router.delete("/profiles/{profile_id}")
|
|
def delete_profile(
|
|
profile_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
p = db.get(DownloadProfile, profile_id)
|
|
if p is None or p.is_builtin or p.user_id != user.id:
|
|
raise HTTPException(status_code=404, detail="Unknown profile")
|
|
db.delete(p)
|
|
db.commit()
|
|
return {"deleted": profile_id}
|
|
|
|
|
|
def _resolve_spec(db: Session, user: User, payload: dict) -> tuple[dict, int | None]:
|
|
"""Pick the format spec for an enqueue: an explicit profile_id (builtin or the user's own),
|
|
an inline spec, or the first builtin as a fallback."""
|
|
pid = payload.get("profile_id")
|
|
if pid is not None:
|
|
p = db.get(DownloadProfile, pid)
|
|
if p is None or (not p.is_builtin and p.user_id != user.id):
|
|
raise HTTPException(status_code=404, detail="Unknown profile")
|
|
return p.spec, p.id
|
|
if isinstance(payload.get("spec"), dict):
|
|
return payload["spec"], None
|
|
fallback = db.execute(
|
|
select(DownloadProfile)
|
|
.where(DownloadProfile.is_builtin.is_(True))
|
|
.order_by(DownloadProfile.sort_order)
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
if fallback is None:
|
|
raise HTTPException(status_code=400, detail="No download profile available.")
|
|
return fallback.spec, fallback.id
|
|
|
|
|
|
# --- enqueue + list + manage ---------------------------------------------------------------
|
|
|
|
@router.post("")
|
|
def enqueue_download(
|
|
payload: dict, user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
source_kind, source_ref = resolve_source(payload.get("source") or "")
|
|
spec, profile_id = _resolve_spec(db, user, payload)
|
|
display_name = (payload.get("display_name") or "").strip() or None
|
|
try:
|
|
job = service.enqueue(
|
|
db, user.id, source_kind, source_ref, spec, profile_id, display_name
|
|
)
|
|
except quota.QuotaExceeded as e:
|
|
raise HTTPException(status_code=422, detail=_quota_detail(e))
|
|
return _serialize_job(db, job)
|
|
|
|
|
|
@router.post("/edit")
|
|
def enqueue_edit(
|
|
payload: dict, user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
"""Queue a phase-2 editor derivative (trim/crop) of a download this user can access.
|
|
|
|
The source may be the user's own download OR one shared with them — either way the resulting
|
|
clip is a NEW per-user derivative in the editor's own library (counts against their quota); the
|
|
source file is only read, never modified, so editing a shared video is safe."""
|
|
source_job = _accessible_job(db, user, int(payload.get("source_job_id") or 0))
|
|
if source_job.status != "done" or source_job.asset_id is None:
|
|
raise HTTPException(status_code=409, detail="You can only edit a finished download.")
|
|
spec = payload.get("edit_spec")
|
|
if not isinstance(spec, dict):
|
|
raise HTTPException(status_code=400, detail="edit_spec must be an object")
|
|
display_name = (payload.get("display_name") or "").strip() or None
|
|
try:
|
|
job = service.enqueue_edit(db, user.id, source_job, spec, display_name)
|
|
except quota.QuotaExceeded as e:
|
|
raise HTTPException(status_code=422, detail=_quota_detail(e))
|
|
except service.EditError as e:
|
|
raise HTTPException(status_code=400, detail={"code": "edit", "reason": e.reason})
|
|
return _serialize_job(db, job)
|
|
|
|
|
|
def _quota_detail(e: quota.QuotaExceeded) -> dict:
|
|
"""Structured error the trilingual client localizes (reason key + numbers), instead of a
|
|
pre-baked English string with a dot-decimal GB value. See lib/api.ts localizeDetail()."""
|
|
return {"code": "quota", "reason": e.reason, "limit": e.limit, "current": e.current}
|
|
|
|
|
|
@router.get("")
|
|
def list_downloads(
|
|
user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> list[dict]:
|
|
jobs = (
|
|
db.execute(
|
|
select(DownloadJob)
|
|
.where(DownloadJob.user_id == user.id)
|
|
.order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc())
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
assets = _assets_for(db, jobs)
|
|
return [_serialize(j, assets.get(j.asset_id)) for j in jobs]
|
|
|
|
|
|
@router.get("/usage")
|
|
def my_usage(user: User = Depends(require_human), db: Session = Depends(get_db)) -> dict:
|
|
return quota.usage(db, user.id)
|
|
|
|
|
|
# Priority when a source has several jobs (e.g. a done one plus a fresh queued one).
|
|
_INDEX_RANK = {"done": 4, "running": 3, "paused": 2, "queued": 1}
|
|
|
|
|
|
@router.get("/index")
|
|
def my_download_index(
|
|
user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
"""source_ref -> best active status, for the feed's per-card "downloaded / queued" badge.
|
|
Small + cheap so the feed can poll it without loading the full job list."""
|
|
rows = db.execute(
|
|
select(DownloadJob.source_ref, DownloadJob.status).where(
|
|
DownloadJob.user_id == user.id,
|
|
DownloadJob.status.in_(("queued", "running", "paused", "done")),
|
|
)
|
|
).all()
|
|
out: dict[str, str] = {}
|
|
for ref, status in rows:
|
|
if _INDEX_RANK.get(status, 0) > _INDEX_RANK.get(out.get(ref, ""), 0):
|
|
out[ref] = status
|
|
return out
|
|
|
|
|
|
@router.get("/shared")
|
|
def shared_with_me(
|
|
user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> list[dict]:
|
|
jobs = (
|
|
db.execute(
|
|
select(DownloadJob)
|
|
.join(DownloadShare, DownloadShare.job_id == DownloadJob.id)
|
|
.where(DownloadShare.to_user_id == user.id)
|
|
.order_by(DownloadShare.created_at.desc())
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
assets = _assets_for(db, jobs)
|
|
out = []
|
|
for j in jobs:
|
|
data = _serialize(j, assets.get(j.asset_id))
|
|
data["shared"] = True
|
|
out.append(data)
|
|
return out
|
|
|
|
|
|
_MAX_EXTRA_LINKS = 8
|
|
|
|
|
|
def _clean_url(raw) -> str | None:
|
|
"""Accept only a plausible http(s) URL (these get rendered as clickable links + embedded in a
|
|
public watch page); anything else is dropped rather than trusted."""
|
|
if not isinstance(raw, str):
|
|
return None
|
|
url = raw.strip()
|
|
if not url:
|
|
return None
|
|
if not url.startswith(("http://", "https://")):
|
|
return None
|
|
parsed = urlparse(url)
|
|
if not parsed.netloc:
|
|
return None
|
|
return url[:2048]
|
|
|
|
|
|
@router.patch("/{job_id}")
|
|
def update_download_meta(
|
|
job_id: int,
|
|
payload: dict,
|
|
user: User = Depends(require_human),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
"""Update a download's display metadata: title, channel name + link, and extra reference URLs.
|
|
All are display-only overrides — the on-disk file keeps its system-decided, id-bound name."""
|
|
job = _own_job(db, user, job_id)
|
|
if "display_name" in payload:
|
|
job.display_name = ((payload.get("display_name") or "").strip())[:255] or None
|
|
if "display_uploader" in payload:
|
|
job.display_uploader = ((payload.get("display_uploader") or "").strip())[:255] or None
|
|
if "display_uploader_url" in payload:
|
|
job.display_uploader_url = _clean_url(payload.get("display_uploader_url"))
|
|
if "extra_links" in payload:
|
|
raw = payload.get("extra_links") or []
|
|
if not isinstance(raw, list):
|
|
raise HTTPException(status_code=400, detail="extra_links must be a list")
|
|
cleaned = [u for u in (_clean_url(x) for x in raw) if u][:_MAX_EXTRA_LINKS]
|
|
job.extra_links = cleaned or None
|
|
db.commit()
|
|
return _serialize_job(db, job)
|
|
|
|
|
|
def _set_status(db: Session, job: DownloadJob, status: str) -> None:
|
|
job.status = status
|
|
db.commit()
|
|
|
|
|
|
@router.post("/{job_id}/pause")
|
|
def pause_download(
|
|
job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
job = _own_job(db, user, job_id)
|
|
if job.status in ("queued", "running"):
|
|
_set_status(db, job, "paused") # a running worker aborts cooperatively via the hook
|
|
return _serialize_job(db, job)
|
|
|
|
|
|
@router.post("/{job_id}/resume")
|
|
def resume_download(
|
|
job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
job = _own_job(db, user, job_id)
|
|
if job.status in ("paused", "error"):
|
|
job.progress = 0
|
|
job.error = None
|
|
# If the shared asset itself failed, reset it to pending so the worker actually
|
|
# re-downloads — otherwise the requeued job would instantly re-inherit the asset's
|
|
# stale error (the worker short-circuits a job whose asset is already errored).
|
|
if job.asset_id:
|
|
asset = db.get(MediaAsset, job.asset_id)
|
|
if asset and asset.status == "error":
|
|
asset.status = "pending"
|
|
asset.error = None
|
|
_set_status(db, job, "queued")
|
|
return _serialize_job(db, job)
|
|
|
|
|
|
@router.post("/{job_id}/cancel")
|
|
def cancel_download(
|
|
job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
job = _own_job(db, user, job_id)
|
|
if job.status not in ("done", "canceled"):
|
|
_release_asset(db, job) # release while the job still counts as holding
|
|
job.status = "canceled"
|
|
db.commit()
|
|
return _serialize_job(db, job)
|
|
|
|
|
|
@router.delete("/{job_id}")
|
|
def delete_download(
|
|
job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
job = _own_job(db, user, job_id)
|
|
_release_asset(db, job)
|
|
db.delete(job)
|
|
db.commit()
|
|
return {"deleted": job_id}
|
|
|
|
|
|
def _release_asset(db: Session, job: DownloadJob) -> None:
|
|
"""Drop this job's hold on its asset when it leaves a holding state. Once NO job holds the
|
|
asset anymore, delete the file + row immediately — a deleted download should free its disk,
|
|
and the shared cache only needs to span *overlapping* holders (a later re-add just downloads
|
|
again). `error` counts as holding: enqueue always +1'd ref_count and the worker never
|
|
decrements on failure (ref_count is the route's job), so an errored job releases here on
|
|
delete — else its +1 leaks and a later resume→ready keeps the file pinned past its last holder.
|
|
Idempotent for the cancel path: cancel already released while the job was holding, then set it
|
|
`canceled` (∉ the set below), so delete-after-cancel is a no-op and never double-releases.
|
|
|
|
We DON'T delete an errored asset row here even at ref==0: another request may be concurrently
|
|
re-enqueuing the same (source,format) — get_or_create_asset would reset that row to `pending` and
|
|
+1 it, and deleting it under that would FK-null the new job's asset (a lost update on ref_count).
|
|
A ready asset is still freed at ref==0 (its file is the point); an orphaned errored row is cheap
|
|
(no file) and gets reused+reset by the next enqueue, or reclaimed by a later GC pass."""
|
|
if not job.asset_id or job.status not in ("queued", "running", "paused", "done", "error"):
|
|
return
|
|
asset = db.get(MediaAsset, job.asset_id)
|
|
if asset is None:
|
|
return
|
|
asset.ref_count = max(0, (asset.ref_count or 0) - 1)
|
|
if asset.ref_count == 0 and asset.status == "ready":
|
|
if asset.rel_path:
|
|
storage.delete_asset_files(settings.download_root, asset.rel_path)
|
|
db.delete(asset)
|
|
|
|
|
|
# --- file download (range-aware, custom display name) --------------------------------------
|
|
|
|
def _accessible_job(db: Session, user: User, job_id: int) -> DownloadJob:
|
|
job = db.get(DownloadJob, job_id)
|
|
if job is None:
|
|
raise HTTPException(status_code=404, detail="Unknown download")
|
|
if job.user_id == user.id:
|
|
return job
|
|
shared = db.execute(
|
|
select(DownloadShare.id).where(
|
|
DownloadShare.job_id == job_id, DownloadShare.to_user_id == user.id
|
|
)
|
|
).first()
|
|
if shared is None:
|
|
raise HTTPException(status_code=404, detail="Unknown download")
|
|
return job
|
|
|
|
|
|
@router.get("/{job_id}/file")
|
|
def download_file(
|
|
job_id: int,
|
|
request: Request,
|
|
user: User = Depends(require_human),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
job = _accessible_job(db, user, job_id)
|
|
asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
|
|
if asset is None or asset.status != "ready" or not asset.rel_path:
|
|
raise HTTPException(status_code=409, detail="This download isn't ready.")
|
|
path = storage.safe_abs_path(settings.download_root, asset.rel_path)
|
|
if path is None:
|
|
raise HTTPException(status_code=404, detail="File is no longer available.")
|
|
|
|
filename = storage.download_filename(
|
|
job.display_name or asset.title or asset.source_ref, asset.container, path
|
|
)
|
|
|
|
asset.last_access_at = datetime.now(timezone.utc)
|
|
db.commit()
|
|
# Starlette's FileResponse honours the Range header (206 partial content) for seek/resume.
|
|
return FileResponse(path, filename=filename)
|
|
|
|
|
|
# --- editor filmstrip (scrub track) --------------------------------------------------------
|
|
|
|
@router.get("/{job_id}/storyboard")
|
|
def storyboard_meta(
|
|
job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
"""Filmstrip geometry for the editor's scrub track, generating the sprite on first use.
|
|
|
|
Best-effort: for a very long/high-res source the sprite may not generate in time — then
|
|
`available` is False and the editor shows a plain timeline."""
|
|
job = _accessible_job(db, user, job_id)
|
|
asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
|
|
if asset is None or asset.status != "ready" or not asset.rel_path:
|
|
raise HTTPException(status_code=409, detail="This download isn't ready.")
|
|
meta = editmod.ensure_storyboard(asset, settings.download_root)
|
|
if meta is None:
|
|
return {"available": False}
|
|
db.commit() # persist storyboard_path if it was just generated
|
|
return {
|
|
"available": True,
|
|
"cols": meta["cols"],
|
|
"rows": meta["rows"],
|
|
"count": meta["count"],
|
|
"interval_s": meta["interval_s"],
|
|
"url": f"/api/downloads/{job_id}/storyboard.jpg",
|
|
}
|
|
|
|
|
|
@router.get("/{job_id}/poster.jpg")
|
|
def poster_image(
|
|
job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
):
|
|
"""The locally-generated poster frame for a download whose source had no thumbnail."""
|
|
job = _accessible_job(db, user, job_id)
|
|
asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
|
|
if asset is None or not asset.poster_path:
|
|
raise HTTPException(status_code=404, detail="No poster.")
|
|
path = storage.safe_abs_path(settings.download_root, asset.poster_path)
|
|
if path is None:
|
|
raise HTTPException(status_code=404, detail="No poster.")
|
|
return FileResponse(path, media_type="image/jpeg")
|
|
|
|
|
|
@router.get("/{job_id}/storyboard.jpg")
|
|
def storyboard_image(
|
|
job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
):
|
|
job = _accessible_job(db, user, job_id)
|
|
asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
|
|
if asset is None or not asset.storyboard_path:
|
|
raise HTTPException(status_code=404, detail="No filmstrip.")
|
|
path = storage.safe_abs_path(settings.download_root, asset.storyboard_path)
|
|
if path is None:
|
|
raise HTTPException(status_code=404, detail="No filmstrip.")
|
|
return FileResponse(path, media_type="image/jpeg")
|
|
|
|
|
|
# --- sharing -------------------------------------------------------------------------------
|
|
|
|
@router.post("/{job_id}/share")
|
|
def share_download(
|
|
job_id: int,
|
|
payload: dict,
|
|
user: User = Depends(require_human),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
job = _own_job(db, user, job_id)
|
|
if job.status != "done" or job.asset_id is None:
|
|
raise HTTPException(status_code=409, detail="Only a finished download can be shared.")
|
|
target = (payload.get("email") or "").strip().lower()
|
|
if not target:
|
|
raise HTTPException(status_code=400, detail="A recipient email is required.")
|
|
recipient = db.execute(
|
|
select(User).where(func.lower(User.email) == target)
|
|
).scalar_one_or_none()
|
|
if recipient is None:
|
|
raise HTTPException(status_code=404, detail="No user with that email.")
|
|
if recipient.id == user.id:
|
|
raise HTTPException(status_code=400, detail="That's already your download.")
|
|
exists = db.execute(
|
|
select(DownloadShare.id).where(
|
|
DownloadShare.job_id == job_id, DownloadShare.to_user_id == recipient.id
|
|
)
|
|
).first()
|
|
if exists is None:
|
|
db.add(
|
|
DownloadShare(job_id=job_id, from_user_id=user.id, to_user_id=recipient.id)
|
|
)
|
|
from app.notifications import create_notification
|
|
|
|
create_notification(
|
|
db,
|
|
user_id=recipient.id,
|
|
type="download_shared",
|
|
title=(job.display_name or "A download") + " was shared with you",
|
|
data={"kind": "download_shared", "from": user.email, "job_id": job_id},
|
|
commit=False,
|
|
)
|
|
db.commit()
|
|
return {"shared_with": recipient.email}
|
|
|
|
|
|
@router.delete("/{job_id}/share/{email}")
|
|
def unshare_download(
|
|
job_id: int,
|
|
email: str,
|
|
user: User = Depends(require_human),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
_own_job(db, user, job_id) # ownership guard (404s if not the caller's job)
|
|
recipient = db.execute(
|
|
select(User).where(func.lower(User.email) == email.strip().lower())
|
|
).scalar_one_or_none()
|
|
if recipient is not None:
|
|
row = db.execute(
|
|
select(DownloadShare).where(
|
|
DownloadShare.job_id == job_id,
|
|
DownloadShare.to_user_id == recipient.id,
|
|
)
|
|
).scalar_one_or_none()
|
|
if row is not None:
|
|
db.delete(row)
|
|
db.commit()
|
|
return {"ok": True}
|
|
|
|
|
|
@router.delete("/shared/{job_id}")
|
|
def remove_shared_with_me(
|
|
job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
"""Remove a download that was shared WITH this user from their 'Shared with me' list. Deletes
|
|
only the recipient's share grant — the owner's job and the physical file are untouched (this is
|
|
a per-user dismissal, not a delete)."""
|
|
row = db.execute(
|
|
select(DownloadShare).where(
|
|
DownloadShare.job_id == job_id, DownloadShare.to_user_id == user.id
|
|
)
|
|
).scalar_one_or_none()
|
|
if row is not None:
|
|
db.delete(row)
|
|
db.commit()
|
|
return {"removed": job_id}
|
|
|
|
|
|
# --- recipients (for the internal "share with a user" picker) ------------------------------
|
|
|
|
@router.get("/recipients")
|
|
def share_recipients(
|
|
user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> list[dict]:
|
|
"""Registered users this user can share a download with (excludes self + the demo account)."""
|
|
rows = (
|
|
db.execute(
|
|
select(User)
|
|
.where(User.id != user.id, User.is_demo.is_(False))
|
|
.order_by(func.lower(User.email))
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
return [{"id": u.id, "email": u.email, "display_name": u.display_name} for u in rows]
|
|
|
|
|
|
# --- public watch links (share by link) ----------------------------------------------------
|
|
|
|
def _own_link(db: Session, user: User, link_id: int) -> DownloadLink:
|
|
link = db.get(DownloadLink, link_id)
|
|
if link is None:
|
|
raise HTTPException(status_code=404, detail="Unknown link")
|
|
job = db.get(DownloadJob, link.job_id)
|
|
if job is None or job.user_id != user.id:
|
|
raise HTTPException(status_code=404, detail="Unknown link")
|
|
return link
|
|
|
|
|
|
def _link_expiry(payload: dict) -> datetime | None:
|
|
days = payload.get("expires_days")
|
|
if days in (None, "", 0, "0"):
|
|
return None
|
|
try:
|
|
d = int(days)
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(status_code=400, detail="expires_days must be a number")
|
|
if d <= 0:
|
|
return None
|
|
return datetime.now(timezone.utc) + timedelta(days=min(d, 3650))
|
|
|
|
|
|
@router.get("/{job_id}/links")
|
|
def list_links(
|
|
job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> list[dict]:
|
|
_own_job(db, user, job_id)
|
|
rows = (
|
|
db.execute(
|
|
select(DownloadLink).where(DownloadLink.job_id == job_id).order_by(DownloadLink.id.desc())
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
return [linksmod.owner_view(row) for row in rows]
|
|
|
|
|
|
@router.post("/{job_id}/links")
|
|
def create_link(
|
|
job_id: int,
|
|
payload: dict,
|
|
user: User = Depends(require_human),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
job = _own_job(db, user, job_id)
|
|
if job.status != "done" or job.asset_id is None:
|
|
raise HTTPException(status_code=409, detail="Only a finished download can be shared.")
|
|
pw = (payload.get("password") or "").strip()
|
|
link = DownloadLink(
|
|
job_id=job_id,
|
|
token=linksmod.new_token(),
|
|
created_by=user.id,
|
|
allow_download=bool(payload.get("allow_download")),
|
|
password_hash=hash_password(pw) if pw else None,
|
|
expires_at=_link_expiry(payload),
|
|
)
|
|
db.add(link)
|
|
db.commit()
|
|
db.refresh(link)
|
|
return linksmod.owner_view(link)
|
|
|
|
|
|
@router.patch("/links/{link_id}")
|
|
def update_link(
|
|
link_id: int,
|
|
payload: dict,
|
|
user: User = Depends(require_human),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
link = _own_link(db, user, link_id)
|
|
if "allow_download" in payload:
|
|
link.allow_download = bool(payload["allow_download"])
|
|
if "expires_days" in payload:
|
|
link.expires_at = _link_expiry(payload)
|
|
if "password" in payload:
|
|
pw = (payload.get("password") or "").strip()
|
|
link.password_hash = hash_password(pw) if pw else None
|
|
db.commit()
|
|
return linksmod.owner_view(link)
|
|
|
|
|
|
@router.delete("/links/{link_id}")
|
|
def revoke_link(
|
|
link_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
link = _own_link(db, user, link_id)
|
|
db.delete(link)
|
|
db.commit()
|
|
return {"deleted": link_id}
|
|
|
|
|
|
# --- admin ---------------------------------------------------------------------------------
|
|
|
|
@admin_router.get("")
|
|
def admin_list_downloads(
|
|
admin: User = Depends(admin_user), db: Session = Depends(get_db)
|
|
) -> list[dict]:
|
|
jobs = (
|
|
db.execute(
|
|
select(DownloadJob).order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc()).limit(500)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
assets = _assets_for(db, jobs)
|
|
emails = {
|
|
u.id: u.email
|
|
for u in db.execute(
|
|
select(User).where(User.id.in_({j.user_id for j in jobs}))
|
|
).scalars()
|
|
}
|
|
out = []
|
|
for j in jobs:
|
|
data = _serialize(j, assets.get(j.asset_id))
|
|
data["user_email"] = emails.get(j.user_id)
|
|
out.append(data)
|
|
return out
|
|
|
|
|
|
@admin_router.get("/storage")
|
|
def admin_storage(
|
|
admin: User = Depends(admin_user), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
ready = db.execute(
|
|
select(func.count(), func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where(
|
|
MediaAsset.status == "ready"
|
|
)
|
|
).one()
|
|
total_assets = db.execute(select(func.count()).select_from(MediaAsset)).scalar()
|
|
per_user = db.execute(
|
|
select(DownloadJob.user_id, func.count(func.distinct(DownloadJob.asset_id)))
|
|
.where(DownloadJob.asset_id.is_not(None))
|
|
.group_by(DownloadJob.user_id)
|
|
).all()
|
|
emails = {
|
|
u.id: u.email
|
|
for u in db.execute(
|
|
select(User).where(User.id.in_([uid for uid, _ in per_user]))
|
|
).scalars()
|
|
}
|
|
return {
|
|
"ready_files": ready[0],
|
|
"total_bytes": int(ready[1]),
|
|
"total_assets": total_assets,
|
|
"total_cap_bytes": sysconfig.get_int(db, "download_total_max_bytes"),
|
|
"per_user": [
|
|
{"user_id": uid, "email": emails.get(uid), "footprint_bytes": quota.footprint(db, uid)}
|
|
for uid, _ in per_user
|
|
],
|
|
}
|
|
|
|
|
|
@admin_router.get("/quota/{user_id}")
|
|
def admin_get_quota(
|
|
user_id: int, admin: User = Depends(admin_user), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
lim = quota.resolve(db, user_id)
|
|
row = db.get(DownloadQuota, user_id)
|
|
return {
|
|
"user_id": user_id,
|
|
"custom": row is not None,
|
|
"max_bytes": lim.max_bytes,
|
|
"max_concurrent": lim.max_concurrent,
|
|
"max_jobs": lim.max_jobs,
|
|
"unlimited": lim.unlimited,
|
|
}
|
|
|
|
|
|
@admin_router.put("/quota/{user_id}")
|
|
def admin_set_quota(
|
|
user_id: int,
|
|
payload: dict,
|
|
admin: User = Depends(admin_user),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
if db.get(User, user_id) is None:
|
|
raise HTTPException(status_code=404, detail="Unknown user")
|
|
cur = quota.resolve(db, user_id)
|
|
row = db.get(DownloadQuota, user_id)
|
|
if row is None:
|
|
row = DownloadQuota(
|
|
user_id=user_id,
|
|
max_bytes=cur.max_bytes,
|
|
max_concurrent=cur.max_concurrent,
|
|
max_jobs=cur.max_jobs,
|
|
unlimited=cur.unlimited,
|
|
)
|
|
db.add(row)
|
|
if "max_bytes" in payload:
|
|
row.max_bytes = max(0, int(payload["max_bytes"]))
|
|
if "max_concurrent" in payload:
|
|
row.max_concurrent = max(1, int(payload["max_concurrent"]))
|
|
if "max_jobs" in payload:
|
|
row.max_jobs = max(1, int(payload["max_jobs"]))
|
|
if "unlimited" in payload:
|
|
row.unlimited = bool(payload["unlimited"])
|
|
db.commit()
|
|
return admin_get_quota(user_id, admin, db)
|
|
|
|
|
|
@admin_router.delete("/quota/{user_id}")
|
|
def admin_reset_quota(
|
|
user_id: int, admin: User = Depends(admin_user), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
row = db.get(DownloadQuota, user_id)
|
|
if row is not None:
|
|
db.delete(row)
|
|
db.commit()
|
|
return admin_get_quota(user_id, admin, db)
|