"""Search over every environment and every indexed task, server side, from the catalog snapshot (app/snapshot.py). GET /api/search?q=&kind=&collection=&f=&sort=&page=&size= a page of cards, the total, the facet counts GET /api/search/tasks?q=&env=&page=&size= tasks across indexed datasets, best match first /api/search answers exactly as the Explore page always filtered its full listing in the browser: kind OpenEnv Spaces require fresh successful API checks. Metadata-only matches are unverified-space; other environment Spaces (ORS included), Harbor datasets (tagged, or an index found task folders), Verifiers, NeMo Gym, OpenEnv datasets, verl and SkyRL, other RL datasets owner exact Hub namespace, case-insensitive; applies to datasets and Spaces collection an admin's collection, "other" (in none) or "mine" (the signed-in visitor's datasets, with mine=1) f facet filters as the page's own URL writes them: `type:Benchmark|Neither;tags:code` (values URI-encoded); values of one facet are alternatives, facets all apply q every word must be part of the id, heading, brief or tags (substrings, any case) sort trending (Hub score, then likes, then monthly dataset downloads; no editorial boosts) | downloads | likes | tasks | rollouts | updated | new; ties keep the listing's order Each facet is counted over what every *other* filter leaves (so picking a value never hides its alternatives). Hidden environments, pins and collections are read from the admin settings on every query, so an admin's change shows at once, between indexer runs. The page's query string works as is: `k`, `c` and `s` are read as kind, collection and sort. Include it with `app.include_router(search_api.router)`; its startup starts the snapshot watcher. """ from __future__ import annotations import hashlib import json import re import threading import time from collections import OrderedDict from dataclasses import dataclass, field, replace from typing import Any from urllib.parse import unquote from fastapi import APIRouter, HTTPException, Request from fastapi.responses import Response from . import snapshot, space_checks router = APIRouter(on_startup=[snapshot.start]) KINDS = ("all", "harbor", "openenv", "unverified-space", "other-space", "verifiers", "nemo-gym", "openenv-data", "verl", "rows") SORTS = ("trending", "downloads", "likes", "tasks", "rollouts", "updated", "new") FACETS = ("type", "size", "health", "mode", "tools", "stage", "mcp", "oe", "oe_source", "evidence", "tags") RANKING = {"version": 2, "source": "Hugging Face Hub", "order": ["trending", "likes", "downloads"], "description": "Hub trending score first; ties use total likes, then dataset downloads over the last month. " "Spaces have no download metric. Remaining ties keep catalog order. Featured picks do not affect rank.", "scope": "Current search and filters", "quality": "Interest signals, not measured environment quality or live availability."} OLD_KINDS = {"dataset": "harbor", "space": "openenv", "ors": "other-space"} # the page's older links FACET_LIMIT = 100 # values returned per facet (most common first); the rest are counted in facets_more MAX_WORDS = 16 MAX_VALUES = 50 # ── the query ──────────────────────────────────────────────────────────────── @dataclass class Query: q: str = "" scope: str = "all" kind: str = "all" owner: str | None = None coll: str | None = None sel: dict[str, list[str]] = field(default_factory=dict) sort: str = "trending" page: int = 1 size: int = 40 @property def words(self) -> list[str]: return self.q.lower().split()[:MAX_WORDS] def public(self) -> dict[str, Any]: return {"q": self.q, "scope": self.scope, "owner": self.owner, "kind": self.kind, "collection": self.coll, "f": self.sel, "sort": self.sort, "page": self.page, "size": self.size} def parse_f(f: str) -> dict[str, list[str]]: """`key:v1|v2;key2:v` (each value URI-encoded, as the page writes it) as {key: [values]}; unknown keys dropped.""" sel: dict[str, list[str]] = {} for pair in (f or "").split(";"): k, _, v = pair.partition(":") if k not in FACETS or not v: continue vals = [unquote(x)[:200] for x in v.split("|") if x] if k == "stage" and "Asleep or stopped" in vals: # saved links from the former broad bucket vals = [x for x in vals if x != "Asleep or stopped"] + ["Sleeping", "Paused", "Building", "Starting", "Error", "Deleting"] if k == "mcp": vals = ["MCP tagged" if x == "Has MCP tools" else x for x in vals] for x in vals: if x not in sel.setdefault(k, []) and len(sel[k]) < MAX_VALUES: sel[k].append(x) return {k: v for k, v in sel.items() if v} def _int(v: Any, default: int, lo: int, hi: int) -> int: try: return max(lo, min(hi, int(v))) except (TypeError, ValueError): return default def parse_query(params: Any, collections: list[dict[str, Any]], mine: bool) -> Query: """The request's filters, normalised the way the page reads its own URL: an unknown kind, sort or collection is no filter rather than an error.""" g = lambda *names: next((params.get(n) for n in names if params.get(n) not in (None, "")), None) kind = g("kind", "k") or "all" kind = OLD_KINDS.get(kind, kind) coll = g("collection", "c") known = {c["id"] for c in collections} | {"other"} | ({"mine"} if mine else set()) sort = g("sort", "s") or "trending" return Query(q=(g("q") or "").strip()[:200], scope="ready" if g("scope") == "ready" else "all", owner=(g("owner") or "").strip()[:200] or None, kind=kind if kind in KINDS else "all", coll=coll if coll in known else None, sel=parse_f(g("f") or ""), sort=sort if sort in SORTS else "trending", page=_int(g("page"), 1, 1, 10_000), size=_int(g("size"), 40, 1, 100)) def envs_fts_query(words: list[str]) -> str | None: """The trigram FTS expression for search words: each one a quoted string (quotes doubled), so nothing a visitor types is read as FTS syntax (operators, NEAR, *, ^, column filters, parentheses). Words under three characters (and non-ASCII ones, whose case folding may differ) are left to the exact check alone.""" terms = ['"' + w.replace('"', '""') + '"' for w in words if len(w) >= 3 and w.isascii() and w.isprintable()] return " ".join(terms) or None _TOKEN = re.compile(r"[^\W_]+", re.UNICODE) def tasks_fts_query(q: str) -> str | None: """The FTS expression for a task search: the words' letters and digits only, each a quoted prefix ("word"*), all required. Quotes, operators, NEAR, column filters and stray asterisks can't get through.""" toks = [t[:40] for t in _TOKEN.findall((q or "").lower())][:8] return " ".join(f'"{t}"*' for t in toks) or None # ── what admins set, read per query ────────────────────────────────────────── _roll: dict[str, Any] = {"at": 0.0, "map": {}} _roll_lock = threading.Lock() def rollout_counts() -> dict[str, int]: """Graded public rollouts per dataset (main._shareable's rule), refreshed every 30 s.""" with _roll_lock: if time.time() - _roll["at"] < 30: return _roll["map"] from . import settings, store hidden = set(settings.get("hidden_runs", [])) counts: dict[str, int] = {} for r in store.list_runs(public=True, limit=100000): if r.get("status") == "done" and r.get("reward") is not None and not r.get("restricted") and r.get("id") not in hidden: k = str(r.get("dataset")) counts[k] = counts.get(k, 0) + 1 with _roll_lock: _roll.update(at=time.time(), map=counts) return counts @dataclass class Ctx: collections: list[dict[str, Any]] cmap: dict[str, str] pins: list[str] hidden: list[str] mine: list[str] rolls: dict[str, int] checks: dict[str, dict] = field(default_factory=dict) def params(self) -> dict[str, str]: """The admin settings as query parameters: each collection's keys, pins, hidden ones, the visitor's, rollouts. (Lists checked with IN, which SQLite evaluates once per statement: no per-row JSON.)""" out = {"pins": json.dumps(self.pins), "hidden": json.dumps(self.hidden), "mine": json.dumps(self.mine), "rolls": json.dumps(self.rolls), "checks": json.dumps([ {**r, "status": space_checks.status(r), "browseable": space_checks.browseable(r)} for r in self.checks.values()])} for i, g in enumerate(self.collections): out[f"c{i}"] = json.dumps([k for k, c in self.cmap.items() if c == g["id"]]) out[f"cid{i}"] = g["id"] return out def base(self) -> str: """The query's common table: every visible environment with what admins set applied (see BASE).""" whens = " ".join(f"WHEN e.key IN (SELECT value FROM json_each(:c{i})) THEN :cid{i}" for i in range(len(self.collections))) return BASE.replace("{coll}", f"CASE {whens} END" if whens else "NULL") def context(mine: list[str] | None = None) -> Ctx: from . import catalog colls = catalog.collections() cmap: dict[str, str] = {} for g in colls: for k in g.get("ids") or []: cmap.setdefault(k, g["id"]) # the first collection naming it, as catalog.collection_of return Ctx(collections=[{k: v for k, v in g.items() if k != "ids"} for g in colls], cmap=cmap, pins=list(catalog.pinned()), hidden=sorted(catalog.hidden()), mine=list(mine or []), rolls=rollout_counts(), checks=space_checks.inventory()) # ── SQL over the snapshot ──────────────────────────────────────────────────── BASE = """ WITH checks AS MATERIALIZED ( SELECT CAST(json_extract(value, '$.id') AS TEXT) AS id, value AS check_json FROM json_each(:checks) ), b0 AS ( SELECT e.rowid AS rid, e.ord, e.id, e.key, e.kind, e.framework, e.openenv, e.heading, e.brief, e.downloads, e.likes, e.trending, e.created, e.updated, e.created_ms, e.updated_ms, e.stage, e.mcp, e.openenv_version, e.manifest, e.hardware, e.badges, e.tags, e.tasks, e.indexed, e.size_f, e.stage_f, e.mcp_f, e.oe_f, e.blob, c.check_json, {coll} AS coll, (e.key IN (SELECT value FROM json_each(:pins))) AS pinned, (e.key IN (SELECT value FROM json_each(:mine))) AS mine FROM envs e LEFT JOIN checks c ON c.id = e.id AND e.kind = 'space' WHERE e.key NOT IN (SELECT value FROM json_each(:hidden)) ), base AS ( SELECT b0.*, CASE WHEN kind = 'space' THEN CASE WHEN {verified} THEN 'openenv' WHEN framework = 'ors' THEN 'ors' ELSE 'space' END ELSE framework END AS framework_q, CASE WHEN kind = 'space' AND {verified} THEN 1 ELSE 0 END AS openenv_q, CASE WHEN kind = 'space' AND NOT COALESCE(({verified}), 0) THEN (SELECT json_group_array(value) FROM json_each(badges) WHERE value != 'OpenEnv') ELSE badges END AS badges_q, CASE WHEN kind = 'space' THEN (CASE WHEN {verified} THEN 'openenv' WHEN framework IS 'openenv' OR openenv THEN 'unverified-space' ELSE 'other-space' END) ELSE COALESCE(framework, 'harbor') END AS fw, CASE WHEN kind = 'space' AND (framework IS 'openenv' OR openenv) THEN CASE WHEN manifest IS NOT NULL THEN 'Manifest present' ELSE 'Hub tag only' END END AS evidence_f, CASE WHEN kind = 'space' THEN {stage} END AS runtime_f, CASE WHEN kind = 'space' THEN CASE WHEN mcp THEN 'MCP tagged' ELSE 'No MCP tag' END END AS mcp_tag_f, CASE WHEN kind = 'space' AND (framework IS 'openenv' OR openenv OR ({verified})) THEN COALESCE(json_extract(check_json, '$.status'), 'Not checked') END AS health_f, CASE WHEN json_extract(check_json, '$.status') = 'API checked' THEN json_extract(check_json, '$.mode') END AS mode_f, CASE WHEN kind = 'space' AND (framework IS 'openenv' OR openenv OR ({verified})) THEN CASE WHEN json_extract(check_json, '$.status') IN ('API checked', 'Checks failed') THEN CASE WHEN json_extract(check_json, '$.tools') > 0 THEN 'Tools discovered' ELSE 'No tools discovered' END ELSE 'Not checked' END END AS tools_f, CASE WHEN kind = 'space' AND (framework IS 'openenv' OR openenv OR ({verified})) THEN COALESCE(NULLIF(json_extract(check_json, '$.version.value'), 'Unknown'), openenv_version, 'Unknown') END AS version_f, CASE WHEN kind = 'space' AND (framework IS 'openenv' OR openenv OR ({verified})) THEN CASE WHEN json_extract(check_json, '$.version.value') IS NOT NULL AND json_extract(check_json, '$.version.value') != 'Unknown' THEN json_extract(check_json, '$.version.source') WHEN openenv_version IS NOT NULL THEN 'Hub tag' ELSE 'Unknown' END END AS version_source_f FROM b0 ) """.replace("{stage}", snapshot.stage_sql("stage")).replace("{verified}", "json_extract(check_json, '$.status') = 'API checked' AND json_extract(check_json, '$.stage') = 'RUNNING'") ROLLOUTS = "COALESCE((SELECT value FROM json_each(:rolls) WHERE key = b.id), 0)" TYPE_VALUES = "json_each(CASE WHEN json_array_length(b.badges_q) = 0 THEN '[\"Neither\"]' ELSE b.badges_q END)" FACET_FILTER = { "type": f"EXISTS (SELECT 1 FROM {TYPE_VALUES} j WHERE j.value IN (SELECT value FROM json_each({{p}})))", "size": "b.size_f IN (SELECT value FROM json_each({p}))", "stage": "b.runtime_f IN (SELECT value FROM json_each({p}))", "health": "b.health_f IN (SELECT value FROM json_each({p}))", "mode": "b.mode_f IN (SELECT value FROM json_each({p}))", "tools": "b.tools_f IN (SELECT value FROM json_each({p}))", "mcp": "b.mcp_tag_f IN (SELECT value FROM json_each({p}))", "oe": "b.version_f IN (SELECT value FROM json_each({p}))", "oe_source": "b.version_source_f IN (SELECT value FROM json_each({p}))", "evidence": "b.evidence_f IN (SELECT value FROM json_each({p}))", "tags": "EXISTS (SELECT 1 FROM json_each(b.tags) j WHERE j.value IN (SELECT value FROM json_each({p})))", } FACET_COUNT = { # FROM, value expression, extra condition "type": (f"base b, {TYPE_VALUES} j", "j.value", "1"), "size": ("base b", "b.size_f", "b.size_f IS NOT NULL"), "stage": ("base b", "b.runtime_f", "b.runtime_f IS NOT NULL"), "health": ("base b", "b.health_f", "b.health_f IS NOT NULL"), "mode": ("base b", "b.mode_f", "b.mode_f IS NOT NULL"), "tools": ("base b", "b.tools_f", "b.tools_f IS NOT NULL"), "mcp": ("base b", "b.mcp_tag_f", "b.mcp_tag_f IS NOT NULL"), "oe": ("base b", "b.version_f", "b.version_f IS NOT NULL"), "oe_source": ("base b", "b.version_source_f", "b.version_source_f IS NOT NULL"), "evidence": ("base b", "b.evidence_f", "b.evidence_f IS NOT NULL"), "tags": ("base b, json_each(b.tags) j", "j.value", "1"), } TRENDING = "b.trending" SORT_SQL = {"trending": TRENDING, "downloads": "b.downloads", "likes": "b.likes", "tasks": "COALESCE(b.tasks, -1)", "rollouts": ROLLOUTS, "updated": "b.updated_ms", "new": "b.created_ms"} def sort_sql(q: Query) -> str: return SORT_SQL[q.sort] def order_sql(q: Query) -> str: return "sk DESC, " + ("b.likes DESC, b.downloads DESC, " if q.sort == "trending" else "") + "b.ord ASC" def order_key(q: Query, d: dict[str, Any]) -> tuple: return (-d["_sk"],) + ((-(d.get("likes") or 0), -(d.get("downloads") or 0)) if q.sort == "trending" else ()) + (d["_ord"],) def where(q: Query, skip: str | None = None, ignore_coll: bool = False, ignore_kind: bool = False) -> tuple[str, dict[str, Any]]: parts: list[str] = [] params: dict[str, Any] = {} if q.scope == "ready": parts.append("(b.kind = 'dataset' OR json_extract(b.check_json, '$.browseable') = 1)") if not ignore_kind and q.kind != "all": parts.append("b.fw = :kind") params["kind"] = q.kind if not ignore_coll and q.coll: if q.coll == "mine": parts.append("b.mine") else: parts.append("COALESCE(b.coll, 'other') = :coll") params["coll"] = q.coll if q.owner: parts.append("substr(b.id, 1, instr(b.id, '/') - 1) = :owner COLLATE NOCASE") params["owner"] = q.owner words = q.words if words: fts = envs_fts_query(words) if fts: # the index narrows it down; the exact check below decides parts.append("b.rid IN (SELECT rowid FROM envs_fts WHERE envs_fts MATCH :fts)") params["fts"] = fts for i, w in enumerate(words): parts.append(f"instr(b.blob, :w{i}) > 0") params[f"w{i}"] = w for k, vals in q.sel.items(): if k == skip or not vals or k not in FACET_FILTER: continue parts.append(FACET_FILTER[k].format(p=f":f_{k}")) params[f"f_{k}"] = json.dumps(vals) return (" AND ".join(parts) or "1"), params def card(r: Any, rolls: dict[str, int]) -> dict[str, Any]: """A row as a card: what the list, the trending grid and the quick look show, nothing more.""" return {"id": r["id"], "key": r["key"], "kind": r["kind"], "framework": r["framework_q"], "fw": r["fw"], "openenv": bool(r["openenv_q"]), "collection": r["coll"], "heading": r["heading"], "brief": r["brief"], "downloads": r["downloads"], "likes": r["likes"], "trending": r["trending"], "updated": r["updated"], "created": r["created"], "stage": r["stage"], "mcp": bool(r["mcp"]), "openenv_version": r["openenv_version"], "manifest": r["manifest"], "hardware": r["hardware"], "badges": json.loads(r["badges_q"]), "tags": json.loads(r["tags"]), "evidence": r["evidence_f"], "stage_label": r["runtime_f"], "api_status": r["health_f"], "api_check": json.loads(r["check_json"]) if r["check_json"] else None, "declared_version": r["version_f"], "version_source": r["version_source_f"], "api_mode": r["mode_f"], "tools_status": r["tools_f"], "pinned": bool(r["pinned"]), "mine": bool(r["mine"]), "indexed": json.loads(r["indexed"]) if r["indexed"] else None, "rollouts": rolls.get(r["id"], 0), "private": False, "_sk": r["sk"], "_ord": r["ord"]} def _rank(counts: dict[str, int]) -> list[list[Any]]: return [[v, n] for v, n in sorted(counts.items(), key=lambda kv: (-kv[1], str(kv[0])))] def sql_search(conn, q: Query, ctx: Ctx, *, facets: bool = True, limit: int | None = None, offset: int | None = None) -> dict[str, Any]: """The query over the snapshot: a page of cards (sorted, `_sk` and `_ord` kept for merging), the total and every facet's counts (raw: {value: n}).""" base, B = ctx.params(), ctx.base() w, p = where(q) lim = q.size if limit is None else limit off = (q.page - 1) * q.size if offset is None else offset rows = conn.execute(f"{B} SELECT b.*, {sort_sql(q)} AS sk FROM base b WHERE {w} ORDER BY {order_sql(q)} LIMIT :lim OFFSET :off", {**base, **p, "lim": lim, "off": off}).fetchall() total = conn.execute(f"{B} SELECT COUNT(*) FROM base b WHERE {w}", {**base, **p}).fetchone()[0] out: dict[str, Any] = {"rows": [card(r, ctx.rolls) for r in rows], "total": total} if not facets: return out fc: dict[str, dict[Any, int]] = {} w, p = where(q, ignore_kind=True) fc["kind"] = {k: n for k, n in conn.execute(f"{B} SELECT b.fw, COUNT(*) FROM base b WHERE {w} GROUP BY b.fw", {**base, **p})} w, p = where(q, ignore_coll=True) coll: dict[str, int] = {} mine = 0 for k, n, m in conn.execute(f"{B} SELECT COALESCE(b.coll, 'other'), COUNT(*), SUM(b.mine) FROM base b WHERE {w} GROUP BY 1", {**base, **p}): coll[k] = n mine += m or 0 if ctx.mine: coll["mine"] = mine fc["collection"] = coll for key in FACETS: frm, val, cond = FACET_COUNT[key] w, p = where(q, skip=key) fc[key] = {v: n for v, n in conn.execute(f"{B} SELECT {val}, COUNT(*) FROM {frm} WHERE {cond} AND {w} GROUP BY {val}", {**base, **p})} out["facets"] = fc return out def sql_extra(conn, ctx: Ctx, trending: int, q: Query | None = None) -> dict[str, Any]: """Counts, task coverage and editorial picks all follow the current search.""" base, B = ctx.params(), ctx.base() w, p = where(q or Query()) params = {**base, **p} ds, sp = conn.execute(f"{B} SELECT SUM(b.kind = 'dataset'), SUM(b.kind = 'space') FROM base b WHERE {w}", params).fetchone() # Count actual searchable records, including adapters (MiMo) with no Harbor # summary. Health discovery must never imply that tasks were downloaded. tasks = conn.execute(f"{B} SELECT COUNT(*) FROM tasks t JOIN base b ON b.key=t.env WHERE b.kind='dataset' AND {w}", params).fetchone()[0] from . import space_tasks visible = [r[0] for r in conn.execute(f"{B} SELECT b.id FROM base b WHERE b.kind='space' AND {w}", params)] space_counts = space_tasks.census(ctx.checks, visible) out: dict[str, Any] = {"stats": {"datasets": ds or 0, "spaces": sp or 0, "tasks": tasks, "dataset_tasks": tasks, "space_tasks": space_counts["tasks"], "space_task_coverage": space_counts}} out["stats"]["openenv"] = dict(conn.execute(f"{B} SELECT b.evidence_f, COUNT(*) FROM base b WHERE b.evidence_f IS NOT NULL AND {w} GROUP BY 1", params)) out["stats"]["openenv_running"] = conn.execute(f"{B} SELECT COUNT(*) FROM base b WHERE b.evidence_f IS NOT NULL AND b.stage = 'RUNNING' AND {w}", params).fetchone()[0] if trending: tq = replace(q or Query(), sort="trending", page=1, size=trending) out["trending"] = sql_search(conn, tq, ctx, facets=False)["rows"] w, p = where(tq) rows = conn.execute(f"{B} SELECT b.*, 0 AS sk FROM base b WHERE b.pinned AND {w} " "ORDER BY (SELECT CAST(key AS INTEGER) FROM json_each(:pins) WHERE value = b.key) LIMIT 12", {**base, **p}) out["featured"] = [card(r, ctx.rolls) for r in rows] # FineEnvs is an explicit editorial section, independent of admin pins # and interest rank. Owner/search filters still apply to it. promoted = conn.execute(f"{B} SELECT b.*, {TRENDING} AS sk FROM base b WHERE {w} " "AND substr(b.id, 1, instr(b.id, '/') - 1) = 'FineEnvs' COLLATE NOCASE " "AND (b.kind = 'dataset' OR json_extract(b.check_json, '$.browseable') = 1) " "ORDER BY (b.kind = 'space') DESC, b.pinned DESC, " + order_sql(tq) + " LIMIT 4", {**base, **p}) out["promoted"] = [card(r, ctx.rolls) for r in promoted] return out # ── the same, in Python: the visitor's own datasets that aren't in the snapshot, and the tests' reference ──────── def py_fw(d: dict[str, Any], check=None) -> str: if d.get("kind") == "space": if space_checks.verified(check): return "openenv" return "unverified-space" if d.get("framework") == "openenv" or d.get("openenv") else "other-space" return d.get("framework") or "harbor" PY_FACETS = { "type": lambda d: d.get("badges") or ["Neither"], "size": lambda d: [snapshot.size_bucket((d.get("indexed") or {}).get("tasks"))] if d.get("kind") == "dataset" and (d.get("framework") or "harbor") == "harbor" else [], "stage": lambda d: [snapshot.stage_value(d.get("stage"))] if d.get("kind") == "space" else [], "mcp": lambda d: (["MCP tagged" if d.get("mcp") else "No MCP tag"]) if d.get("kind") == "space" else [], "health": lambda d: [d["api_status"]] if d.get("api_status") else [], "mode": lambda d: [d["api_mode"]] if d.get("api_mode") else [], "tools": lambda d: [d["tools_status"]] if d.get("tools_status") else [], "oe": lambda d: [d["declared_version"]] if d.get("declared_version") else [], "oe_source": lambda d: [d["version_source"]] if d.get("version_source") else [], "evidence": lambda d: [d["evidence"]] if d.get("evidence") else [], "tags": lambda d: d.get("tags") or [], } def py_cards(rows: list[dict[str, Any]], ctx: Ctx, ord0: int = 0) -> list[dict[str, Any]]: """Apply visibility and editorial metadata without changing framework evidence or rank.""" hidden, pins, mine = set(ctx.hidden), set(ctx.pins), set(ctx.mine) out = [] for i, r in enumerate(rows): key = r.get("key") or r["id"] if key in hidden: continue coll = ctx.cmap.get(key) d = {**r, "key": key, "collection": coll, "pinned": key in pins, "mine": key in mine or bool(r.get("mine")), "rollouts": ctx.rolls.get(str(r["id"]), 0), "_ord": ord0 + i, "_blob": snapshot.blob_of(r), "badges": list(r.get("badges") or []), "tags": list(r.get("tags") or [])} check = ctx.checks.get(d["id"], {}) if d.get("kind") == "space" else {} declared = d.get("kind") == "space" and (d.get("framework") == "openenv" or d.get("openenv")) d["fw"] = py_fw(d, check) d["evidence"] = ("Manifest present" if d.get("manifest") else "Hub tag only") if declared else None d["stage_label"] = snapshot.stage_value(d.get("stage")) if d.get("kind") == "space" else None health = space_checks.status(check) candidate = declared or d["fw"] == "openenv" ver = check.get("version") or {} known_ver = ver.get("value") and ver["value"] != "Unknown" d.update(api_check={**check, "status": health} if check else None, api_status=health if candidate else None, api_mode=check.get("mode") if health == space_checks.PASS else None, tools_status=("Tools discovered" if check.get("tools", 0) > 0 else "No tools discovered") if candidate and health in (space_checks.PASS, "Checks failed") else "Not checked" if candidate else None, declared_version=(ver["value"] if known_ver else d.get("openenv_version") or "Unknown") if candidate else None, version_source=(ver.get("source") if known_ver else "Hub tag" if d.get("openenv_version") else "Unknown") if candidate else None) if d.get("kind") == "space": d["openenv"] = d["fw"] == "openenv" d["framework"] = "openenv" if d["openenv"] else "ors" if d.get("framework") == "ors" else "space" if not d["openenv"]: d["badges"] = [b for b in d["badges"] if b != "OpenEnv"] out.append(d) return out def _time(iso: Any) -> int: return snapshot.iso_ms(iso) def py_key(q: Query): keys = {"trending": lambda d: d.get("trending") or 0, "downloads": lambda d: d.get("downloads") or 0, "likes": lambda d: d.get("likes") or 0, "tasks": lambda d: ((d.get("indexed") or {}).get("tasks") if (d.get("indexed") or {}).get("tasks") is not None else -1), "rollouts": lambda d: d["rollouts"], "updated": lambda d: _time(d.get("updated")), "new": lambda d: _time(d.get("created"))} return keys[q.sort] def py_passes(d: dict[str, Any], q: Query, skip: str | None = None, ignore_coll: bool = False, ignore_kind: bool = False) -> bool: """The page's own `passes()`, line for line.""" if q.scope == "ready" and d.get("kind") == "space" and not space_checks.browseable(d.get("api_check")): return False if not ignore_kind and q.kind != "all" and d["fw"] != q.kind: return False if not ignore_coll and q.coll and not (d["mine"] if q.coll == "mine" else (d.get("collection") or "other") == q.coll): return False if q.owner and d["id"].split("/", 1)[0].lower() != q.owner.lower(): return False if q.words and not all(w in d["_blob"] for w in q.words): return False for k, vals in q.sel.items(): if k == skip or not vals or k not in PY_FACETS: continue if not any(v in vals for v in PY_FACETS[k](d)): return False return True def py_search(cards: list[dict[str, Any]], q: Query, *, with_mine: bool = False) -> dict[str, Any]: """`cards` (from py_cards) filtered, sorted and counted as the page did it: every match (sorted) and raw facets.""" key = py_key(q) matches = [d for d in cards if py_passes(d, q)] for d in matches: d["_sk"] = key(d) matches.sort(key=lambda d: order_key(q, d)) fc: dict[str, dict[Any, int]] = {"kind": {}, "collection": {}} for d in cards: if py_passes(d, q, ignore_kind=True): fc["kind"][d["fw"]] = fc["kind"].get(d["fw"], 0) + 1 if py_passes(d, q, ignore_coll=True): c = d.get("collection") or "other" fc["collection"][c] = fc["collection"].get(c, 0) + 1 if with_mine and d["mine"]: fc["collection"]["mine"] = fc["collection"].get("mine", 0) + 1 if with_mine: fc["collection"].setdefault("mine", 0) for k in FACETS: counts: dict[Any, int] = {} for d in cards: if py_passes(d, q, skip=k): for v in PY_FACETS[k](d): counts[v] = counts.get(v, 0) + 1 fc[k] = counts return {"matches": matches, "total": len(matches), "facets": fc} def _public_card(d: dict[str, Any]) -> dict[str, Any]: keep = ("id", "key", "kind", "framework", "fw", "openenv", "collection", "heading", "brief", "downloads", "likes", "trending", "updated", "created", "stage", "mcp", "openenv_version", "manifest", "hardware", "badges", "tags", "pinned", "mine", "indexed", "rollouts", "private", "evidence", "stage_label", "api_status", "api_check", "declared_version", "version_source", "api_mode", "tools_status") out = {k: d.get(k) for k in keep} out["openenv"], out["mcp"], out["private"] = bool(out["openenv"]), bool(out["mcp"]), bool(out["private"]) ix = d.get("indexed") out["indexed"] = {k: ix.get(k) for k in ("tasks", "graded", "image") if ix.get(k) is not None} if isinstance(ix, dict) else None return out def environments(include_hidden: bool = False) -> list[dict[str, Any]]: """Every public environment as catalog.environments() shapes it (collection, pins and hidden ones as admins set them now), from the snapshot, else the live catalog: for code that wants the whole listing (app/seo.py).""" ctx = context() if include_hidden: ctx.hidden = [] with snapshot.use() as (_, conn): rows = conn.execute(f"{ctx.base()} SELECT b.*, 0 AS sk FROM base b ORDER BY b.ord", ctx.params()).fetchall() out = [card(r, ctx.rolls) for r in rows] for d in out: d.pop("_sk", None) d.pop("_ord", None) d["gated"] = False return out # ── the endpoints ──────────────────────────────────────────────────────────── ANSWER_TTL, ANSWERS = 15.0, 256 # recent public answers kept in memory: the first page everyone opens is computed once _answers: OrderedDict[str, tuple[float, dict[str, Any]]] = OrderedDict() _answers_lock = threading.Lock() def _respond(request: Request, payload: dict[str, Any], cache: str, ms: float | None = None) -> Response: """Compact JSON with a weak ETag (a repeat comes back 304), the given Cache-Control, and how long the query took as Server-Timing (not in the body, so the ETag only changes when the answer does).""" body = json.dumps(payload, separators=(",", ":"), ensure_ascii=False, default=str).encode() etag = 'W/"' + hashlib.sha1(body).hexdigest()[:20] + '"' headers = {"Cache-Control": cache, "ETag": etag} if ms is not None: headers["Server-Timing"] = f"db;dur={ms:.1f}" if cache.startswith("private"): headers["Vary"] = "Cookie" if request.headers.get("if-none-match") == etag: return Response(status_code=304, headers=headers) return Response(body, media_type="application/json", headers=headers) MINE_FRESH, MINE_STALE = 120.0, 3600.0 _mine: OrderedDict[str, tuple[float, list[dict[str, Any]]]] = OrderedDict() # by token hash, never the token _mine_lock = threading.Lock() _mine_refreshing: set[str] = set() def _mine_rows(request: Request) -> list[dict[str, Any]] | None: """The signed-in visitor's own datasets (catalog.mine), or None when signed out. Kept per token: after two minutes the last list is still answered at once while a fresh one is fetched in the background, so a visitor's searches never wait on the Hub (catalog.mine is a few Hub calls).""" from . import auth, catalog u = auth.current_user(request) if not u or not u.get("token"): return None token = u["token"] who = hashlib.sha256(token.encode()).hexdigest()[:16] def fetch() -> list[dict[str, Any]]: try: rows = catalog.mine(token) except Exception: # noqa: BLE001 - the Hub is down: no "Yours" this time return [] with _mine_lock: _mine[who] = (time.time(), rows) _mine.move_to_end(who) while len(_mine) > 500: _mine.popitem(last=False) return rows with _mine_lock: hit = _mine.get(who) age = time.time() - hit[0] if hit else None if hit is None or age > MINE_STALE: return fetch() if age > MINE_FRESH and who not in _mine_refreshing: _mine_refreshing.add(who) def refresh() -> None: try: fetch() finally: _mine_refreshing.discard(who) threading.Thread(target=refresh, daemon=True, name="mine-refresh").start() return hit[1] @router.get("/api/search") def search(request: Request): """A page of environment cards, the total and the facet counts; the page's header numbers and trending row on request (`trending=`). `facets=0` leaves the counts out (paging), `mine=1` adds the signed-in visitor's own datasets (a private response).""" params = request.query_params want_mine = params.get("mine") in ("1", "true") mine_rows = _mine_rows(request) if want_mine else None ctx = context() q = parse_query(params, ctx.collections, mine=bool(mine_rows)) facets = params.get("facets", "1") not in ("0", "false") trending = _int(params.get("trending"), 0, 0, 48) t0 = time.time() cache = "private, max-age=30" if want_mine else "public, max-age=30, stale-while-revalidate=120" key = None if not want_mine: # the same question about the same snapshot and settings: answered from memory for a while try: key = json.dumps([snapshot.get().name, ctx.params(), q.public(), facets, trending], sort_keys=True) except snapshot.SnapshotError as e: raise HTTPException(503, f"the catalog isn't ready: {e}") hit = _answers.get(key) if hit and time.time() - hit[0] < ANSWER_TTL: return _respond(request, hit[1], cache, (time.time() - t0) * 1000) try: with snapshot.use() as (snap, conn): extras: list[dict[str, Any]] = [] if mine_rows: keys = [r.get("key") or r["id"] for r in mine_rows] ctx.mine = keys present = {k for (k,) in conn.execute("SELECT key FROM envs WHERE key IN (SELECT value FROM json_each(?))", (json.dumps(keys),))} extras = [r for r in mine_rows if (r.get("key") or r["id"]) not in present] if extras: # the visitor's datasets the snapshot doesn't have (private ones): merged in, same rules off = (q.page - 1) * q.size res = sql_search(conn, q, ctx, facets=facets, limit=off + q.size, offset=0) more = py_search(py_cards([{**r, "key": r.get("key") or r["id"], "kind": "dataset", "mine": True} for r in extras], ctx, ord0=10**9), q, with_mine=True) merged = sorted(res["rows"] + more["matches"], key=lambda d: order_key(q, d)) res["rows"], res["total"] = merged[off:off + q.size], res["total"] + more["total"] if facets: for k, counts in more["facets"].items(): for v, n in counts.items(): res["facets"][k][v] = res["facets"][k].get(v, 0) + n else: res = sql_search(conn, q, ctx, facets=facets) head = sql_extra(conn, ctx, trending, q) if trending or facets else {} if extras and head: head["stats"]["datasets"] += more["total"] private_tasks = sum((r.get("indexed") or {}).get("tasks") or 0 for r in more["matches"]) for metric in ("tasks", "dataset_tasks"): head["stats"][metric] += private_tasks if trending: tq = replace(q, sort="trending") more = py_search(py_cards([{**r, "key": r.get("key") or r["id"], "kind": "dataset", "mine": True} for r in extras], ctx, ord0=10**9), tq)["matches"] head["trending"] = sorted(head["trending"] + more, key=lambda d: order_key(tq, d))[:trending] info = {"built_at": snap.built_at, "source": snap.source} except snapshot.SnapshotError as e: raise HTTPException(503, f"the catalog isn't ready: {e}") out: dict[str, Any] = {"rows": [_public_card(d) for d in res["rows"]], "total": res["total"], "query": q.public()} if facets: fc = res["facets"] out["facets"] = {"kind": {**fc["kind"], "all": sum(fc["kind"].values())}, "collection": fc["collection"]} out["facets_more"] = {} for k in FACETS: ranked = _rank(fc[k]) out["facets"][k] = ranked[:FACET_LIMIT] if len(ranked) > FACET_LIMIT: out["facets_more"][k] = len(ranked) - FACET_LIMIT out["collections"] = ctx.collections out["mine"] = len(mine_rows or []) if head.get("stats"): out["stats"] = head["stats"] if trending: out["trending"] = [_public_card(d) for d in head.get("trending", [])] out["featured"] = [_public_card(d) for d in head.get("featured", [])] out["promoted"] = [_public_card(d) for d in head.get("promoted", [])] out["ranking"] = RANKING out["api_checks"] = {"fresh_seconds": space_checks.FRESH, "description": "Running on the Hub and OpenEnv API checks passed within the last hour. " "No episode or reward validation. Versions come from repository dependencies or Hub tags, not the installed runtime."} out["snapshot"] = info if key is not None: with _answers_lock: _answers[key] = (time.time(), out) while len(_answers) > ANSWERS: _answers.popitem(last=False) return _respond(request, out, cache, (time.time() - t0) * 1000) @router.get("/api/search/tasks") def search_tasks(request: Request, q: str = "", env: str = "", page: int = 1, size: int = 20): """Tasks across every indexed dataset, best match first: the dataset, the task's ref (its page is /t//), title, brief and facets. Every word must start a word of the title, brief, path, category or tags. At most 10,000 are counted.""" expr = tasks_fts_query(q[:200]) page, size = _int(page, 1, 1, 500), _int(size, 20, 1, 50) if env and not re.fullmatch(r"[A-Za-z0-9][\w.-]*/[\w.-]+", env): raise HTTPException(400, "env: a dataset id, like org/name") if not expr: return _respond(request, {"rows": [], "total": 0, "q": q, "snapshot": None}, "public, max-age=60") from . import catalog t0 = time.time() p = {"m": expr, "hidden": json.dumps(sorted(catalog.hidden())), "env": env, "lim": size, "off": (page - 1) * size} w = "tasks_fts MATCH :m AND t.env NOT IN (SELECT value FROM json_each(:hidden))" + (" AND t.env = :env" if env else "") try: with snapshot.use() as (snap, conn): rows = conn.execute("SELECT t.env, t.ref, t.title, t.brief, t.category, t.difficulty, t.grading, t.run " f"FROM tasks_fts JOIN tasks t ON t.rowid = tasks_fts.rowid WHERE {w} " "ORDER BY tasks_fts.rank LIMIT :lim OFFSET :off", p).fetchall() total = conn.execute(f"SELECT COUNT(*) FROM (SELECT 1 FROM tasks_fts JOIN tasks t ON t.rowid = tasks_fts.rowid WHERE {w} LIMIT 10001)", p).fetchone()[0] info = {"built_at": snap.built_at, "source": snap.source} except snapshot.SnapshotError as e: raise HTTPException(503, f"the catalog isn't ready: {e}") out = {"rows": [{"env": r["env"], "ref": r["ref"], "title": r["title"], "brief": r["brief"], "category": r["category"], "difficulty": r["difficulty"], "grading": r["grading"], "run": r["run"]} for r in rows], "total": min(total, 10000), "more": total > 10000, "q": q, "page": page, "size": size, "snapshot": info} return _respond(request, out, "public, max-age=60, stale-while-revalidate=300", (time.time() - t0) * 1000) @router.get("/api/search/rank") def search_rank(key: str, request: Request, scope: str = "all"): """Where one environment stands among its kind (Spaces, or datasets) by the page's trending order: {n, of}. One query, so a page can say "#599 of 6,818 Spaces" without loading the whole listing.""" ctx = context() w, params = where(Query(scope="ready" if scope == "ready" else "all")) sql = ctx.base() + f""", ranked AS ( SELECT key, ROW_NUMBER() OVER (PARTITION BY kind ORDER BY trending DESC, likes DESC, downloads DESC, ord) AS n, COUNT(*) OVER (PARTITION BY kind) AS total FROM base b WHERE {w}) SELECT n, total FROM ranked WHERE key = :key""" try: with snapshot.use() as (_snap, conn): row = conn.execute(sql, {**ctx.params(), **params, "key": key}).fetchone() except snapshot.SnapshotError as e: raise HTTPException(503, f"the catalog isn't ready: {e}") if not row: raise HTTPException(404, "not in the catalog") return _respond(request, {"n": row[0], "of": row[1]}, "public, max-age=60, stale-while-revalidate=300") @router.get("/api/search/status") def search_status(): """Which snapshot this process reads, and when it last looked for a new one.""" return snapshot.status()