src/server.py

"""
FastAPI server for accompy.

Endpoints:
  GET  /                        → serve the web UI
  GET  /api/scores              → list available scores
  GET  /api/scores/{name}       → score data as JSON
  GET  /api/scores/{name}/sheet → sheet music HTML
  GET  /api/corpus/search?q=    → search music21 corpus
  POST /api/convert             → convert a corpus piece
"""

import json
import os
import re
import shlex
import shutil
import tempfile
import subprocess
import sys
import atexit
import zipfile
import hashlib
import hmac
import secrets
import threading
import time
from pathlib import Path
from functools import lru_cache
from datetime import datetime, timedelta, timezone
from urllib.parse import quote
from xml.etree import ElementTree as ET
from xml.sax.saxutils import escape as xml_escape
import requests
from fastapi import FastAPI, HTTPException, UploadFile, File, Form, Request, Response
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse, RedirectResponse
from starlette.exceptions import HTTPException as StarletteHTTPException
from pydantic import BaseModel
from src.env import load_local_env
from src.convert_score import convert_lilypond_parts_to_musicxml, convert_score_source, render_html, slugify_score_name
from src.fingering import (
    apply_auto_fingering,
    normalize_fingering_state,
    piano_part_indices,
    stack_fingering_chord_numbers_in_html,
)
from src.paths import get_static_dir, get_traces_dir
from src.storage import create_score_store, SupabaseScoreStore, _score_row_to_payload
from src.r2_artifacts import (
    get_r2_artifact_download_url,
    get_score_storage_keys,
    r2_configured,
    upload_r2_artifact,
)

load_local_env()

app = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=1000)


class SPAStaticFiles(StaticFiles):
    # THE HTML MUST NEVER BE CACHED. Every other asset is cache-busted by a
    # `?v=<tag>` query that index.html carries -- which means a STALE index.html
    # defeats the entire scheme, because it keeps asking for the old tag and the
    # browser keeps answering from cache. That is not hypothetical: on
    # 2026-09-17 a player spent a whole session on an app.js from 2026-09-15
    # (trace 20260917-130128 records `tag: 20260915-ff17` against a current
    # `version: 2026-09-17.2`), so four fixes landed that day were never in the
    # session being tested, and the bug they were reporting was one of the fixed
    # ones. FastAPI/Starlette send only etag + last-modified for static files,
    # and a browser is free to reuse those heuristically without revalidating.
    _NO_STORE = {"Cache-Control": "no-store, must-revalidate", "Pragma": "no-cache", "Expires": "0"}

    @staticmethod
    def _is_html_response(response) -> bool:
        # Keyed off the RESPONSE, not the request path: the mount uses html=True,
        # so "/" is resolved to index.html inside StaticFiles and the path this
        # method sees for the root is "." -- which no sensible path test matches.
        return "text/html" in (response.headers.get("content-type") or "")

    async def get_response(self, path: str, scope):
        try:
            response = await super().get_response(path, scope)
        except StarletteHTTPException as exc:
            headers = dict(scope.get("headers") or [])
            accept = headers.get(b"accept", b"").decode("latin-1")
            if exc.status_code == 404 and "text/html" in accept:
                response = await super().get_response("index.html", scope)
                response.headers.update(self._NO_STORE)
                return response
            raise
        if self._is_html_response(response):
            response.headers.update(self._NO_STORE)
        return response


# ── Corpus index (built once on first search) ─────────────────────────────────

@lru_cache(maxsize=1)
def _corpus_index():
    """Return list of {path, composer, title} for all .mxl files in the corpus."""
    from music21 import corpus
    pkg_dir = os.path.dirname(corpus.__file__)
    entries = []
    for p in corpus.getPaths():
        s = str(p)
        if not s.endswith('.mxl'):
            continue
        rel   = os.path.relpath(s, pkg_dir).replace(os.sep, '/').replace('.mxl', '')
        parts = rel.split('/')
        composer = parts[0] if len(parts) >= 2 else ''
        title    = '/'.join(parts[1:]) if len(parts) >= 2 else rel
        entries.append({'path': rel, 'composer': composer, 'title': title})
    return entries

STATIC_DIR = str(get_static_dir())
SHEET_RENDER_CACHE_DIR = Path(
    os.getenv("NOTEPILOT_SHEET_RENDER_CACHE_DIR", ".accompy_sheet_cache")
).expanduser()
SHEET_HTML_CONTENT_TYPE = "text/html; charset=utf-8"
ALLOWED_IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg"}
ALLOWED_PDF_SUFFIXES = {".pdf"}
ALLOWED_MUSICXML_SUFFIXES = {".xml", ".mxl", ".musicxml"}
ALLOWED_MUSESCORE_SUFFIXES = {".mscx", ".mscz"}
ALLOWED_LILYPOND_SUFFIXES = {".ily", ".ly"}
ALLOWED_ZIP_SUFFIXES = {".zip"}
ALLOWED_DIRECT_SCORE_SUFFIXES = ALLOWED_MUSICXML_SUFFIXES | ALLOWED_MUSESCORE_SUFFIXES
LILYPOND_HELPER_STEMS = {"global", "music", "midi"}
_score_store = create_score_store()
SESSION_COOKIE_NAME = "accompy_session"
SESSION_DAYS = 30
_fingering_jobs: dict[str, dict] = {}
_active_fingering_jobs: dict[tuple[str, str], str] = {}
_fingering_jobs_lock = threading.Lock()
_import_jobs: dict[str, dict] = {}
_import_jobs_lock = threading.Lock()
_session_user_cache: dict[str, tuple[float, dict | None]] = {}
_session_user_cache_lock = threading.Lock()
SESSION_USER_CACHE_TTL_SEC = 60


def hash_password(password: str, salt: str | None = None) -> str:
    salt = salt or secrets.token_hex(16)
    digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), 200_000)
    return f"{salt}${digest.hex()}"


def verify_password(password: str, stored: str) -> bool:
    try:
        salt, digest = stored.split("$", 1)
    except ValueError:
        return False
    calculated = hash_password(password, salt).split("$", 1)[1]
    return hmac.compare_digest(calculated, digest)


def create_session_token() -> str:
    return secrets.token_urlsafe(48)


def set_app_session_cookie(response: Response, session_token: str):
    response.set_cookie(
        key=SESSION_COOKIE_NAME,
        value=session_token,
        httponly=True,
        samesite="lax",
        secure=False,
        max_age=SESSION_DAYS * 24 * 60 * 60,
        expires=SESSION_DAYS * 24 * 60 * 60,
        path="/",
    )


def create_app_session_response(response: Response, user: dict) -> dict:
    session_token = create_session_token()
    expires_at = datetime.now(timezone.utc) + timedelta(days=SESSION_DAYS)
    _score_store.create_app_session(
        user["id"],
        hashlib.sha256(session_token.encode("utf-8")).hexdigest(),
        expires_at.isoformat(),
    )
    public_user = {"id": user["id"], "username": user["username"]}
    _cache_app_user(session_token, public_user)
    set_app_session_cookie(response, session_token)
    return {"ok": True, "user": public_user}


def _get_cached_app_user(raw_token: str) -> dict | None:
    with _session_user_cache_lock:
        cached = _session_user_cache.get(raw_token)
        if not cached:
            return None
        expires_at, user = cached
        if expires_at <= time.monotonic():
            _session_user_cache.pop(raw_token, None)
            return None
        return dict(user) if user else None


def _cache_app_user(raw_token: str, user: dict | None):
    with _session_user_cache_lock:
        _session_user_cache[raw_token] = (
            time.monotonic() + SESSION_USER_CACHE_TTL_SEC,
            dict(user) if user else None,
        )


def _clear_cached_app_user(raw_token: str):
    with _session_user_cache_lock:
        _session_user_cache.pop(raw_token, None)


def require_supabase_user_id(request: Request, action: str) -> str:
    user_id = current_user_id_for_request(request)
    if user_id:
        return user_id
    raise HTTPException(
        status_code=401,
        detail=f"Authenticated user is required for Supabase-backed {action}."
    )


def current_user_id_for_request(request: Request) -> str | None:
    if isinstance(_score_store, SupabaseScoreStore):
        token = request.cookies.get(SESSION_COOKIE_NAME, "").strip()
        if token:
            user = _get_cached_app_user(token)
            if user is None:
                user = _score_store.get_app_session_user(token)
                _cache_app_user(token, user)
            if user:
                return user.get("id")
    auth_header = request.headers.get("authorization", "").strip()
    if auth_header.lower().startswith("bearer "):
        token = auth_header.split(" ", 1)[1].strip()
        if token and os.getenv("SUPABASE_URL", "").strip():
            response = requests.get(
                f"{os.getenv('SUPABASE_URL').rstrip('/')}/auth/v1/user",
                headers={
                    "apikey": os.getenv("SUPABASE_ANON_KEY", "").strip(),
                    "Authorization": f"Bearer {token}",
                },
                timeout=15,
            )
            if response.status_code == 200:
                return response.json().get("id")
            raise HTTPException(status_code=401, detail="Invalid Supabase session.")
    return None


def current_app_user_for_request(request: Request) -> dict | None:
    if not isinstance(_score_store, SupabaseScoreStore):
        return None
    token = request.cookies.get(SESSION_COOKIE_NAME, "").strip()
    if not token:
        return None
    user = _get_cached_app_user(token)
    if user is None:
        user = _score_store.get_app_session_user(token)
        _cache_app_user(token, user)
    return user


def score_name_from_input(raw: str) -> str:
    return slugify_score_name(raw)


def ensure_fingering_state(score: dict) -> dict:
    normalized = dict(score)
    has_fingered_sheet = bool(
        normalized.get("has_fingered_sheet")
        or normalized.get("fingered_musicxml_source")
        or normalized.get("fingered_sheet_html")
    )
    fingering = normalize_fingering_state(
        normalized.get("parts") or [],
        normalized.get("fingering"),
        has_fingered_sheet=has_fingered_sheet,
    )
    normalized["fingering"] = fingering
    normalized["has_sheet"] = bool(normalized.get("has_sheet") or normalized.get("musicxml_source"))
    normalized["has_fingered_sheet"] = has_fingered_sheet
    return normalized


