checks.py

"""Deterministic, grounded evaluation of a Build Small hackathon Space.

Everything here is plain HF Hub API + README parsing — no LLM. The output is
a facts dict plus a checklist of (rule, status, evidence) rows that the UI
renders directly and the LLM stage uses as grounding.
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field, asdict

import requests

from guide import (
    ORG,
    TRACK_TAGS,
    SPONSOR_TAGS,
    ACHIEVEMENT_TAGS,
    MANAGED_PREFIXES,
)

API = "https://huggingface.co/api"
RAW = "https://huggingface.co/spaces/{space_id}/raw/main/{path}"
PARAM_CAP = 32_000_000_000
TINY_CAP = 4_000_000_000
TIMEOUT = 20

PASS, WARN, FAIL = "pass", "warn", "fail"
STATUS_ICON = {PASS: "✅", WARN: "⚠️", FAIL: "❌"}


@dataclass
class Check:
    rule: str
    status: str
    evidence: str


@dataclass
class Evaluation:
    space_id: str
    exists: bool = False
    error: str = ""
    checks: list[Check] = field(default_factory=list)
    facts: dict = field(default_factory=dict)
    verdict: str = ""

    def to_dict(self) -> dict:
        return asdict(self)


def normalize_space_name(value: str) -> str:
    """Collapse a pasted URL / owner-prefixed path to the bare space name.

    Mirrors `normalizeSpaceName` in the field guide's readme.ts.
    """
    s = value.strip()
    s = re.sub(r"^https?://huggingface\.co/", "", s, flags=re.I)
    s = s.lstrip("/")
    s = re.sub(r"^spaces/", "", s, flags=re.I)
    s = re.sub(rf"^{ORG}/", "", s, flags=re.I)
    return s.split("/")[0].strip()


def _get_json(url: str) -> dict | list | None:
    try:
        r = requests.get(url, timeout=TIMEOUT)
        if r.status_code != 200:
            return None
        return r.json()
    except requests.RequestException:
        return None


def _get_text(url: str) -> str | None:
    try:
        r = requests.get(url, timeout=TIMEOUT)
        if r.status_code != 200:
            return None
        return r.text
    except requests.RequestException:
        return None


def _fetch_commits(space_id: str, limit: int = 100) -> list[dict]:
    data = _get_json(f"{API}/spaces/{space_id}/commits/main?limit={limit}")
    return data if isinstance(data, list) else []


def _codex_commits(commits: list[dict]) -> list[str]:
    """Titles of commits with Codex attribution (title, message, or author)."""
    hits = []
    for c in commits:
        blob = " ".join([
            c.get("title", ""), c.get("message", ""),
            " ".join((a.get("user") or "") for a in c.get("authors", [])),
        ])
        if re.search(r"\bcodex\b", blob, re.I):
            hits.append(c.get("title", "")[:80])
    return hits


def list_org_spaces(org: str = ORG, limit: int = 300) -> list[str]:
    data = _get_json(f"{API}/spaces?author={org}&limit={limit}")
    if not isinstance(data, list):
        return []
    return sorted(s["id"].split("/", 1)[1] for s in data if "id" in s)


# ---- README parsing (same tolerant rules as the field guide's readme.ts) ----

FRONTMATTER_RE = re.compile(r"^?---[ \t]*\r?\n(.*?)\r?\n---[ \t]*\r?\n?", re.S)


def parse_readme(raw: str) -> tuple[dict, str]:
    """Return (frontmatter-ish dict with at least 'tags', body)."""
    m = FRONTMATTER_RE.match(raw)
    if not m:
        return {"tags": []}, raw
    fm_text, body = m.group(1), raw[m.end():]
    fm: dict = {"tags": _find_tags(fm_text)}
    try:
        import yaml

        loaded = yaml.safe_load(fm_text)
        if isinstance(loaded, dict):
            loaded.setdefault("tags", fm["tags"])
            fm = loaded
            if not isinstance(fm.get("tags"), list):
                fm["tags"] = _find_tags(fm_text)
    except Exception:
        pass
    fm["tags"] = [str(t) for t in (fm.get("tags") or [])]
    return fm, body


def _find_tags(fm_text: str) -> list[str]:
    unquote = lambda s: s.strip().strip("'\"")
    lines = fm_text.split("\n")
    for i, line in enumerate(lines):
        m = re.match(r"^tags:[ \t]*(.*)$", line)
        if not m:
            continue
        inline = m.group(1).strip()
        if inline.startswith("["):
            return [t for t in (unquote(x) for x in inline.strip("[]").split(",")) if t]
        if inline:
            return [unquote(inline)]
        tags = []
        for nxt in lines[i + 1:]:
            lm = re.match(r"^[ \t]+-[ \t]+(.*)$", nxt)
            if not lm:
                break
            tags.append(unquote(lm.group(1)))
        return tags
    return []


def split_managed(tags: list[str]) -> dict[str, list[str]]:
    out = {"track": [], "sponsor": [], "achievement": []}
    for t in tags:
        for prefix in MANAGED_PREFIXES:
            if t.startswith(prefix):
                out[prefix.rstrip(":")].append(t)
    return out


# ---- model discovery + parameter counts ----

# org/name pairs quoted in source code; permissive but filtered below
MODEL_ID_RE = re.compile(r"['\"]([A-Za-z0-9][\w.\-]*/[A-Za-z0-9][\w.\-]*)['\"]")
NON_MODEL_EXT = (".py", ".json", ".txt", ".md", ".csv", ".png", ".jpg", ".gguf",
                 ".bin", ".safetensors", ".yaml", ".yml", ".html", ".css", ".js",
                 ".wav", ".mp3", ".mp4", ".env")
SIZE_IN_NAME_RE = re.compile(r"(\d+(?:\.\d+)?)\s*[bB](?![A-Za-z0-9])")


def _candidate_model_ids(space: dict, code_blobs: list[str], readme_body: str) -> list[str]:
    ids: list[str] = list(space.get("models") or [])
    card = space.get("cardData") or {}
    ids += list(card.get("models") or [])
    for blob in code_blobs + [readme_body]:
        for m in MODEL_ID_RE.findall(blob or ""):
            ids.append(m)
    seen, out = set(), []
    for i in ids:
        low = i.lower()
        if i in seen or low.endswith(NON_MODEL_EXT) or low.startswith(("http", "./", "../")):
            continue
        seen.add(i)
        out.append(i)
    return out[:25]


def _model_params(model_id: str) -> tuple[int | None, str]:
    """Return (total params or None, how we know)."""
    data = _get_json(f"{API}/models/{model_id}")
    if not isinstance(data, dict) or "id" not in data:
        return None, "not found on the Hub"
    st = data.get("safetensors") or {}
    if st.get("total"):
        return int(st["total"]), "safetensors metadata"
    gguf = data.get("gguf") or {}
    if gguf.get("total"):
        return int(gguf["total"]), "gguf metadata"
    m = SIZE_IN_NAME_RE.search(model_id.split("/", 1)[1])
    if m:
        return int(float(m.group(1)) * 1e9), "size parsed from model name"
    return None, "no parameter metadata on the Hub"


# ---- link detection ----

VIDEO_PATTERNS = (
    r"youtube\.com/(?:watch|shorts|embed)", r"youtu\.be/", r"vimeo\.com/\d",
    r"loom\.com/share", r"\.mp4\b", r"\.webm\b", r"\.mov\b",
)
SOCIAL_PATTERNS = (
    r"(?<![\w.])x\.com/\w+/status", r"twitter\.com/\w+/status",
    r"linkedin\.com/(?:posts|feed/update)", r"bsky\.app/profile/.+/post",
    r"threads\.net/@", r"mastodon\.[\w.]+/@\w+/\d", r"reddit\.com/r/.+/comments",
)
VIDEO_FILE_EXT = (".mp4", ".webm", ".mov", ".mkv")
EXTERNAL_API_HOSTS = (
    "api.openai.com", "api.anthropic.com", "generativelanguage.googleapis",
    "api.mistral.ai", "openrouter.ai", "api.cohere", "api.together",
    "api.groq.com", "api.deepseek.com",
)


def _find_links(text: str, patterns: tuple[str, ...]) -> list[str]:
    found = []
    for pat in patterns:
        for m in re.finditer(rf"\S*{pat}\S*", text, flags=re.I):
            found.append(m.group(0).rstrip(").,]>\"'"))
    return list(dict.fromkeys(found))


# ---- the evaluation itself ----

def evaluate_space(name_or_url: str) -> Evaluation:
    name = normalize_space_name(name_or_url)
    space_id = f"{ORG}/{name}"
    ev = Evaluation(space_id=space_id)
    if not name:
        ev.error = "Empty space name."
        return ev

    space = _get_json(f"{API}/spaces/{space_id}")
    if not isinstance(space, dict) or "id" not in space:
        ev.error = (
            f"Space “{space_id}” was not found (or is private). "
            "Rule 2 requires the submission to live as a public Space in the "
            f"`{ORG}` org."
        )
        ev.checks.append(Check("Space in the Build Small org", FAIL, ev.error))
        ev.verdict = "NOT READY"
        return ev
    ev.exists = True

    sdk = space.get("sdk", "")
    siblings = [s.get("rfilename", "") for s in space.get("siblings", [])]
    readme_raw = _get_text(RAW.format(space_id=space_id, path="README.md")) or ""
    fm, body = parse_readme(readme_raw)
    tags = fm.get("tags", [])
    managed = split_managed(tags)

    # source files worth scanning for model ids / runtime hints
    code_files = [f for f in siblings
                  if f.endswith((".py", ".txt", ".toml", ".cfg", ".ts", ".js"))
                  and not f.startswith((".git", "node_modules"))][:12]
    code_blobs = []
    for f in code_files:
        blob = _get_text(RAW.format(space_id=space_id, path=f))
        if blob:
            code_blobs.append(blob[:200_000])
    all_code = "\n".join(code_blobs)

    # -- rule 2: gradio app in the org --
    if sdk == "gradio":
        ev.checks.append(Check("Ship a Gradio app (in the org)", PASS,
                               f"`{space_id}` is a Gradio Space (sdk: gradio)."))
    elif sdk == "docker":
        has_gradio = bool(re.search(r"\bimport gradio\b|\bfrom gradio\b|gradio", all_code, re.I))
        ev.checks.append(Check(
            "Ship a Gradio app (in the org)", PASS if has_gradio else WARN,
            "Docker Space — allowed as long as the interface is Gradio. "
            + ("Found Gradio usage in the source." if has_gradio
               else "Could not confirm a Gradio interface in the source; judges must see a Gradio app.")))
    else:
        ev.checks.append(Check("Ship a Gradio app (in the org)", FAIL,
                               f"Space sdk is `{sdk or 'unknown'}` — submissions must be Gradio apps."))

    # -- rule 6a: track tags --
    if managed["track"]:
        known = [t for t in managed["track"] if t in TRACK_TAGS]
        unknown = [t for t in managed["track"] if t not in TRACK_TAGS]
        if known:
            desc = "; ".join(f"`{t}` ({TRACK_TAGS[t].split('—')[0].strip()})" for t in known)
            status = PASS if not unknown else WARN
            extra = f" Unknown track tag(s): {', '.join(f'`{t}`' for t in unknown)} — valid ids are `track:backyard`, `track:wood`." if unknown else ""
            ev.checks.append(Check("Track tagged in README", status, f"Tagged {desc}.{extra}"))
        else:
            ev.checks.append(Check("Track tagged in README", FAIL,
                                   f"Track tag(s) {', '.join(f'`{t}`' for t in unknown)} are not valid — use `track:backyard` and/or `track:wood`."))
    else:
        ev.checks.append(Check("Track tagged in README", FAIL,
                               "No `track:` tag in the README frontmatter — add `track:backyard` and/or `track:wood` to the `tags:` list."))

    # invalid sponsor / achievement tags
    bad = ([t for t in managed["sponsor"] if t not in SPONSOR_TAGS]
           + [t for t in managed["achievement"] if t not in ACHIEVEMENT_TAGS])
    if bad:
        ev.checks.append(Check("Tag spelling", WARN,
                               f"Unrecognized tag(s): {', '.join(f'`{t}`' for t in bad)} — the judges' tooling only knows the canonical ids from the field guide."))

    # -- rule 3: demo video --
    video_links = _find_links(body, VIDEO_PATTERNS)
    video_files = [f for f in siblings if f.lower().endswith(VIDEO_FILE_EXT)]
    if video_links:
        ev.checks.append(Check("Demo video linked", PASS, f"Found in README: {video_links[0]}"))
    elif video_files:
        ev.checks.append(Check("Demo video linked", WARN,
                               f"Video file `{video_files[0]}` is in the repo but not linked from the README — link it so judges can find it."))
    else:
        ev.checks.append(Check("Demo video linked", FAIL,
                               "No video link (YouTube/Vimeo/Loom/.mp4) in the README and no video file in the Space."))

    # -- rule 4: social post --
    social_links = _find_links(body, SOCIAL_PATTERNS)
    if social_links:
        ev.checks.append(Check("Social post linked", PASS, f"Found in README: {social_links[0]}"))
    else:
        ev.checks.append(Check("Social post linked", FAIL,
                               "No social-media post link (X/LinkedIn/Bluesky/Threads/Mastodon/Reddit) found in the README."))

    # -- rule 6b: write-up --
    prose = re.sub(r"Check out the configuration reference.*", "", body, flags=re.I)
    prose = re.sub(r"https?://\S+|[#*`\->\[\]()|!]", " ", prose)
    words = len(prose.split())
    if words >= 80:
        ev.checks.append(Check("README write-up", PASS, f"README body has a write-up (~{words} words)."))
    elif words >= 25:
        ev.checks.append(Check("README write-up", WARN,
                               f"README body is thin (~{words} words) — the rules ask for the idea, how it was built, and the tech used."))
    else:
        ev.checks.append(Check("README write-up", FAIL,
                               "README body is essentially the default template — add a short write-up of the idea, the build, and the tech."))

    # -- rule 1: every model under 32B --
    model_ids = _candidate_model_ids(space, code_blobs, body)
    models_info: list[dict] = []
    for mid in model_ids:
        params, source = _model_params(mid)
        if params is None and source == "not found on the Hub":
            continue  # regex false positive, almost certainly not a model
        models_info.append({"id": mid, "params": params, "source": source})
    over = [m for m in models_info if m["params"] and m["params"] >= PARAM_CAP]
    unknown_size = [m for m in models_info if m["params"] is None]
    if over:
        ev.checks.append(Check("Every model under 32B", FAIL,
                               "; ".join(f"`{m['id']}` is {m['params']/1e9:.1f}B (>= 32B cap)" for m in over)))
    elif models_info:
        sized = [m for m in models_info if m["params"]]
        detail = ", ".join(f"`{m['id']}` ({m['params']/1e9:.1f}B)" for m in sized) or "none with a known size"
        status = PASS if not unknown_size else WARN
        extra = (" Could not size: " + ", ".join(f"`{m['id']}`" for m in unknown_size) + "."
                 if unknown_size else "")
        ev.checks.append(Check("Every model under 32B", status,
                               f"Detected models all under the cap: {detail}.{extra}"))
    else:
        ev.checks.append(Check("Every model under 32B", WARN,
                               "Could not detect any Hub model ids in the Space metadata, README, or source — judges may ask which models power the app; list them in the README."))

    # -- rule 5: zero gpu (info only — per-user count is not public) --
    hardware = ((space.get("runtime") or {}).get("hardware") or {}).get("current") or ""
    if str(hardware).startswith("zero"):
        ev.checks.append(Check("Zero GPU limit (max 10 apps/user)", WARN,
                               f"This Space runs on Zero GPU (`{hardware}`). The 10-apps-per-user cap can't be verified publicly — double-check your own count."))
    else:
        ev.checks.append(Check("Zero GPU limit (max 10 apps/user)", PASS,
                               f"Hardware is `{hardware or 'cpu-basic'}` — the Zero GPU cap is not a concern for this Space."))

    # -- commit history: authors + Codex attribution --
    commits = _fetch_commits(space_id)
    commit_authors = sorted({(a.get("user") or "").lower()
                             for c in commits for a in c.get("authors", [])} - {""})
    codex_hits = _codex_commits(commits)

    # -- sponsor-tag requirements vs evidence --
    code_and_models = (all_code + " " + " ".join(m["id"] for m in models_info)).lower()
    sponsor_evidence = {
        "sponsor:openbmb": ("minicpm" in code_and_models, "a MiniCPM model"),
        "sponsor:nvidia": ("nemotron" in code_and_models, "a Nemotron model"),
        "sponsor:modal": (bool(re.search(r"\bimport modal\b|modal\.com", all_code + " " + body, re.I)), "Modal usage noted"),
        "sponsor:openai": (bool(codex_hits), "Codex-attributed commits in the Space's history"),
    }
    for tag in managed["sponsor"]:
        if tag in sponsor_evidence:
            ok, what = sponsor_evidence[tag]
            if ok and tag == "sponsor:openai":
                ev.checks.append(Check(f"Sponsor requirement: `{tag}`", PASS,
                                       f"{len(codex_hits)} Codex-attributed commit(s) found — e.g. “{codex_hits[0]}”."))
            elif not ok:
                ev.checks.append(Check(f"Sponsor requirement: `{tag}`", WARN,
                                       f"Tagged for {SPONSOR_TAGS[tag].split('—')[0].strip()} but no evidence of {what} was found in the Space."))

    # -- achievement-tag claims vs evidence (verify every claimed tag; surface
    #    evidenced-but-unclaimed ones as opportunities) --
    external = sorted({h for h in EXTERNAL_API_HOSTS if h in all_code})
    uses_llamacpp = bool(re.search(r"llama[_\-]?cpp|\.gguf", code_and_models))
    org_or_author_models = [
        m["id"] for m in models_info
        if m["id"].split("/")[0].lower() in commit_authors + [ORG]
        or re.search(r"lora|sft|finetun|-ft\b", m["id"], re.I)
    ]
    hub_dataset_links = re.findall(r"huggingface\.co/datasets/[\w.\-]+/[\w.\-]+", body)
    writeup_links = _find_links(body, (
        r"huggingface\.co/blog", r"medium\.com/", r"dev\.to/",
        r"\w+\.substack\.com", r"hashnode\.\w+", r"\w+\.bearblog\.dev",
    ))
    extra_md = [f for f in siblings
                if f.lower().endswith(".md") and f.lower() != "readme.md"]
    custom_ui = bool(re.search(r"gr\.Server|gradio\.Server", all_code)) or any(
        f.lower().endswith((".html", ".css")) or f.startswith(("frontend/", "static/", "templates/"))
        for f in siblings)
    achievement_evidence = {
        "achievement:offgrid": (
            not external and bool(models_info),
            "no external API hosts in the source" if not external
            else f"the source references external API host(s): {', '.join(external)}"),
        "achievement:welltuned": (
            bool(org_or_author_models),
            f"fine-tune-looking / own model(s): {', '.join('`' + m + '`' for m in org_or_author_models[:3])}"
            if org_or_author_models else "no published fine-tune by the author was detected among the models"),
        "achievement:offbrand": (
            custom_ui,
            "custom frontend files / gr.Server usage found" if custom_ui
            else "no custom-UI evidence (gr.Server, .html/.css, frontend dir) found"),
        "achievement:llama": (
            uses_llamacpp,
            "llama.cpp / GGUF usage found in the source" if uses_llamacpp
            else "no llama.cpp or GGUF usage found"),
        "achievement:sharing": (
            bool(space.get("datasets") or hub_dataset_links),
            f"linked Hub dataset(s): {', '.join((space.get('datasets') or hub_dataset_links)[:3])}"
            if (space.get("datasets") or hub_dataset_links) else "no Hub dataset / trace link found"),
        "achievement:fieldnotes": (
            bool(writeup_links or extra_md),
            f"write-up found: {(writeup_links or extra_md)[0]}"
            if (writeup_links or extra_md) else "no blog/report link or extra write-up file found"),
    }
    for tag in managed["achievement"]:
        if tag in achievement_evidence:
            ok, why = achievement_evidence[tag]
            if not ok:
                ev.checks.append(Check(f"Achievement claim: `{tag}`", WARN,
                                       f"Tagged {ACHIEVEMENT_TAGS[tag].split('—')[0].strip()} but {why}."))

    # -- opportunities (not rules; surfaced to the LLM and the UI) --
    tiny = [m for m in models_info if m["params"] and m["params"] <= TINY_CAP]
    opportunities = []
    if tiny:
        opportunities.append(
            f"Tiny Titan bonus ($1.5k, judged): uses {', '.join('`' + m['id'] + '`' for m in tiny)} (<=4B) — make the small-model story loud in the README/demo.")
    achievement_pitch = {
        "achievement:offgrid": "if everything truly runs locally, tag it (Off the Grid)",
        "achievement:welltuned": "tag it if one of these is your own published fine-tune (Well-Tuned)",
        "achievement:offbrand": "tag it — custom UI past the stock Gradio look (Off-Brand)",
        "achievement:llama": "tag it (Llama Champion)",
        "achievement:sharing": "tag it if that's a shared trace (Sharing is Caring)",
        "achievement:fieldnotes": "tag it (Field Notes)",
    }
    for tag, (ok, why) in achievement_evidence.items():
        if ok and tag not in managed["achievement"]:
            opportunities.append(f"`{tag}`: {why} — {achievement_pitch[tag]}.")
    if "minicpm" in code_and_models and "sponsor:openbmb" not in managed["sponsor"]:
        opportunities.append("MiniCPM detected — tag `sponsor:openbmb` to enter the OpenBMB prize.")
    if "nemotron" in code_and_models and "sponsor:nvidia" not in managed["sponsor"]:
        opportunities.append("Nemotron detected — tag `sponsor:nvidia` to enter the NVIDIA prize.")
    if codex_hits and "sponsor:openai" not in managed["sponsor"]:
        opportunities.append(
            f"{len(codex_hits)} Codex-attributed commit(s) in the history — tag `sponsor:openai` to enter the OpenAI prize.")

    ev.facts = {
        "space_id": space_id,
        "sdk": sdk,
        "hardware": hardware,
        "likes": space.get("likes"),
        "last_modified": space.get("lastModified"),
        "tags": tags,
        "managed_tags": managed,
        "models_detected": models_info,
        "video_links": video_links,
        "video_files": video_files,
        "social_links": social_links,
        "external_api_hosts_in_code": external,
        "readme_words": words,
        "commit_authors": commit_authors,
        "codex_commits": codex_hits[:5],
        "opportunities": opportunities,
        "readme_body_excerpt": body[:4000],
    }

    statuses = [c.status for c in ev.checks]
    ev.verdict = ("NOT READY" if FAIL in statuses
                  else "ALMOST READY" if WARN in statuses
                  else "READY TO SUBMIT")
    return ev


def checklist_markdown(ev: Evaluation) -> str:
    rows = ["| | Rule | Evidence |", "|---|---|---|"]
    for c in ev.checks:
        rows.append(f"| {STATUS_ICON[c.status]} | **{c.rule}** | {c.evidence} |")
    return "\n".join(rows)