alembic/app/routers/dedup.py

import os
import re

from fastapi import APIRouter, BackgroundTasks, 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
from app.settings import settings

router = APIRouter(prefix="/dedup", tags=["dedup"])
templates = Jinja2Templates(directory="app/templates")

# Every library track lives under this prefix -- showing it on every row of
# every candidate adds nothing but width, so strip it for display (the full
# path is still available in the title attribute on hover).
_LIBRARY_PREFIX = str(settings.music_data_dir / "Library") + "/"

# dedup-library.sh's four tag-based passes (numbered_sibling, case_insensitive,
# normalized, cross_album_fuzzy) all match on filenames/tags; find-fuzzy-dupes.py's
# one pass (fuzzy_audio) matches on acoustic fingerprint. The specific pass
# name is an implementation detail -- the table just needs to say which of
# the two matching methods caught it.
_ACOUSTIC_PASS = "fuzzy_audio"


def _pass_label(pass_name: str) -> str:
    return "Acoustically Similar" if pass_name == _ACOUSTIC_PASS else "File Naming"


def _display_path(path: str) -> str:
    return path[len(_LIBRARY_PREFIX):] if path.startswith(_LIBRARY_PREFIX) else path


# Compilation-style import paths carry a "$track " prefix on the filename
# (see pipeline/configs/beets config.yaml paths: comp/albumtype_soundtrack);
# singleton paths (the vast majority of this library) don't. Strip it either
# way so the title reads clean.
_TRACK_PREFIX_RE = re.compile(r"^\d{1,3}[\s.\-]+")

# rank_file() in dedup-library.sh always ranks FLAC above every other format
# and WAV below MP3 (a download glitch, never desirable) -- mirror that
# judgment in the format pill's color so it reads as "this one's the keeper"
# at a glance, not just a bare extension string.
_FORMAT_BADGE = {"flac": "badge-success", "wav": "badge-warning"}


def _split_display(path: str) -> dict:
    """Break a library display path into (artist, album, title, ext).

    Beets' singleton path template is always
    %the{$albumartist}/$album/$title (pipeline/configs/beets config.yaml,
    paths:) -- every track in this library lands at exactly that shape, so
    the last two segments are reliably album/filename. A path that doesn't
    fit (an unexpected root-level file) degrades to showing the raw display
    path as the title instead of guessing at structure that isn't there.
    """
    display = _display_path(path)
    parts = display.split("/")
    if len(parts) >= 2:
        artist, album, filename = parts[0], parts[-2], parts[-1]
        title, ext = os.path.splitext(filename)
        title = _TRACK_PREFIX_RE.sub("", title)
    else:
        artist, album = "", ""
        title, ext = os.path.splitext(display)
    return {
        "path": path,
        "display": display,
        "artist": artist,
        "album": album,
        "title": title,
        "ext": ext.lstrip(".").lower(),
    }


def _file_size(path: str) -> int | None:
    try:
        return os.path.getsize(path)
    except OSError:
        return None


def _to_row(c) -> dict:
    keep = _split_display(c.keep_path)
    delete = _split_display(c.delete_path)
    keep["size_bytes"] = _file_size(c.keep_path)
    keep["badge"] = _FORMAT_BADGE.get(keep["ext"], "badge-muted")
    delete["size_bytes"] = c.delete_size_bytes if c.delete_size_bytes is not None else _file_size(c.delete_path)
    delete["badge"] = _FORMAT_BADGE.get(delete["ext"], "badge-muted")
    return {
        "id": c.id,
        "pass_name": c.pass_name,
        "pass_label": _pass_label(c.pass_name),
        "keep": keep,
        "delete": delete,
        # Same-tag passes (numbered_sibling/case_insensitive/normalized) share
        # artist/title almost always; cross_album_fuzzy and fuzzy_audio can
        # legitimately differ (different credit, different album entirely) --
        # the template only collapses shared context onto one line when it's
        # actually shared, otherwise shows both sides in full.
        "same_title": keep["title"].lower() == delete["title"].lower(),
        "same_artist": keep["artist"].lower() == delete["artist"].lower(),
        "same_album": keep["album"].lower() == delete["album"].lower(),
    }


@router.get("")
async def dedup_index(request: Request, user: dict = Depends(require_auth)):
    candidates = [_to_row(c) for c in dedup_review_service.list_pending_candidates()]
    ignored = [_to_row(c) for c in dedup_review_service.list_ignored_candidates()]
    return templates.TemplateResponse(
        request, "dedup/index.html", {"candidates": candidates, "ignored": ignored}
    )


@router.post("/scan")
async def scan(background: BackgroundTasks, user: dict = Depends(require_auth)):
    # Runs both the file-naming and acoustic passes -- see
    # dedup_review_service.scan_all(). The acoustic pass can take a while on a
    # large library, so run it in the background and return immediately; the
    # candidates appear here once it finishes (reload), and the run shows up
    # under Settings then Jobs. If the pipeline is busy it records skipped_lock
    # there rather than blocking.
    background.add_task(dedup_review_service.scan_all, triggered_by="manual")
    return RedirectResponse(url="/dedup?started=1", 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)


@router.post("/ignore")
async def ignore(request: Request, user: dict = Depends(require_auth)):
    form = await request.form()
    ignored_by = user.get("email") or user.get("sub", "unknown")
    for v in form.getlist("candidate_id"):
        dedup_review_service.ignore_candidate(int(v), ignored_by)
    return RedirectResponse(url="/dedup", status_code=303)


@router.post("/unignore")
async def unignore(request: Request, user: dict = Depends(require_auth)):
    form = await request.form()
    for v in form.getlist("candidate_id"):
        dedup_review_service.unignore_candidate(int(v))
    return RedirectResponse(url="/dedup", status_code=303)