def render_sheet_html_from_musicxml_text(
    score_name: str,
    title: str,
    xml_text: str,
    *,
    stack_fingering_chords: bool = False,
) -> str:
    if not (xml_text or "").strip():
        return ""
    with tempfile.TemporaryDirectory(prefix="accompy_sheet_variant_") as tmp_dir:
        tmp_path = Path(tmp_dir)
        xml_path = tmp_path / f"{score_name}.musicxml"
        html_path = tmp_path / f"{score_name}.html"
        xml_path.write_text(xml_text, encoding="utf-8")
        render_html(str(xml_path), str(html_path), title)
        html = html_path.read_text(encoding="utf-8") if html_path.exists() else ""
        return stack_fingering_chord_numbers_in_html(html) if stack_fingering_chords else html


def sheet_render_cache_path(
    score_name: str,
    variant: str,
    parts_key: str,
    xml_text: str,
    *,
    stack_fingering_chords: bool = False,
) -> Path:
    safe_score = re.sub(r"[^A-Za-z0-9_.-]+", "_", score_name or "score").strip("._") or "score"
    safe_variant = re.sub(r"[^A-Za-z0-9_.-]+", "_", variant or "base").strip("._") or "base"
    safe_parts = re.sub(r"[^A-Za-z0-9_.-]+", "_", parts_key or "full").strip("._") or "full"
    digest = hashlib.sha256()
    digest.update((xml_text or "").encode("utf-8"))
    digest.update(b"\0stack=")
    digest.update(b"1" if stack_fingering_chords else b"0")
    return SHEET_RENDER_CACHE_DIR / safe_score / f"{safe_variant}-{safe_parts}-{digest.hexdigest()[:20]}.html"


def load_cached_sheet_render(cache_path: Path) -> str:
    try:
        if cache_path.exists() and cache_path.is_file():
            html = cache_path.read_text(encoding="utf-8")
            if "<svg" in html:
                return html
    except Exception as exc:
        print(f"Could not read sheet render cache {cache_path}: {exc}")
    return ""


def save_cached_sheet_render(cache_path: Path, html: str):
    if not html or "<svg" not in html:
        return
    try:
        cache_path.parent.mkdir(parents=True, exist_ok=True)
        cache_path.write_text(html, encoding="utf-8")
    except Exception as exc:
        print(f"Could not write sheet render cache {cache_path}: {exc}")


def strip_sheet_title(html: str) -> str:
    return re.sub(r"\s*<h1>.*?</h1>\s*", "\n", html or "", count=1, flags=re.IGNORECASE | re.DOTALL)


def sheet_artifact_type_for_variant(variant: str) -> str:
    return "fingered_html" if variant == "fingered" else "full_html"


def score_artifact_key_for_type(user_id: str, score_id: str, artifact_type: str) -> str:
    keys = get_score_storage_keys(user_id, score_id)
    if artifact_type == "fingered_html":
        return keys["fingered_html"]
    return keys["full_html"]


def persist_score_render_artifact(user_id: str, score_name: str, artifact_type: str, html: str):
    if not (
        r2_configured()
        and isinstance(_score_store, SupabaseScoreStore)
        and html
        and "<svg" in html
    ):
        return None
    row = _score_store.load_score_row(user_id, score_name)
    score_id = row.get("id")
    if not score_id:
        return None
    key = score_artifact_key_for_type(user_id, score_id, artifact_type)
    upload_r2_artifact(
        key=key,
        body=html,
        content_type=SHEET_HTML_CONTENT_TYPE,
    )
    _score_store.upsert_score_artifact(
        score_id=score_id,
        user_id=user_id,
        artifact_type=artifact_type,
        storage_provider="r2",
        object_key=key,
        content_type=SHEET_HTML_CONTENT_TYPE,
        size_bytes=len(html.encode("utf-8")),
    )
    return key


def persist_score_render_artifacts(user_id: str, score_name: str, artifacts: list[tuple[str, str]]):
    for artifact_type, html in artifacts:
        try:
            persist_score_render_artifact(user_id, score_name, artifact_type, html)
        except Exception as exc:
            print(f"Could not persist {artifact_type} R2 artifact for {score_name}: {exc}")


def save_score_with_render_artifacts(user_id: str, payload: dict) -> dict:
    artifact_candidates: list[tuple[str, str]] = []
    should_externalize = r2_configured() and isinstance(_score_store, SupabaseScoreStore)
    save_payload = dict(payload)

    if should_externalize:
        base_html = save_payload.get("sheet_html") or ""
        fingered_html = save_payload.get("fingered_sheet_html") or ""
        if base_html and "<svg" in base_html:
            artifact_candidates.append(("full_html", base_html))
            save_payload["sheet_html"] = ""
        if fingered_html and "<svg" in fingered_html:
            artifact_candidates.append(("fingered_html", fingered_html))
            save_payload["fingered_sheet_html"] = ""

    saved = _score_store.save_score(user_id, save_payload)
    if artifact_candidates:
        persist_score_render_artifacts(user_id, saved["name"], artifact_candidates)
    return saved


def get_score_artifact(row: dict, user_id: str, artifact_type: str) -> dict | None:
    if not (r2_configured() and isinstance(_score_store, SupabaseScoreStore)):
        return None
    score_id = row.get("id")
    if not score_id:
        return None
    try:
        artifact = _score_store.get_score_artifact_by_id(
            score_id=score_id,
            user_id=user_id,
            artifact_type=artifact_type,
        )
    except Exception as exc:
        print(f"Could not load {artifact_type} artifact metadata for {row.get('slug') or score_id}: {exc}")
        return None
    if artifact and artifact.get("storage_provider") == "r2" and artifact.get("object_key"):
        return artifact
    return None


def fetch_score_artifact_html(row: dict, user_id: str, artifact_type: str) -> str:
    artifact = get_score_artifact(row, user_id, artifact_type)
    if not artifact:
        return ""
    try:
        url = get_r2_artifact_download_url(artifact["object_key"], expires_in=60)
        response = requests.get(url, timeout=20)
        if response.status_code == 200 and "<svg" in response.text:
            return response.text
    except Exception as exc:
        print(f"Could not fetch {artifact_type} artifact HTML for {row.get('slug')}: {exc}")
    return ""


def render_full_score_html_for_artifact(row: dict, variant: str) -> str:
    score = ensure_fingering_state(_score_row_to_payload(row))
    score_data = row.get("score_data") or {}
    title = score.get("title") or score.get("name") or row.get("slug") or "score"
    name = score.get("name") or row.get("slug") or "score"
    if variant == "fingered":
        source = score_data.get("fingered_musicxml_source") or score.get("fingered_musicxml_source") or ""
        cached_html = score_data.get("fingered_sheet_html") or score.get("fingered_sheet_html") or ""
        if source:
            return render_sheet_html_from_musicxml_text(
                name,
                title,
                source,
                stack_fingering_chords=True,
            )
        return stack_fingering_chord_numbers_in_html(cached_html)

    cached_html = row.get("sheet_html") or score.get("sheet_html") or ""
    if cached_html and "<svg" in cached_html:
        return cached_html
    source = score_data.get("musicxml_source") or score.get("musicxml_source") or ""
    if source:
        return render_sheet_html_from_musicxml_text(name, title, source)
    return ""


def parse_sheet_part_indices(raw: str | None) -> list[int] | None:
    if raw is None:
        return None
    indices = []
    for item in str(raw).split(","):
        item = item.strip()
        if not item:
            continue
        try:
            idx = int(item)
        except ValueError:
            continue
        if idx >= 0 and idx not in indices:
            indices.append(idx)
    return indices or None


def _xml_local_name(tag: str) -> str:
    return tag.rsplit("}", 1)[-1] if "}" in tag else tag


def _musicxml_child(parent, local_name: str):
    return next((child for child in list(parent) if _xml_local_name(child.tag) == local_name), None)


def _musicxml_child_text(parent, local_name: str) -> str | None:
    child = _musicxml_child(parent, local_name)
    return child.text.strip() if child is not None and child.text else None


def _set_musicxml_child_text(parent, local_name: str, value: str):
    child = _musicxml_child(parent, local_name)
    if child is not None:
        child.text = value


def _source_part_selections(
    part_indices: list[int] | None,
    score_parts_meta: list[dict] | None,
    score_part_ids: list[str],
    staff_slots: list[tuple[str, int]],
) -> dict[str, set[int] | None]:
    if not part_indices:
        return {}

    selections: dict[str, set[int] | None] = {}
    if score_parts_meta:
        has_source_metadata = any(
            (part or {}).get("source_part_id") or (part or {}).get("source_staff")
            for part in score_parts_meta
        )
        if not has_source_metadata and len(staff_slots) >= len(score_parts_meta) > 1:
            for idx in part_indices:
                if 0 <= idx < len(staff_slots):
                    part_id, staff_num = staff_slots[idx]
                    selections.setdefault(part_id, set()).add(staff_num)
            if selections:
                return selections

        if not has_source_metadata and len(score_part_ids) == 1 and len(score_parts_meta) > 1:
            selections[score_part_ids[0]] = {idx + 1 for idx in part_indices if 0 <= idx < len(score_parts_meta)}
            return selections

        for idx in part_indices:
            if idx < 0 or idx >= len(score_parts_meta):
                continue
            meta = score_parts_meta[idx] or {}
            part_id = str(meta.get("source_part_id") or "").strip()
            if not part_id and idx < len(score_part_ids):
                part_id = score_part_ids[idx]
            if not part_id:
                continue
            staff = meta.get("source_staff")
            try:
                staff_num = int(staff) if staff is not None else None
            except (TypeError, ValueError):
                staff_num = None
            if staff_num:
                current = selections.setdefault(part_id, set())
                if current is not None:
                    current.add(staff_num)
            else:
                selections[part_id] = None

    if selections:
        return selections

    if len(score_part_ids) == 1 and score_parts_meta and len(score_parts_meta) > 1:
        selections[score_part_ids[0]] = {idx + 1 for idx in part_indices if 0 <= idx < len(score_parts_meta)}
        return selections

    for idx in part_indices:
        if 0 <= idx < len(score_part_ids):
            selections[score_part_ids[idx]] = None
    return selections


