"""marathonmvp — one app. This week's world, any track measured against it, and every track already measured kept ready to re-measure. Five tabs behind one passphrase (design/ONE_APP.md): The world what is charting everywhere this week, coloured by genre family, with the records charting in the most markets. The landing view: the global picture comes before anything catalogue-relevant. This week the weekly drop sheet, rendered natively from the same JSON the static sheet was built from, so the app reads on a Monday with no file to hand. Analyse upload a track, or find one by name. The report. Saved every analysis run so far. Open, re-run against this week, re-score the whole list, delete. How it works the methods and the limits, in plain language. The three surfaces this replaces were a static encrypted Space for the sheet, a local prototype for the trend space, and this Gradio app for the matcher. Framing the static Space inside a tab was checked and rejected: it would ask for the passphrase a second time inside the frame, and put two scroll containers inside a nested iframe on a phone. design/ONE_APP.md §2 has the headers that were measured. Gated: the Space repo can be public because nothing runs without the team passphrase, which lives only in the Space secret MARATHON_LIVE_PASSPHRASE (same value as the drop sheet's). No passphrase set → the app refuses to analyse anything rather than silently serving an open endpoint. The corpus (corpus.npz + corpus.json) is this week's scan exported by `marathon.cli export-live`. It is NOT in this public repo: it lives in the private dataset repo kalamishere/marathon-corpus and is fetched at startup with the HF_TOKEN Space secret, so repo browsers see code but no data. Local dev keeps working: files already present next to app.py win. No database, no network at match time. Look: the Gradio chrome is themed to the drop sheet's palette (paper background, near-black accent, the same greens and oranges) so the two surfaces read as one product. The results block is rendered by `render.py`, which owns the drop sheet's card and chip styling. """ from __future__ import annotations import hmac import os import re import shutil import tempfile import threading import time import traceback import urllib.parse from collections import OrderedDict from pathlib import Path import gradio as gr import soundfile as sf import analysis import clap_embed import howitworks import livematch import lookup as lookupmod import reachview import render import savedview import store import tags as tagmod import worldmap import worldview HERE = Path(__file__).resolve().parent CORPUS_DATASET = "kalamishere/marathon-corpus" # Snippet clips are written here and served back through Gradio's own file # route, which is what lets them sit inside the results HTML instead of in a # detached row of players underneath it. The directory is passed to # `launch(allowed_paths=...)`; without that the route refuses to serve them. SNIP_ROOT = Path(tempfile.gettempdir()) / "marathon_snippets" SNIP_TTL_S = 3600.0 def _fetch_corpus(dest: Path) -> Path: """Fetch corpus files from the private dataset repo unless they are already present locally (dev checkout / staged verify dir).""" if (dest / "corpus.npz").exists() and (dest / "corpus.json").exists(): return dest from huggingface_hub import hf_hub_download token = os.environ.get("HF_TOKEN") if not token: raise RuntimeError( "corpus files missing and HF_TOKEN not set — cannot fetch " f"{CORPUS_DATASET}") for name in ("corpus.npz", "corpus.json"): got = hf_hub_download(repo_id=CORPUS_DATASET, filename=name, repo_type="dataset", token=token) (dest / name).unlink(missing_ok=True) (dest / name).symlink_to(got) return dest CORPUS = livematch.Corpus.load(_fetch_corpus(HERE)) # Singapore, Hong Kong, Malta and Cape Verde reach the corpus with their ISO # code where their name should be — the export reads names out of the map's # country file, and the map has no outline for them. Named once, here, so # every tab reading this corpus reads names. worldmap.name_regions(CORPUS.regions) PASSPHRASE = os.environ.get("MARATHON_LIVE_PASSPHRASE", "") SNIPPETS_PER_SET = 3 # two sets: closest-to-trend and strongest-hook _state = {"embedder": None, "vocab": None, "zero_shot_ok": None} _model_lock = threading.Lock() _run_lock = threading.Lock() _warm = {"done": False, "seconds": None, "error": None} def _embedder(): """One model, loaded once. The load is ~90 seconds of checkpoint fetch and setup. `warm_up()` starts it on a background thread the moment the app begins serving, so it is normally finished before anyone uploads anything. An analysis that arrives while that thread is still working waits on this lock rather than loading a second two-gigabyte copy. """ with _model_lock: if _state["embedder"] is None: t0 = time.time() emb = clap_embed.Embedder() try: _state["vocab"] = tagmod.TagVocab(emb) except Exception as exc: # text tower missing → group two only print(f"[tags] text tower unavailable, zero-shot disabled: {exc}") _state["vocab"] = None _state["embedder"] = emb _warm["seconds"] = round(time.time() - t0, 1) _warm["done"] = True print(f"[warm] model ready in {_warm['seconds']}s", flush=True) return _state["embedder"] def warm_up() -> threading.Thread: """Load the model at startup instead of on the first upload. The thread is started just before `launch()`, so the gate and the upload zone are on screen while the checkpoint is still coming down. A failure here is not fatal: the embedder stays unset and the next analysis tries the load again, reporting through the progress line instead of a log nobody reads. """ def go(): try: _embedder() except Exception as exc: _warm["error"] = str(exc) print(f"[warm] load failed, the first analysis will retry: {exc}", flush=True) traceback.print_exc() t = threading.Thread(target=go, name="warm-clap", daemon=True) t.start() return t # What a phone or a notes app does to pasted text: curls the quotes, # stretches the dashes, and rides along invisible spaces. The passphrase a # reader pastes is the passphrase they were sent — fold these before # comparing, on both sides, so the gate judges the phrase and not the app # it was copied out of (reported 23 Aug: the phrase refused on mobile after # a copy from Notion). _FOLD = str.maketrans({"‘": "'", "’": "'", "“": '"', "”": '"', "–": "-", "—": "-", " ": " ", "​": "", "": ""}) def _fold(text: str) -> str: return text.translate(_FOLD).strip() def _ok(passphrase: str | None) -> bool: """The one place the gate is decided. Every path that computes anything goes through here first, including the named API route.""" return bool(PASSPHRASE) and hmac.compare_digest( _fold(passphrase or ""), _fold(PASSPHRASE)) def _notice(message: str) -> str: """Gate and input errors in the same visual language as the results.""" return (render.CSS + '
' f'{message}
') def _snip_dir() -> Path: """A fresh directory for this run's clips, with anything older than an hour swept up. The Space box is small and every analysis writes ~2.5 MB.""" SNIP_ROOT.mkdir(parents=True, exist_ok=True) cutoff = time.time() - SNIP_TTL_S for old in SNIP_ROOT.iterdir(): try: if old.is_dir() and old.stat().st_mtime < cutoff: shutil.rmtree(old, ignore_errors=True) except OSError: pass return Path(tempfile.mkdtemp(prefix="run_", dir=str(SNIP_ROOT))) def _clip(path: str, start_s: float, end_s: float, out_dir: str) -> str: """A real slice of the uploaded file, written out so the browser can play it. Original sample rate and channels — the point is to judge the clip. MP3 where libsndfile can write it: four 30-second stereo WAVs is ~23 MB down a phone connection, and this is a listening check, not a master. """ info = sf.info(path) sr = info.samplerate data, _ = sf.read(path, start=int(start_s * sr), stop=int(end_s * sr), dtype="float32") stem = os.path.join(out_dir, f"snippet_{int(start_s)}-{int(end_s)}") try: out = stem + ".mp3" sf.write(out, data, sr, format="MP3") return out except Exception: out = stem + ".wav" sf.write(out, data, sr) return out def _servable(paths: list) -> list: """Copies of stored files that Gradio's file route will actually serve. `hf_hub_download` lands files in the HF cache, which sits outside `allowed_paths` — the file route refused every one of them and reopened reports drew players that would not play (seen live, 22 Aug). Copying into the clips root puts them inside the allowlist; the hourly sweep cleans them up like any run's clips. """ if not any(paths): return list(paths) d = _snip_dir() out = [] for p in paths: if not p: out.append(None) continue try: dest = Path(d) / Path(p).name shutil.copy(p, dest) out.append(str(dest)) except OSError: out.append(None) return out def _clip_url(path: str) -> str: """Gradio 5+ serves allowed local files from /gradio_api/file=. Relative on purpose: the Space is embedded in an iframe, so the URL has to resolve against whatever host is serving the app.""" return "/gradio_api/file=" + urllib.parse.quote(str(Path(path).resolve())) def _sections(path: str) -> int: """How many 30-second windows this file yields on the 5-second grid. Read off the file header so the progress line can quote a real number rather than a guess. Same arithmetic as `Embedder.embed_windows`. """ try: dur = float(sf.info(path).duration) except Exception: return 0 if dur < analysis.WINDOW_S: return 0 return int((dur - analysis.WINDOW_S) // clap_embed.GRID_S) + 1 # The name of the setting is deliberately not here. Whoever reads this cannot # act on it — they are a label reader, not the person who set the app up — and # the deploy notes carry the name for the person who can. NO_SECRET = ("No team passphrase has been set for this app. Whoever " "set it up has to add one before anything will run.") WRONG_PASS = ("That passphrase is not right. It is the same one the drop " "sheet uses.") # Shown on every tab, so it must be true on every tab — This week and Saved # have no Analyse button to press. NEED_PASS = ("Enter the team passphrase at the top to unlock this. One " "passphrase opens every tab.") def _analyse(audio_path: str, progress, label: str | None = None, source: str = "upload", resolved: dict | None = None, lead_html: str = "", snippet_note: str | None = None): """The work, with the progress line naming the stage it is actually in. Four stages, each announced when the work it names begins. Nothing here animates between them: a bar that moves while the model is thinking would be inventing information the app does not have. The report is yielded twice. Cutting six 30-second clips costs 8.3 seconds on the Space — measured, not assumed — and none of the markets, the snippet times or the tags depend on it, so the whole report goes on screen first and the players arrive on the cards afterwards. On a local machine that gap is one second and the second yield lands almost on top of the first, which costs nothing. `listen` and `score` are called separately rather than through `analyse_track`, because the archive needs the `listen` half — those vectors are what make a re-score next Monday free. """ if not _warm["done"]: progress(0.02, desc="warming up — about a minute, the first time " "after a quiet spell") emb = _embedder() n = _sections(audio_path) progress(0.20, desc=( f"listening to the track — {n} sections of it against " f"{len(CORPUS.sounds):,} charting sounds" if n else "listening to the track")) t0 = time.time() heard = analysis.listen(audio_path, emb) res = analysis.score(heard, CORPUS, vocab=_state["vocab"], snippet_top=SNIPPETS_PER_SET) t_analysis = time.time() - t0 label = label or Path(audio_path).stem progress(0.85, desc="cutting the clips") t0 = time.time() yield lead_html + render.render(res, label, snippet_note=snippet_note) t_first = time.time() - t0 # Each snippet card carries its own player, so the clips are cut and # handed to the renderer as URLs. The union of both sets is at most 6 # clips; picks reference them by clip_i. t0 = time.time() clips: list[str | None] = [] clip_files: list[str | None] = [] if not snippet_note: out_dir = _snip_dir() for p in res["snippets"]: try: # Cut to the last bar line inside the window when there is # one, so what the team hears finishes its phrase instead of # stopping flat. The delivered window stays the full 30 # seconds — that is the field a distributor asks a label to # fill — and the card prints both. end = p.get("natural_end_s") or p["end_s"] f = _clip(audio_path, p["start_s"], end, str(out_dir)) clip_files.append(f) clips.append(_clip_url(f)) except Exception as exc: print(f"[clip] {exc}") clip_files.append(None) clips.append(None) t_clips = time.time() - t0 # The whole track, playable at the top of the report — comparing means # hearing the analysed track next to the trending sounds it is scored # against. track_url = None try: track_copy = _servable([audio_path])[0] track_url = _clip_url(track_copy) if track_copy else None except Exception as exc: print(f"[track-player] {exc}") progress(0.98, desc="adding the players") t0 = time.time() html = lead_html + render.render(res, label, snippet_clips=clips, snippet_note=snippet_note, track_audio_url=track_url) # Kept without being asked. A team running a list of forty tracks and # tapping save on each one would save none of them by track ten, and the # vectors are the whole reason a track never needs analysing twice. # A failed save must not lose the report that is already on screen. t_save = time.time() try: store.save(res, heard, label, source=source, resolved=resolved, clip_paths=clip_files, source_audio=audio_path) except Exception as exc: traceback.print_exc() html += _notice( "This analysis could not be added to Saved. The report above is " "complete — copy anything you need from it.") # Printed so the split between the stages, and the gap the early yield # actually buys, can be read off the Space's own log rather than guessed # at from a stopwatch. print(f"[timing] analysis {t_analysis:.1f}s · first report {t_first:.1f}s " f"· clips {t_clips:.1f}s · players {t_save - t0:.1f}s " f"· save {time.time() - t_save:.1f}s", flush=True) yield html def run(passphrase: str, audio_path: str, progress=gr.Progress()): """The named API route. Signature is (passphrase, audio) and stays that way — the deploy check drives the live Space through it. A generator, so the report can reach the screen before the clips are cut. A client that only wants the answer, `gradio_client.predict` included, gets the last value and never sees the difference. """ if not PASSPHRASE: yield _notice(NO_SECRET) return if not _ok(passphrase): yield _notice(WRONG_PASS) return if not audio_path: yield _notice("Choose an audio file to analyse — mp3, wav or m4a.") return # One analysis at a time on a box this size. The queue already serialises # the two buttons; this catches anything arriving through the API route # alongside a browser run. if not _run_lock.acquire(blocking=False): yield _notice("Another analysis is already running. Give it a " "moment, then press Analyse.") return try: yield from _analyse(audio_path, progress) except Exception as exc: traceback.print_exc() yield _notice("That file could not be read. Try another export of it — " "mp3, wav or m4a.") finally: _run_lock.release() def run_lookup(passphrase: str, artist: str, title: str, progress=gr.Progress()): """Find a track by name and analyse its official 30-second preview. For the case where the team has a name and no file. Audio is never taken from YouTube — see `lookup.py` for why — so this resolves the same kind of licensed preview every corpus sound is embedded from. """ if not PASSPHRASE: yield _notice(NO_SECRET) return if not _ok(passphrase): yield _notice(WRONG_PASS) return if not (title or "").strip(): yield _notice("Type a title to search for. An artist name as well " "makes the match much more reliable.") return if not _run_lock.acquire(blocking=False): yield _notice("Another analysis is already running. Give it a " "moment, then search again.") return try: progress(0.05, desc=f"looking for “{artist} — {title}”") try: match = lookupmod.search(artist, title) except lookupmod.NotFound as exc: yield _notice(render.ESC(str(exc))) return except lookupmod.Unavailable as exc: # Never reported as "your track is not there" — it is not the # same fact, and the reader would go and look for the file. The # cause goes to the log; "HTTP 503" on screen tells a label # reader nothing they can act on. print(f"[lookup] search unavailable: {exc}", flush=True) yield _notice( "The music search is not answering just now. Try again in a " "minute, or upload the file.") return card = savedview.match_card(match, preview_url=match.get("preview")) yield card # before any verdict progress(0.15, desc="fetching the 30-second preview") path = lookupmod.download(match["preview"]) label = lookupmod.label_for(match) yield from _analyse( path, progress, label=label, source="lookup", resolved={k: match.get(k) for k in ("artist", "title", "source", "source_id", "link")}, lead_html=card, snippet_note=savedview.PREVIEW_NO_SNIPPETS) except Exception as exc: traceback.print_exc() yield _notice("That search did not finish. Try it again, or upload " "the file.") finally: _run_lock.release() def run_batch(passphrase: str, paths, progress=gr.Progress()): """Several files, one after another, each saved as it finishes. A batch that dies on file nine leaves the first eight in Saved. Nothing is held back to the end. """ if not _ok(passphrase): yield _notice(NEED_PASS if PASSPHRASE else NO_SECRET) return files = [p for p in (paths or []) if p] if not files: yield _notice("Choose some audio files first.") return if not _run_lock.acquire(blocking=False): yield _notice("Another analysis is already running. Give it a " "moment, then press Run.") return try: emb = _embedder() done, failed = [], [] for n, path in enumerate(files, 1): path = getattr(path, "name", path) label = Path(path).stem progress((n - 1) / len(files), desc=f"{label} — {n} of {len(files)}") try: heard = analysis.listen(path, emb) res = analysis.score(heard, CORPUS, vocab=_state["vocab"], snippet_top=SNIPPETS_PER_SET) out_dir = _snip_dir() clip_files: list[str | None] = [] for p in res["snippets"]: try: # To the last bar line inside the window, exactly as # a single upload is cut. Cutting to the flat 30 # seconds here left the card labelled "PLAYS TO THE # BAR" over audio that did no such thing (23 Aug # audit). end = p.get("natural_end_s") or p["end_s"] clip_files.append(_clip(path, p["start_s"], end, str(out_dir))) except Exception: clip_files.append(None) # And the track itself, so a report opened out of a batch # can play the record its market cards are comparing. # `store.save` drops anything over its own size cap. aid = store.save(res, heard, label, clip_paths=clip_files, source_audio=path) row = store.meta(aid) or {} done.append({"label": label, "top_markets": row.get("top_markets") or []}) except Exception as exc: traceback.print_exc() failed.append({"label": label, "error": str(exc)}) # after every file, so a long batch shows its progress as results yield savedview.batch_summary(done, failed) if not done and not failed: yield _notice("Nothing was analysed.") except Exception as exc: traceback.print_exc() yield _notice("That run stopped. Anything finished before it stopped " "is in Saved.") finally: _run_lock.release() def auto_run(passphrase: str, audio_path: str, progress=gr.Progress()): """Upload starts the analysis on its own once the passphrase is good. With no passphrase yet, the upload leaves the screen alone: `_pending` has already put the "enter the passphrase" line there, and replacing it with "that passphrase is not right" would be wrong — nothing was typed. """ if not _ok(passphrase): yield gr.skip() return yield from run(passphrase, audio_path, progress) # -------------------------------------------------------------------------- # Saved — the archive, and the weekly re-read that is the point of it def _rows() -> list[dict]: """The index. Raises `store.StoreUnavailable` when it cannot be read. It used to catch everything and return an empty list, so an archive behind a bad minute at the hub was drawn as "Nothing saved yet. Every track you analyse is kept here automatically…" — a team whose forty analyses had vanished being told they had never saved one (23 Aug audit). `store.py` states the rule it was breaking: emptiness is never reported as failure, nor failure as emptiness. """ try: return store.index() except store.StoreUnavailable: raise except Exception as exc: # Anything else is still a failure to read, not an empty archive. raise store.StoreUnavailable(str(exc)) from exc def saved_refresh(passphrase: str): """Redraw the Saved tab. The list is a function of the archive, drawn by `@gr.render`, so a refresh is a new nonce rather than a second copy of the list HTML. The clock is the nonce: any new value differs from the last one, and no caller has to know what the previous value was. """ return "", time.time() # A reopened report comes out the same every time. It is drawn from the # stored analysis and the corpus this process has loaded, and nothing about # who is reading it enters the drawing — so what is kept below can be shared # by every reader without a passphrase ever being part of the key. The gate # is checked before any of it is reached, as it is on every other path. # # Drawing one costs a download of the stored result, a copy of the source # audio and up to six clip copies into the directory the player can reach, # and roughly 0.2 MB of HTML. The copies are most of the wait, and the Saved # tab redraws itself on every open and every close, so a team stepping # through a list paid that wait on each step (reported 23 Aug). # # Staleness is handled by the key rather than by trust: it carries the # reading the archive holds now — when it was saved, when it was last # re-read, how many times — so a re-score here, a re-score on the Monday # sweep, or a re-score by another process writing the same archive all # produce a different key and a fresh drawing. `_forget` is the belt to that # braces, called wherever this app rewrites a stored reading itself. _REPORTS: "OrderedDict[tuple, dict]" = OrderedDict() _REPORTS_MAX = 32 # ~0.2 MB of HTML each, measured _reports_lock = threading.Lock() def _report_key(aid: str, row: dict) -> tuple: return (aid, str(row.get("saved_at") or ""), str(row.get("rescored_at") or ""), str(row.get("runs") or ""), str(row.get("corpus_week") or "")) def _remembered(key: tuple) -> str | None: with _reports_lock: hit = _REPORTS.get(key) if hit is None: return None # The served copies are swept an hour after they are written, so a # drawing that has outlived its files is redrawn rather than handed # back with players that will not play. if not all(os.path.exists(p) for p in hit["files"]): _REPORTS.pop(key, None) return None _REPORTS.move_to_end(key) return hit["html"] def _remember(key: tuple, html: str, files: list) -> None: with _reports_lock: _REPORTS[key] = {"html": html, "files": [f for f in files if f]} _REPORTS.move_to_end(key) while len(_REPORTS) > _REPORTS_MAX: _REPORTS.popitem(last=False) def _forget(aid: str | None = None) -> None: """Drop what was drawn for one analysis, or for all of them.""" with _reports_lock: for k in [k for k in _REPORTS if aid is None or k[0] == aid]: _REPORTS.pop(k, None) def saved_open(passphrase: str, aid: str, row: dict | None = None): """Re-render a stored report. No model, no corpus arithmetic, no audio — the answer was computed once and written down. `row` is the index row where the caller already holds it, which is what the Saved tab hands in. The index carries everything drawn here, so opening a track off the list does not read its meta a second time. """ if not _ok(passphrase): return _notice(NEED_PASS if PASSPHRASE else NO_SECRET) if not aid: return _notice("Choose a saved analysis first.") row = row or store.meta(aid) if not row: return _notice("That analysis could not be read back.") key = _report_key(aid, row) already = _remembered(key) if already is not None: return already res = store.result(aid) if not res: return _notice("That analysis could not be read back.") # By window, not by position: a re-score re-orders the shortlist and # re-cuts nothing, so position is not identity (`store.clip_for`). files = _servable(store.clips_for(aid, res)) src = _servable([store.source_path(aid)])[0] html = (savedview.reopened_note(row, str(CORPUS.week)) + render.render(res, row.get("label") or aid, snippet_clips=[_clip_url(p) if p else None for p in files], track_audio_url=_clip_url(src) if src else None)) _remember(key, html, list(files) + [src]) return html def saved_rerun(passphrase: str, aid: str, progress=gr.Progress()): """This track against the corpus loaded right now. The stored vectors carry everything: the market ranking is a matrix multiply, and the snippet ranking recomputes because the hook and downbeat halves of it were never about the corpus in the first place. """ if not _ok(passphrase): return _notice(NEED_PASS if PASSPHRASE else NO_SECRET) if not aid: return _notice("Choose a saved analysis first.") progress(0.3, desc=f"re-reading against week {CORPUS.week}") t0 = time.time() try: res = store.rescore(aid, CORPUS, vocab=_state["vocab"]) except Exception as exc: traceback.print_exc() return _notice("That re-run did not finish. The saved report is " "unchanged — try it again.") if res is None: return _notice("That analysis has no saved reading to re-read. " "Analyse the track again to bring it back.") _forget(aid) # the stored reading it was drawn from is gone row = store.meta(aid) or {} print(f"[timing] re-score {row.get('label')} {time.time() - t0:.2f}s", flush=True) # Clips were cut for the windows the ORIGINAL run picked, and this # re-score can promote a window that was never cut. `clips_for` matches # them by window, so a card offers a player only where the file on it is # the 30 seconds the card is headed with — the old version checked that # the index existed and nothing more (23 Aug audit). stored = _servable(store.clips_for(aid, res)) clips = [_clip_url(stored[p["clip_i"]]) if p.get("clip_i") is not None and p["clip_i"] < len(stored) and stored[p["clip_i"]] else None for p in res["snippets"]] src = _servable([store.source_path(aid)])[0] return (savedview.reopened_note(row, str(CORPUS.week)) + render.render(res, row.get("label") or aid, snippet_clips=clips, track_audio_url=_clip_url(src) if src else None)) def saved_rescore_all(passphrase: str, progress=gr.Progress()): """The whole list against this week. This is the weekly mechanism. When Monday's corpus lands, the catalogue's standing against the new world is one press away, and the same function is callable from `marathon.cli` when the cron picks it up. """ if not _ok(passphrase): return _notice(NEED_PASS if PASSPHRASE else NO_SECRET), time.time() _forget() # every stored reading on the list is moving try: out = store.rescore_all(CORPUS, vocab=_state["vocab"], progress=progress) except Exception as exc: traceback.print_exc() return _notice("The re-score did not finish. Nothing on the list was " "changed — try it again."), time.time() return savedview.rescore_summary(out), time.time() def saved_delete(passphrase: str, aid: str): if not _ok(passphrase): return _notice(NEED_PASS if PASSPHRASE else NO_SECRET), aid, time.time() if not aid: return _notice("Nothing to delete."), "", time.time() label = (store.meta(aid) or {}).get("label", aid) try: store.delete(aid) _forget(aid) except Exception as exc: traceback.print_exc() return (_notice("That could not be deleted. It is still on the list."), aid, time.time()) # Back to the list: the report that was open is no longer an analysis. return (_notice(f"Deleted {render.ESC(str(label))}, with its clips."), "", time.time()) # -------------------------------------------------------------------------- # This week — the drop sheet, rendered natively def week_parts(passphrase: str) -> tuple[str, str]: """`(header, sheet)` — the tab in two blocks, with room between them. Two rather than one because "Analyse a track →" belongs in the header and only a Gradio button can switch a Gradio tab: an `` inside an HTML component has nowhere to point, and no script runs there to help. So the header is one component, the button is the next, and the sheet follows. `weekly.py` may not be present yet in a checkout mid-build, and a missing weekly export is a normal state on a fresh Space, so neither is allowed to take the tab down. """ if not _ok(passphrase): return "", _notice(NEED_PASS if PASSPHRASE else NO_SECRET) try: import weekly data = weekly.fetch() if not data: return "", _notice( "No drop sheet has been published for this week yet. The " "scan runs on Monday; the markets and the matcher are " "available now.") return weekly.head(data), weekly.body(data) except Exception as exc: print(f"[weekly] {exc}", flush=True) traceback.print_exc() return "", _notice("This week's sheet could not be loaded. The other " "tabs are working — try this one again in a minute.") def world_html(passphrase: str) -> str: """The world tab: the global trend picture, before any catalogue. Kalam has asked three times for the global view to come before anything catalogue-relevant, and this is it with a tab of its own. It reads the corpus, which is always loaded, plus this week's sheet if one has been published — so the tab is not empty on a week whose scan has not run, it is simply missing the cross-market records, which is the one part of it the database rather than the corpus knows. Gated and error-wrapped exactly like `week_parts`: a failure here shows a line in this tab and leaves the rest of the app standing. """ if not _ok(passphrase): return _notice(NEED_PASS if PASSPHRASE else NO_SECRET) try: data = None try: import weekly data = weekly.fetch() except Exception as exc: # The map and the sentences do not need the sheet. Losing it # costs one section, not the tab. print(f"[world] no weekly sheet: {exc}", flush=True) return worldview.render_block(CORPUS, data) except Exception as exc: print(f"[world] {exc}", flush=True) traceback.print_exc() return _notice("The world this week could not be drawn. The other " "tabs are working — try this one again in a minute.") def week_html(passphrase: str) -> str: """The whole tab as one block, for any caller that wants it in one.""" head, sheet = week_parts(passphrase) return head + sheet # -------------------------------------------------------------------------- # Share links — one address that opens where it says it opens # # "Look at this one" between two people on a label is a link, not a set of # directions. The app's own address plus `?tab=saved` opens on Saved, and # `?open=` opens that saved report itself. The passphrase still applies: # a link is a shortcut through the app, never around the gate. TAB_IDS = ("world", "week", "analyse", "saved", "how") # What somebody writing the link by hand is likely to type instead. TAB_ALIASES = {"analyze": "analyse", "howitworks": "how", "how-it-works": "how", "how_it_works": "how", "thisweek": "week", "this-week": "week", "theworld": "world", "the-world": "world", "map": "world"} # The shape `store.new_id` writes: a timestamp, a slug, six hex characters. # Nothing outside it is ever used to build a path into the archive. AID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,119}$") def link_target(params) -> tuple[str | None, str]: """`(tab, saved analysis id)` for the query on a share link. `?tab=saved` opens the Saved tab. `?open=` opens one saved report, which lives on the Saved tab, so it asks for that tab too. A tab name the app does not have reads as no request at all, and an id that no longer exists is dealt with where the list is drawn: the reader gets the list, which is where they were going anyway, rather than a line telling them off for following a link somebody sent them. """ try: params = dict(params or {}) except Exception: return None, "" aid = str(params.get("open") or "").strip() tab = str(params.get("tab") or "").strip().lower() if aid: return "saved", (aid if AID_RE.match(aid) else "") tab = TAB_ALIASES.get(tab, tab) return (tab if tab in TAB_IDS else None), "" # The drop sheet's palette, applied to Gradio's own chrome. Values are the # ones in marathon/static/signals.html; the `_dark` twins are its dark theme. PAPER, CARD, INSET, LINE = "#fbfbf9", "#ffffff", "#f3f2ef", "#e6e4df" INK, INK2, INK3 = "#141417", "#55534e", "#97948c" D_PAPER, D_CARD, D_INSET, D_LINE = "#161615", "#1e1e1c", "#262624", "#343330" D_INK, D_INK2, D_INK3 = "#f5f4f0", "#b8b5ac", "#7c7a72" THEME = gr.themes.Base().set( body_background_fill=PAPER, body_background_fill_dark=D_PAPER, body_text_color=INK, body_text_color_dark=D_INK, body_text_color_subdued=INK2, body_text_color_subdued_dark=D_INK2, background_fill_primary=PAPER, background_fill_primary_dark=D_PAPER, background_fill_secondary=INSET, background_fill_secondary_dark=D_INSET, border_color_primary=LINE, border_color_primary_dark=D_LINE, block_background_fill=CARD, block_background_fill_dark=D_CARD, block_border_color=LINE, block_border_color_dark=D_LINE, block_label_text_color=INK3, block_label_text_color_dark=D_INK3, block_title_text_color=INK2, block_title_text_color_dark=D_INK2, block_radius="16px", panel_background_fill=CARD, panel_background_fill_dark=D_CARD, input_background_fill=INSET, input_background_fill_dark=D_INSET, input_border_color=LINE, input_border_color_dark=D_LINE, input_placeholder_color=INK3, input_placeholder_color_dark=D_INK3, input_radius="10px", button_large_radius="12px", button_small_radius="10px", # the sheet's CTA: near-black on paper, paper on near-black button_primary_background_fill=INK, button_primary_background_fill_hover="#2c2c31", button_primary_background_fill_dark=D_INK, button_primary_background_fill_hover_dark="#e3e1da", button_primary_border_color=INK, button_primary_border_color_dark=D_INK, button_primary_text_color="#ffffff", button_primary_text_color_dark=INK, button_secondary_background_fill=CARD, button_secondary_background_fill_dark=D_CARD, button_secondary_border_color=LINE, button_secondary_border_color_dark=D_LINE, color_accent="#0e9f6e", color_accent_soft="#e3f5ec", color_accent_soft_dark="#123124", ) CSS = f""" footer {{ display: none !important; }} /* Six tabs on a phone: at Gradio's default sizing only two fit and the rest disappear behind the dots menu — including Saved, the tab a shared link is usually about (reported 23 Aug, on mobile). Compact sizing fits four on a normal phone; the dots stay as the way to the last two. */ @media (max-width: 640px) {{ .tabs .tab-container button {{ font-size: 12.5px !important; padding: 8px 7px !important; }} }} /* Gradio's default is a Google font; the sheet is on the system stack. `!important` because the theme stylesheet loads after custom CSS. */ :root, .dark {{ --font: -apple-system, BlinkMacSystemFont, "Segoe UI", system-ui, sans-serif !important; /* the progress line is drawn in --font-mono, which is a Google font by default — a sentence about what the app is doing should read in the same face as the rest of the sentences */ --font-mono: -apple-system, BlinkMacSystemFont, "Segoe UI", system-ui, sans-serif !important; }} /* `width: 100%` is load-bearing on a phone. Gradio lays the container out as a flex item, and with only a max-width it takes its max-content size and overflows the viewport the moment the audio widget has a file in it: 760px of page inside a 375px screen, the whole report sliding sideways. */ .gradio-container {{ max-width: 760px !important; width: 100% !important; margin: 0 auto !important; }} #intro {{ margin-bottom: 4px; }} #intro .kick {{ font-size: 11px; font-weight: 700; letter-spacing: .18em; text-transform: uppercase; color: {INK3}; }} #intro h1 {{ font-size: 30px; font-weight: 800; letter-spacing: -.02em; margin: 3px 0 8px; line-height: 1.1; }} #intro p {{ font-size: 13.5px; line-height: 1.5; color: {INK2}; margin: 0 0 3px; max-width: 62ch; }} #intro .meta {{ font-size: 11.5px; font-weight: 700; letter-spacing: .1em; text-transform: uppercase; color: {INK3}; margin-top: 8px; }} .dark #intro .kick, .dark #intro .meta {{ color: {D_INK3}; }} .dark #intro p {{ color: {D_INK2}; }} /* The upload widget has two states sharing one container: the drop zone before a file, the player after one. Every rule below is scoped to the state it belongs to. `#up .audio-container button {{ min-height: 104px }}` — unscoped — is what made the loaded player look broken: it hit the transport, the volume control and the download/share/clear icons too, so a 104px-tall toolbar floated over the corner, the play triangle sat in a field of dead space and the speed control showed as a bordered "1x" box. Keep the drop-zone rules on the drop button. Empty state. Gradio's stock copy is loose text nodes between spans, so `display: none` on the children leaves it on screen; `font-size: 0` on the wrapper is what actually blanks it, and our prompt comes back in `::after` at its own size. Styling around the widget, not inside it. */ #up .audio-container {{ background: transparent; border: 0; }} #up .audio-container > button[aria-dropeffect] {{ background: {INSET}; border: 1px dashed {LINE}; border-radius: 14px; min-height: 104px; padding: 26px 14px 18px; }} #up .audio-container > button[aria-dropeffect] .wrap {{ font-size: 0; min-height: 0; gap: 0; }} #up .audio-container > button[aria-dropeffect] .wrap > * {{ display: none; }} #up .audio-container > button[aria-dropeffect] .wrap::after {{ content: "Tap to choose an audio file — mp3, wav or m4a"; font-size: 14px; font-weight: 600; color: {INK2}; }} .dark #up .audio-container > button[aria-dropeffect] {{ background: {D_INSET}; border-color: {D_LINE}; }} .dark #up .audio-container > button[aria-dropeffect] .wrap::after {{ color: {D_INK2}; }} /* Loaded state: one inset card — waveform, the times, a centred transport. `buttons=[]` drops download and share (neither means anything for a file you just picked); `editable=False` drops the trim tools. Volume is hidden because the browser and the phone both already have one, and a lone speaker icon under a centred transport reads as a stray control. */ #up .component-wrapper {{ background: {INSET}; border-radius: 14px; padding: 10px 12px 8px; }} #up .waveform-container {{ min-height: 0 !important; }} #up .timestamps {{ padding: 2px 0 0 !important; }} #up .timestamps time {{ font-size: 11.5px; color: {INK3}; font-variant-numeric: tabular-nums; }} #up .controls {{ padding: 0 !important; margin: 0 !important; min-height: 0 !important; flex-wrap: nowrap !important; }} #up .controls button {{ min-height: 0 !important; height: 30px !important; padding: 0 6px !important; }} #up .controls .control-wrapper, #up .settings-wrapper {{ display: none !important; }} #up .icon-button-wrapper {{ background: transparent !important; box-shadow: none !important; }} #up .icon-button-wrapper button {{ min-height: 0 !important; height: 26px !important; padding: 0 5px !important; }} .dark #up .component-wrapper {{ background: {D_INSET}; }} .dark #up .timestamps time {{ color: {D_INK3}; }} #go {{ font-weight: 700; letter-spacing: .01em; }} /* The line under the gate: whether the passphrase is remembered here, and whether the model is still loading. Small, quiet, never a box. */ #hint {{ margin: -4px 0 0; }} #hint .hint {{ font-size: 12.5px; line-height: 1.45; color: {INK3}; margin: 0; }} .dark #hint .hint {{ color: {D_INK3}; }} #forget {{ font-size: 12px !important; font-weight: 600; min-height: 0 !important; padding: 5px 11px !important; width: auto !important; align-self: flex-start; }} /* The stage line. Gradio draws it absolutely over the top of the results block, which is fine while the block holds the "Analysing" notice and a collision once the report arrives underneath it — two sets of words in the same place. An opaque ground turns it into a status strip sitting on the report instead. */ #out .wrap.center {{ background: {PAPER} !important; padding: 6px 0 10px; border-radius: 12px; }} .dark #out .wrap.center {{ background: {D_PAPER} !important; }} #out .progress-level-inner {{ font-size: 12.5px !important; letter-spacing: 0; color: {INK2}; }} .dark #out .progress-level-inner {{ color: {D_INK2}; }} /* Saved cards are the control. The card itself is HTML, so the thing that can carry a click back to Python is a Button — it is stretched over the whole card at zero opacity, and the card underneath draws the hover and the chevron. A phone gets a 100%-wide, full-height tap target instead of a picker at the foot of the tab that nobody found. The fallback matters: if this positioning ever fails, the button is still there, visible, reading "Open full report". Nothing becomes unreachable. */ .savedcard {{ position: relative; margin-bottom: 14px; }} /* Gradio puts elem_classes on the