33fad8911b
A download whose source title exceeded 255 chars (e.g. a recipe baked into a Facebook video's title) failed with a StringDataRightTruncation: the worker auto-fills an empty display_name from the title, which overflows the 255-char column. Clip the title to 255 at every point it flows into display_name (worker completion paths + enqueue); the full title is untouched on the asset. Pre-existing latent bug, surfaced by an unusually long title. Release v0.31.1.
200 lines
7 KiB
Python
200 lines
7 KiB
Python
"""Enqueue + cache resolution for the download center (the per-user layer over MediaAsset).
|
|
|
|
`enqueue` is the single entry point used by the API: it resolves (or creates) the shared
|
|
MediaAsset for the requested source+format, links a per-user DownloadJob to it, and — on a
|
|
cache hit (asset already ready) — marks the job done instantly with no re-download. The worker
|
|
handles everything else. State transitions used by the routes (pause/resume/cancel/delete) and
|
|
the worker live in `worker.py` / the routes; this module owns enqueue + asset bookkeeping.
|
|
"""
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.downloads import edit as editmod
|
|
from app.downloads import formats, quota
|
|
from app.models import Channel, DownloadJob, MediaAsset, Video
|
|
|
|
|
|
class EditError(Exception):
|
|
"""Bad edit request (empty spec / source not ready); `reason` is an i18n-mappable key."""
|
|
|
|
def __init__(self, reason: str):
|
|
self.reason = reason
|
|
super().__init__(reason)
|
|
|
|
|
|
def source_url(source_kind: str, source_ref: str) -> str:
|
|
"""The URL yt-dlp should fetch. YouTube stores the 11-char id; ad-hoc stores the URL."""
|
|
if source_kind == "youtube":
|
|
return f"https://www.youtube.com/watch?v={source_ref}"
|
|
return source_ref
|
|
|
|
|
|
def get_or_create_asset(
|
|
db: Session, source_kind: str, source_ref: str, format_sig: str
|
|
) -> MediaAsset:
|
|
"""Fetch the cache row for this (source, format) or create a fresh `pending` one.
|
|
|
|
Races on the unique cache key are resolved by re-selecting after an IntegrityError, so two
|
|
simultaneous enqueues of the same video+format converge on one shared asset."""
|
|
stmt = select(MediaAsset).where(
|
|
MediaAsset.source_kind == source_kind,
|
|
MediaAsset.source_ref == source_ref,
|
|
MediaAsset.format_sig == format_sig,
|
|
)
|
|
asset = db.execute(stmt).scalar_one_or_none()
|
|
if asset is not None:
|
|
return asset
|
|
asset = MediaAsset(
|
|
source_kind=source_kind,
|
|
source_ref=source_ref,
|
|
format_sig=format_sig,
|
|
status="pending",
|
|
ref_count=0,
|
|
)
|
|
db.add(asset)
|
|
try:
|
|
db.flush()
|
|
except IntegrityError:
|
|
db.rollback()
|
|
asset = db.execute(stmt).scalar_one()
|
|
return asset
|
|
|
|
|
|
def populate_from_catalog(db: Session, asset: MediaAsset) -> bool:
|
|
"""Fill an asset's display metadata (title/uploader/thumbnail/date/duration) from our own
|
|
catalog when the source is a known YouTube video — so a queued/downloading row shows the
|
|
real title + thumbnail immediately instead of the bare id. 0 network cost. Returns True if
|
|
it found a catalog match."""
|
|
if asset.source_kind != "youtube" or asset.title:
|
|
return False
|
|
video = db.get(Video, asset.source_ref)
|
|
if video is None:
|
|
return False
|
|
asset.title = video.title
|
|
asset.thumbnail_url = video.thumbnail_url
|
|
asset.duration_s = video.duration_seconds
|
|
if video.published_at:
|
|
asset.upload_date = video.published_at.date()
|
|
channel = db.get(Channel, video.channel_id)
|
|
if channel is not None:
|
|
asset.uploader = channel.title
|
|
return True
|
|
|
|
|
|
def _next_queue_pos(db: Session) -> int:
|
|
return (db.execute(select(func.coalesce(func.max(DownloadJob.queue_pos), 0))).scalar() or 0) + 1
|
|
|
|
|
|
def enqueue(
|
|
db: Session,
|
|
user_id: int,
|
|
source_kind: str,
|
|
source_ref: str,
|
|
spec: dict,
|
|
profile_id: int | None = None,
|
|
display_name: str | None = None,
|
|
) -> DownloadJob:
|
|
"""Queue a download for `user_id`. Returns the DownloadJob (already `done` on a cache hit).
|
|
|
|
Raises quota.QuotaExceeded if the user is at their job-count or footprint cap."""
|
|
quota.check_enqueue(db, user_id)
|
|
spec = formats.normalize(spec)
|
|
sig = formats.format_sig(spec)
|
|
asset = get_or_create_asset(db, source_kind, source_ref, sig)
|
|
# Show a real title + thumbnail the moment it's queued (feed downloads are catalog videos).
|
|
populate_from_catalog(db, asset)
|
|
|
|
job = DownloadJob(
|
|
user_id=user_id,
|
|
asset_id=asset.id,
|
|
source_kind=source_kind,
|
|
source_ref=source_ref,
|
|
profile_id=profile_id,
|
|
profile_snapshot=spec,
|
|
display_name=display_name or (asset.title[:255] if asset.status == "ready" and asset.title else None),
|
|
status="queued",
|
|
queue_pos=_next_queue_pos(db),
|
|
)
|
|
db.add(job)
|
|
asset.ref_count = (asset.ref_count or 0) + 1
|
|
|
|
# Cache hit: the file already exists — no worker round-trip needed.
|
|
if asset.status == "ready":
|
|
job.status = "done"
|
|
job.progress = 100
|
|
asset.last_access_at = datetime.now(timezone.utc)
|
|
|
|
db.commit()
|
|
db.refresh(job)
|
|
return job
|
|
|
|
|
|
def _clip_title(source_title: str | None, display_name: str | None) -> str:
|
|
if display_name:
|
|
return display_name[:255]
|
|
base = source_title or "Clip"
|
|
return f"{base} (clip)"[:255]
|
|
|
|
|
|
def enqueue_edit(
|
|
db: Session,
|
|
user_id: int,
|
|
source_job: DownloadJob,
|
|
edit_spec: dict,
|
|
display_name: str | None = None,
|
|
) -> DownloadJob:
|
|
"""Queue a phase-2 editor derivative (trim/crop) of `source_job` for `user_id`.
|
|
|
|
The derivative is per-user (never shared-cached): its asset uses `source_kind='edit'` with a
|
|
cache key that embeds the user id, so an identical re-edit by the same user is a cache hit but
|
|
it never dedups across users. Reuses the same worker/quota/GC path as a download.
|
|
|
|
Raises quota.QuotaExceeded (holdings cap) or EditError (empty spec / source not ready)."""
|
|
src_asset = db.get(MediaAsset, source_job.asset_id) if source_job.asset_id else None
|
|
if src_asset is None or src_asset.status != "ready" or not src_asset.rel_path:
|
|
raise EditError("source_not_ready")
|
|
|
|
spec = editmod.normalize_edit_spec(edit_spec)
|
|
if not spec:
|
|
raise EditError("empty_edit")
|
|
|
|
quota.check_enqueue(db, user_id)
|
|
|
|
sig = editmod.edit_sig(spec)
|
|
ref = f"{user_id}:{src_asset.id}"[:512]
|
|
asset = get_or_create_asset(db, "edit", ref, sig)
|
|
if not asset.title:
|
|
asset.title = _clip_title(src_asset.title, display_name)
|
|
asset.uploader = src_asset.uploader
|
|
asset.thumbnail_url = src_asset.thumbnail_url
|
|
asset.upload_date = src_asset.upload_date
|
|
asset.duration_s = editmod.clip_duration(spec, src_asset.duration_s)
|
|
|
|
job = DownloadJob(
|
|
user_id=user_id,
|
|
asset_id=asset.id,
|
|
job_kind="edit",
|
|
source_kind="edit",
|
|
source_ref=ref,
|
|
source_job_id=source_job.id,
|
|
source_asset_id=src_asset.id,
|
|
edit_spec=spec,
|
|
profile_snapshot={},
|
|
display_name=display_name or asset.title,
|
|
status="queued",
|
|
queue_pos=_next_queue_pos(db),
|
|
)
|
|
db.add(job)
|
|
asset.ref_count = (asset.ref_count or 0) + 1
|
|
|
|
if asset.status == "ready": # identical clip already produced for this user → instant
|
|
job.status = "done"
|
|
job.progress = 100
|
|
asset.last_access_at = datetime.now(timezone.utc)
|
|
|
|
db.commit()
|
|
db.refresh(job)
|
|
return job
|