def _musicxml_part_staff_counts(root, score_part_ids: list[str]) -> dict[str, int]:
    counts = {part_id: 1 for part_id in score_part_ids if part_id}
    for part_el in [child for child in list(root) if _xml_local_name(child.tag) == "part"]:
        part_id = part_el.attrib.get("id")
        if not part_id:
            continue
        max_staves = 1
        for attributes in part_el.iter():
            if _xml_local_name(attributes.tag) != "attributes":
                continue
            staves_text = _musicxml_child_text(attributes, "staves")
            if staves_text:
                try:
                    max_staves = max(max_staves, int(staves_text))
                except ValueError:
                    pass
        counts[part_id] = max_staves
    return counts


def _musicxml_staff_slots(staff_counts: dict[str, int], score_part_ids: list[str]) -> list[tuple[str, int]]:
    slots: list[tuple[str, int]] = []
    for part_id in score_part_ids:
        if not part_id:
            continue
        staves = max(1, int(staff_counts.get(part_id) or 1))
        for staff_num in range(1, staves + 1):
            slots.append((part_id, staff_num))
    return slots


def _selected_staff_set_for_part(
    part_id: str,
    selected_staves: set[int] | None,
    total_staves: int,
) -> set[int] | None:
    if selected_staves is None:
        return None
    valid = {staff for staff in selected_staves if staff >= 1}
    if not valid or total_staves <= 1 or len(valid) >= total_staves:
        return None
    return valid


def _filter_part_to_staves(part_el, selected_staves: set[int]):
    for measure in [child for child in list(part_el) if _xml_local_name(child.tag) == "measure"]:
        for child in list(measure):
            local_name = _xml_local_name(child.tag)
            if local_name == "note":
                staff_text = _musicxml_child_text(child, "staff")
                if staff_text:
                    try:
                        staff_num = int(staff_text)
                    except ValueError:
                        staff_num = None
                    if staff_num is not None and staff_num not in selected_staves:
                        measure.remove(child)
                        continue
                    _set_musicxml_child_text(child, "staff", "1")
            elif local_name == "direction":
                staff_text = _musicxml_child_text(child, "staff")
                if staff_text:
                    try:
                        staff_num = int(staff_text)
                    except ValueError:
                        staff_num = None
                    if staff_num is not None and staff_num not in selected_staves:
                        measure.remove(child)
                        continue
                    _set_musicxml_child_text(child, "staff", "1")
            elif local_name == "attributes":
                _set_musicxml_child_text(child, "staves", "1")
                for attr_child in list(child):
                    attr_local = _xml_local_name(attr_child.tag)
                    if attr_local not in {"clef", "staff-details", "transpose"}:
                        continue
                    number = attr_child.attrib.get("number")
                    if number:
                        try:
                            staff_num = int(number)
                        except ValueError:
                            staff_num = None
                        if staff_num is not None and staff_num not in selected_staves:
                            child.remove(attr_child)
                            continue
                    attr_child.attrib["number"] = "1"


def _measure_has_sounding_note(measure) -> bool:
    for note_el in [child for child in list(measure) if _xml_local_name(child.tag) == "note"]:
        if _musicxml_child(note_el, "pitch") is not None or _musicxml_child(note_el, "unpitched") is not None:
            return True
    return False


def _measure_has_note(measure) -> bool:
    return any(_xml_local_name(child.tag) == "note" for child in list(measure))


def _measure_can_be_folded_after_start(measure) -> bool:
    allowed = {"note", "backup", "forward", "print", "attributes", "barline"}
    for child in list(measure):
        local_name = _xml_local_name(child.tag)
        if local_name == "note":
            if _musicxml_child(child, "pitch") is not None or _musicxml_child(child, "unpitched") is not None:
                return False
            continue
        if local_name not in allowed:
            return False
    return True


def _ensure_measure_attributes(measure):
    attributes = _musicxml_child(measure, "attributes")
    if attributes is not None:
        return attributes
    attributes = ET.Element("attributes")
    insert_at = 0
    for idx, child in enumerate(list(measure)):
        if _xml_local_name(child.tag) in {"print", "direction"}:
            insert_at = idx + 1
    measure.insert(insert_at, attributes)
    return attributes


def _measure_duration_from_context(measure, context: dict[str, int]) -> int:
    attributes = _musicxml_child(measure, "attributes")
    if attributes is not None:
        divisions_text = _musicxml_child_text(attributes, "divisions")
        if divisions_text:
            try:
                context["divisions"] = max(1, int(float(divisions_text)))
            except ValueError:
                pass
        time_el = _musicxml_child(attributes, "time")
        if time_el is not None:
            beats_text = _musicxml_child_text(time_el, "beats")
            beat_type_text = _musicxml_child_text(time_el, "beat-type")
            try:
                context["beats"] = max(1, int(float(beats_text or context["beats"])))
                context["beat_type"] = max(1, int(float(beat_type_text or context["beat_type"])))
            except ValueError:
                pass
    return max(1, round(context["divisions"] * context["beats"] * 4 / context["beat_type"]))


def _measure_durations_by_part(measures_by_part) -> list[list[int]]:
    durations_by_part = []
    for measures in measures_by_part:
        context = {"divisions": 1, "beats": 4, "beat_type": 4}
        durations_by_part.append([
            _measure_duration_from_context(measure, context)
            for measure in measures
        ])
    return durations_by_part


def _ensure_full_measure_rest(measure, duration: int):
    if _measure_has_note(measure):
        return
    rest_note = ET.Element("note")
    rest_el = ET.SubElement(rest_note, "rest", {"measure": "yes"})
    rest_el.text = None
    duration_el = ET.SubElement(rest_note, "duration")
    duration_el.text = str(max(1, int(duration)))
    voice_el = ET.SubElement(rest_note, "voice")
    voice_el.text = "1"

    insert_at = 0
    for idx, child in enumerate(list(measure)):
        if _xml_local_name(child.tag) in {"print", "attributes", "direction"}:
            insert_at = idx + 1
    measure.insert(insert_at, rest_note)


def _reset_measure_to_full_rest(measure, duration: int):
    for child in list(measure):
        measure.remove(child)
    _ensure_full_measure_rest(measure, duration)


def _ensure_multiple_rest_marker(measure, count: int):
    attributes = _ensure_measure_attributes(measure)
    measure_style = _musicxml_child(attributes, "measure-style")
    if measure_style is None:
        measure_style = ET.Element("measure-style")
        attributes.append(measure_style)
    multiple_rest = _musicxml_child(measure_style, "multiple-rest")
    if multiple_rest is None:
        multiple_rest = ET.Element("multiple-rest", {"use-symbols": "yes"})
        measure_style.append(multiple_rest)
    multiple_rest.text = str(max(2, int(count)))


def collapse_empty_musicxml_measures(xml_text: str, *, min_measures: int = 2, include_measure_map: bool = False):
    try:
        root = ET.fromstring(xml_text.encode("utf-8"))
    except ET.ParseError:
        return ("", []) if include_measure_map else ""

    parts = [child for child in list(root) if _xml_local_name(child.tag) == "part"]
    if not parts:
        return (xml_text, []) if include_measure_map else xml_text

    measures_by_part = [
        [child for child in list(part) if _xml_local_name(child.tag) == "measure"]
        for part in parts
    ]
    if not measures_by_part or not measures_by_part[0]:
        return (xml_text, []) if include_measure_map else xml_text

    measure_count = min(len(measures) for measures in measures_by_part)
    durations_by_part = _measure_durations_by_part(measures_by_part)
    foldable = []
    for idx in range(measure_count):
        measures_at_idx = [measures[idx] for measures in measures_by_part]
        is_empty = all(not _measure_has_sounding_note(measure) for measure in measures_at_idx)
        can_fold = idx == 0 or all(_measure_can_be_folded_after_start(measure) for measure in measures_at_idx)
        foldable.append(is_empty and can_fold)

    runs: list[tuple[int, int]] = []
    idx = 0
    while idx < measure_count:
        if not foldable[idx]:
            idx += 1
            continue
        start = idx
        while idx < measure_count and foldable[idx]:
            idx += 1
        length = idx - start
        if length >= max(2, min_measures):
            runs.append((start, length))

    if not runs:
        measure_map = [
            {"startMeasure": idx, "endMeasure": idx + 1}
            for idx in range(measure_count)
        ]
        return (xml_text, measure_map) if include_measure_map else xml_text

    for start, length in runs:
        for part_idx, measures in enumerate(measures_by_part):
            _ensure_full_measure_rest(measures[start], durations_by_part[part_idx][start])
            _ensure_multiple_rest_marker(measures[start], length)
            for hidden_idx in range(start + 1, start + length):
                if hidden_idx < len(measures):
                    _reset_measure_to_full_rest(
                        measures[hidden_idx],
                        durations_by_part[part_idx][hidden_idx],
                    )

    collapsed_ranges = {start: length for start, length in runs}
    hidden_measure_indices = {
        idx
        for start, length in runs
        for idx in range(start + 1, start + length)
    }
    measure_map = []
    idx = 0
    while idx < measure_count:
        if idx in collapsed_ranges:
            length = collapsed_ranges[idx]
            measure_map.append({"startMeasure": idx, "endMeasure": idx + length})
            idx += length
            continue
        if idx not in hidden_measure_indices:
            measure_map.append({"startMeasure": idx, "endMeasure": idx + 1})
        idx += 1

    output = ET.tostring(root, encoding="unicode")
    return (output, measure_map) if include_measure_map else output


def inject_sheet_measure_map(html: str, measure_map: list[dict]) -> str:
    if not measure_map:
        return html
    import json

    payload = json.dumps(measure_map, separators=(",", ":"))
    script = f'<script id="notepilot-measure-map" type="application/json">{payload}</script>'
    if "</body>" in html:
        return html.replace("</body>", f"{script}\n</body>", 1)
    return html + script


