cb170dfd32
Two visual gaps for non-catalog downloads: - Channel link: YouTube already exposes channel_url; Facebook exposes none but a numeric uploader_id that resolves at facebook.com/<id>. `_uploader_url` derives it so the auto-detected channel renders as a real clickable link. - Poster: a source with no thumbnail (e.g. a direct reddit HLS URL) showed a blank image box. The worker now cuts a representative frame with ffmpeg (`ensure_poster`) into the `<base>.jpg` sidecar and records `poster_path` (migration 0050). The card, the public watch page (<video poster> + og:image), and link previews fall back to it via new authed + public poster endpoints. Adds `app.downloads.backfill` (one-off, re-run-safe) to fill uploader_url (re-extract YouTube/Facebook metadata) and posters for pre-existing downloads.
97 lines
3.7 KiB
Python
97 lines
3.7 KiB
Python
"""Public "share by link" helpers for the download center.
|
|
|
|
A DownloadLink is a capability URL: the unguessable `token` grants anyone read access to one
|
|
download's file on the login-free `/watch/{token}` page. Password-protected links additionally
|
|
require a short-lived signed **grant** on the file request — because a `<video src>` can't send a
|
|
header, the password is exchanged once at `/unlock` for an HMAC grant appended to the file URL,
|
|
so the raw password never rides in the media URL (or server logs).
|
|
"""
|
|
import hashlib
|
|
import hmac
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
|
|
from app.config import settings
|
|
from app.models import DownloadLink
|
|
|
|
_GRANT_TTL = 6 * 3600 # a watch session's grant is good for 6h, then the viewer re-unlocks
|
|
|
|
|
|
def new_token() -> str:
|
|
return secrets.token_urlsafe(24) # ~32 chars, 192 bits
|
|
|
|
|
|
def is_expired(link: DownloadLink) -> bool:
|
|
return link.expires_at is not None and link.expires_at <= datetime.now(timezone.utc)
|
|
|
|
|
|
# --- signed grants (password-protected file access) ----------------------------------------
|
|
|
|
def _sign(msg: str) -> str:
|
|
return hmac.new(
|
|
settings.secret_key.encode(), msg.encode(), hashlib.sha256
|
|
).hexdigest()[:32]
|
|
|
|
|
|
def make_grant(token: str, now: float | None = None) -> str:
|
|
"""A short-lived HMAC grant proving the viewer unlocked `token`. Format: `<exp>.<sig>`."""
|
|
exp = int((now if now is not None else datetime.now(timezone.utc).timestamp())) + _GRANT_TTL
|
|
return f"{exp}.{_sign(f'{token}.{exp}')}"
|
|
|
|
|
|
def check_grant(token: str, grant: str | None) -> bool:
|
|
if not grant or "." not in grant:
|
|
return False
|
|
exp_s, sig = grant.split(".", 1)
|
|
try:
|
|
exp = int(exp_s)
|
|
except ValueError:
|
|
return False
|
|
if exp < datetime.now(timezone.utc).timestamp():
|
|
return False
|
|
return hmac.compare_digest(sig, _sign(f"{token}.{exp}"))
|
|
|
|
|
|
# --- serialization -------------------------------------------------------------------------
|
|
|
|
def owner_view(link: DownloadLink, base_url: str = "") -> dict:
|
|
"""What the file owner sees when managing their links."""
|
|
return {
|
|
"id": link.id,
|
|
"token": link.token,
|
|
"url": f"{base_url}/watch/{link.token}",
|
|
"allow_download": link.allow_download,
|
|
"has_password": link.password_hash is not None,
|
|
"expires_at": link.expires_at.isoformat() if link.expires_at else None,
|
|
"view_count": link.view_count,
|
|
"created_at": link.created_at.isoformat() if link.created_at else None,
|
|
}
|
|
|
|
|
|
def public_meta(link: DownloadLink, job, asset, file_url: str, poster_url: str | None = None) -> dict:
|
|
"""Metadata for the public watch page (only ever returned once access is authorized).
|
|
|
|
Resolves the owner's per-download display overrides (custom title/channel) over the asset's
|
|
auto-extracted values, and surfaces the source + any extra reference links as clickable URLs."""
|
|
from app.downloads import service
|
|
|
|
source_url = None
|
|
if job is not None and job.job_kind != "edit":
|
|
source_url = asset.source_webpage_url or (
|
|
service.source_url(job.source_kind, job.source_ref)
|
|
if job.source_kind in ("youtube", "url")
|
|
else None
|
|
)
|
|
return {
|
|
"title": (job and job.display_name) or asset.title,
|
|
"uploader": (job and job.display_uploader) or asset.uploader,
|
|
"uploader_url": (job and job.display_uploader_url) or asset.uploader_url,
|
|
"source_url": source_url,
|
|
"extra_links": (job and job.extra_links) or [],
|
|
"duration_s": asset.duration_s,
|
|
"width": asset.width,
|
|
"height": asset.height,
|
|
"allow_download": link.allow_download,
|
|
"file_url": file_url,
|
|
"poster_url": poster_url,
|
|
}
|