diff --git a/backend/app/demo.py b/backend/app/demo.py new file mode 100644 index 00000000..80f2d10f --- /dev/null +++ b/backend/app/demo.py @@ -0,0 +1,77 @@ +"""Demo-account service: reset the shared demo sandbox to a clean, seeded baseline. + +Shared by the admin's manual reset endpoint (routes/admin.py) and the scheduled demo-reset job +(scheduler.py). Lives in the service layer, not routes, so the scheduler can call it without an +`app.routes -> app.scheduler` import cycle (the reason its import used to be lazy). +""" +from sqlalchemy import delete, func, select +from sqlalchemy.orm import Session + +from app.models import Playlist, PlaylistItem, Subscription, User, Video, VideoState + + +def _seed_demo_playlists(db: Session, demo: User) -> int: + """Seed a few sample playlists from the shared catalog so a freshly-reset demo isn't + empty — gives visitors something to play with. Best-effort: skips silently if the + catalog is too small.""" + recent = ( + db.execute( + select(Video.id) + .where(Video.is_short.is_(False), Video.live_status == "none") + .order_by(Video.published_at.desc().nullslast()) + .limit(60) + ) + .scalars() + .all() + ) + if not recent: + return 0 + samples = [ + ("Fresh picks", recent[0:10]), + ("Quick watch", recent[10:18]), + ("Saved for later", recent[18:26]), + ] + created = 0 + for pos, (name, vids) in enumerate(samples): + if not vids: + continue + pl = Playlist(user_id=demo.id, name=name, position=pos) + db.add(pl) + db.flush() + for ipos, vid in enumerate(vids): + db.add(PlaylistItem(playlist_id=pl.id, video_id=vid, position=ipos)) + created += 1 + return created + + +def _seed_demo_subscriptions(db: Session, demo: User, n: int = 14) -> int: + """Seed the demo with a handful of catalog channels (the ones with the most videos) so its + Channel manager isn't empty — gives visitors something to explore and matches the landing-page + screenshot. Idempotent: clears the demo's subscriptions first, then re-adds the top N.""" + db.execute(delete(Subscription).where(Subscription.user_id == demo.id)) + top = db.execute( + select(Video.channel_id) + .where(Video.channel_id.is_not(None)) + .group_by(Video.channel_id) + .order_by(func.count().desc()) + .limit(n) + ).scalars().all() + for channel_id in top: + db.add(Subscription(user_id=demo.id, channel_id=channel_id)) + return len(top) + + +def reset_demo_state(db: Session, demo: User) -> int: + """Wipe the demo account's per-user state (watch/save/hide states, playlists, preferences) + back to a clean baseline and re-seed a few sample playlists + channel subscriptions. Returns + the seeded playlist count. Shared by the admin's manual reset button and the scheduled demo + reset (see scheduler).""" + db.execute(delete(VideoState).where(VideoState.user_id == demo.id)) + # Playlist items cascade on playlist delete (ondelete="CASCADE"). + db.execute(delete(Playlist).where(Playlist.user_id == demo.id)) + demo.preferences = {"language": "en"} + db.commit() + seeded = _seed_demo_playlists(db, demo) + _seed_demo_subscriptions(db, demo) + db.commit() + return seeded diff --git a/backend/app/downloads/backfill.py b/backend/app/downloads/backfill.py index de99c412..af751966 100644 --- a/backend/app/downloads/backfill.py +++ b/backend/app/downloads/backfill.py @@ -21,7 +21,7 @@ from app.db import SessionLocal from app.downloads import edit as editmod from app.downloads import formats, service from app.models import MediaAsset -from app.worker import _uploader_url +from app.worker import uploader_url log = logging.getLogger("siftlode.backfill") @@ -73,7 +73,7 @@ def run() -> dict: if a.uploader_url is None and _should_refetch(a): url = a.source_webpage_url or service.source_url(a.source_kind, a.source_ref) info = _extract_meta(url) - uu = _uploader_url(info) if info else None + uu = uploader_url(info) if info else None if uu: a.uploader_url = uu filled_url += 1 diff --git a/backend/app/downloads/service.py b/backend/app/downloads/service.py index 1e4514de..d6fe7a17 100644 --- a/backend/app/downloads/service.py +++ b/backend/app/downloads/service.py @@ -12,8 +12,9 @@ from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session +from app.config import settings from app.downloads import edit as editmod -from app.downloads import formats, quota +from app.downloads import formats, quota, storage from app.models import Channel, DownloadJob, MediaAsset, Video @@ -220,3 +221,30 @@ def enqueue_edit( db.commit() db.refresh(job) return job + + +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) diff --git a/backend/app/plex/people.py b/backend/app/plex/people.py index 1142e7dd..7f5652c7 100644 --- a/backend/app/plex/people.py +++ b/backend/app/plex/people.py @@ -74,7 +74,7 @@ def enrich_row_people(db: Session, row, cast: list[dict]) -> bool: # items in the same batch: SessionLocal is autoflush=False, so the ORM query above can't see a # same-batch pending insert, and this path also races the concurrent info-page enrichment. The # Core execute runs immediately (not deferred to the caller's commit), so a later row in the - # batch sees the row and skips it. Mirrors app/sync/videos.py:_insert_stubs. + # batch sees the row and skips it. Mirrors app/sync/videos.py:insert_stubs. db.execute( pg_insert(PlexPerson) .values([{"name": nm, "photo_thumb": th} for nm, th in missing.items()]) diff --git a/backend/app/routes/admin.py b/backend/app/routes/admin.py index 3a1f5ad9..c4bbfa3b 100644 --- a/backend/app/routes/admin.py +++ b/backend/app/routes/admin.py @@ -4,7 +4,7 @@ from datetime import datetime, timezone from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException from pydantic import BaseModel -from sqlalchemy import delete, func, select +from sqlalchemy import select from sqlalchemy.orm import Session from app import audit @@ -18,15 +18,11 @@ from app.auth import ( purge_user, ) from app.db import get_db +from app.demo import reset_demo_state from app.models import ( DemoWhitelist, Invite, - Playlist, - PlaylistItem, - Subscription, User, - Video, - VideoState, ) from app.utils import iso, valid_email @@ -365,73 +361,6 @@ def remove_demo_whitelist( return {"deleted": entry_id} -def _seed_demo_playlists(db: Session, demo: User) -> int: - """Seed a few sample playlists from the shared catalog so a freshly-reset demo isn't - empty — gives visitors something to play with. Best-effort: skips silently if the - catalog is too small.""" - recent = ( - db.execute( - select(Video.id) - .where(Video.is_short.is_(False), Video.live_status == "none") - .order_by(Video.published_at.desc().nullslast()) - .limit(60) - ) - .scalars() - .all() - ) - if not recent: - return 0 - samples = [ - ("Fresh picks", recent[0:10]), - ("Quick watch", recent[10:18]), - ("Saved for later", recent[18:26]), - ] - created = 0 - for pos, (name, vids) in enumerate(samples): - if not vids: - continue - pl = Playlist(user_id=demo.id, name=name, position=pos) - db.add(pl) - db.flush() - for ipos, vid in enumerate(vids): - db.add(PlaylistItem(playlist_id=pl.id, video_id=vid, position=ipos)) - created += 1 - return created - - -def _seed_demo_subscriptions(db: Session, demo: User, n: int = 14) -> int: - """Seed the demo with a handful of catalog channels (the ones with the most videos) so its - Channel manager isn't empty — gives visitors something to explore and matches the landing-page - screenshot. Idempotent: clears the demo's subscriptions first, then re-adds the top N.""" - db.execute(delete(Subscription).where(Subscription.user_id == demo.id)) - top = db.execute( - select(Video.channel_id) - .where(Video.channel_id.is_not(None)) - .group_by(Video.channel_id) - .order_by(func.count().desc()) - .limit(n) - ).scalars().all() - for channel_id in top: - db.add(Subscription(user_id=demo.id, channel_id=channel_id)) - return len(top) - - -def reset_demo_state(db: Session, demo: User) -> int: - """Wipe the demo account's per-user state (watch/save/hide states, playlists, preferences) - back to a clean baseline and re-seed a few sample playlists + channel subscriptions. Returns - the seeded playlist count. Shared by the admin's manual reset button and the scheduled demo - reset (see scheduler).""" - db.execute(delete(VideoState).where(VideoState.user_id == demo.id)) - # Playlist items cascade on playlist delete (ondelete="CASCADE"). - db.execute(delete(Playlist).where(Playlist.user_id == demo.id)) - demo.preferences = {"language": "en"} - db.commit() - seeded = _seed_demo_playlists(db, demo) - _seed_demo_subscriptions(db, demo) - db.commit() - return seeded - - @router.post("/demo/reset") def reset_demo( admin: User = Depends(admin_user), db: Session = Depends(get_db) diff --git a/backend/app/routes/downloads.py b/backend/app/routes/downloads.py index 7c1c9655..aa10e5ce 100644 --- a/backend/app/routes/downloads.py +++ b/backend/app/routes/downloads.py @@ -511,9 +511,9 @@ def redownload_download( asset = service.get_or_create_asset(db, job.source_kind, job.source_ref, sig) service.populate_from_catalog(db, asset) - # Now drop the old hold — `_release_asset` deletes the old file + row only if this job was its + # Now drop the old hold — `service.release_asset` deletes the old file + row only if this job was its # last holder (a co-held asset is left intact for its other holders). - _release_asset(db, job) + service.release_asset(db, job) asset.ref_count = (asset.ref_count or 0) + 1 job.asset_id = asset.id @@ -538,7 +538,7 @@ def cancel_download( ) -> 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 + service.release_asset(db, job) # release while the job still counts as holding job.status = "canceled" db.commit() return _serialize_job(db, job) @@ -549,39 +549,12 @@ 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) + service.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: diff --git a/backend/app/routes/feed.py b/backend/app/routes/feed.py index 89d0367a..5591ee77 100644 --- a/backend/app/routes/feed.py +++ b/backend/app/routes/feed.py @@ -91,7 +91,7 @@ def feed_columns(user: User, state) -> list: ] -def _serialize(row) -> dict: +def serialize(row) -> dict: return { "id": row.id, "title": row.title, @@ -258,7 +258,7 @@ def _filtered_query( # channel searches work. `rank_expr` (weight-aware ts_rank) drives the "relevance" sort. rank_expr = None if q: - ts_str = _to_tsquery_str(q) + ts_str = to_tsquery_str(q) if ts_str: tsq = func.to_tsquery("public.unaccent_simple", ts_str) query = query.where( @@ -341,7 +341,7 @@ def _filtered_query( return query, rank_expr -def _to_tsquery_str(q: str) -> str | None: +def to_tsquery_str(q: str) -> str | None: """Turn free-text into a to_tsquery() string: word runs ANDed together, the LAST word prefix-matched (`:*`) for as-you-type search — e.g. "ti amo magyar" → "ti & amo & magyar:*". Only `\\w` runs containing an alphanumeric are kept, so no tsquery operators can leak in @@ -512,7 +512,7 @@ def get_feed( has_more = len(rows) > limit page = rows[:limit] return { - "items": [_serialize(r) for r in page], + "items": [serialize(r) for r in page], "next_cursor": _encode_cursor(keys, page[-1]) if (has_more and page) else None, "has_more": has_more, "limit": limit, diff --git a/backend/app/routes/playlists.py b/backend/app/routes/playlists.py index d61a1078..08621ea8 100644 --- a/backend/app/routes/playlists.py +++ b/backend/app/routes/playlists.py @@ -10,7 +10,7 @@ from app import quota from app.auth import current_user, has_read_scope, has_write_scope, require_write_scope from app.db import get_db from app.models import Channel, Playlist, PlaylistItem, User, Video, VideoState -from app.routes.feed import _serialize, feed_columns +from app.routes.feed import serialize, feed_columns from app.sync.playlists import ( plan_push, push_playlist, @@ -400,7 +400,7 @@ def get_playlist( "source": pl.source, "yt_playlist_id": pl.yt_playlist_id, "dirty": pl.dirty, - "items": [_serialize(r) for r in rows], + "items": [serialize(r) for r in rows], } diff --git a/backend/app/routes/plex/_shared.py b/backend/app/routes/plex/_shared.py index 53c68c58..ecd7f8c2 100644 --- a/backend/app/routes/plex/_shared.py +++ b/backend/app/routes/plex/_shared.py @@ -12,7 +12,7 @@ from app.models import ( log = logging.getLogger("siftlode.plex") -def _item_or_404(db: Session, rating_key: str) -> PlexItem: +def item_or_404(db: Session, rating_key: str) -> PlexItem: it = db.query(PlexItem).filter_by(rating_key=str(rating_key)).first() if it is None: raise HTTPException(status_code=404, detail="Unknown Plex item") diff --git a/backend/app/routes/plex/browse.py b/backend/app/routes/plex/browse.py index 54a46ef2..9d0efef1 100644 --- a/backend/app/routes/plex/browse.py +++ b/backend/app/routes/plex/browse.py @@ -32,8 +32,8 @@ from app.plex import sync as plex_sync from app.plex.client import PlexClient, PlexError, PlexNotConfigured from app.routes._common import owned_or_404 from app.routes.admin import admin_user -from app.routes.feed import _to_tsquery_str -from app.routes.plex._shared import _item_or_404 +from app.routes.feed import to_tsquery_str +from app.routes.plex._shared import item_or_404 router = APIRouter(prefix="/api/plex", tags=["plex"]) @@ -258,7 +258,7 @@ def unified_library( want_movies = scope in ("movie", "both") and bool(movie_lib_ids) want_shows = scope in ("show", "both") and bool(show_lib_ids) - ts = _to_tsquery_str(q) if q else None + ts = to_tsquery_str(q) if q else None tsq = func.to_tsquery(_TS_CONFIG, ts) if ts else None selects = [] @@ -1561,7 +1561,7 @@ def item_detail( ) -> dict: """Everything the player needs: metadata, cast, intro/credit markers (from Plex), episode prev/next, and the per-user resume position + watch status.""" - it = _item_or_404(db, rating_key) + it = item_or_404(db, rating_key) st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first() rich: dict = {} diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index eb0523ac..cbb7d6e2 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -33,7 +33,7 @@ from app.plex import people as plex_people from app.plex import stream as plex_stream from app.plex import watch_sync as plex_watch from app.plex.client import PlexClient, PlexError, PlexNotConfigured -from app.routes.plex._shared import _item_or_404 +from app.routes.plex._shared import item_or_404 router = APIRouter(prefix="/api/plex", tags=["plex"]) @@ -105,7 +105,7 @@ def subtitle( return Response(subtitles.shift_vtt(hit, offset), media_type="text/vtt") try: with SessionLocal() as db: - it = _item_or_404(db, rating_key) + it = item_or_404(db, rating_key) with PlexClient(db) as plex: part = ((plex.metadata(it.rating_key) or {}).get("Media") or [{}])[ 0 @@ -236,7 +236,7 @@ def item_progress( `final` (sent on the pause/pagehide/unmount flush, not the periodic 10s checkpoint) is what debounces the Plex resume push: we mirror the settled position to Plex only on a final flush, so a single watch doesn't spray `/:/timeline` calls at the server every 10 seconds.""" - it = _item_or_404(db, rating_key) + it = item_or_404(db, rating_key) position = max(0, int(body.position_seconds)) duration = int(body.duration_seconds) final = body.final @@ -307,7 +307,7 @@ def item_state( """Set the per-user watch status (new|watched|hidden). watched→scrobble / →new→unscrobble push to a linked Plex account immediately; `hidden` is a Siftlode-only concept with no Plex analogue, so it never touches Plex.""" - it = _item_or_404(db, rating_key) + it = item_or_404(db, rating_key) status = body.status if status not in ("new", "watched", "hidden"): raise HTTPException(status_code=400, detail="Invalid status") @@ -434,7 +434,7 @@ def stream_session( non-default audio track (`audio` = stream ordinal) forces the HLS path so the chosen audio can be muxed in. Subtitles are independent WebVTT tracks (GET /subtitle) — no session restart.""" plex_stream.reap_idle() # opportunistic cleanup of idle sessions on each new playback - it = _item_or_404(db, rating_key) + it = item_or_404(db, rating_key) if it.playable == "transcode": raise HTTPException( status_code=501, detail="This format needs transcoding (a later phase)." @@ -471,7 +471,7 @@ def stream_file( """Serve the raw local media file with HTTP range support. Used both for direct-play (inline) and for the player's "download original" button (`?download=1` → attachment). The download is the untouched physical file — no re-encode/repackage.""" - it = _item_or_404(db, rating_key) + it = item_or_404(db, rating_key) src = plex_paths.local_media_path(db, it.file_path) if src is None: raise HTTPException(status_code=404, detail="Media file not found on disk") @@ -512,7 +512,7 @@ def stream_hls_file( if s is None: # No session yet (e.g. reaped): the ENTRY playlist may auto-start one at 0 — master.m3u8 for a # multi-audio item, index.m3u8 for a single-audio remux item. - it = _item_or_404(db, rating_key) + it = item_or_404(db, rating_key) if filename == "master.m3u8": s = plex_stream.start_session(db, it, 0.0, multi=True) elif filename == "index.m3u8" and it.playable == "remux": diff --git a/backend/app/routes/search.py b/backend/app/routes/search.py index 7f4809a4..8c860ac3 100644 --- a/backend/app/routes/search.py +++ b/backend/app/routes/search.py @@ -36,9 +36,9 @@ from app.models import ( Video, VideoState, ) -from app.routes.feed import _serialize, feed_columns +from app.routes.feed import serialize, feed_columns from app.sync.subscriptions import apply_channel_details -from app.sync.videos import _insert_stubs, apply_video_details, parse_dt +from app.sync.videos import insert_stubs, apply_video_details, parse_dt from app.youtube import search_scrape from app.youtube.client import YouTubeClient, YouTubeError from app.youtube.shorts import make_client, probe_is_short @@ -102,7 +102,7 @@ def _serialize_ids(db: Session, user: User, ids: list[str]) -> list[dict]: .outerjoin(state, and_(state.video_id == Video.id, state.user_id == user.id)) .where(Video.id.in_(ids)) ).all() - by_id = {r.id: _serialize(r) for r in rows} + by_id = {r.id: serialize(r) for r in rows} return [by_id[i] for i in ids if i in by_id] @@ -136,7 +136,7 @@ def _ingest_candidates( # 2) Video stubs, flagged via_search. Existing rows keep their flags (ON CONFLICT DO # NOTHING), so an organically-synced video stays organic. - _insert_stubs( + insert_stubs( db, [ { diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index 2bf5a660..1723cfc3 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -13,8 +13,9 @@ from app import audit, progress, quota from app.audit import AuditAction from app.config import settings from app.db import SessionLocal +from app.demo import reset_demo_state from app.downloads.gc import run_download_gc -from app.models import SchedulerSetting +from app.models import SchedulerSetting, User from app.plex.sync import sync as run_plex_sync from app.plex.watch_sync import run_plex_watch_reconcile, run_plex_watch_sync from app.notifications import create_notification @@ -310,14 +311,11 @@ def _maintenance_job() -> None: def _demo_reset_job() -> None: - # Auto-clean the shared demo sandbox. No-op (and never creates one) if there's no demo - # account. Lazy import avoids a routes<->scheduler import cycle. + # Auto-clean the shared demo sandbox. No-op (and never creates one) if there's no demo account. def work(db): - from app.models import User demo = db.query(User).filter(User.is_demo.is_(True)).one_or_none() if demo is None: return {"skipped": "no demo account"} - from app.routes.admin import reset_demo_state return {"playlists_seeded": reset_demo_state(db, demo)} _job("demo_reset", work) diff --git a/backend/app/sync/explore.py b/backend/app/sync/explore.py index 9faf8007..b9c8ee93 100644 --- a/backend/app/sync/explore.py +++ b/backend/app/sync/explore.py @@ -23,7 +23,7 @@ from app.models import ( Video, VideoState, ) -from app.sync.videos import _insert_stubs, _stub_from_playlist_item, apply_video_details +from app.sync.videos import insert_stubs, stub_from_playlist_item, apply_video_details from app.utils import now_utc as _now from app.youtube.client import YouTubeClient @@ -71,11 +71,11 @@ def explore_ingest_page( data = yt.get_playlist_items(channel.uploads_playlist_id, page_token) stubs: list[dict] = [] for item in data.get("items", []): - stub = _stub_from_playlist_item(item, channel.id) + stub = stub_from_playlist_item(item, channel.id) if stub is not None: stub["via_explore"] = True stubs.append(stub) - inserted = _insert_stubs(db, stubs) + inserted = insert_stubs(db, stubs) self_enriched = _enrich_new(db, yt, [s["id"] for s in stubs]) log.info( "explore ingest channel=%s ingested=%s enriched=%s", channel.id, inserted, self_enriched diff --git a/backend/app/sync/videos.py b/backend/app/sync/videos.py index 2a8cba93..3b2f605f 100644 --- a/backend/app/sync/videos.py +++ b/backend/app/sync/videos.py @@ -44,7 +44,7 @@ def parse_dt(value: str | None) -> datetime | None: return None -def _insert_stubs(db: Session, rows: list[dict]) -> int: +def insert_stubs(db: Session, rows: list[dict]) -> int: """Idempotently insert video stubs; returns the number of newly inserted rows. Uses RETURNING so the count is exact (ON CONFLICT DO NOTHING leaves rowcount @@ -76,7 +76,7 @@ def apply_rss_feed(db: Session, channel: Channel, entries: list[dict]) -> int: """Insert new video stubs from an already-fetched feed and stamp the channel's last RSS time. DB-only — must run on the session's own thread; the network fetch (fetch_channel_feed) is kept separate so callers like run_rss_poll can parallelise it.""" - inserted = _insert_stubs(db, entries) + inserted = insert_stubs(db, entries) channel.last_rss_at = _now() db.add(channel) db.commit() @@ -84,7 +84,7 @@ def apply_rss_feed(db: Session, channel: Channel, entries: list[dict]) -> int: # --- Backfill (uploads playlist; costs quota) --- -def _stub_from_playlist_item(item: dict, channel_id: str) -> dict | None: +def stub_from_playlist_item(item: dict, channel_id: str) -> dict | None: content = item.get("contentDetails", {}) snippet = item.get("snippet", {}) video_id = content.get("videoId") @@ -120,7 +120,7 @@ def backfill_channel_recent(db: Session, yt: YouTubeClient, channel: Channel) -> data = yt.get_playlist_items(channel.uploads_playlist_id, page_token) stopped_mid_page = False for item in data.get("items", []): - stub = _stub_from_playlist_item(item, channel.id) + stub = stub_from_playlist_item(item, channel.id) if stub is None: continue if stub["published_at"] and stub["published_at"] < cutoff: @@ -147,7 +147,7 @@ def backfill_channel_recent(db: Session, yt: YouTubeClient, channel: Channel) -> channel.recent_synced_at = _now() channel.backfill_cursor = next_cursor channel.backfill_done = done - inserted = _insert_stubs(db, collected) + inserted = insert_stubs(db, collected) db.add(channel) db.commit() return inserted @@ -166,11 +166,11 @@ def backfill_channel_deep( rows = [ s for s in ( - _stub_from_playlist_item(it, channel.id) for it in data.get("items", []) + stub_from_playlist_item(it, channel.id) for it in data.get("items", []) ) if s is not None ] - total += _insert_stubs(db, rows) + total += insert_stubs(db, rows) page_token = data.get("nextPageToken") channel.backfill_cursor = page_token if not page_token: diff --git a/backend/app/worker.py b/backend/app/worker.py index 846f7d7b..aecd3432 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -262,7 +262,7 @@ def _finish_from_cache(job_id: int, asset_id: int) -> None: db.commit() -def _uploader_url(info: dict) -> str | None: +def uploader_url(info: dict) -> str | None: """The channel/uploader page URL for the auto-clickable channel line. YouTube gives a `channel_url`/`uploader_url` directly; Facebook gives neither but exposes a numeric `uploader_id` (the page id) that resolves at facebook.com/. Returns None for sources with @@ -290,7 +290,7 @@ def _fill_asset_meta_early(asset_id: int, info: dict) -> None: # the final title with its own source_ref fallback. asset.title = normalize_title(info.get("title")) or info.get("id") asset.uploader = info.get("uploader") or info.get("channel") - asset.uploader_url = _uploader_url(info) + asset.uploader_url = uploader_url(info) asset.thumbnail_url = info.get("thumbnail") asset.duration_s = int(info["duration"]) if info.get("duration") else None asset.upload_date = _parse_upload_date(info.get("upload_date")) @@ -638,7 +638,7 @@ def _download(job_id: int, asset_id: int, source_kind: str, source_ref: str, spe asset.acodec = info.get("acodec") if info.get("acodec") not in (None, "none") else None asset.title = meta.title asset.uploader = meta.uploader - asset.uploader_url = _uploader_url(info) + asset.uploader_url = uploader_url(info) asset.upload_date = meta.upload_date asset.thumbnail_url = meta.thumbnail_url # Record a self-hosted poster for every download: it's the reliable link-preview