def filter_musicxml_parts(
    xml_text: str,
    part_indices: list[int] | None,
    score_parts_meta: list[dict] | None = None,
) -> str:
    if not part_indices:
        return xml_text
    try:
        root = ET.fromstring(xml_text.encode("utf-8"))
    except ET.ParseError:
        return ""

    part_list = next((child for child in list(root) if _xml_local_name(child.tag) == "part-list"), None)
    if part_list is None:
        return ""

    score_parts = [child for child in list(part_list) if _xml_local_name(child.tag) == "score-part"]
    score_part_ids = [part.attrib.get("id", "") for part in score_parts]
    staff_counts = _musicxml_part_staff_counts(root, score_part_ids)
    staff_slots = _musicxml_staff_slots(staff_counts, score_part_ids)
    selections = _source_part_selections(part_indices, score_parts_meta, score_part_ids, staff_slots)
    selected_ids = {part_id for part_id in selections if part_id}
    if not selected_ids:
        return ""
    has_staff_filter = any(staves is not None for staves in selections.values())
    if len(selected_ids) == len(score_parts) and not has_staff_filter:
        return xml_text

    for child in list(part_list):
        local_name = _xml_local_name(child.tag)
        if local_name == "part-group":
            part_list.remove(child)
        elif local_name == "score-part" and child.attrib.get("id") not in selected_ids:
            part_list.remove(child)

    for child in list(root):
        if _xml_local_name(child.tag) != "part":
            continue
        part_id = child.attrib.get("id")
        if part_id not in selected_ids:
            root.remove(child)
            continue
        staves = selections.get(part_id)
        staves_count = staff_counts.get(part_id or "", 1)
        selected_staves = _selected_staff_set_for_part(part_id or "", staves, staves_count)
        if selected_staves:
            _filter_part_to_staves(child, selected_staves)

    return ET.tostring(root, encoding="unicode")


def midi_to_musicxml_pitch(midi: int) -> str:
    steps = [
        ("C", 0),
        ("C", 1),
        ("D", 0),
        ("D", 1),
        ("E", 0),
        ("F", 0),
        ("F", 1),
        ("G", 0),
        ("G", 1),
        ("A", 0),
        ("A", 1),
        ("B", 0),
    ]
    midi = int(midi)
    step, alter = steps[midi % 12]
    octave = midi // 12 - 1
    alter_xml = f"<alter>{alter}</alter>" if alter else ""
    return f"<pitch><step>{step}</step>{alter_xml}<octave>{octave}</octave></pitch>"


def event_pitches(event) -> list[int]:
    if not isinstance(event, list) or not event:
        return []
    raw = event[0]
    pitches = raw if isinstance(raw, list) else [raw]
    result = []
    for pitch in pitches:
        try:
            result.append(int(pitch))
        except (TypeError, ValueError):
            continue
    return result


