"""
ASR Annotation — Team Progress Dashboard
========================================
A tiny FastAPI service deployed as a Hugging Face Space. Annotators using the
client-side labeling webapp click "Share Progress", which POSTs a *counts-only*
snapshot here. Each report is persisted as one file per annotator inside a
private HF Dataset repo (the Space filesystem is ephemeral; the Dataset is the
durable source of truth). The dashboard reads all reports and shows per-person
and aggregate stats.
No transcripts or audio ever reach this service — only tallies.
Environment / Space secrets
---------------------------
HF_TOKEN Write token for the account that owns the Dataset repo.
DATASET_REPO e.g. "shaanilmo/asr-annotation-progress".
REPORT_TOKEN Shared string the webapp must send in the X-Report-Token header.
"""
import json
import os
import re
import time
from datetime import datetime, timezone
from fastapi import FastAPI, Header, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse, JSONResponse
from huggingface_hub import HfApi, hf_hub_download, list_repo_files
from pydantic import BaseModel
# ----------------------------------------------------------------------------
# Config
# ----------------------------------------------------------------------------
HF_TOKEN = os.environ.get("HF_TOKEN")
DATASET_REPO = os.environ.get("DATASET_REPO", "shaanilmo/asr-annotation-progress")
REPORT_TOKEN = os.environ.get("REPORT_TOKEN", "")
REPORTS_DIR = "reports"
STATUS_KEYS = ["pending", "delete", "correct", "to_correct", "edited"]
api = HfApi(token=HF_TOKEN)
app = FastAPI(title="ASR Annotation Progress")
# Annotators open the labeling page from file:// or some static host, so allow any
# origin. Only POST/GET are used and the payload carries no secrets beyond the
# shared report token, which is validated explicitly below.
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
@app.on_event("startup")
def ensure_dataset_repo():
"""Create the private Dataset repo on first boot if it doesn't exist yet."""
if not HF_TOKEN:
# Let the app still boot so the dashboard can render a helpful message.
print("WARNING: HF_TOKEN not set — reports cannot be persisted.")
return
try:
api.create_repo(
repo_id=DATASET_REPO,
repo_type="dataset",
private=True,
exist_ok=True,
)
except Exception as exc: # pragma: no cover - best effort
print(f"WARNING: could not ensure dataset repo {DATASET_REPO}: {exc}")
# ----------------------------------------------------------------------------
# Models
# ----------------------------------------------------------------------------
class Counts(BaseModel):
pending: int = 0
delete: int = 0
correct: int = 0
to_correct: int = 0
edited: int = 0
class Report(BaseModel):
annotator: str
team: str = ""
intern: bool = False
recoveryUsed: bool = False
folder: str = ""
total: int = 0
counts: Counts
# Per-status audio duration in seconds (mirrors counts) for the time view.
# Optional/back-compatible: old reports omit these and render as no-time.
durations: dict = {}
durationTotal: float = 0
durationKnown: int = 0
updatedAt: int = 0
# Known teams; anything else is treated as Unassigned.
TEAMS = {"Research", "Investments", "Growth", "ConvAI", "HyperPersonalization"}
# ----------------------------------------------------------------------------
# Helpers
# ----------------------------------------------------------------------------
def slugify(name: str) -> str:
slug = re.sub(r"[^a-zA-Z0-9_-]+", "_", name.strip()).strip("_").lower()
return slug or "anonymous"
def _read_all_reports() -> list[dict]:
"""Download every reports/*.json from the Dataset and return parsed dicts."""
try:
files = list_repo_files(DATASET_REPO, repo_type="dataset", token=HF_TOKEN)
except Exception:
return []
reports = []
for path in files:
if not (path.startswith(REPORTS_DIR + "/") and path.endswith(".json")):
continue
try:
local = hf_hub_download(
DATASET_REPO,
filename=path,
repo_type="dataset",
token=HF_TOKEN,
)
with open(local, "r", encoding="utf-8") as fh:
reports.append(json.load(fh))
except Exception:
continue
reports.sort(key=lambda r: r.get("updatedAt", 0), reverse=True)
return reports
def _dedupe_per_folder(reports: list[dict]) -> list[dict]:
"""Collapse to the latest report per (annotator, folder). Handles re-shares and
any legacy-flat vs new-nested duplicate of the same folder (no double-counting)."""
latest: dict[tuple, dict] = {}
for r in reports:
key = (slugify(r.get("annotator", "")), (r.get("folder") or "").strip().lower())
cur = latest.get(key)
if cur is None or int(r.get("updatedAt", 0) or 0) >= int(cur.get("updatedAt", 0) or 0):
latest[key] = r
return list(latest.values())
def _group_by_annotator(reports: list[dict]) -> list[dict]:
"""Sum a person's deduped per-folder reports into one row + a folders breakdown."""
groups: dict[str, dict] = {}
for r in reports:
slug = slugify(r.get("annotator", ""))
counts = r.get("counts", {}) or {}
g = groups.get(slug)
if g is None:
g = groups[slug] = {
"annotator": r.get("annotator", "Anonymous"),
"slug": slug,
"team": r.get("team", "") or "",
"intern": bool(r.get("intern", False)),
"recoveryUsed": False,
"total": 0,
"counts": {k: 0 for k in STATUS_KEYS},
"durations": {k: 0.0 for k in STATUS_KEYS},
"durationTotal": 0.0,
"durationKnown": 0,
"updatedAt": 0,
"folders": [],
}
durs = r.get("durations", {}) or {}
g["total"] += int(r.get("total", 0) or 0)
for k in STATUS_KEYS:
g["counts"][k] += int(counts.get(k, 0) or 0)
g["durations"][k] += float(durs.get(k, 0) or 0)
g["durationTotal"] += float(r.get("durationTotal", 0) or 0)
g["durationKnown"] += int(r.get("durationKnown", 0) or 0)
g["recoveryUsed"] = g["recoveryUsed"] or bool(r.get("recoveryUsed", False))
if bool(r.get("intern", False)):
g["intern"] = True
ru = int(r.get("updatedAt", 0) or 0)
if ru >= g["updatedAt"]:
g["updatedAt"] = ru
if r.get("team"):
g["team"] = r.get("team") # latest non-empty team wins
g["folders"].append({
"folder": r.get("folder", "") or "",
"total": int(r.get("total", 0) or 0),
"counts": counts,
"updatedAt": ru,
})
for g in groups.values():
g["folders"].sort(key=lambda f: f.get("updatedAt", 0), reverse=True)
return list(groups.values())
def _aggregate(reports: list[dict]) -> dict:
totals = {k: 0 for k in STATUS_KEYS}
dur_totals = {k: 0.0 for k in STATUS_KEYS}
grand_total = 0
dur_grand_total = 0.0
dur_known = 0
for r in reports:
grand_total += int(r.get("total", 0) or 0)
counts = r.get("counts", {}) or {}
durs = r.get("durations", {}) or {}
for k in STATUS_KEYS:
totals[k] += int(counts.get(k, 0) or 0)
dur_totals[k] += float(durs.get(k, 0) or 0)
dur_grand_total += float(r.get("durationTotal", 0) or 0)
dur_known += int(r.get("durationKnown", 0) or 0)
# "Done" = files whose pipeline is actually finished. A 'to_correct' file is
# triaged but still needs editing, so it does NOT count toward completion
# (neither does 'pending'). The remaining backlog is pending + to_correct.
done = totals["delete"] + totals["correct"] + totals["edited"]
return {
"total": grand_total,
"counts": totals,
"durations": dur_totals,
"durationTotal": dur_grand_total,
"durationKnown": dur_known,
"done": done,
"to_fix": totals["to_correct"],
"percent": round(100 * done / grand_total) if grand_total else 0,
}
# ----------------------------------------------------------------------------
# Routes
# ----------------------------------------------------------------------------
@app.post("/report")
def submit_report(report: Report, x_report_token: str = Header(default="")):
if not REPORT_TOKEN or x_report_token != REPORT_TOKEN:
raise HTTPException(status_code=403, detail="Invalid or missing report token.")
if not HF_TOKEN:
raise HTTPException(status_code=500, detail="Server missing HF_TOKEN.")
slug = slugify(report.annotator)
# One file per (annotator, folder) so a person's folders accumulate on the
# dashboard instead of overwriting each other.
folder_slug = slugify(report.folder) if report.folder else "unknown"
payload = report.model_dump()
payload["annotator"] = report.annotator.strip() or "Anonymous"
payload["team"] = report.team if report.team in TEAMS else ""
payload["slug"] = slug
payload["receivedAt"] = int(time.time() * 1000)
data = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
try:
api.upload_file(
path_or_fileobj=data,
path_in_repo=f"{REPORTS_DIR}/{slug}/{folder_slug}.json",
repo_id=DATASET_REPO,
repo_type="dataset",
commit_message=f"progress: {payload['annotator']} · {report.folder or '?'} ({report.total} segs)",
)
except Exception as exc:
raise HTTPException(status_code=502, detail=f"Could not persist report: {exc}")
return {"ok": True, "annotator": payload["annotator"], "slug": slug}
@app.get("/api/stats")
def stats():
deduped = _dedupe_per_folder(_read_all_reports())
return JSONResponse(
{
"annotators": _group_by_annotator(deduped),
"aggregate": _aggregate(deduped),
"generatedAt": int(time.time() * 1000),
}
)
@app.get("/", response_class=HTMLResponse)
def dashboard():
return HTMLResponse(DASHBOARD_HTML)
# ----------------------------------------------------------------------------
# Dashboard (single self-contained page; fetches /api/stats and re-renders)
# ----------------------------------------------------------------------------
DASHBOARD_HTML = """
ASR Annotation — Team Progress
ASR Annotation — Team Progress
Live dashboard · auto-refreshes every 15s · 0 annotators reporting