refactor: R12 S3 — invert scheduler->routes, publish cross-module privates

Layering cleanup, no behaviour change (ruff clean backend+scripts, pytest 193):

- Invert the scheduler->routes dependency: reset_demo_state (+ its demo-seed
  helpers) move from routes/admin.py to a new service module app/demo.py.
  Both routes/admin.py and scheduler.py now import it from app.demo, so the
  scheduler no longer reaches into the routes layer — its lazy import (the
  routes<->scheduler cycle workaround) is gone, hoisted to a top-level import.

- Publish the private symbols that were imported across module boundaries (the
  underscore convention was lying): feed._serialize -> serialize,
  feed._to_tsquery_str -> to_tsquery_str, videos._insert_stubs -> insert_stubs,
  videos._stub_from_playlist_item -> stub_from_playlist_item,
  worker._uploader_url -> uploader_url, plex._shared._item_or_404 ->
  item_or_404. Scoped renames — the 5 other route modules keep their own truly
  private _serialize helpers.

- _release_asset moves from routes/downloads.py to app/downloads/service.py as
  the public release_asset (asset bookkeeping is a service concern); the routes
  call it via service.release_asset.
This commit is contained in:
2026-08-11 22:58:10 +02:00
parent ac6383e92c
commit 51463b226e
16 changed files with 153 additions and 148 deletions
+2 -73
View File
@@ -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)
+4 -31
View File
@@ -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:
+4 -4
View File
@@ -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,
+2 -2
View File
@@ -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],
}
+1 -1
View File
@@ -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")
+4 -4
View File
@@ -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 = {}
+7 -7
View File
@@ -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":
+4 -4
View File
@@ -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,
[
{