def synthesize_selected_parts_musicxml(score: dict, part_indices: list[int], title: str) -> str:
    parts = score.get("parts") or []
    selected = [
        (idx, parts[idx])
        for idx in part_indices
        if 0 <= idx < len(parts)
    ]
    if not selected:
        return ""

    divisions = 480
    measure_starts = [
        float(beat)
        for beat in (score.get("measure_beats") or [])
        if isinstance(beat, (int, float))
    ]
    max_end = 0.0
    for _, part in selected:
        for event in part.get("notes") or []:
            if not isinstance(event, list) or len(event) < 2:
                continue
            beat = float(event[1] or 0)
            duration = float(event[2] or 1) if len(event) > 2 else 1
            max_end = max(max_end, beat + max(duration, 0))
    if not measure_starts:
        measure_starts = [0.0]
        while measure_starts[-1] < max_end:
            measure_starts.append(measure_starts[-1] + 4.0)
    if len(measure_starts) == 1 or measure_starts[-1] < max_end:
        span = 4.0
        if len(measure_starts) >= 2:
            spans = [b - a for a, b in zip(measure_starts, measure_starts[1:]) if b > a]
            if spans:
                span = sorted(spans)[len(spans) // 2]
        while measure_starts[-1] < max_end:
            measure_starts.append(measure_starts[-1] + span)

    def duration_xml(beats: float) -> str:
        duration = max(1, int(round(max(0.0, beats) * divisions)))
        return f"<duration>{duration}</duration>"

    def rest_xml(beats: float) -> str:
        if beats <= 0.0001:
            return ""
        return f"<note><rest />{duration_xml(beats)}<voice>1</voice></note>"

    part_list_xml = []
    parts_xml = []
    for output_idx, (_part_idx, part) in enumerate(selected, start=1):
        part_id = f"P{output_idx}"
        part_name = xml_escape(str(part.get("name") or f"Part {output_idx}"))
        part_list_xml.append(f'<score-part id="{part_id}"><part-name>{part_name}</part-name></score-part>')
        events = sorted(
            [event for event in (part.get("notes") or []) if isinstance(event, list) and len(event) >= 2],
            key=lambda event: float(event[1] or 0),
        )
        measures_xml = []
        for measure_idx, start in enumerate(measure_starts):
            end = measure_starts[measure_idx + 1] if measure_idx + 1 < len(measure_starts) else max(max_end, start + 4.0)
            attrs = ""
            if measure_idx == 0:
                attrs = (
                    f"<attributes><divisions>{divisions}</divisions>"
                    "<key><fifths>0</fifths></key><time><beats>4</beats><beat-type>4</beat-type></time>"
                    "<clef><sign>G</sign><line>2</line></clef></attributes>"
                )
            cursor = start
            notes_xml = []
            for event in events:
                beat = float(event[1] or 0)
                if beat < start - 0.0001 or beat >= end - 0.0001:
                    continue
                if beat > cursor:
                    notes_xml.append(rest_xml(beat - cursor))
                event_duration = float(event[2] or 1) if len(event) > 2 else 1
                clipped_duration = max(0.0, min(event_duration, end - beat))
                pitches = event_pitches(event)
                for pitch_idx, pitch in enumerate(pitches):
                    chord = "<chord />" if pitch_idx else ""
                    notes_xml.append(
                        f"<note>{chord}{midi_to_musicxml_pitch(pitch)}"
                        f"{duration_xml(clipped_duration)}<voice>1</voice></note>"
                    )
                cursor = max(cursor, beat + clipped_duration)
            if cursor < end:
                notes_xml.append(rest_xml(end - cursor))
            measures_xml.append(
                f'<measure number="{measure_idx + 1}">{attrs}{"".join(notes_xml)}</measure>'
            )
        parts_xml.append(f'<part id="{part_id}">{"".join(measures_xml)}</part>')

    escaped_title = xml_escape(title or "Selected parts")
    return (
        '<?xml version="1.0" encoding="UTF-8"?>'
        '<score-partwise version="3.1">'
        f"<movement-title>{escaped_title}</movement-title>"
        f"<part-list>{''.join(part_list_xml)}</part-list>"
        f"{''.join(parts_xml)}"
        "</score-partwise>"
    )


def build_fingered_score_variant(score: dict, progress_callback=None) -> tuple[str, str, dict]:
    progress = progress_callback or (lambda *_args, **_kwargs: None)
    musicxml_source = (score.get("musicxml_source") or "").strip()
    if not musicxml_source:
        raise HTTPException(status_code=400, detail="This score has no MusicXML source to annotate.")

    score_name = score.get("name") or "score"
    title = score.get("title") or score_name
    with tempfile.TemporaryDirectory(prefix="accompy_fingering_") as tmp_dir:
        work_dir = Path(tmp_dir)
        base_path = work_dir / f"{score_name}.musicxml"
        progress(10, "Preparing MusicXML")
        parts_data = score.get("parts") or []
        selected_piano_indices = piano_part_indices(parts_data)
        fingering_source = musicxml_source
        fingering_parts_data = parts_data
        if selected_piano_indices and len(selected_piano_indices) < len(parts_data):
            filtered_source = filter_musicxml_parts(musicxml_source, selected_piano_indices, parts_data)
            if filtered_source:
                fingering_source = filtered_source
                fingering_parts_data = [
                    parts_data[idx] for idx in selected_piano_indices if 0 <= idx < len(parts_data)
                ]

        base_path.write_text(fingering_source, encoding="utf-8")

        fingered_path, fingering = apply_auto_fingering(
            str(base_path),
            out_dir=str(work_dir),
            score_name=score_name,
            parts_data=fingering_parts_data,
            progress_callback=progress,
        )
        if not fingering.get("applied"):
            reason = fingering.get("reason") or "generation_failed"
            if reason == "unsupported_parts":
                detail = "Automatic fingering is available for piano or keyboard parts."
            elif reason == "missing_dependency":
                detail = "PianoPlayer is not installed in the backend environment."
            else:
                detail = f"Could not generate fingering ({reason})."
            raise HTTPException(status_code=400, detail=detail)

        fingered_html_path = work_dir / f"{score_name}__fingered.html"
        progress(80, "Rendering fingered sheet")
        render_html(str(fingered_path), str(fingered_html_path), title)

        progress(92, "Finalizing output")
        fingered_musicxml_source = Path(fingered_path).read_text(encoding="utf-8")
        fingered_sheet_html = fingered_html_path.read_text(encoding="utf-8") if fingered_html_path.exists() else ""
        return fingered_musicxml_source, fingered_sheet_html, fingering


def _fingering_job_public_payload(job: dict) -> dict:
    return {
        "id": job["id"],
        "score_name": job["score_name"],
        "status": job["status"],
        "progress": job["progress"],
        "message": job["message"],
        "error": job.get("error"),
        "annotations": job.get("annotations") or 0,
        "created_at": job["created_at"],
        "updated_at": job["updated_at"],
    }


def _update_fingering_job(job_id: str, **updates) -> dict | None:
    with _fingering_jobs_lock:
        job = _fingering_jobs.get(job_id)
        if not job:
            return None
        job.update(updates)
        job["updated_at"] = datetime.now(timezone.utc).isoformat()
        return dict(job)


def _create_or_get_active_fingering_job(user_id: str, score_name: str) -> tuple[dict, bool]:
    with _fingering_jobs_lock:
        active_job_id = _active_fingering_jobs.get((user_id, score_name))
        if active_job_id:
            active_job = _fingering_jobs.get(active_job_id)
            if active_job and active_job.get("status") in {"queued", "running"}:
                return dict(active_job), False
            _active_fingering_jobs.pop((user_id, score_name), None)

        now_iso = datetime.now(timezone.utc).isoformat()
        job_id = secrets.token_urlsafe(12)
        job = {
            "id": job_id,
            "user_id": user_id,
            "score_name": score_name,
            "status": "queued",
            "progress": 0,
            "message": "Queued",
            "error": None,
            "annotations": 0,
            "created_at": now_iso,
            "updated_at": now_iso,
        }
        _fingering_jobs[job_id] = job
        _active_fingering_jobs[(user_id, score_name)] = job_id
        return dict(job), True


def _load_fingering_job_for_user(user_id: str, score_name: str, job_id: str) -> dict:
    with _fingering_jobs_lock:
        job = _fingering_jobs.get(job_id)
        if not job or job.get("user_id") != user_id or job.get("score_name") != score_name:
            raise HTTPException(status_code=404, detail="Fingering job not found.")
        return dict(job)


def _finish_fingering_job(job_id: str):
    with _fingering_jobs_lock:
        job = _fingering_jobs.get(job_id)
        if not job:
            return
        _active_fingering_jobs.pop((job["user_id"], job["score_name"]), None)


def _import_job_public_payload(job: dict) -> dict:
    return {
        "id": job["id"],
        "status": job["status"],
        "progress": job["progress"],
        "message": job["message"],
        "error": job.get("error"),
        "result": job.get("result"),
        "created_at": job["created_at"],
        "updated_at": job["updated_at"],
    }


def _create_import_job(user_id: str, score_name: str, upload_paths: list[Path], work_dir: Path, output_dir: Path) -> dict:
    now_iso = datetime.now(timezone.utc).isoformat()
    job_id = secrets.token_urlsafe(12)
    job = {
        "id": job_id,
        "user_id": user_id,
        "score_name": score_name,
        "upload_paths": [str(path) for path in upload_paths],
        "work_dir": str(work_dir),
        "output_dir": str(output_dir),
        "status": "queued",
        "progress": 4,
        "message": "Queued import",
        "error": None,
        "result": None,
        "created_at": now_iso,
        "updated_at": now_iso,
    }
    with _import_jobs_lock:
        _import_jobs[job_id] = job
    return dict(job)


def _update_import_job(job_id: str, **updates) -> dict | None:
    with _import_jobs_lock:
        job = _import_jobs.get(job_id)
        if not job:
            return None
        job.update(updates)
        job["updated_at"] = datetime.now(timezone.utc).isoformat()
        return dict(job)


def _load_import_job_for_user(user_id: str, job_id: str) -> dict:
    with _import_jobs_lock:
        job = _import_jobs.get(job_id)
        if not job or job.get("user_id") != user_id:
            raise HTTPException(status_code=404, detail="Import job not found.")
        return dict(job)


def _run_fingering_job(job_id: str):
    job = _fingering_jobs.get(job_id)
    if not job:
        return

    user_id = job["user_id"]
    score_name = job["score_name"]

    try:
        _update_fingering_job(job_id, status="running", progress=5, message="Loading score")
        score = ensure_fingering_state(_score_store.load_score(user_id, score_name))
        if score.get("fingering", {}).get("applied"):
            _update_fingering_job(
                job_id,
                status="completed",
                progress=100,
                message="Fingering already generated",
                annotations=(score.get("fingering") or {}).get("annotations") or 0,
            )
            return

        fingered_musicxml_source, fingered_sheet_html, fingering = build_fingered_score_variant(
            score,
            progress_callback=lambda percent, message: _update_fingering_job(
                job_id,
                status="running",
                progress=max(0, min(99, int(percent))),
                message=message,
            ),
        )
        _update_fingering_job(job_id, progress=96, message="Saving fingering")
        save_score_with_render_artifacts(user_id, {
            "name": score["name"],
            "title": score.get("title") or score["name"],
            "parts": score.get("parts") or [],
            "measure_beats": score.get("measure_beats") or [],
            "sheet_html": score.get("sheet_html") or "",
            "musicxml_source": score.get("musicxml_source") or "",
            "fingered_sheet_html": fingered_sheet_html,
            "fingered_musicxml_source": fingered_musicxml_source,
            "fingering": fingering,
            "source_type": score.get("source_type") or "converted",
        })
        _update_fingering_job(
            job_id,
            status="completed",
            progress=100,
            message="Fingering ready",
            annotations=fingering.get("annotations") or 0,
        )
    except HTTPException as exc:
        detail = exc.detail if isinstance(exc.detail, str) else str(exc.detail)
        _update_fingering_job(job_id, status="failed", progress=100, message=detail, error=detail)
    except Exception as exc:
        _update_fingering_job(
            job_id,
            status="failed",
            progress=100,
            message="Fingering generation failed.",
            error=str(exc),
        )
    finally:
        _finish_fingering_job(job_id)


async def save_uploaded_file(upload: UploadFile, directory: Path, index: int) -> Path:
    suffix = Path(upload.filename or "").suffix.lower()
    stem = score_name_from_input(Path(upload.filename or f"upload_{index}").stem)
    dest = directory / f"{index:02d}_{stem}{suffix}"
    data = await upload.read()
    dest.write_bytes(data)
    await upload.close()
    return dest


def combine_images_to_pdf(image_paths: list[Path], out_path: Path) -> Path:
    import fitz

    doc = fitz.open()
    try:
        for image_path in image_paths:
            img = fitz.open(image_path)
            try:
                pdf_bytes = img.convert_to_pdf()
            finally:
                img.close()
            img_pdf = fitz.open("pdf", pdf_bytes)
            try:
                doc.insert_pdf(img_pdf)
            finally:
                img_pdf.close()
        doc.save(out_path)
    finally:
        doc.close()
    return out_path


def extract_lilypond_zip(zip_path: Path, work_dir: Path) -> list[Path]:
    lily_dir = work_dir / "lilypond"
    lily_dir.mkdir(parents=True, exist_ok=True)
    try:
        with zipfile.ZipFile(zip_path) as zf:
            lily_members = [
                member for member in zf.infolist()
                if not member.is_dir()
                and not member.filename.startswith("__MACOSX/")
                and Path(member.filename).suffix.lower() == ".ily"
            ]
            members = lily_members or [
                member for member in zf.infolist()
                if not member.is_dir()
                and not member.filename.startswith("__MACOSX/")
                and Path(member.filename).suffix.lower() == ".ly"
            ]
            if not members:
                raise HTTPException(status_code=400, detail="The .zip file must contain one or more .ily or .ly files.")

            extracted = []
            part_members = [
                member for member in members
                if Path(member.filename).stem.lower() not in LILYPOND_HELPER_STEMS
            ] or members
            for index, member in enumerate(sorted(part_members, key=lambda item: item.filename.lower())):
                safe_stem = score_name_from_input(Path(member.filename).stem or f"part_{index + 1}")
                dest = lily_dir / f"{index:02d}_{safe_stem}{Path(member.filename).suffix.lower()}"
                with zf.open(member) as src, open(dest, "wb") as out:
                    shutil.copyfileobj(src, out)
                extracted.append(dest)
            return extracted
    except zipfile.BadZipFile as exc:
        raise HTTPException(status_code=400, detail="The uploaded .zip file could not be read.") from exc


def prepare_omr_input(upload_paths: list[Path], work_dir: Path) -> Path | list[Path]:
    suffixes = {path.suffix.lower() for path in upload_paths}
    if suffixes & ALLOWED_DIRECT_SCORE_SUFFIXES:
        if len(upload_paths) != 1 or not suffixes <= ALLOWED_DIRECT_SCORE_SUFFIXES:
            raise HTTPException(status_code=400, detail="Upload one MusicXML/MuseScore file, LilyPond file set, .zip bundle, one PDF, or one/more image files.")
        return upload_paths[0]

    if suffixes & ALLOWED_LILYPOND_SUFFIXES:
        if not suffixes <= ALLOWED_LILYPOND_SUFFIXES:
            raise HTTPException(status_code=400, detail="Upload only .ily/.ly files when importing LilyPond parts.")
        part_paths = [
            path for path in upload_paths
            if path.stem.lower() not in LILYPOND_HELPER_STEMS
        ] or upload_paths
        return sorted(part_paths)

    if suffixes & ALLOWED_ZIP_SUFFIXES:
        if len(upload_paths) != 1 or not suffixes <= ALLOWED_ZIP_SUFFIXES:
            raise HTTPException(status_code=400, detail="Upload one .zip bundle, not a mix of .zip and other files.")
        return extract_lilypond_zip(upload_paths[0], work_dir)

    if suffixes & ALLOWED_PDF_SUFFIXES:
        if len(upload_paths) != 1 or not suffixes <= ALLOWED_PDF_SUFFIXES:
            raise HTTPException(status_code=400, detail="Upload one MusicXML/MuseScore file, LilyPond file set, .zip bundle, one PDF, or one/more image files.")
        return upload_paths[0]

    if not suffixes or not suffixes <= ALLOWED_IMAGE_SUFFIXES:
        raise HTTPException(status_code=400, detail="Supported uploads are MusicXML (.xml, .mxl, .musicxml), MuseScore (.mscx, .mscz), LilyPond (.ily/.ly or .zip), PDF, PNG, JPG, and JPEG.")

    return combine_images_to_pdf(upload_paths, work_dir / "input.pdf")


def run_audiveris(input_path: Path, output_dir: Path):
    audiveris_bin = os.getenv("AUDIVERIS_BIN", "audiveris").strip() or "audiveris"
    command = [
        *shlex.split(audiveris_bin),
        "-batch",
        "-transcribe",
        "-export",
        "-output",
        str(output_dir),
        str(input_path),
    ]
    try:
        completed = subprocess.run(command, capture_output=True, text=True, check=True)
    except FileNotFoundError as exc:
        raise HTTPException(
            status_code=500,
            detail="Audiveris is not installed or AUDIVERIS_BIN is not set.",
        ) from exc
    except subprocess.CalledProcessError as exc:
        message = (exc.stderr or exc.stdout or "Audiveris failed").strip()
        raise HTTPException(status_code=500, detail=f"Audiveris failed: {message}") from exc
    return completed


def find_musicxml_output(output_dir: Path) -> Path:
    candidates = sorted(
        [
            *output_dir.rglob("*.mxl"),
            *output_dir.rglob("*.musicxml"),
            *output_dir.rglob("*.xml"),
        ],
        key=lambda path: (path.suffix.lower() != ".mxl", -path.stat().st_size, str(path)),
    )
    for candidate in candidates:
        if candidate.name.lower().endswith("opus.xml"):
            continue
        return candidate
    raise HTTPException(status_code=500, detail="Audiveris completed but no MusicXML output was found.")


def convert_and_save_import(
    user_id: str,
    score_name: str,
    upload_paths: list[Path],
    work_dir: Path,
    output_dir: Path,
    progress_callback=None,
) -> dict:
    def progress(percent: int, message: str):
        if progress_callback:
            progress_callback(max(0, min(99, int(percent))), message)

    progress(10, "Preparing upload")
    import_input = prepare_omr_input(upload_paths, work_dir)
    if isinstance(import_input, list):
        progress(34, f"Preparing {len(import_input)} LilyPond parts")
        musicxml_path = Path(convert_lilypond_parts_to_musicxml(
            [str(path) for path in import_input],
            str(output_dir),
            score_name,
            progress_callback=progress,
        ))
    elif import_input.suffix.lower() in ALLOWED_DIRECT_SCORE_SUFFIXES:
        progress(34, "Reading uploaded score")
        musicxml_path = import_input
    else:
        progress(20, "Starting Audiveris")
        run_audiveris(import_input, output_dir)
        progress(68, "Collecting Audiveris output")
        musicxml_path = find_musicxml_output(output_dir)

    try:
        progress(76, "Building NotePilot score")
        result = convert_score_source(str(musicxml_path), name=score_name, out_dir=str(output_dir))
    except Exception as exc:
        suffix = musicxml_path.suffix.lower()
        source_label = "uploaded MusicXML/MuseScore file" if suffix in ALLOWED_DIRECT_SCORE_SUFFIXES else "Audiveris output"
        raise HTTPException(status_code=400, detail=f"Could not convert {source_label}: {exc}") from exc

    progress(92, "Saving imported score")
    saved = save_score_with_render_artifacts(user_id, {
        "name": result["name"],
        "title": result["title"],
        "parts": result["parts"],
        "measure_beats": result["measure_beats"],
        "sheet_html": Path(result["out_html"]).read_text(encoding="utf-8") if os.path.exists(result["out_html"]) else "",
        "musicxml_source": Path(result["render_source_path"]).read_text(encoding="utf-8") if os.path.exists(result["render_source_path"]) else "",
        "fingered_sheet_html": "",
        "fingered_musicxml_source": "",
        "fingering": result.get("fingering") or {},
        "source_type": "upload",
    })
    return {
        "name": saved["name"],
        "parts": len(saved["parts"]),
        "total_notes": sum(len(part.get("notes", [])) for part in saved["parts"]),
        "has_sheet": saved["has_sheet"],
    }


def _run_import_job(job_id: str):
    with _import_jobs_lock:
        job = dict(_import_jobs.get(job_id) or {})
    if not job:
        return

    work_dir = Path(job["work_dir"])
    try:
        _update_import_job(job_id, status="running", progress=8, message="Preparing upload")
        result = convert_and_save_import(
            job["user_id"],
            job["score_name"],
            [Path(path) for path in job["upload_paths"]],
            work_dir,
            Path(job["output_dir"]),
            progress_callback=lambda percent, message: _update_import_job(
                job_id,
                status="running",
                progress=percent,
                message=message,
            ),
        )
        _update_import_job(
            job_id,
            status="completed",
            progress=100,
            message="Import complete",
            result=result,
        )
    except HTTPException as exc:
        detail = exc.detail if isinstance(exc.detail, str) else str(exc.detail)
        _update_import_job(job_id, status="failed", progress=100, message=detail, error=detail)
    except Exception as exc:
        _update_import_job(
            job_id,
            status="failed",
            progress=100,
            message="Import failed.",
            error=str(exc),
        )
    finally:
        shutil.rmtree(work_dir, ignore_errors=True)


@app.get("/api/corpus/search")
def search_corpus(q: str = ""):
    index   = _corpus_index()
    q_lower = q.lower().strip()
    if not q_lower:
        # Return a sample: first 5 from each composer
        from collections import defaultdict
        by_composer = defaultdict(list)
        for e in index:
            by_composer[e['composer']].append(e)
        results = []
        for entries in by_composer.values():
            results.extend(entries[:5])
        return {"results": results[:60]}

    results = [
        e for e in index
        if q_lower in e['path'].lower()
        or q_lower in e['composer'].lower()
        or q_lower in e['title'].lower()
    ]
    return {"results": results[:60]}


@app.get("/api/config")
def get_config():
    return {
        "supabase_enabled": isinstance(_score_store, SupabaseScoreStore),
        "auth_enabled": isinstance(_score_store, SupabaseScoreStore),
        "google_auth_enabled": bool(
            isinstance(_score_store, SupabaseScoreStore)
            and os.getenv("SUPABASE_URL", "").strip()
            and os.getenv("SUPABASE_ANON_KEY", "").strip()
        ),
    }


class SimpleAuthRequest(BaseModel):
    username: str
    password: str


class SupabaseTokenRequest(BaseModel):
    access_token: str


@app.get("/api/session")
def get_session(request: Request):
    try:
        user = current_app_user_for_request(request)
    except HTTPException as exc:
        return {
            "authenticated": False,
            "user": None,
            "session_error": exc.detail,
        }
    return {
        "authenticated": bool(user),
        "user": {
            "id": user["id"],
            "username": user["username"],
        } if user else None,
    }


@app.post("/api/signup")
def signup(req: SimpleAuthRequest, response: Response):
    if not isinstance(_score_store, SupabaseScoreStore):
        raise HTTPException(status_code=400, detail="Signup requires Supabase-backed storage.")
    username = req.username.strip()
    password = req.password
    if not re.fullmatch(r"[A-Za-z0-9_]{3,32}", username):
        raise HTTPException(status_code=400, detail="Username must be 3-32 characters using letters, numbers, or underscore.")
    if len(password) < 4:
        raise HTTPException(status_code=400, detail="Password must be at least 4 characters.")
    existing = _score_store.get_app_user_by_username(username)
    if existing:
        raise HTTPException(status_code=400, detail="Username already exists.")
    created = _score_store.create_app_user(username, hash_password(password))
    if not created:
        raise HTTPException(status_code=500, detail="Could not create user.")
    return create_app_session_response(response, created)


@app.post("/api/login")
def login(req: SimpleAuthRequest, response: Response):
    if not isinstance(_score_store, SupabaseScoreStore):
        raise HTTPException(status_code=400, detail="Login requires Supabase-backed storage.")
    username = req.username.strip()
    password = req.password
    user = _score_store.get_app_user_by_username(username)
    if not user or not verify_password(password, user.get("password_hash") or ""):
        raise HTTPException(status_code=401, detail="Invalid username or password.")
    return create_app_session_response(response, user)


@app.get("/api/auth/google/start")
def start_google_auth(request: Request):
    if not isinstance(_score_store, SupabaseScoreStore):
        raise HTTPException(status_code=400, detail="Google sign-in requires Supabase-backed storage.")
    supabase_url = os.getenv("SUPABASE_URL", "").strip().rstrip("/")
    anon_key = os.getenv("SUPABASE_ANON_KEY", "").strip()
    if not (supabase_url and anon_key):
        raise HTTPException(status_code=400, detail="Google sign-in requires SUPABASE_URL and SUPABASE_ANON_KEY.")

    origin = public_app_origin(request)
    redirect_to = f"{origin}/auth/callback"
    auth_url = (
        f"{supabase_url}/auth/v1/authorize"
        f"?provider=google"
        f"&redirect_to={quote(redirect_to, safe='')}"
    )
    return RedirectResponse(auth_url, status_code=302)


def public_app_origin(request: Request) -> str:
    configured_origin = os.getenv("APP_PUBLIC_URL", "").strip().rstrip("/")
    if configured_origin:
        return configured_origin

    forwarded_proto = request.headers.get("x-forwarded-proto", "").split(",", 1)[0].strip()
    forwarded_host = request.headers.get("x-forwarded-host", "").split(",", 1)[0].strip()
    forwarded_port = request.headers.get("x-forwarded-port", "").split(",", 1)[0].strip()
    if forwarded_proto and forwarded_host:
        host = forwarded_host
        if forwarded_port and ":" not in host and forwarded_port not in {"80", "443"}:
            host = f"{host}:{forwarded_port}"
        return f"{forwarded_proto}://{host}"

    for header_name in ("x-original-host", "x-real-host", "host"):
        host = request.headers.get(header_name, "").split(",", 1)[0].strip()
        if host and not host.startswith(("127.0.0.1", "localhost")):
            proto = forwarded_proto or "https"
            return f"{proto}://{host}"

    space_host = os.getenv("SPACE_HOST", "").strip().rstrip("/")
    if space_host:
        return space_host if space_host.startswith(("http://", "https://")) else f"https://{space_host}"

    space_id = os.getenv("SPACE_ID", "").strip()
    if "/" in space_id:
        owner, space = space_id.split("/", 1)
        return f"https://{owner}-{space.replace('_', '-')}.hf.space"

    return str(request.base_url).rstrip("/")


@app.post("/api/auth/supabase-token")
def login_with_supabase_token(req: SupabaseTokenRequest, response: Response):
    if not isinstance(_score_store, SupabaseScoreStore):
        raise HTTPException(status_code=400, detail="Google sign-in requires Supabase-backed storage.")
    supabase_url = os.getenv("SUPABASE_URL", "").strip().rstrip("/")
    anon_key = os.getenv("SUPABASE_ANON_KEY", "").strip()
    access_token = req.access_token.strip()
    if not (supabase_url and anon_key and access_token):
        raise HTTPException(status_code=400, detail="Missing Google sign-in token.")

    auth_response = requests.get(
        f"{supabase_url}/auth/v1/user",
        headers={
            "apikey": anon_key,
            "Authorization": f"Bearer {access_token}",
        },
        timeout=15,
    )
    if auth_response.status_code != 200:
        raise HTTPException(status_code=401, detail="Google sign-in token was not accepted.")

    supabase_user = auth_response.json()
    app_metadata = supabase_user.get("app_metadata") or {}
    user_metadata = supabase_user.get("user_metadata") or {}
    provider = app_metadata.get("provider")
    providers = app_metadata.get("providers") or []
    issuer = str(user_metadata.get("iss") or "")
    if provider != "google" and "google" not in providers and "accounts.google.com" not in issuer:
        raise HTTPException(status_code=400, detail="This sign-in token is not from Google.")

    supabase_id = str(supabase_user.get("id") or "").replace("-", "")
    if not supabase_id:
        raise HTTPException(status_code=400, detail="Google sign-in did not return a user id.")
    email = str(supabase_user.get("email") or user_metadata.get("email") or "").strip().lower()
    username = email or f"google_{supabase_id[:24]}"
    user = _score_store.get_app_user_by_username(username)
    if not user:
        user = _score_store.create_app_user(username, hash_password(secrets.token_urlsafe(32)))
    if not user:
        raise HTTPException(status_code=500, detail="Could not create Google user.")
    return create_app_session_response(response, user)


# ── Debug: session traces ────────────────────────────────────────────────────
# Local development aid. The browser records what the follower heard, believed
# and played during a run (static/trace.js) and posts it here; it lands in
# .accompy_traces/ next to the repo so it can be read straight off disk.
# Unauthenticated on purpose: the server only binds to 127.0.0.1 in run.py.

MAX_TRACE_BYTES = 64 * 1024 * 1024


def _trace_filename(payload: dict, *dirs: Path) -> str:
    meta = payload.get("meta") or {}
    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    name = str(meta.get("score") or "session")
    slug = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")[:48] or "session"
    # The stamp has one-second resolution, so two saves of the same score in
    # the same second (Save pressed twice, trace.js's start() rescue re-posting)
    # resolve to the same name -- and the previous save was chmodded 0o444 by
    # save_debug_trace, so write_text on it raises PermissionError [Errno 13]
    # and the request 500s, losing a recording that cannot be made again. Never
    # hand back a name that already exists in any of `dirs`.
    filename = f"{stamp}_{slug}.json"
    n = 1
    while any((d / filename).exists() for d in dirs):
        n += 1
        filename = f"{stamp}_{slug}-{n}.json"
    return filename


@app.post("/api/debug/trace")
async def save_debug_trace(request: Request):
    raw = await request.body()
    if len(raw) > MAX_TRACE_BYTES:
        raise HTTPException(status_code=413, detail="Trace too large.")
    try:
        payload = json.loads(raw)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=f"Invalid JSON: {exc}") from exc
    if not isinstance(payload, dict) or not isinstance(payload.get("events"), list):
        raise HTTPException(status_code=400, detail="Trace must be an object with an events list.")

    traces_dir = get_traces_dir()
    backup_dir = Path.home() / ".notepilot_trace_backups"
    filename = _trace_filename(payload, traces_dir, backup_dir)
    target = traces_dir / filename
    text = json.dumps(payload, separators=(",", ":"))
    # This is the save path for a recording that cannot be made again: say what
    # went wrong rather than letting the write surface as an opaque 500.
    try:
        target.write_text(text, encoding="utf-8")
        (traces_dir / "latest.json").write_text(text, encoding="utf-8")
    except OSError as exc:
        raise HTTPException(status_code=500, detail=f"Could not write trace to {target}: {exc}") from exc

    # A recording is the only record of a session that cannot be reproduced, so
    # every save is mirrored OUTSIDE the repo immediately and the copy is made
    # read-only. Tooling that cleans up .accompy_traces/ cannot reach it there.
    backup_note = None
    try:
        backup_dir.mkdir(parents=True, exist_ok=True)
        backup = backup_dir / filename
        backup.write_text(text, encoding="utf-8")
        os.chmod(backup, 0o444)
        backup_note = str(backup)
    except OSError as exc:
        backup_note = f"backup failed: {exc}"
    # The timestamped original is never rewritten (_trace_filename never hands
    # back an existing name), so it can be read-only too; latest.json stays
    # writable. OUTSIDE the backup try: when the mirror failed, the except
    # swallowed this too and silently left the repo copy writable -- exactly the
    # protection the comment above promises.
    try:
        os.chmod(target, 0o444)
    except OSError:
        pass

    return {
        "ok": True,
        "backup": backup_note,
        "path": str(target.relative_to(traces_dir.parent)) if target.is_relative_to(traces_dir.parent) else str(target),
        "events": len(payload["events"]),
        "markers": len(payload.get("markers") or []),
    }


