diff --git a/app/main.py b/app/main.py index 3a4392d..72c049e 100644 --- a/app/main.py +++ b/app/main.py @@ -5,7 +5,7 @@ from fastapi.staticfiles import StaticFiles from starlette.middleware.sessions import SessionMiddleware from app.db import enable_beets_db_wal, init_db -from app.routers import auth, dashboard, import_, library +from app.routers import auth, dashboard, dedup, genres, import_, library from app.services import scheduler_service from app.settings import settings @@ -35,6 +35,8 @@ def create_app() -> FastAPI: app.include_router(dashboard.router) app.include_router(library.router) app.include_router(import_.router) + app.include_router(dedup.router) + app.include_router(genres.router) return app diff --git a/app/routers/dedup.py b/app/routers/dedup.py new file mode 100644 index 0000000..1beea99 --- /dev/null +++ b/app/routers/dedup.py @@ -0,0 +1,30 @@ +from fastapi import APIRouter, Depends, Request +from fastapi.responses import RedirectResponse +from fastapi.templating import Jinja2Templates + +from app.security.deps import require_auth +from app.services import dedup_review_service + +router = APIRouter(prefix="/dedup", tags=["dedup"]) +templates = Jinja2Templates(directory="app/templates") + + +@router.get("") +async def dedup_index(request: Request, user: dict = Depends(require_auth)): + candidates = dedup_review_service.list_pending_candidates() + return templates.TemplateResponse(request, "dedup/index.html", {"candidates": candidates}) + + +@router.post("/scan") +async def scan(user: dict = Depends(require_auth)): + await dedup_review_service.scan(triggered_by="manual") + return RedirectResponse(url="/dedup", status_code=303) + + +@router.post("/confirm") +async def confirm(request: Request, user: dict = Depends(require_auth)): + form = await request.form() + candidate_ids = [int(v) for k, v in form.multi_items() if k == "candidate_id"] + confirmed_by = user.get("email") or user.get("sub", "unknown") + await dedup_review_service.confirm_and_apply(candidate_ids, confirmed_by) + return RedirectResponse(url="/dedup", status_code=303) diff --git a/app/routers/genres.py b/app/routers/genres.py new file mode 100644 index 0000000..ee3a539 --- /dev/null +++ b/app/routers/genres.py @@ -0,0 +1,44 @@ +from fastapi import APIRouter, Depends, Form, Request +from fastapi.responses import RedirectResponse +from fastapi.templating import Jinja2Templates +from sqlalchemy import select + +from app.db import get_db +from app.models import GenreCandidate, GenreRun +from app.security.deps import require_auth +from app.services import genre_review_service + +router = APIRouter(prefix="/genres", tags=["genres"]) +templates = Jinja2Templates(directory="app/templates") + + +@router.get("") +async def genres_index(request: Request, user: dict = Depends(require_auth), db=Depends(get_db)): + latest_run = db.execute(select(GenreRun).order_by(GenreRun.id.desc())).scalars().first() + candidates = [] + if latest_run is not None: + candidates = list( + db.execute( + select(GenreCandidate).where(GenreCandidate.genre_run_id == latest_run.id) + ).scalars() + ) + return templates.TemplateResponse( + request, "genres/index.html", {"latest_run": latest_run, "candidates": candidates} + ) + + +@router.post("/scan") +async def scan(force: bool = Form(False), user: dict = Depends(require_auth)): + await genre_review_service.run(apply=False, force=force, triggered_by="manual") + return RedirectResponse(url="/genres", status_code=303) + + +@router.post("/lock") +async def lock( + artist: str = Form(...), + genre: str = Form(...), + user: dict = Depends(require_auth), +): + changed_by = user.get("email") or user.get("sub", "unknown") + await genre_review_service.lock_artist_genre(artist, genre, changed_by) + return RedirectResponse(url="/genres", status_code=303) diff --git a/app/services/dedup_review_service.py b/app/services/dedup_review_service.py new file mode 100644 index 0000000..7b06896 --- /dev/null +++ b/app/services/dedup_review_service.py @@ -0,0 +1,151 @@ +import json +import time +from pathlib import Path + +from sqlalchemy import select + +from app.db import SessionLocal +from app.models import DedupCandidate, DedupRun +from app.services import pipeline_runner +from app.settings import settings + +_SCRIPT = "dedup-library.sh" + + +def _parse_json_lines(output: str) -> list[dict]: + candidates = [] + for line in output.splitlines(): + line = line.strip() + if not line.startswith("{"): + continue + try: + candidates.append(json.loads(line)) + except json.JSONDecodeError: + continue + return candidates + + +async def scan(triggered_by: str = "manual") -> DedupRun: + """Dry-run dedup-library.sh --json, persist every candidate deletion + into a fresh dedup_runs/dedup_candidates pair. Never deletes anything + -- the scheduled maintenance:dedup job also only ever calls this (no + --apply), matching the false-negative-biased dedup preference; actual + deletion only ever happens through confirm_and_apply() below.""" + script = str(settings.pipeline_dir / "lib" / _SCRIPT) + job_run, output = await pipeline_runner.run_job_capture( + "dedup:scan", [script, "--json"], triggered_by=triggered_by + ) + candidates = _parse_json_lines(output) + + db = SessionLocal() + try: + dedup_run = DedupRun( + started_at=job_run.started_at, + finished_at=job_run.finished_at, + mode="dry_run", + groups_found=len({(c["pass"], c["keep_path"]) for c in candidates}), + kept=len({(c["pass"], c["keep_path"]) for c in candidates}), + deleted=0, + log_path=job_run.log_path, + ) + db.add(dedup_run) + db.commit() + db.refresh(dedup_run) + + for c in candidates: + db.add( + DedupCandidate( + dedup_run_id=dedup_run.id, + pass_name=c.get("pass", "unknown"), + keep_path=c["keep_path"], + delete_path=c["delete_path"], + delete_size_bytes=c.get("delete_size_bytes"), + ) + ) + db.commit() + db.refresh(dedup_run) + return dedup_run + finally: + db.close() + + +async def confirm_and_apply(candidate_ids: list[int], confirmed_by: str) -> DedupRun | None: + """Apply only the confirmed candidate deletions. + + Cheap pre-check here: skip anything where delete_path or keep_path no + longer exists (something already changed it since the scan). The real + re-verification of ranking happens for free inside dedup-library.sh + itself: --apply --only-paths re-runs all 4 passes from scratch and + recomputes keep/delete for every group before consulting the only-paths + allowlist, so if a group's ranking flipped since the scan (e.g. the old + keep_path is gone and delete_path is now the last copy), the script's + fresh pass will assign delete_path the KEEP role instead -- it never + reaches a DELETE branch for it, so --only-paths naming it is simply + never consulted. No duplicate ranking logic needed here. + """ + db = SessionLocal() + try: + candidates = [db.get(DedupCandidate, cid) for cid in candidate_ids] + candidates = [c for c in candidates if c is not None and not c.applied] + + now = time.time() + confirmed_candidates = [] + for c in candidates: + if not Path(c.delete_path).exists() or not Path(c.keep_path).exists(): + continue + c.confirmed = True + c.confirmed_by = confirmed_by + c.confirmed_at = now + confirmed_candidates.append(c) + db.commit() + + if not confirmed_candidates: + return None + + confirm_file = settings.logs_dir / f"dedup-confirm-{int(now * 1000)}.txt" + confirm_file.parent.mkdir(parents=True, exist_ok=True) + confirm_file.write_text("\n".join(c.delete_path for c in confirmed_candidates) + "\n") + + script = str(settings.pipeline_dir / "lib" / _SCRIPT) + job_run, _output = await pipeline_runner.run_job_capture( + "dedup:apply", + [script, "--apply", "--only-paths", str(confirm_file), "--json"], + triggered_by=f"manual:{confirmed_by}", + ) + + still_there = {c.delete_path for c in confirmed_candidates if Path(c.delete_path).exists()} + actually_deleted = 0 + for c in confirmed_candidates: + if c.delete_path not in still_there: + c.applied = True + actually_deleted += 1 + db.commit() + + apply_run = DedupRun( + started_at=job_run.started_at, + finished_at=job_run.finished_at, + mode="apply", + deleted=actually_deleted, + kept=len(confirmed_candidates) - actually_deleted, + log_path=job_run.log_path, + ) + db.add(apply_run) + db.commit() + db.refresh(apply_run) + return apply_run + finally: + db.close() + + +def list_pending_candidates(dedup_run_id: int | None = None) -> list[DedupCandidate]: + db = SessionLocal() + try: + query = select(DedupCandidate).where( + DedupCandidate.applied == False, # noqa: E712 + DedupCandidate.confirmed == False, # noqa: E712 + ) + if dedup_run_id is not None: + query = query.where(DedupCandidate.dedup_run_id == dedup_run_id) + return list(db.execute(query).scalars()) + finally: + db.close() diff --git a/app/services/genre_review_service.py b/app/services/genre_review_service.py new file mode 100644 index 0000000..3f3823d --- /dev/null +++ b/app/services/genre_review_service.py @@ -0,0 +1,83 @@ +import json + +from app.db import SessionLocal +from app.models import GenreCandidate, GenreRun +from app.services import genre_fix, pipeline_runner +from app.settings import settings + +_SCRIPT = "spotify-genre.py" + + +def _parse_json_lines(output: str) -> list[dict]: + candidates = [] + for line in output.splitlines(): + line = line.strip() + if not line.startswith("{"): + continue + try: + candidates.append(json.loads(line)) + except json.JSONDecodeError: + continue + return candidates + + +async def run(apply: bool, force: bool = False, triggered_by: str = "schedule") -> GenreRun: + """Run spotify-genre.py --json (dry-run or --apply), persisting the + diff into a fresh genre_runs/genre_candidates pair either way. + + Unlike dedup, genre writes stay auto-apply on the weekly schedule + (low-risk/reversible, GENRE_LOCK-protected) -- this wraps that same + scheduled run so its diff becomes reviewable after the fact, rather + than being a separate preview-only mode. scheduler_service's + maintenance:spotify_genre job calls this with apply=True.""" + args = ["--json"] + if apply: + args.append("--apply") + if force: + args.append("--force") + + script = str(settings.pipeline_dir / "lib" / _SCRIPT) + job_run, output = await pipeline_runner.run_job_capture( + "genre:run", [script] + args, triggered_by=triggered_by + ) + candidates = _parse_json_lines(output) + + db = SessionLocal() + try: + genre_run = GenreRun( + started_at=job_run.started_at, + finished_at=job_run.finished_at, + mode="apply" if apply else "dry_run", + written=len(candidates) if apply else 0, + unchanged=None, + no_match=None, + no_meta=None, + log_path=job_run.log_path, + ) + db.add(genre_run) + db.commit() + db.refresh(genre_run) + + for c in candidates: + db.add( + GenreCandidate( + genre_run_id=genre_run.id, + artist=c.get("artist"), + title=c.get("title"), + old_genre=c.get("old_genre"), + new_genre=c.get("new_genre"), + ) + ) + db.commit() + db.refresh(genre_run) + return genre_run + finally: + db.close() + + +async def lock_artist_genre(artist: str, genre: str, changed_by: str): + """One-click 'lock this artist's genre' action from the review UI, for + when a scheduled run got something wrong -- wraps + genre_fix.set_artist_genre (fix-genre.sh sets GENRE_LOCK=1 as part of + applying, which is what makes future spotify-genre runs leave it alone).""" + return await genre_fix.set_artist_genre(artist, genre, changed_by) diff --git a/app/services/pipeline_runner.py b/app/services/pipeline_runner.py index 449fa7b..34a9eb4 100644 --- a/app/services/pipeline_runner.py +++ b/app/services/pipeline_runner.py @@ -143,6 +143,95 @@ async def run_job( _lock.release() +async def run_job_capture( + job_key: str, + argv: list[str], + triggered_by: str = "manual", + timeout: float | None = None, +) -> tuple[JobRun, str]: + """Like run_job(), but captures stdout+stderr as text and returns it + alongside the JobRun, instead of only writing it to the log file -- + for callers that need to parse structured (JSON-lines) output, e.g. + dedup_review_service's dry-run scan. The full output is still written + to a log file afterward so job_runs.log_path works the same as any + other job. Shares the same lock as run_job().""" + started_at = time.time() + + if not await _try_acquire_nowait(): + db = SessionLocal() + run = JobRun( + job_key=job_key, + started_at=started_at, + finished_at=started_at, + status="skipped_lock", + triggered_by=triggered_by, + ) + db.add(run) + db.commit() + db.refresh(run) + db.close() + return run, "" + + try: + log_dir = settings.logs_dir / job_key.replace(":", "_") + log_dir.mkdir(parents=True, exist_ok=True) + log_path = log_dir / f"{time.strftime('%Y%m%d-%H%M%S')}.log" + + db = SessionLocal() + run = JobRun( + job_key=job_key, + started_at=started_at, + status="running", + triggered_by=triggered_by, + log_path=str(log_path), + ) + db.add(run) + db.commit() + db.refresh(run) + run_id = run.id + db.close() + + output_text = "" + exit_code: int | None + try: + proc = await asyncio.create_subprocess_exec( + *argv, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + env=_subprocess_env(), + ) + try: + stdout_bytes, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout) + exit_code = proc.returncode + output_text = stdout_bytes.decode(errors="replace") + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + exit_code = -1 + output_text += f"\n[pipeline_runner] TIMEOUT after {timeout}s -- process killed\n" + status = "success" if exit_code == 0 else "failed" + except Exception as exc: + exit_code = None + status = "failed" + output_text += f"\n[pipeline_runner] exception before/while running: {exc!r}\n" + + log_path.write_text(output_text) + + finished_at = time.time() + db = SessionLocal() + run = db.get(JobRun, run_id) + run.finished_at = finished_at + run.status = status + run.exit_code = exit_code + run.summary = _summarize_log(log_path) + db.commit() + db.refresh(run) + db.close() + return run, output_text + finally: + _lock.release() + + async def run_playlist(playlist_name: str, no_m3u: bool = False, triggered_by: str = "schedule") -> JobRun: script = settings.pipeline_dir / "bin" / "run-playlist.sh" argv = [str(script)] diff --git a/app/services/scheduler_service.py b/app/services/scheduler_service.py index 105a1a0..8cc6f77 100644 --- a/app/services/scheduler_service.py +++ b/app/services/scheduler_service.py @@ -6,7 +6,7 @@ from sqlalchemy import select from app.db import SessionLocal from app.models import Playlist, ScheduledJob -from app.services import pipeline_runner +from app.services import dedup_review_service, genre_review_service, pipeline_runner from app.settings import settings TIMEZONE = "America/Edmonton" @@ -44,6 +44,21 @@ def _beet(job_key: str, args: list[str]): return _run +async def _dedup_scan(triggered_by: str = "schedule"): + """Dry-run only, ever, on the schedule -- see module docstring. Populates + dedup_candidates for review; deletion only ever happens through a + confirmed dedup_review_service.confirm_and_apply() call from the UI.""" + return await dedup_review_service.scan(triggered_by=triggered_by) + + +async def _genre_run(triggered_by: str = "schedule"): + """Keeps its current auto-apply behavior (unlike dedup) -- genre writes + are low-risk/reversible and GENRE_LOCK-protected -- but now goes through + genre_review_service so every run's diff is persisted for after-the-fact + review instead of only living in a log file.""" + return await genre_review_service.run(apply=True, force=True, triggered_by=triggered_by) + + async def _log_rotation(triggered_by: str = "schedule"): """Replaces `find /var/log/sldl -name '*.log' -mtime +30 -delete`.""" cutoff = time.time() - 30 * 86400 @@ -68,7 +83,7 @@ MAINTENANCE_JOBS: dict[str, tuple[dict, callable]] = { ), "maintenance:dedup": ( dict(minute=55, hour=8), - _lib("maintenance:dedup", "dedup-library.sh"), # no --apply: dry-run only + _dedup_scan, # no --apply, ever, on schedule -- see _dedup_scan docstring ), "maintenance:gen_djmix_playlist": ( dict(minute=57, hour=8), @@ -125,7 +140,7 @@ MAINTENANCE_JOBS: dict[str, tuple[dict, callable]] = { ), "maintenance:spotify_genre": ( dict(minute=50, hour=8, day_of_week="sun"), - _lib("maintenance:spotify_genre", "spotify-genre.py", ["--apply", "--force"]), + _genre_run, # goes through genre_review_service -- see its docstring ), "maintenance:beets_update_sync": ( dict(minute=53, hour=8, day_of_week="sun"), diff --git a/app/templates/dedup/index.html b/app/templates/dedup/index.html new file mode 100644 index 0000000..7853a7a --- /dev/null +++ b/app/templates/dedup/index.html @@ -0,0 +1,38 @@ +{% extends "base.html" %} +{% block title %}Dedup — alembic{% endblock %} +{% block content %} +
+ Scheduled dedup runs dry-run only, every day. Nothing is ever deleted + without an explicit confirm below. +
+ + + ++ The weekly scheduled run still writes genres automatically (low-risk, + reversible, GENRE_LOCK-protected) -- this shows what the last run changed. +
+ + + +{{ latest_run.mode }} — {{ candidates | length }} change(s)
+| Artist | Title | Old genre | New genre | |
|---|---|---|---|---|
| {{ c.artist }} | +{{ c.title }} | +{{ c.old_genre or '(empty)' }} | +{{ c.new_genre }} | ++ + | +
| No changes in the last run. | ||||
No runs yet.
+{% endif %} +{% endblock %} diff --git a/pipeline/lib/dedup-library.sh b/pipeline/lib/dedup-library.sh index 088d736..71b6c6b 100755 --- a/pipeline/lib/dedup-library.sh +++ b/pipeline/lib/dedup-library.sh @@ -16,16 +16,57 @@ # "Best" is always: FLAC > MP3/other, then largest file within the same format. # # DRY RUN by default. Pass --apply to actually delete. +# +# --json: additionally emit one JSON line per candidate deletion to stdout +# (NDJSON), on top of the normal human log -- for dedup_review_service to +# parse into dedup_runs/dedup_candidates. Purely additive; the human log +# format is unchanged. +# --only-paths FILE: in --apply mode, only actually delete a candidate if +# its delete_path appears (one per line) in FILE -- everything else in a +# group still gets ranked/logged/reported as normal, just not deleted. +# This lets the confirm step in the dedup UI re-run the exact same +# ranking/safety logic and apply only what a human explicitly approved, +# without --only-paths this flag is a no-op and --apply deletes everything +# as before (preserves existing behavior for anyone invoking this +# directly). set -u APPLY=0 -[[ "${1:-}" == "--apply" ]] && APPLY=1 +JSON_MODE=0 +ONLY_PATHS_FILE="" +while [[ $# -gt 0 ]]; do + case "$1" in + --apply) APPLY=1; shift ;; + --json) JSON_MODE=1; shift ;; + --only-paths) ONLY_PATHS_FILE="$2"; shift 2 ;; + *) shift ;; + esac +done + +declare -A ONLY_PATHS=() +if [[ -n "$ONLY_PATHS_FILE" ]]; then + while IFS= read -r p; do + [[ -n "$p" ]] && ONLY_PATHS["$p"]=1 + done < "$ONLY_PATHS_FILE" +fi LOG=${ALEMBIC_CONFIG_DIR:-/config}/logs/dedup-$(date +%Y%m%d-%H%M%S).log mkdir -p "$(dirname "$LOG")" log() { echo "$*" | tee -a "$LOG"; } +# JSON lines go to stdout ONLY (never through log()/tee, so they never end +# up interleaved with the human-readable $LOG file -- a caller wanting +# structured output should capture this process's stdout separately). +json_emit() { [[ $JSON_MODE -eq 1 ]] && printf '%s\n' "$1"; } + +json_escape() { + local s="$1" + s="${s//\\/\\\\}" + s="${s//\"/\\\"}" + printf '%s' "$s" +} + if [[ $APPLY -eq 1 ]]; then log "[$(date -Iseconds)] === DEDUP (APPLY) ===" else @@ -106,17 +147,35 @@ process_group() { log "" log "--- $label ---" + # Pass name for JSON output, matching dedup_candidates.pass_name's + # convention (numbered_sibling|case_insensitive|normalized|cross_album_fuzzy). + local pass_name="unknown" + case "$label" in + "P1 numbered-sibling"*) pass_name="numbered_sibling" ;; + "P2 case-insensitive"*) pass_name="case_insensitive" ;; + "P3 normalized"*) pass_name="normalized" ;; + "P4 cross-album"*) pass_name="cross_album_fuzzy" ;; + esac + local first=1 + local keep_path="" while IFS="$SEP" read -r _sk id hp; do [[ -z "$hp" ]] && continue - local size_h; size_h=$(numfmt --to=iec --suffix=B "$(stat -c '%s' "$hp" 2>/dev/null || echo 0)") + local size_bytes; size_bytes=$(stat -c '%s' "$hp" 2>/dev/null || echo 0) + local size_h; size_h=$(numfmt --to=iec --suffix=B "$size_bytes") if [[ $first -eq 1 ]]; then log " KEEP ($size_h) $hp" + keep_path="$hp" first=0 else log " DELETE ($size_h) $hp" + json_emit "{\"pass\":\"${pass_name}\",\"keep_path\":\"$(json_escape "$keep_path")\",\"delete_path\":\"$(json_escape "$hp")\",\"delete_size_bytes\":${size_bytes}}" if [[ $APPLY -eq 1 ]]; then - if [[ -n "$id" ]]; then + # With --only-paths given, only delete entries the caller explicitly + # confirmed; without it, delete everything (original behavior). + if [[ -n "$ONLY_PATHS_FILE" && -z "${ONLY_PATHS[$hp]:-}" ]]; then + log " (skipped — not in --only-paths confirm list)" + elif [[ -n "$id" ]]; then # In beets — beet remove -d removes from DB + disk. beet remove -d -f "id:${id}" >> "$LOG" 2>&1 else diff --git a/pipeline/lib/spotify-genre.py b/pipeline/lib/spotify-genre.py index 64804c9..52c8449 100755 --- a/pipeline/lib/spotify-genre.py +++ b/pipeline/lib/spotify-genre.py @@ -277,6 +277,10 @@ def main(): help="Overwrite existing GENRE tags (default: only fill empty)") ap.add_argument("--query", help="Beets query to limit which tracks are processed") ap.add_argument("--limit", type=int, help="Stop after N tracks (debug)") + ap.add_argument("--json", action="store_true", + help="Additionally emit one JSON line per candidate change to stdout, " + "for genre_review_service to parse into genre_runs/genre_candidates. " + "Purely additive -- normal output is unchanged.") args = ap.parse_args() whitelist = load_whitelist() @@ -333,11 +337,21 @@ def main(): ok = write_genre(p, new_value) if ok: written += 1 + if args.json: + print(json.dumps({ + "artist": primary_artist, "title": title, + "old_genre": current_genre, "new_genre": new_value, + })) # Quiet — only print every 50th if written % 50 == 0: print(f" [{written}] {primary_artist} - {title}: {new_value}", flush=True) else: print(f" WOULD-SET {primary_artist} - {title} → {new_value}") + if args.json: + print(json.dumps({ + "artist": primary_artist, "title": title, + "old_genre": current_genre, "new_genre": new_value, + })) written += 1 # count as 'planned' print(f"\n[spotify-genre] {('written' if args.apply else 'plan')}: {written}, "