""" 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 os import re import shlex import shutil import tempfile import subprocess 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 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): async def get_response(self, path: str, scope): try: return 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: return await super().get_response("index.html", scope) raise # ── 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 " str: return re.sub(r"\s*

.*?

\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 " 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 " 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 " 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 " 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'' if "" in html: return html.replace("", f"{script}\n", 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}" if alter else "" return f"{step}{alter_xml}{octave}" 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}" def rest_xml(beats: float) -> str: if beats <= 0.0001: return "" return f"{duration_xml(beats)}1" 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'{part_name}') 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"{divisions}" "0" "G2" ) 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 = "" if pitch_idx else "" notes_xml.append( f"{chord}{midi_to_musicxml_pitch(pitch)}" f"{duration_xml(clipped_duration)}1" ) cursor = max(cursor, beat + clipped_duration) if cursor < end: notes_xml.append(rest_xml(end - cursor)) measures_xml.append( f'{attrs}{"".join(notes_xml)}' ) parts_xml.append(f'{"".join(measures_xml)}') escaped_title = xml_escape(title or "Selected parts") return ( '' '' f"{escaped_title}" f"{''.join(part_list_xml)}" f"{''.join(parts_xml)}" "" ) 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) @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 " 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 "