# ── macOS alert sounds while the microphone is listening ──────────────────
# A notification "ding" played through the speaker lands in the microphone and
# the follower. A web page cannot silence other applications, but this server
# runs on the player's own machine, so it can turn the macOS ALERT volume --
# the channel system alerts and notification sounds use -- down to zero while
# the microphone is on and put it back afterwards. Notification banners still
# appear; only their sound is muted. Localhost and macOS only.
_ALERT_VOLUME_LOCK = threading.Lock()
_ALERT_VOLUME_SAVED: int | None = None   # alert volume before muting; None = not muted by us


def _osascript(expression: str) -> subprocess.CompletedProcess:
    return subprocess.run(["osascript", "-e", expression], capture_output=True, text=True, timeout=5, check=False)


def _restore_alert_volume() -> None:
    global _ALERT_VOLUME_SAVED
    with _ALERT_VOLUME_LOCK:
        if _ALERT_VOLUME_SAVED is None:
            return
        try:
            _osascript(f"set volume alert volume {int(_ALERT_VOLUME_SAVED)}")
        finally:
            _ALERT_VOLUME_SAVED = None


atexit.register(_restore_alert_volume)


@app.post("/api/system/alert-sounds")
async def set_alert_sounds(request: Request):
    global _ALERT_VOLUME_SAVED
    host = request.client.host if request.client else ""
    if host not in ("127.0.0.1", "::1", "localhost"):
        raise HTTPException(status_code=403, detail="Only the machine running the server may change its alert volume.")
    if sys.platform != "darwin":
        return JSONResponse({"ok": False, "supported": False, "reason": "macOS only"})
    try:
        payload = json.loads((await request.body()) or b"{}")
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=f"Invalid JSON: {exc}") from exc
    mute = bool(payload.get("mute")) if isinstance(payload, dict) else False
    if not mute:
        _restore_alert_volume()
        return {"ok": True, "supported": True, "muted": False}
    with _ALERT_VOLUME_LOCK:
        if _ALERT_VOLUME_SAVED is None:
            current = _osascript("alert volume of (get volume settings)")
            try:
                saved = int((current.stdout or "").strip())
            except ValueError:
                saved = 100
            done = _osascript("set volume alert volume 0")
            if done.returncode != 0:
                return JSONResponse({"ok": False, "supported": True, "reason": (done.stderr or "osascript failed").strip()})
            _ALERT_VOLUME_SAVED = saved
        return {"ok": True, "supported": True, "muted": True, "restoreTo": _ALERT_VOLUME_SAVED}


