Add manual fix / manual import services and routers

library_edit.update_track_fields: single-track tag edits via `beet modify
-y -m` (subprocess with an argument list, not beets.library.Library
in-process -- an arg-list subprocess has no shell-injection surface to
avoid in the first place, and correctly reuses beets' own configured
path-format/move logic rather than reimplementing it against a hand-built
Library object with guessed config). Every changed field gets a
manual_fix_audit row with old/new values; triggers a Navidrome rescan
afterward if anything changed. Uses "genres" (plural) as the field name,
matching this deployment's actual beets column and every existing script's
convention, not the generic beets docs' singular "genre".

genre_fix.set_artist_genre and retag.retag_from_url wrap the existing
fix-genre.sh and fix-track-metadata.py as callable services rather than
reimplementing their per-format mutagen tag-writing logic.

manual_import: ties pipeline_runner.run_import_track to the manual_imports
table, mirrors import-track.sh's own argument-count-based disambiguation
(0/1/2 args), plus import-me/ listing and upload-with-path-traversal-guard.

routers/library.py + routers/import_.py: browsable library list, a track
detail/edit page (tag edit, artist-wide genre override, retag-from-URL
forms), and the import page (upload + trigger). Minimal templates for now,
Task 9 covers full UI polish.

Verified update_track_fields end-to-end against a REAL beets library (not
mocked): generated a tagged FLAC with ffmpeg/metaflac, imported it via a
real `beet import`, edited artist+title through the service, and confirmed
the file was physically moved to the new path-template location, tags were
rewritten on disk, the DB updated, and both changes landed correctly in
manual_fix_audit. Also verified the full app boots with the new routers
registered and all four new routes correctly redirect to login when
unauthenticated.

app/services/genre_fix.py

@@ -0,0 +1,22 @@
+from app.models import JobRun
+from app.services import pipeline_runner
+
+
+async def set_artist_genre(artist: str, genre: str, changed_by: str, dry_run: bool = False) -> JobRun:
+    """Wraps the existing fix-genre.sh (per-format FLAC/MP3/MP4/OGG tag
+    writing + GENRE_LOCK + beets DB sync) rather than reimplementing its
+    mutagen logic here -- it's already tested and handles every format this
+    library contains.
+
+    No manual_fix_audit row: this is an artist-wide bulk edit (could touch
+    dozens of tracks), not a single beets_item_id the audit table's schema
+    is shaped for. The job_runs row (with its full log) is the audit trail
+    for this action instead.
+    """
+    args = []
+    if dry_run:
+        args.append("--dry")
+    args.extend([artist, genre])
+    return await pipeline_runner.run_lib_script(
+        "manual:fix_genre", "fix-genre.sh", args, triggered_by=f"manual:{changed_by}"
+    )

app/services/library_edit.py

@@ -0,0 +1,89 @@
+import asyncio
+import subprocess
+import time
+
+from sqlalchemy.orm import Session
+
+from app.models import ManualFixAudit
+from app.services import beets_service, pipeline_runner
+
+# "genres" (plural) matches this deployment's actual beets DB column and
+# every existing script's `beet ls -f '$genres'` convention -- not the
+# generic beets docs' singular "genre".
+EDITABLE_FIELDS = ["artist", "title", "album", "albumartist", "genres", "grouping", "track", "year"]
+
+
+def _run_beet_modify(item_id: int, changes: dict[str, str]) -> subprocess.CompletedProcess:
+    """subprocess with an argument list (no shell=True) -- user-submitted
+    field values never pass through a shell, so there's no injection risk
+    to escape against. This also reuses beets' own configured path-format/
+    move logic exactly as the `beet` CLI does, rather than reimplementing
+    it against a hand-built beets.library.Library object with guessed
+    config values (directory/path_formats aren't reliably knowable from
+    here during the migration's transitional beets-mount period)."""
+    field_args = [f"{field}={value}" for field, value in changes.items()]
+    return subprocess.run(
+        ["beet", "modify", "-y", "-m", f"id:{item_id}"] + field_args,
+        capture_output=True,
+        text=True,
+    )
+
+
+async def update_track_fields(
+    db: Session, item_id: int, changes: dict[str, str], changed_by: str
+) -> list[str]:
+    """Apply one or more field edits to a single beets item via `beet
+    modify -y -m` (writes tags to the file, updates the DB, and moves the
+    file if the new fields change its path template -- all three in one
+    beets-native operation).
+
+    Every field that actually changed value gets one manual_fix_audit row
+    with old/new values. Triggers a (share-health-gated) Navidrome rescan
+    afterward if anything changed, so the edit shows up without waiting for
+    the next scheduled scan.
+
+    Returns the list of field names that were actually different after the
+    edit (a value re-submitted unchanged doesn't generate an audit row).
+    """
+    unknown = set(changes) - set(EDITABLE_FIELDS)
+    if unknown:
+        raise ValueError(f"not editable: {unknown}")
+    if not changes:
+        return []
+
+    before = beets_service.get_item(item_id)
+    if before is None:
+        raise ValueError(f"no beets item with id={item_id}")
+
+    result = await asyncio.to_thread(_run_beet_modify, item_id, changes)
+    if result.returncode != 0:
+        raise RuntimeError(f"beet modify failed: {result.stderr.strip()}")
+
+    after = beets_service.get_item(item_id) or {}
+    changed_fields = []
+    now = time.time()
+    for field in changes:
+        old_value = before.get(field)
+        new_value = after.get(field, changes[field])
+        if old_value != new_value:
+            db.add(
+                ManualFixAudit(
+                    beets_item_id=item_id,
+                    file_path=after.get("path", before.get("path", "")),
+                    field=field,
+                    old_value=str(old_value) if old_value is not None else None,
+                    new_value=str(new_value) if new_value is not None else None,
+                    changed_by=changed_by,
+                    changed_at=now,
+                    source="tag_edit",
+                )
+            )
+            changed_fields.append(field)
+    db.commit()
+
+    if changed_fields:
+        await pipeline_runner.run_lib_script(
+            "manual:navidrome_scan_after_edit", "navidrome-scan.sh", triggered_by="manual"
+        )
+
+    return changed_fields

