feat(downloads): M4 — REST API (enqueue, manage, file-serve, sharing, admin)

routes/downloads.py (require_human; admin_router adds admin_user):
- profiles: list (builtins + own), create/update/delete custom
- enqueue: resolve_source (bare id / watch / youtu.be / shorts URLs -> youtube id; else
  raw URL for the generic extractor), profile_id or inline spec or builtin fallback; maps
  quota.QuotaExceeded -> 422 speaking message
- list / usage / shared-with-me; rename (display name only); pause/resume/cancel/delete
  (ref_count bookkeeping)
- GET /{id}/file: ownership-or-share check, path-traversal guard, range-aware FileResponse
  with the user's custom display name (Content-Disposition), bumps last_access
- share/unshare by email + a download_shared notification
- admin: all-jobs, storage dashboard (totals + per-user footprint), per-user quota GET/PUT/DELETE
- cast SUM(bigint) footprints to int for clean numeric JSON
- wired both routers in main.py

Verified via TestClient: full enqueue->download->206 range fetch->share->access-control
(cross-user delete 404); unauth = 401 on user + admin routes; all 15 paths registered.
This commit is contained in:
npeter83 2026-07-03 00:34:08 +02:00
parent 7148335c09
commit 51158ad9cf
3 changed files with 623 additions and 1 deletions

View file

@ -63,7 +63,7 @@ def footprint(db: Session, user_id: int) -> int:
) )
.distinct() .distinct()
) )
return ( return int(
db.execute( db.execute(
select(func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where( select(func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where(
MediaAsset.id.in_(held_assets), MediaAsset.status == "ready" MediaAsset.id.in_(held_assets), MediaAsset.status == "ready"

View file

@ -31,6 +31,7 @@ from app.routes import (
admin, admin,
channels, channels,
config as config_routes, config as config_routes,
downloads,
feed, feed,
health, health,
me, me,
@ -122,6 +123,8 @@ app.include_router(messages.router)
app.include_router(channels.router) app.include_router(channels.router)
app.include_router(playlists.router) app.include_router(playlists.router)
app.include_router(saved_views.router) app.include_router(saved_views.router)
app.include_router(downloads.router)
app.include_router(downloads.admin_router)
app.include_router(admin.router) app.include_router(admin.router)
app.include_router(config_routes.router) app.include_router(config_routes.router)
app.include_router(scheduler_routes.router) app.include_router(scheduler_routes.router)

View file

@ -0,0 +1,619 @@
"""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, timezone
from pathlib import Path
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.auth import admin_user, require_human
from app.config import settings
from app.db import get_db
from app.downloads import quota, service, storage
from app.models import (
DownloadJob,
DownloadProfile,
DownloadQuota,
DownloadShare,
MediaAsset,
User,
)
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}$")
_BASENAME_ILLEGAL = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
# --- 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,
"profile_id": job.profile_id,
"spec": job.profile_snapshot,
"display_name": job.display_name,
"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,
"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 _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_message(e))
asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
return _serialize(job, asset)
def _quota_message(e: quota.QuotaExceeded) -> str:
if e.reason == "max_jobs":
return f"You've reached your download limit ({e.limit} items). Remove some first."
if e.reason == "max_bytes":
gb = e.limit / 1_073_741_824
return f"You've used your download storage quota ({gb:.1f} GB). Free up space first."
return "Download quota exceeded."
@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)
@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
@router.patch("/{job_id}")
def rename_download(
job_id: int,
payload: dict,
user: User = Depends(require_human),
db: Session = Depends(get_db),
) -> dict:
job = _own_job(db, user, job_id)
name = (payload.get("display_name") or "").strip()
# Display name only — the on-disk file keeps its system-decided, id-bound name.
job.display_name = name[:255] or None
db.commit()
asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
return _serialize(job, asset)
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
asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
return _serialize(job, asset)
@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
_set_status(db, job, "queued")
asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
return _serialize(job, asset)
@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"):
job.status = "canceled"
_decrement_ref(db, job)
db.commit()
asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
return _serialize(job, asset)
@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)
_decrement_ref(db, job)
db.delete(job)
db.commit()
return {"deleted": job_id}
def _decrement_ref(db: Session, job: DownloadJob) -> None:
"""Drop this job's hold on its asset. The asset itself lingers as cache until its TTL /
eviction (so a re-add is still a cache hit); only the reference count changes here."""
if job.asset_id and job.status not in ("canceled", "error"):
asset = db.get(MediaAsset, job.asset_id)
if asset and asset.ref_count > 0:
asset.ref_count -= 1
# --- file download (range-aware, custom display name) --------------------------------------
def _clean_basename(name: str) -> str:
return _BASENAME_ILLEGAL.sub("", name).strip() or "download"
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.abs_path(settings.download_root, asset.rel_path).resolve()
root = Path(settings.download_root).resolve()
if root not in path.parents or not path.exists():
raise HTTPException(status_code=404, detail="File is no longer available.")
ext = asset.container or path.suffix.lstrip(".")
base = _clean_basename(job.display_name or asset.title or asset.source_ref)
if base.lower().endswith(f".{ext.lower()}"):
base = base[: -(len(ext) + 1)]
filename = f"{base}.{ext}"
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)
# --- 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:
job = _own_job(db, user, job_id)
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}
# --- 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": settings.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)