@app.post("/api/logout")
def logout(request: Request, response: Response):
    if isinstance(_score_store, SupabaseScoreStore):
        token = request.cookies.get(SESSION_COOKIE_NAME, "").strip()
        if token:
            _score_store.delete_app_session(token)
            _clear_cached_app_user(token)
    response.delete_cookie(SESSION_COOKIE_NAME, path="/")
    return {"ok": True}


@app.get("/api/scores")
def list_scores(request: Request):
    user_id = require_supabase_user_id(request, "score listing")
    return _score_store.list_scores(user_id)


@app.get("/api/scores/{name}")
def get_score(name: str, request: Request):
    user_id = require_supabase_user_id(request, "score loading")
    return ensure_fingering_state(_score_store.load_score(user_id, name, include_sheet_assets=False))


class InstrumentUpdate(BaseModel):
    part_index: int
    instrument: str


class ScoreTitleUpdate(BaseModel):
    title: str


@app.patch("/api/scores/{name}/instrument")
def update_instrument(name: str, req: InstrumentUpdate, request: Request):
    """Persist an instrument change for a part in the stored score."""
    user_id = require_supabase_user_id(request, "score updates")
    score = ensure_fingering_state(_score_store.load_score(user_id, name))
    parts = score["parts"]
    if req.part_index < 0 or req.part_index >= len(parts):
        raise HTTPException(status_code=400, detail="Invalid part index")
    parts[req.part_index]["instrument"] = req.instrument
    _score_store.save_score(user_id, {
        "name": score["name"],
        "title": score.get("title") or score["name"],
        "parts": parts,
        "measure_beats": score.get("measure_beats") or [],
        "sheet_html": score.get("sheet_html") or "",
        "musicxml_source": score.get("musicxml_source") or "",
        "fingering": score.get("fingering") or {},
        "source_type": score.get("source_type") or "converted",
    })
    return {"updated": True}


@app.patch("/api/scores/{name}/title")
def update_score_title(name: str, req: ScoreTitleUpdate, request: Request):
    user_id = require_supabase_user_id(request, "score renaming")
    title = req.title.strip()
    if not title:
        raise HTTPException(status_code=400, detail="Title cannot be empty.")
    if len(title) > 120:
        raise HTTPException(status_code=400, detail="Title must be 120 characters or fewer.")
    renamed = _score_store.rename_score_title(user_id, name, title)
    return {"name": renamed["name"], "title": renamed["title"]}


@app.delete("/api/scores/{name}")
def delete_score(name: str, request: Request):
    user_id = require_supabase_user_id(request, "deletes")
    _score_store.delete_score(user_id, name)
    return {"deleted": [name]}


@app.get("/api/scores/{name}/meta")
def get_score_meta(name: str, request: Request):
    return {"name": name, "mtime": 0}


@app.get("/api/scores/{name}/musicxml")
def get_score_musicxml(name: str, request: Request, variant: str = "base"):
    user_id = require_supabase_user_id(request, "sheet source loading")
    row = _score_store.load_score_row(user_id, name)
    score_data = row.get("score_data") or {}
    if variant == "fingered":
        xml_source = score_data.get("fingered_musicxml_source") or ""
        if not xml_source:
            raise HTTPException(status_code=404, detail="No fingered MusicXML source for this score")
    else:
        xml_source = score_data.get("musicxml_source") or ""
        if not xml_source:
            raise HTTPException(status_code=404, detail="No MusicXML source for this score")
    return PlainTextResponse(
        content=xml_source,
        media_type="application/vnd.recordare.musicxml+xml",
        headers={"Cache-Control": "no-store"},
    )


@app.get("/api/scores/{name}/artifact-url")
def get_score_artifact_url(name: str, request: Request, variant: str = "base"):
    user_id = require_supabase_user_id(request, "sheet artifact loading")
    if not r2_configured():
        raise HTTPException(status_code=503, detail="R2 artifact storage is not configured.")
    row = _score_store.load_score_row(user_id, name)
    normalized_variant = "fingered" if variant == "fingered" else "base"
    artifact_type = sheet_artifact_type_for_variant(normalized_variant)
    artifact = get_score_artifact(row, user_id, artifact_type)

    if not artifact:
        html = render_full_score_html_for_artifact(row, normalized_variant)
        if html and "<svg" in html:
            persist_score_render_artifact(user_id, name, artifact_type, html)
            row = _score_store.load_score_row(user_id, name)
            artifact = get_score_artifact(row, user_id, artifact_type)

    if not artifact:
        raise HTTPException(status_code=404, detail="Score render artifact not found")

    return {
        "url": get_r2_artifact_download_url(artifact["object_key"]),
        "storage_provider": artifact.get("storage_provider") or "r2",
        "content_type": artifact.get("content_type") or SHEET_HTML_CONTENT_TYPE,
        "size_bytes": artifact.get("size_bytes") or 0,
    }