app/services/manual_import.py

@@ -0,0 +1,66 @@
+import time
+from pathlib import Path
+
+from sqlalchemy.orm import Session
+
+from app.models import ManualImport
+from app.services import pipeline_runner
+from app.settings import settings
+
+
+def import_me_dir() -> Path:
+    return settings.music_data_dir / "import-me"
+
+
+def list_import_me_contents() -> list[str]:
+    root = import_me_dir()
+    if not root.exists():
+        return []
+    return sorted(p.name for p in root.iterdir())
+
+
+def save_uploaded_file(filename: str, content: bytes) -> Path:
+    """Write an uploaded file into import-me/ alongside whatever's already
+    there via SMB. Path-traversal guard: only the basename of the supplied
+    filename is used, so an upload can't escape import-me/."""
+    root = import_me_dir()
+    root.mkdir(parents=True, exist_ok=True)
+    safe_name = Path(filename).name
+    if not safe_name:
+        raise ValueError("empty filename")
+    dest = root / safe_name
+    dest.write_bytes(content)
+    return dest
+
+
+async def import_track(
+    db: Session, source: str | None, playlist_name: str | None, imported_by: str
+) -> ManualImport:
+    """source: a filename/subdir relative to import-me/, or None/empty to
+    import everything currently sitting in import-me/. Mirrors
+    import-track.sh's own argument-count-based disambiguation (see its
+    docstring): 0 args = import everything untagged, 1 arg = either a real
+    path under import-me/ or a playlist name (the script figures out which),
+    2 args = path + playlist name explicitly."""
+    args: list[str] = []
+    if source:
+        args.append(source)
+        if playlist_name:
+            args.append(playlist_name)
+    elif playlist_name:
+        args.append(playlist_name)
+
+    job_run = await pipeline_runner.run_import_track(args, triggered_by="manual")
+
+    record = ManualImport(
+        source_path=source or "(all of import-me/)",
+        playlist_name=playlist_name,
+        imported_by=imported_by,
+        imported_at=time.time(),
+        status=job_run.status,
+        job_run_id=job_run.id,
+    )
+    db.add(record)
+    db.commit()
+    db.refresh(record)
+    return record

app/services/retag.py

@@ -0,0 +1,16 @@
+from app.models import JobRun
+from app.services import pipeline_runner
+
+
+async def retag_from_url(file_path: str, url: str, changed_by: str, keep_genre: bool = False) -> JobRun:
+    """Wraps the existing fix-track-metadata.py (Spotify/iTunes URL -> fetch
+    canonical metadata, write tags, replace cover art, beet update + move,
+    refill genre, trigger Navidrome scan) rather than reimplementing its
+    metadata-fetch and multi-format tag-writing logic here."""
+    args = []
+    if keep_genre:
+        args.append("--keep-genre")
+    args.extend([file_path, url])
+    return await pipeline_runner.run_lib_script(
+        "manual:retag_from_url", "fix-track-metadata.py", args, triggered_by=f"manual:{changed_by}"
+    )