From 5d9952ab1e6cd83854afcb3b19020e8451b909d9 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Fri, 23 Oct 2026 17:30:28 +0200 Subject: [PATCH] feat(reader): a Takeout video the catalog learns later is marked watched then; the import's status A watched video the catalog did not know at import time waits in `takeout_items`. The Shorts job now settles those rows right after its probe (the only place a video's Short-ness becomes known), every user's, a commit each: `reader.takeout.apply_waiting`, its count in the job's result. `GET /api/takeout` is what the settings page reads: the user's last import run (`jobruns.last_run`) and the kept videos by kind and outcome, "waiting" for one not known yet. --- backend/app/jobruns.py | 10 +++ backend/app/reader/queries.py | 27 +++++++ backend/app/reader/takeout.py | 24 ++++++ backend/app/routes/feed.py | 24 +++++- backend/app/scheduler.py | 8 +- backend/tests/test_takeout.py | 85 ++++++++++++++++++++- backend/tests/test_unit_fences.py | 2 + frontend/src/i18n/locales/en/scheduler.json | 5 +- frontend/src/i18n/locales/hu/scheduler.json | 5 +- 9 files changed, 181 insertions(+), 9 deletions(-) diff --git a/backend/app/jobruns.py b/backend/app/jobruns.py index 7beb7572..5596e651 100644 --- a/backend/app/jobruns.py +++ b/backend/app/jobruns.py @@ -119,6 +119,16 @@ def recent_runs(db: Session, jobs: list[str], per_job: int) -> dict[str, list[Jo return out +def last_run(db: Session, job: str, target: str) -> JobRun | None: + """The newest run of an ad-hoc `job` on `target` (a user's own import), or None.""" + return db.execute( + select(JobRun) + .where(JobRun.job == job, JobRun.target == target) + .order_by(JobRun.started_at.desc(), JobRun.id.desc()) + .limit(1) + ).scalar_one_or_none() + + def gc(db: Session, retention_days: int) -> int: """Delete finished runs older than `retention_days` (0 = keep forever). A run still going is never reaped. The caller commits.""" diff --git a/backend/app/reader/queries.py b/backend/app/reader/queries.py index ffb7302f..88a6550a 100644 --- a/backend/app/reader/queries.py +++ b/backend/app/reader/queries.py @@ -422,3 +422,30 @@ def takeout_outcomes(db: Session, user: User) -> dict[tuple[str, str | None], in .group_by(TakeoutItem.kind, TakeoutItem.outcome) ).all() return {(kind, outcome): n for kind, outcome, n in rows} + + +def takeout_summary(db: Session, user: User) -> dict[str, dict[str, int]]: + """The user's kept Takeout videos by kind and outcome, for their settings page: `{"watched": + {"watched": n, "short": n, "waiting": n, …}, "watch_later": {…}}` — the import's outcome words, + "waiting" for a video not known yet.""" + out: dict[str, dict[str, int]] = {kind.value: {} for kind in TakeoutKind} + for (kind, outcome), n in takeout_outcomes(db, user).items(): + out[kind][outcome or "waiting"] = n + return out + + +def takeout_users_ready(db: Session) -> list[int]: + """Every user with a waiting "watched" row whose video is now known and probed for Shorts: the + rows a later run of the Shorts probe made ready.""" + return list( + db.execute( + select(TakeoutItem.user_id) + .join(Video, Video.id == TakeoutItem.video_id) + .where( + TakeoutItem.kind == TakeoutKind.WATCHED, + TakeoutItem.outcome.is_(None), + Video.shorts_probed, + ) + .distinct() + ).scalars() + ) diff --git a/backend/app/reader/takeout.py b/backend/app/reader/takeout.py index 9a752a1e..7a7b12b9 100644 --- a/backend/app/reader/takeout.py +++ b/backend/app/reader/takeout.py @@ -25,6 +25,8 @@ from contextlib import contextmanager from dataclasses import dataclass, field from datetime import datetime +from sqlalchemy.orm import Session + from app import quota from app.bodylimit import TAKEOUT_MAX_BYTES from app.db import SessionLocal @@ -174,6 +176,11 @@ def _one_import(user_id: int) -> Iterator[None]: yield +def run_target(user_id: int) -> str: + """The import run's target, `job_runs.target`: one user's import.""" + return f"user:{user_id}" + + def run_import(user_id: int) -> dict: """The import job: apply what the user's kept Takeout rows can say now, as one recorded run. The watched part commits before the Watch later part asks YouTube for the videos the catalog @@ -198,3 +205,20 @@ def run_import(user_id: int) -> dict: if outcomes.get((TakeoutKind.WATCH_LATER, None)): result["watch_later_waiting"] = outcomes[(TakeoutKind.WATCH_LATER, None)] return result + + +def apply_waiting(db: Session) -> dict: + """Mark watched the waiting Takeout rows whose video the catalog has come to know and the Shorts + probe has cleared since the import — every user's, one commit each. Run after each Shorts probe + (the scheduler's `shorts` job), the only place a video's Short-ness becomes known. Returns its + counts, or an empty dict when nothing was waiting on a probed video.""" + users = queries.takeout_users_ready(db) + marked = 0 + for user_id in users: + user = db.get(User, user_id) + if user is None: + continue + with _one_import(user_id): + marked += service.apply_takeout_watched(db, user)["watched"] + db.commit() + return {"takeout_watched": marked} if users else {} diff --git a/backend/app/routes/feed.py b/backend/app/routes/feed.py index a8fd811a..d6a41e24 100644 --- a/backend/app/routes/feed.py +++ b/backend/app/routes/feed.py @@ -6,7 +6,7 @@ from pydantic import BaseModel from sqlalchemy.orm import Session from sqlalchemy.orm.exc import StaleDataError -from app import quota, scheduler +from app import jobruns, quota, scheduler from app.auth import current_user from app.db import get_db, release_read_connection from app.mailtext import lang_for_user @@ -334,7 +334,7 @@ def upload_takeout( scheduler.TAKEOUT_IMPORT, partial(takeout.run_import, user.id), actor_id=user.id, - target=f"user:{user.id}", + target=takeout.run_target(user.id), notify=True, ) return { @@ -346,3 +346,23 @@ def upload_takeout( "other": parsed.other, "malformed": parsed.malformed, } + + +@router.get("/takeout") +def takeout_status(user: User = Depends(current_user), db: Session = Depends(get_db)): + """What the user's Takeout import has done: its last run (none before the first upload), and + every video kept from the export, by kind and outcome — a video the catalog did not know yet + counts as "waiting" until it does, and is marked then.""" + run = jobruns.last_run(db, scheduler.TAKEOUT_IMPORT, takeout.run_target(user.id)) + return { + "run": None + if run is None + else { + "status": run.status, + "started_at": iso(run.started_at), + "finished_at": iso(run.finished_at), + "result": run.result, + "failure": run.failure, + }, + **queries.takeout_summary(db, user), + } diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index dc5e2506..44d395ac 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -22,6 +22,7 @@ from app.plex.tmdb import run_people_enrich from app.plex.watch_sync import run_plex_watch_reconcile, run_plex_watch_sync from app.notifications import create_notification from app.runstatus import RunStatus, RunTrigger +from app.reader import takeout from app.state import is_sync_paused from app.sync.autotag import run_autotag_all from app.sync.explore import purge_ephemeral @@ -403,7 +404,12 @@ def _autotag_job() -> None: def _shorts_job() -> None: - _job("shorts", run_shorts) + def work(db): + # A video waiting in someone's Takeout import becomes markable the moment the probe has + # said it is not a Short, so the import's waiting rows are settled right after it. + return run_shorts(db) | takeout.apply_waiting(db) + + _job("shorts", work) def _subscriptions_job() -> None: diff --git a/backend/tests/test_takeout.py b/backend/tests/test_takeout.py index 39d65834..f603c306 100644 --- a/backend/tests/test_takeout.py +++ b/backend/tests/test_takeout.py @@ -8,9 +8,10 @@ from datetime import datetime, timedelta, timezone import pytest from sqlalchemy import select -from app import scheduler +from app import jobruns, scheduler from app.catalog.models import Channel, Video -from app.models import Notification +from app.models import JobRun, Notification +from app.runstatus import RunStatus, RunTrigger from app.reader import service, takeout from app.reader.models import ( Playlist, @@ -360,3 +361,83 @@ def test_an_ad_hoc_run_that_asks_for_it_tells_its_user_how_it_went(db, make_user assert [(n.data["job_id"], n.data["result"]) for n in notices] == [ ("takeout_import", {"watched": 3}) ] + + +# --- the videos that wait ---------------------------------------------------------------------- + + +def test_waiting_videos_are_marked_once_the_shorts_probe_has_cleared_them(db, make_user): + a, b = make_user("a@example.com"), make_user("b@example.com") + _video(db, "later000000", probed=False) + _video(db, "notyet00000", probed=False) + db.commit() + service.stage_takeout( + db, a, TakeoutKind.WATCHED, {"later000000": (T0, None), "notyet00000": (T0, None)} + ) + service.stage_takeout(db, b, TakeoutKind.WATCHED, {"later000000": (T0, None)}) + db.commit() + assert takeout.apply_waiting(db) == {} + + # The probe clears one of them; the other is still unprobed. + db.get(Video, "later000000").shorts_probed = True + db.commit() + assert takeout.apply_waiting(db) == {"takeout_watched": 2} + + assert _states(db, a)["later000000"].watched_at == T0 + assert _states(db, b)["later000000"].status == "watched" + assert _rows(db, a)["notyet00000"].outcome is None + assert takeout.apply_waiting(db) == {} + + +def test_the_shorts_job_settles_the_waiting_rows_its_probe_made_ready(db, make_user): + user = make_user("job-shorts@example.com") + _video(db, "ready000000") # probed already: the job's probe has nothing left to do + db.commit() + service.stage_takeout(db, user, TakeoutKind.WATCHED, {"ready000000": (T0, None)}) + db.commit() + + scheduler._shorts_job() + + db.expire_all() + run = db.execute(select(JobRun).where(JobRun.job == "shorts")).scalar_one() + assert run.result == {"probed": 0, "shorts": 0, "takeout_watched": 1} + assert _states(db, user)["ready000000"].status == "watched" + + +# --- what the settings page reads -------------------------------------------------------------- + + +def test_the_status_is_the_users_own_last_run_and_counts(db, make_user, as_user): + user, other = make_user("status@example.com"), make_user("other@example.com") + assert as_user(user).get("/api/takeout").json() == { + "run": None, + "watched": {}, + "watch_later": {}, + } + _video(db, "plain000000") + _video(db, "short000000", short=True) + db.commit() + service.stage_takeout( + db, + user, + TakeoutKind.WATCHED, + {v: (T0, None) for v in ("plain000000", "short000000", "unknown0000")}, + ) + service.stage_takeout(db, user, TakeoutKind.WATCH_LATER, {"plain000000": (T0, None)}) + service.stage_takeout(db, other, TakeoutKind.WATCHED, {"plain000000": (T0, None)}) + service.apply_takeout_watched(db, user) + db.commit() + older = jobruns.start("takeout_import", RunTrigger.AD_HOC, user.id, f"user:{user.id}") + jobruns.finish(older, RunStatus.OK, summary=None, result={"watched": 0}) + newest = jobruns.start("takeout_import", RunTrigger.AD_HOC, user.id, f"user:{user.id}") + jobruns.finish(newest, RunStatus.OK, summary=None, result={"watched": 1}) + theirs = jobruns.start("takeout_import", RunTrigger.AD_HOC, other.id, f"user:{other.id}") + jobruns.finish(theirs, RunStatus.OK, summary=None, result={"watched": 99}) + + body = as_user(user).get("/api/takeout").json() + + assert body["run"]["status"] == "ok" + assert body["run"]["result"] == {"watched": 1} + assert body["run"]["finished_at"] is not None + assert body["watched"] == {"watched": 1, "short": 1, "waiting": 1} + assert body["watch_later"] == {"waiting": 1} diff --git a/backend/tests/test_unit_fences.py b/backend/tests/test_unit_fences.py index f49b3a0d..14f11f33 100644 --- a/backend/tests/test_unit_fences.py +++ b/backend/tests/test_unit_fences.py @@ -364,6 +364,8 @@ UNIT_COMMITS = { # The Takeout import is a recorded run with its own session: the watched part lands before # the Watch later part asks YouTube (which commits the videos it fetched). "reader/takeout.py:run_import", + # Run inside the Shorts job, on its session: each user's waiting rows land on their own. + "reader/takeout.py:apply_waiting", }, # The catalog's ingest jobs commit per page or batch, a YouTube call between each: a page # fetched and stored is not lost when the next call fails, and an enrich that fails rolls back diff --git a/frontend/src/i18n/locales/en/scheduler.json b/frontend/src/i18n/locales/en/scheduler.json index 7c1a7101..fbd6410b 100644 --- a/frontend/src/i18n/locales/en/scheduler.json +++ b/frontend/src/i18n/locales/en/scheduler.json @@ -67,7 +67,7 @@ "enrich": "Fetches video details (duration, views, live/Shorts flags) and re-checks live videos until they end. If it stops, videos stay without a duration/stats and ended livestreams stay stuck as “live”.", "backfill": "Pulls in recent uploads, then the full back-catalogue for opted-in channels, within the shared quota. If it stops, older videos and full history don't fill in.", "autotag": "Detects each channel's language/topics and applies automatic tags. If it stops, new channels get no auto tags, so language/topic filters miss them.", - "shorts": "Probes youtube.com/shorts to mark which videos are Shorts. If it stops, Shorts aren't separated from normal videos in the Feed.", + "shorts": "Probes youtube.com/shorts to mark which videos are Shorts, then marks watched the videos a Google Takeout import was waiting on. If it stops, Shorts aren't separated from normal videos in the Feed, and those videos keep waiting.", "subscriptions": "Re-imports every user's YouTube subscriptions. If it stops, channels subscribed to or dropped on YouTube aren't reflected here.", "playlist_sync": "Mirrors each user's YouTube playlists into the app. If it stops, changes to your YouTube playlists don't show up locally.", "maintenance": "Finds videos that can no longer be played (deleted, made private, abandoned premieres), hides them from the Feed and removes them after a grace period; re-checks the least-recently-checked videos. If it stops, dead videos linger in the catalogue.", @@ -186,7 +186,8 @@ "watch_later_added": "added to Watch later", "watch_later_present": "already in Watch later", "watch_later_missing": "no longer on YouTube", - "watch_later_waiting": "Watch later waiting" + "watch_later_waiting": "Watch later waiting", + "takeout_watched": "marked watched from Takeout" }, "skip": { "disabled": "disabled", diff --git a/frontend/src/i18n/locales/hu/scheduler.json b/frontend/src/i18n/locales/hu/scheduler.json index 244cf1b0..8f04643d 100644 --- a/frontend/src/i18n/locales/hu/scheduler.json +++ b/frontend/src/i18n/locales/hu/scheduler.json @@ -67,7 +67,7 @@ "enrich": "Lekéri a videók adatait (hossz, megtekintés, élő/Shorts jelzők), és frissíti az élő videókat, amíg véget nem érnek. Ha leáll, a videók hossz/statisztika nélkül maradnak, a véget ért adások pedig „élő” állapotban ragadnak.", "backfill": "Behúzza a friss feltöltéseket, majd a kijelölt csatornák teljes előzményét, a közös kvótán belül. Ha leáll, a régebbi videók és a teljes előzmény nem töltődik fel.", "autotag": "Felismeri a csatornák nyelvét/témáit, és automatikus címkéket ad. Ha leáll, az új csatornák címke nélkül maradnak, így a nyelv/téma szűrők kihagyják őket.", - "shorts": "A youtube.com/shorts próbával jelöli, mely videók Shorts-ok. Ha leáll, a Shorts-ok nem különülnek el a normál videóktól a Feedben.", + "shorts": "A youtube.com/shorts próbával jelöli, mely videók Shorts-ok, utána megnézettre jelöli azokat a videókat, amelyekre egy Google Takeout import várt. Ha leáll, a Shorts-ok nem különülnek el a normál videóktól a Feedben, és ezek a videók tovább várnak.", "subscriptions": "Újraimportálja minden felhasználó YouTube-feliratkozásait. Ha leáll, a YouTube-on felvett vagy törölt feliratkozások nem tükröződnek itt.", "playlist_sync": "Tükrözi a felhasználók YouTube lejátszási listáit az appba. Ha leáll, a YouTube-listák változásai nem jelennek meg lokálisan.", "maintenance": "Megkeresi a már sehol le nem játszható videókat (törölt, priváttá tett, elhagyott premier), elrejti őket a Feedből, és türelmi idő után eltávolítja; újraellenőrzi a legrégebben ellenőrzött videókat. Ha leáll, a halott videók a katalógusban maradnak.", @@ -186,7 +186,8 @@ "watch_later_added": "Watch laterbe téve", "watch_later_present": "már a Watch laterben", "watch_later_missing": "már nincs a YouTube-on", - "watch_later_waiting": "Watch later vár" + "watch_later_waiting": "Watch later vár", + "takeout_watched": "Takeoutból megnézettre jelölve" }, "skip": { "disabled": "kikapcsolva",