@app.get("/api/scores/{name}/sheet")
def get_sheet(name: str, request: Request, variant: str = "base", parts: str | None = None):
    user_id = require_supabase_user_id(request, "sheet loading")
    row = _score_store.load_score_row(user_id, name)
    score = ensure_fingering_state(_score_row_to_payload(row))
    title = score.get("title") or score.get("name") or name
    score_data = row.get("score_data") or {}
    part_indices = parse_sheet_part_indices(parts)
    score_parts_meta = score.get("parts") or []
    cached_base_html = (row.get("sheet_html") or score.get("sheet_html") or "").strip()
    cached_fingered_html = (score_data.get("fingered_sheet_html") or "").strip()

    def render_filtered_source(
        xml_source: str,
        *,
        cache_variant: str,
        stack_fingering_chords: bool = False,
    ) -> str:
        filtered_source = ""
        try:
            filtered_source = filter_musicxml_parts(xml_source, part_indices, score_parts_meta)
        except Exception as exc:
            print(f"Could not filter sheet parts for {name}: {exc}")
        if not filtered_source:
            filtered_source = xml_source if not part_indices else ""
        measure_map = []
        if part_indices:
            try:
                collapsed_source, measure_map = collapse_empty_musicxml_measures(
                    filtered_source,
                    include_measure_map=True,
                )
                filtered_source = collapsed_source or filtered_source
            except Exception as exc:
                print(f"Could not collapse empty selected-part measures for {name}: {exc}")
                measure_map = []
        html = ""
        if filtered_source:
            parts_key = ",".join(str(idx) for idx in part_indices) if part_indices else "full"
            cache_path = sheet_render_cache_path(
                score.get("name") or name,
                cache_variant,
                parts_key,
                filtered_source,
                stack_fingering_chords=stack_fingering_chords,
            )
            html = load_cached_sheet_render(cache_path)
            if not html:
                try:
                    html = render_sheet_html_from_musicxml_text(
                        score.get("name") or name,
                        title,
                        filtered_source,
                        stack_fingering_chords=stack_fingering_chords,
                    )
                    html = inject_sheet_measure_map(html, measure_map)
                    save_cached_sheet_render(cache_path, html)
                except Exception as exc:
                    print(f"Could not render MusicXML sheet for {name}: {exc}")
            if html and "<svg" in html:
                return html
        if (not html or "<svg" not in html) and part_indices:
            try:
                synthetic_source = synthesize_selected_parts_musicxml(score, part_indices, title)
                if synthetic_source:
                    parts_key = "synthetic-" + ",".join(str(idx) for idx in part_indices)
                    cache_path = sheet_render_cache_path(
                        score.get("name") or name,
                        cache_variant,
                        parts_key,
                        synthetic_source,
                        stack_fingering_chords=stack_fingering_chords,
                    )
                    html = load_cached_sheet_render(cache_path)
                    if not html:
                        html = render_sheet_html_from_musicxml_text(
                            score.get("name") or name,
                            title,
                            synthetic_source,
                            stack_fingering_chords=stack_fingering_chords,
                        )
                        save_cached_sheet_render(cache_path, html)
            except Exception as exc:
                print(f"Could not render synthesized selected sheet for {name}: {exc}")
        return html

    if variant == "fingered":
        html = ""
        if not part_indices:
            html = fetch_score_artifact_html(row, user_id, "fingered_html")
        if score.get("fingered_musicxml_source"):
            if not html or "<svg" not in html:
                html = render_filtered_source(
                    score.get("fingered_musicxml_source") or "",
                    cache_variant="fingered",
                    stack_fingering_chords=True,
                )
                if html and "<svg" in html and not part_indices:
                    persist_score_render_artifacts(user_id, name, [("fingered_html", html)])
        if (not html or "<svg" not in html) and not part_indices:
            html = stack_fingering_chord_numbers_in_html(cached_fingered_html)
    else:
        html = "" if part_indices else cached_base_html
        if (not html or "<svg" not in html) and not part_indices:
            html = fetch_score_artifact_html(row, user_id, "full_html")
        if (not html or "<svg" not in html) and score.get("musicxml_source"):
            html = render_filtered_source(
                score.get("musicxml_source") or "",
                cache_variant="base",
            )
            if html and "<svg" in html and not part_indices:
                try:
                    if r2_configured():
                        persist_score_render_artifact(user_id, name, "full_html", html)
                    else:
                        _score_store.update_score_sheet_html(user_id, name, html)
                except Exception as exc:
                    print(f"Could not cache rendered sheet HTML for {name}: {exc}")
    if not html or "<svg" not in html:
        raise HTTPException(status_code=404, detail="No sheet music for this score")
    html = strip_sheet_title(html)
    return HTMLResponse(content=html, headers={"Cache-Control": "no-store"})


@app.post("/api/scores/{name}/fingering/generate")
def generate_score_fingering(name: str, request: Request):
    user_id = require_supabase_user_id(request, "fingering generation")
    score = ensure_fingering_state(_score_store.load_score(user_id, name))
    if score.get("fingering", {}).get("applied"):
        raise HTTPException(status_code=400, detail="This score already has generated fingering.")

    job, created = _create_or_get_active_fingering_job(user_id, name)
    if created:
        worker = threading.Thread(target=_run_fingering_job, args=(job["id"],), daemon=True)
        worker.start()
    return JSONResponse(content=_fingering_job_public_payload(job), status_code=202)


@app.get("/api/scores/{name}/fingering/jobs/{job_id}")
def get_score_fingering_job(name: str, job_id: str, request: Request):
    user_id = require_supabase_user_id(request, "fingering status checks")
    job = _load_fingering_job_for_user(user_id, name, job_id)
    return _fingering_job_public_payload(job)


class ConvertRequest(BaseModel):
    corpus_path: str
    name: str


@app.post("/api/convert")
def convert_score(req: ConvertRequest, request: Request):
    temp_dir = tempfile.TemporaryDirectory(prefix="accompy_convert_")
    out_dir = temp_dir.name
    try:
        result = convert_score_source(f"corpus:{req.corpus_path}", name=score_name_from_input(req.name), out_dir=out_dir)
    except Exception as exc:
        temp_dir.cleanup()
        raise HTTPException(status_code=400, detail=str(exc)) from exc

    user_id = require_supabase_user_id(request, "score saving")
    saved = save_score_with_render_artifacts(user_id, {
        "name": result["name"],
        "title": result["title"],
        "parts": result["parts"],
        "measure_beats": result["measure_beats"],
        "sheet_html": Path(result["out_html"]).read_text(encoding="utf-8") if os.path.exists(result["out_html"]) else "",
        "musicxml_source": Path(result["render_source_path"]).read_text(encoding="utf-8") if os.path.exists(result["render_source_path"]) else "",
        "fingered_sheet_html": "",
        "fingered_musicxml_source": "",
        "fingering": result.get("fingering") or {},
        "source_type": "corpus",
    })
    temp_dir.cleanup()
    return {
        "name": saved["name"],
        "parts": len(saved["parts"]),
        "total_notes": sum(len(part.get("notes", [])) for part in saved["parts"]),
        "has_sheet": saved["has_sheet"],
    }


@app.post("/api/import")
async def import_score(
    request: Request,
    files: list[UploadFile] = File(...),
    name: str = Form(""),
):
    if not files:
        raise HTTPException(status_code=400, detail="No files uploaded.")

    score_name = score_name_from_input(name or Path(files[0].filename or "imported_score").stem)

    with tempfile.TemporaryDirectory(prefix="accompy_import_") as tmp_dir:
        work_dir = Path(tmp_dir)
        uploads_dir = work_dir / "uploads"
        output_dir = work_dir / "audiveris"
        uploads_dir.mkdir(parents=True, exist_ok=True)
        output_dir.mkdir(parents=True, exist_ok=True)

        upload_paths = [await save_uploaded_file(upload, uploads_dir, idx) for idx, upload in enumerate(files)]
        user_id = require_supabase_user_id(request, "score saving")
        return convert_and_save_import(user_id, score_name, upload_paths, work_dir, output_dir)


@app.post("/api/import/start")
async def start_import_score(
    request: Request,
    files: list[UploadFile] = File(...),
    name: str = Form(""),
):
    if not files:
        raise HTTPException(status_code=400, detail="No files uploaded.")

    user_id = require_supabase_user_id(request, "score saving")
    score_name = score_name_from_input(name or Path(files[0].filename or "imported_score").stem)
    work_dir = Path(tempfile.mkdtemp(prefix="accompy_import_"))
    uploads_dir = work_dir / "uploads"
    output_dir = work_dir / "audiveris"
    uploads_dir.mkdir(parents=True, exist_ok=True)
    output_dir.mkdir(parents=True, exist_ok=True)

    try:
        upload_paths = [await save_uploaded_file(upload, uploads_dir, idx) for idx, upload in enumerate(files)]
        job = _create_import_job(user_id, score_name, upload_paths, work_dir, output_dir)
        thread = threading.Thread(target=_run_import_job, args=(job["id"],), daemon=True)
        thread.start()
        return _import_job_public_payload(job)
    except Exception:
        shutil.rmtree(work_dir, ignore_errors=True)
        raise


@app.get("/api/import/jobs/{job_id}")
def get_import_job(job_id: str, request: Request):
    user_id = require_supabase_user_id(request, "score importing")
    return _import_job_public_payload(_load_import_job_for_user(user_id, job_id))


def serve_spa_index():
    html = (Path(STATIC_DIR) / "index.html").read_text(encoding="utf-8")
    return HTMLResponse(content=html, headers={"Cache-Control": "no-store"})


@app.get("/signin", include_in_schema=False)
@app.get("/login", include_in_schema=False)
@app.get("/auth/callback", include_in_schema=False)
def serve_auth_spa_routes():
    return serve_spa_index()


# Serve static files and fallback to index.html for browser-routed score URLs.
app.mount("/", SPAStaticFiles(directory=STATIC_DIR, html=True), name="static")