"""
Stage 5 - The GUI.
A Gradio Blocks interface over the four stages. It holds no detection logic: it uploads bytes,
calls pipeline -> detector -> analysis -> embedder, and renders what comes back.
**The interface comes up first, before anything is downloaded.** The user is asked for a Hugging
Face read token, and only then does the app fetch anything - one part at a time, each ticked off as
it finishes. The scanner appears when loading is done; the result tabs appear when a scan is done.
Nothing empty is ever on screen.
That ordering is deliberate. Gemma-2-9B in 4-bit is several gigabytes: loading it before `launch()`
means minutes of blank screen and, on a Space, a container the platform may consider dead. It also
turns a gated-repo refusal into a startup stack trace rather than a sentence in the UI.
Four models, MiMo-7B by default. MiMo is not the most accurate - Gemma-2-9B is, by 0.024 F1 - but
it is the only one of the four that anybody can download, so it is the one the app fetches at
start-up and the one a user with no Hugging Face account can still use. Gemma is gated by Google
and asks for a token; Qwen and Phi are open and load on request. Each entry in the picker states
what it costs you, including Phi-4-mini's, which is that it flags 95% of clean files.
"""
import os
import random
import sys
import threading
import time
from pathlib import Path
import gradio as gr
sys.path.insert(0, str(Path(__file__).resolve().parent))
import analysis
import detector
import embedder
import pipeline
# Bumped whenever this file changes in a way a running notebook would care about, so a stale
# import is visible in the bootstrap output instead of surfacing as a missing attribute later.
__version__ = "5.1-arrows-with-card"
DEFAULT_MODEL = detector.ACTIVE # "mimo" - ungated, so it can be fetched before any token
# Which model the loaded session is using. Written by connect(), read by analyse(), so a scan can
# never run on a different model from the one whose name is on the screen.
READY = {"ok": False, "model": DEFAULT_MODEL}
# ZeroGPU: the decorator only exists inside a Space. Everywhere else this is a no-op, so the same
# file runs in a notebook without a Spaces install.
ON_ZEROGPU = False
try:
import spaces
ON_ZEROGPU = True
except Exception:
class _NoSpaces:
@staticmethod
def GPU(*a, **kw):
def wrap(fn):
return fn
return wrap
spaces = _NoSpaces()
GPU_SLICE = 8 # windows per ZeroGPU allocation
GPU_DURATION = 120
# Per-window cost now lives on each ModelSpec: MiMo is 2.6x faster than Gemma, so one estimate
# for all four would be wrong by minutes on a long document.
MAX_NEIGHBOURS = 5 # Part A's precision@5 is the only figure measured, so 5 is the cap
CSS = """
/* Centre everything, at a width that keeps prose readable. */
.gradio-container {max-width: 1000px !important; margin: 0 auto !important;}
.gradio-container .prose {max-width: none !important;}
#hero {text-align: center; margin: 0.4rem auto 1rem auto;}
#hero h1 {margin-bottom: 0.3rem; font-size: 2rem;}
#hero p {opacity: 0.75; margin: 0;}
/* The start panel is a narrow card, centred. */
#setup {max-width: 560px; margin: 0 auto 0.5rem auto; padding: 1.1rem 1.3rem;
border-radius: 14px;}
#setup h3 {margin-top: 0;}
#steps {max-width: 560px; margin: 0.6rem auto; font-size: 0.94rem; line-height: 1.9;}
#steps p {margin: 0.15rem 0;}
/* The rotating note shown while something loads: a card that sits in front of the page, not a
tint behind it. The colours come from the theme's
own variables so it is opaque in both light and dark mode - the earlier rgba() tint let the
background show through and the text read as if it were behind everything. */
.waiting {max-width: 560px; margin: 1rem auto; padding: 1rem 1.15rem 1.1rem 1.15rem;
border-radius: 14px; position: relative; z-index: 5;
background: var(--block-background-fill, #ffffff);
border: 1px solid var(--border-color-primary, rgba(128,128,128,0.35));
box-shadow: 0 6px 20px rgba(0,0,0,0.22);}
.waiting p {margin: 0.35rem 0; font-size: 0.95rem; line-height: 1.6;}
/* The first line is the label - "While you're waiting" - and is set apart from the paragraph
under it by a rule, so the eye finds the family name rather than a wall of text. */
.waiting p:first-child {margin: 0; padding-bottom: 0.5rem; margin-bottom: 0.6rem;
border-bottom: 1px solid var(--border-color-primary,
rgba(128,128,128,0.25));
font-size: 0.82rem; letter-spacing: 0.08em; text-transform: uppercase;
opacity: 0.75;}
.waiting p:first-child strong {font-weight: 700;}
.waiting strong {font-weight: 700;}
/* The one progress bar in this app. A class, not an id, because both phases - loading and
scanning - draw their own, and Gradio's built-in progress is switched off everywhere
(show_progress="hidden") so that this is the only bar a user ever sees. Plain divs rather than
a component: it has to be redrawn from inside a generator, and markup is the only thing a
generator can repaint. */
.pbar {max-width: 560px; margin: 0.7rem auto 0.2rem auto;}
.pbar .track {height: 8px; border-radius: 999px; overflow: hidden;
background: var(--border-color-primary, rgba(128,128,128,0.25));}
.pbar .fill {height: 100%; border-radius: 999px; transition: width 0.35s ease;
background: linear-gradient(90deg, #f97316, #fb923c);}
.pbar .label {display: flex; justify-content: space-between; gap: 1rem; margin-top: 0.45rem;
font-size: 0.88rem; opacity: 0.8;}
.pbar .label .pct {font-variant-numeric: tabular-nums; opacity: 0.7;}
/* The arrows under a waiting card. Kept small, centred and quiet: they are an offer, not the
thing to look at, and the card auto-advances perfectly well if they are never touched. */
.wnav {max-width: 560px; margin: -0.4rem auto 0.8rem auto !important; justify-content: center;
gap: 0.5rem;}
.wnav button {flex: 0 0 auto !important; border-radius: 999px !important; font-size: 1.1rem;
line-height: 1; padding: 0.25rem 0.9rem !important; opacity: 0.7;}
.wnav button:hover {opacity: 1;}
/* Tabs as tabs: a resting row of shapes with the active one lifted out of the panel, rather than
three words with one underlined. The panel below joins the active tab, so the two read as one
surface. */
.gradio-container .tab-nav {border-bottom: 1px solid var(--border-color-primary,
rgba(128,128,128,0.3)) !important;
gap: 0.25rem; padding: 0 0.2rem;}
.gradio-container .tab-nav button {border: 1px solid transparent !important;
border-bottom: none !important;
border-radius: 10px 10px 0 0 !important;
padding: 0.55rem 1.05rem !important;
font-size: 0.95rem; font-weight: 500; opacity: 0.72;
margin-bottom: -1px;
transition: background 0.15s ease, opacity 0.15s ease;}
.gradio-container .tab-nav button:hover {opacity: 1;
background: var(--background-fill-secondary,
rgba(128,128,128,0.10)) !important;}
.gradio-container .tab-nav button.selected {opacity: 1; font-weight: 600;
background: var(--block-background-fill, rgba(128,128,128,0.16)) !important;
border-color: var(--border-color-primary, rgba(128,128,128,0.3)) !important;
box-shadow: inset 0 2px 0 0 #f97316;}
.gradio-container .tabitem {border: 1px solid var(--border-color-primary,
rgba(128,128,128,0.3)) !important;
border-top: none !important; border-radius: 0 0 12px 12px !important;
padding: 1rem 1.1rem !important;}
#verdict {text-align: center; margin: 0.8rem 0 0.2rem 0;}
#verdict h2 {margin: 0.2rem 0;}
#cost {font-size: 0.92rem; opacity: 0.85;}
/* Result tables: a quiet header row and hairline rules, so a five-column table reads as data
rather than as a grid of boxes. */
#report table th {background: var(--background-fill-secondary, rgba(128,128,128,0.10));
font-weight: 600; font-size: 0.9rem; letter-spacing: 0.02em;}
#report table, #report table td, #report table th {
border-color: var(--border-color-primary, rgba(128,128,128,0.25)) !important;}
#report h2 {font-size: 1.35rem; margin-top: 0.2rem;}
#report h3 {font-size: 1.1rem;}
#report hr {border: none; border-top: 1px solid var(--border-color-primary,
rgba(128,128,128,0.25)); margin: 1.4rem 0;}
/* Wrap everything: evidence strings and file ids are long and must not force a sideways scroll. */
#report, #report * {overflow-wrap: anywhere; word-break: normal;}
#report table {width: 100%; table-layout: fixed;}
#report td, #report th {white-space: normal !important; overflow-wrap: anywhere;
vertical-align: top; padding: 0.4rem 0.55rem;}
#report code {overflow-wrap: anywhere; white-space: pre-wrap;}
.gradio-container table td, .gradio-container table th {white-space: normal !important;
overflow-wrap: anywhere;}
"""
# Gradio 6 moved `css` from the Blocks constructor to launch(); 4 and 5 want it on Blocks. Colab
# and Spaces do not run the same version, so pass it wherever this install expects it.
GRADIO_MAJOR = int(gr.__version__.split(".")[0])
_BLOCKS_KW = {} if GRADIO_MAJOR >= 6 else {"css": CSS}
_LAUNCH_KW = {"css": CSS} if GRADIO_MAJOR >= 6 else {}
# --------------------------------------------------------------------------------------------
# Start-up: one part at a time
# --------------------------------------------------------------------------------------------
def _steps(model: str) -> list:
"""
The checklist for one model. The first step exists only for a gated one.
Showing "Check the token" while MiMo loads would be a lie about what the app is doing, and
would suggest a token is needed when the whole point of the default is that none is.
"""
spec = detector.DETECTORS[model]
token_step = ["Check the token"] if spec.gated else []
return token_step + ["Download the embedding index",
f"Download {spec.label.split(' - ')[0]}",
"Quantise to 4-bit"]
def _checklist(model: str, done: int, current: str = "", failed: str = "") -> str:
"""The step list as it stands: finished ticked, current running, the rest waiting."""
lines = []
for i, name in enumerate(_steps(model)):
if failed and i == done:
lines.append(f"✗ **{name}** - {failed}")
elif i < done:
lines.append(f"✓ {name}")
elif i == done and current:
lines.append(f"● **{name}** - {current}")
else:
lines.append(f"○ {name}")
return "\n\n".join(lines)
FACT_SECONDS = 20.0 # how long one family stays up when nobody touches the arrows
FACT_POLL = 0.4 # how often a waiting loop re-reads the rotation
def new_rotation():
"""
A fresh reading order, shuffled, starting at the first entry.
Re-made at the start of every load and every scan, so two runs in a row do not open on the
same family. A shuffled list rather than an independent random draw each time: drawing
independently repeats families and can leave some never shown at all.
"""
order = list(detector.FAMILY_NOTES)
random.shuffle(order)
return {"order": order, "i": 0, "touched": time.monotonic()}
def _ensure(rot):
"""Rotations live in gr.State, which starts as whatever the first caller finds."""
if not rot:
rot.update(new_rotation())
return rot
def rotation_note(rot, advance: bool = True) -> str:
"""
The note to show now, advancing only if nobody has pressed an arrow for FACT_SECONDS.
The clock is reset by the arrow handlers, so pressing one buys another full interval to read
on - an auto-advance two seconds after a deliberate click would be the app arguing with the
person using it.
"""
_ensure(rot)
if advance and time.monotonic() - rot["touched"] >= FACT_SECONDS:
rot["i"] = (rot["i"] + 1) % len(rot["order"])
rot["touched"] = time.monotonic()
return _waiting_note(rot)
def step_rotation(rot, delta: int):
"""An arrow press: move by hand, and restart the idle clock."""
_ensure(rot)
rot["i"] = (rot["i"] + delta) % len(rot["order"])
rot["touched"] = time.monotonic()
return _panel(_waiting_note(rot))
def _family_title(family: str) -> str:
"""
The family name as a person would write it.
`.title()` on the registry key gives "Ssrf" and "Llm Prompt Injection" - acronyms are the one
thing it cannot do, and these keys are mostly acronyms, so the exceptions are spelled out.
"""
special = {
"ssrf": "SSRF - server-side request forgery",
"llm_prompt_injection": "LLM prompt injection",
"xfa_acroform_injection": "XFA / AcroForm injection",
"dde_template_injection": "DDE template injection",
"uri_redirect_phishing": "URI redirect phishing",
"cross_site_scripting": "Cross-site scripting (XSS)",
}
return special.get(family, family.replace("_", " ").capitalize())
def _waiting_note(rot) -> str:
"""
One family, as three paragraphs: label, name, explanation.
Split into paragraphs rather than one block because the CSS styles the first one as the card's
heading; a single paragraph would give a wall of text with a bold phrase buried in it. The
counter tells the reader the arrows lead somewhere and how far round they are.
"""
family = rot["order"][rot["i"]]
counter = f"{rot['i'] + 1} of {len(rot['order'])}"
return (f"**While you're waiting** · {counter}\n\n"
f"**{_family_title(family)}**\n\n"
f"{detector.FAMILY_NOTES[family]}")
def _nav(text: str):
"""
The arrow row that belongs to a waiting card: on screen exactly when the card is.
Arrows with nothing to page through are a control that does nothing, which is worse than no
control at all - so they are tied to the same string the card is.
"""
return gr.update(visible=bool(text))
def _panel(text: str):
"""
A card that is on screen only when it has something to say.
An empty Markdown still draws its padding and border, so without this the loading screen
carries an empty box between every note.
"""
return gr.update(value=text, visible=bool(text))
def _while_working(work, render, rot):
"""
Run `work()` on a thread and keep the screen talking while it runs.
A generator delegated to with `yield from`: it yields whatever `render(note)` builds - a full
set of UI updates - and returns whatever `work()` returned. Downloading six gigabytes is one
blocking call that cannot report from inside itself, so the only way to say anything during it
is to put it on another thread and watch.
It re-reads the rotation every FACT_POLL and yields only when the text has actually changed,
which covers both ways the note can move: the idle timer running out, and the reader pressing
an arrow - the arrow handler mutates the same rotation object this loop is reading.
An exception raised on the thread is re-raised here, so the caller's existing `except` still
sees a failed download as a failed download.
"""
box = {}
def runner():
try:
box["value"] = work()
except BaseException as e: # re-raised below, on the generator's own stack
box["error"] = e
thread = threading.Thread(target=runner, daemon=True)
thread.start()
shown = rotation_note(rot, advance=False)
yield render(shown)
while thread.is_alive():
current = rotation_note(rot)
if current != shown:
shown = current
yield render(shown)
time.sleep(FACT_POLL)
thread.join()
if "error" in box:
raise box["error"]
return box.get("value")
def preflight(model: str) -> tuple:
"""
Is the token real, and does this account have this model? Returns (status, message).
Three answers, not two, because a user's next move differs in each: `ok`, `no_access` (the
token is fine, the licence has not been accepted) and `bad_token` (the string is not a token).
Collapsing the middle case into "it failed" is the commonest mistake here - a perfectly good
token on an unaccepted licence then reads as a typo, and the user retypes it forever.
Two wrong ways to ask, both of which pass when they should not:
* `model_info` - Gemma is `gated=manual`, which gates the *weights* and leaves the model card
and metadata public, so this succeeds for anybody, garbage token included.
* `hf_hub_download("config.json")` - answers from the local cache once the file has been
fetched before, so it keeps saying yes after access is revoked.
`auth_check` asks the one question that matters and never answers from cache.
"""
spec = detector.DETECTORS[model]
if not spec.gated:
return "ok", "no token needed"
token = detector.hf_token()
if not token:
return "bad_token", "no token given, and none found in this runtime"
# Is the token itself real? Asked separately so a refusal can be attributed to the right
# thing: whoami fails on a bad string, auth_check fails on a good string without the licence.
try:
from huggingface_hub import HfApi
HfApi().whoami(token=token)
except Exception as e:
return "bad_token", f"{type(e).__name__}: {e}"
try:
try:
from huggingface_hub import auth_check
auth_check(spec.repo, token=token)
except ImportError: # older hub: same question, asked by hand
from huggingface_hub import get_hf_file_metadata, hf_hub_url
get_hf_file_metadata(hf_hub_url(spec.repo, "config.json"), token=token)
return "ok", "accepted"
except Exception as e:
name = type(e).__name__
if "Gated" in name or "Repository" in name or "401" in str(e) or "403" in str(e):
return "no_access", detector.gate_url(model)
return "error", f"{name}: {e}"
def token_hint() -> str:
if detector.hf_token():
return ("A token was found in this runtime. Leave the box empty and press **Send token** "
"to use it.")
return ("Paste a **read** token from huggingface.co/settings/tokens. It is used only to "
"download the model, and is never stored.")
# --------------------------------------------------------------------------------------------
# The default model, fetched at start-up
# --------------------------------------------------------------------------------------------
_PREFETCH = {"thread": None}
def prefetch_default():
"""
Start pulling MiMo into the cache as the app boots, on a background thread.
On a thread rather than inline: the interface has to be answering requests within seconds or a
Space is considered dead, and a user who intends to pick Gemma anyway should not wait behind a
download they will not use. By the time anyone has read the page and pressed a button the
fetch is usually finished, and `connect()` then ticks the download step off immediately.
Only the default is prefetched, and only because it is ungated - there is nothing to prefetch
for Gemma until a token exists, and prefetching all four would pull twenty-odd gigabytes for
three models nobody asked for.
"""
if _PREFETCH["thread"] is not None:
return _PREFETCH["thread"]
def work():
try:
from huggingface_hub import snapshot_download
snapshot_download(detector.DETECTORS[DEFAULT_MODEL].repo,
allow_patterns=["*.json", "*.safetensors", "*.model", "tokenizer*"])
except Exception as e: # a cold cache at Start time, not a crash
print(f"[prefetch] {type(e).__name__}: {e}")
thread = threading.Thread(target=work, daemon=True)
thread.start()
_PREFETCH["thread"] = thread
return thread
def connect(model: str, token: str, rot):
"""
Everything between "the page loaded" and "you can scan a PDF", one part at a time.
A generator: each part reports before it starts and ticks off when it finishes, so the bar
advances four times rather than sitting still through a six-gigabyte download.
Yields (checklist, setup panel, scanner panel, waiting note, token verdict, progress bar).
No `gr.Progress`: Gradio draws its own bar over every output component of an event, which on
a five-output event meant the same percentage rendered several times down the page. One bar,
drawn by `_bar`, with Gradio's own switched off at the event (`show_progress="hidden"`).
"""
steps = _steps(model)
spec = detector.DETECTORS[model]
rot.update(new_rotation()) # a fresh reading order for every load
def state(done, current="", failed="", waiting="", verdict=""):
finished = done >= len(steps)
return (_checklist(model, done, current, failed),
gr.update(visible=not finished),
gr.update(visible=finished),
_panel(waiting), # each card is hidden when empty, so the
_panel(verdict), # loading screen never shows a blank box
"" if finished else _bar(done, len(steps), steps[done]),
_nav(waiting))
READY["ok"], READY["model"] = False, model
# 1 - the token, for a gated model only -------------------------------------------------
step = 0
if spec.gated:
token = (token or "").strip()
if token:
os.environ["HF_TOKEN"] = token # detector.hf_token() reads the environment first
yield state(0, "asking Hugging Face")
status, detail = preflight(model)
if status == "no_access":
yield state(0, failed="the licence has not been accepted",
verdict=(f"### 🔒 Access not granted\n\nYour token works, but "
f"this account has not been given {spec.label.split(' - ')[0]}. Accept "
f"the licence here, signed in as the same account, then press "
f"**Send token** again:\n\nURL: {detail}"))
return
if status == "bad_token":
yield state(0, failed="token refused",
verdict=("### ❌ The Hugging Face token was not approved\n\n"
"Check it at huggingface.co/settings/tokens and try again, or "
"pick a different model - MiMo, Qwen and Phi need no token."))
return
if status != "ok":
yield state(0, failed=detail,
verdict=f"### ⚠️ Could not reach Hugging Face\n\n`{detail}`")
return
step = 1
yield state(1, verdict=f"### ✅ Token approved, loading {spec.label.split(' - ')[0]}…")
# 2 - the embedding index ----------------------------------------------------------------
keep = f"### ✅ Token approved, loading {spec.label.split(' - ')[0]}…" if spec.gated else ""
yield state(step, "5 MB", verdict=keep)
try:
idx = yield from _while_working(
embedder.load_index, lambda note: state(step, "5 MB", waiting=note, verdict=keep), rot)
n = idx["matrix"].shape[0]
except Exception as e:
yield state(step, failed=f"{type(e).__name__}: {e}",
verdict=f"### ⚠️ Could not load the embedding index\n\n`{e}`")
return
# 3 - the weights ------------------------------------------------------------------------
# The long one. Several gigabytes on a cold cache is minutes of nothing happening, which is
# where the twelve families come in - the wait is spent explaining what the model is about to
# look for, rather than on a bar the user cannot read anything into.
#
# For the default model this is usually instant: prefetch_default() has been pulling it since
# the app booted, and snapshot_download returns from cache.
size_note = "about 6 GB, cached after the first run"
yield state(step + 1, size_note, verdict=keep)
def _fetch_weights():
from huggingface_hub import snapshot_download
return snapshot_download(spec.repo, token=detector.hf_token(),
allow_patterns=["*.json", "*.safetensors", "*.model",
"tokenizer*"])
try:
yield from _while_working(
_fetch_weights,
lambda note: state(step + 1, size_note, waiting=note, verdict=keep), rot)
except Exception as e:
yield state(step + 1, failed=f"{type(e).__name__}: {e}",
verdict=f"### ⚠️ Download failed\n\n`{e}`")
return
# 4 - onto the GPU -----------------------------------------------------------------------
if ON_ZEROGPU:
# No GPU exists outside a scan on ZeroGPU, so quantising has to wait for the first one.
yield state(step + 2, "deferred to the first scan (ZeroGPU has no device until then)",
verdict=keep)
time.sleep(0.4)
else:
yield state(step + 2, "loading onto the GPU", verdict=keep)
try:
yield from _while_working(
lambda: detector.load(model),
lambda note: state(step + 2, "loading onto the GPU", waiting=note, verdict=keep),
rot)
except Exception as e:
# detector.load() already explains the two ways this fails - no GPU, or one too
# small - so its message is shown as written rather than wrapped in another guess.
yield state(step + 2, failed=f"{type(e).__name__}",
verdict=f"### ⚠️ The model could not be loaded\n\n{e}")
return
READY["ok"] = True
yield state(len(steps),
verdict=f"### ✅ {spec.label.split(' - ')[0]} is loaded and ready.")
# --------------------------------------------------------------------------------------------
# The scan
# --------------------------------------------------------------------------------------------
@spaces.GPU(duration=GPU_DURATION)
def _scan_slice(windows):
"""One GPU allocation's worth of windows. The only function that touches the GPU."""
return detector.scan(windows, name=READY["model"])
def _bar(done: int, total: int, note: str = "") -> str:
"""
The progress bar, drawn from the section counts rather than from elapsed time.
A time-based bar has to guess, and guesses wrongly on the first run of a document twice the
size of the last one. Sections are the unit of work the scan actually consumes, and their
number is known before the first one starts, so the bar is a measurement.
"""
pct = 100.0 * done / max(total, 1)
label = note or f"Section {min(done + 1, total)} of {total}"
return (f""
f"
{label}"
f"{pct:.0f}%