| """Search engines and link previews. Every page of the app has its own path, and the server answers each with the app |
| plus, in its head, the page's own title, description, canonical address, Open Graph and Twitter card, and schema.org |
| data (a Dataset for an environment, a SoftwareApplication for a Space, a task as part of its environment, breadcrumbs); |
| in its body, the page's content as plain HTML, for crawlers that don't run JavaScript (the app replaces it as it |
| starts). Plus robots.txt, a sitemap of every environment and every indexed task, and a preview image per page. |
| |
| Catalog pages use public indexes. Row and Space task pages may make bounded anonymous |
| reads of the same withheld task views shown in the UI. Crawlers never start indexing, |
| wake a Space, run an episode or receive a visitor's credentials. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import html |
| import io |
| import json |
| import re |
| import threading |
| import time |
| from functools import lru_cache |
| from typing import Any |
| from urllib.parse import quote, urlencode |
|
|
| from fastapi import Request |
| from fastapi.responses import HTMLResponse, PlainTextResponse, Response |
|
|
| from . import catalog, config, seo_tasks, snapshot, space_checks, spaces_live |
|
|
| SITE = "HF RL Explorer" |
| TAGLINE = "RL environments on the Hugging Face Hub" |
| DESCRIPTION = ("Explore reinforcement learning environments and tasks on the Hugging Face Hub. Browse OpenEnv, Harbor, " |
| "MiMo, NeMo Gym and Verifiers, inspect rewards, and run supported agent rollouts.") |
| KIND = {"harbor": "Harbor dataset", "verifiers": "Verifiers environment", "nemo-gym": "NeMo Gym dataset", "rows": "RL dataset", |
| "mimo": "MiMo RL release", "openenv": "OpenEnv Space", "space": "environment Space"} |
| MIMO = "XiaomiMiMo/MiMo-V2.6-RL-oss" |
| NOINDEX = re.compile(r"^/(run/|runs$|compare/)") |
| SITEMAP_CHUNK = 10_000 |
|
|
|
|
| |
| def base_url(request: Request) -> str: |
| """The site's own origin, for canonical addresses: the Space's public URL, else what the request came to.""" |
| if config.PUBLIC_URL: |
| return config.PUBLIC_URL |
| proto = request.headers.get("x-forwarded-proto") or request.url.scheme |
| host = request.headers.get("x-forwarded-host") or request.headers.get("host") or request.url.netloc |
| return f"{proto}://{host}" |
|
|
|
|
| def enc(spec: str) -> str: |
| return "/".join(quote(p, safe="") for p in spec.split("/")) |
|
|
|
|
| def esc(s: Any) -> str: |
| return html.escape(str(s if s is not None else ""), quote=True) |
|
|
|
|
| def clip(text: str, n: int) -> str: |
| text = re.sub(r"\s+", " ", re.sub(r"[#*_`>|\[\]]+", " ", str(text or ""))).strip() |
| return text if len(text) <= n else text[: n - 1].rsplit(" ", 1)[0].rstrip(",.;:") + "…" |
|
|
|
|
| @lru_cache(maxsize=4) |
| def _template(mtime: float) -> str: |
| return (config.WEB_DIR / "index.html").read_text() |
|
|
|
|
| def template() -> str: |
| p = config.WEB_DIR / "index.html" |
| return _template(p.stat().st_mtime) |
|
|
|
|
| |
| _listing: dict[str, Any] = {"at": 0.0, "by": {}, "rows": []} |
| _lock = threading.Lock() |
|
|
|
|
| def listing() -> tuple[dict[str, dict[str, Any]], list[dict[str, Any]]]: |
| """The catalog's listing of public environments, by key (`org/name`, `space:org/name`), refreshed every 10 min.""" |
| with _lock: |
| if time.time() - _listing["at"] < 600 and _listing["rows"]: |
| return _listing["by"], _listing["rows"] |
| try: |
| from . import search_api |
|
|
| rows = search_api.environments() |
| except Exception: |
| rows = [] |
| by = {r["key"]: r for r in rows} |
| with _lock: |
| _listing.update(at=time.time(), by=by, rows=rows) |
| return by, rows |
|
|
|
|
| def public_rows(rows): |
| hidden = set(catalog.hidden()) |
| return [r for r in rows if r["key"] not in hidden and not any(r.get(k) for k in ("private", "gated", "restricted"))] |
|
|
|
|
| def discoverable(r, records=None): |
| return r.get("kind") == "dataset" or space_checks.browseable((space_checks.inventory() if records is None else records).get(r["id"])) |
|
|
|
|
| def index_rows(spec: str) -> list[dict[str, Any]]: |
| """An environment's tasks, if they are already known here (never built for a crawler): [{path, title, brief, category}].""" |
| if spec == MIMO: |
| return _mimo_rows() |
| try: |
| p = catalog._index_path(spec) |
| mtime = p.stat().st_mtime |
| except (OSError, ValueError): |
| try: |
| return _snapshot_rows(spec, snapshot.get().name) |
| except snapshot.SnapshotError: |
| return [] |
| return _harbor_rows(spec, mtime) |
|
|
|
|
| @lru_cache(maxsize=1) |
| def _mimo_rows() -> list[dict[str, Any]]: |
| from .mimo import catalog as mcat |
|
|
| return [{"path": e["id"], "title": e["t"], "brief": e["s"], "category": e["d"]} for e in mcat.index()["envs"]] |
|
|
|
|
| @lru_cache(maxsize=32) |
| def _harbor_rows(spec: str, mtime: float) -> list[dict[str, Any]]: |
| idx = catalog._read_index(spec) |
| if not idx or (idx.get("info") or {}).get("restricted"): |
| return [] |
| return [{"path": t["path"], "title": t.get("title"), "brief": t.get("brief"), "category": t.get("category")} for t in (idx or {}).get("tasks") or []] |
|
|
|
|
| @lru_cache(maxsize=32) |
| def _snapshot_rows(spec: str, revision: str) -> list[dict[str, Any]]: |
| with snapshot.use() as (_, conn): |
| return [dict(r) for r in conn.execute("SELECT ref AS path, title, brief, category FROM tasks WHERE env = ? ORDER BY ref", (spec,))] |
|
|
|
|
| def display_name(row: dict[str, Any] | None, spec: str) -> str: |
| """What to call an environment: its card's heading when that names it, else the repository name with the heading.""" |
| repo = spec.split("/")[-1] |
| heading = (row or {}).get("heading") or "" |
| norm = lambda x: re.sub(r"[^a-z0-9]", "", x.lower()) |
| if not heading or norm(heading) == norm(repo): |
| return repo |
| return heading if norm(repo) in norm(heading) else f"{repo}: {heading}" |
|
|
|
|
| def kind_of(row: dict[str, Any] | None, spec: str = "") -> str: |
| if spec == MIMO: |
| return KIND["mimo"] |
| if not row: |
| return "RL environment" |
| if row.get("kind") == "space": |
| if row.get("fw") == "unverified-space": |
| return "Unverified Space" |
| return {"openenv": KIND["openenv"], "ors": "ORS Space"}.get(row.get("framework") or ("openenv" if row.get("openenv") else ""), KIND["space"]) |
| return KIND.get(row.get("framework") or "", "RL dataset") |
|
|
|
|
| |
| def page(request: Request, *, title: str, description: str, path: str, body: str = "", jsonld: list[dict[str, Any]] | None = None, |
| image: str | None = None, noindex: bool = False, og_type: str = "website", status: int = 200) -> HTMLResponse: |
| base = base_url(request) |
| url = base + path |
| full = title if SITE in title else f"{title} · {SITE}" |
| img = base + (image or "/social/rl-explorer.png") |
| head = "\n".join([ |
| f'<link rel="canonical" href="{esc(url)}">', |
| f'<meta name="rlx-page-url" content="{esc(url)}">', |
| '<meta name="robots" content="noindex, follow">' if noindex else '<meta name="robots" content="index, follow, max-image-preview:large, max-snippet:-1">', |
| f'<meta property="og:site_name" content="{SITE}">', |
| f'<meta property="og:type" content="{og_type}">', |
| f'<meta property="og:title" content="{esc(full)}">', |
| f'<meta property="og:description" content="{esc(description)}">', |
| f'<meta property="og:url" content="{esc(url)}">', |
| f'<meta property="og:image" content="{esc(img)}">', |
| '<meta property="og:image:type" content="image/png">', |
| '<meta property="og:locale" content="en_US">', |
| '<meta property="og:image:width" content="1200">', |
| '<meta property="og:image:height" content="630">', |
| f'<meta property="og:image:alt" content="{esc(full)}">', |
| '<meta name="twitter:card" content="summary_large_image">', |
| f'<meta name="twitter:title" content="{esc(full)}">', |
| f'<meta name="twitter:description" content="{esc(description)}">', |
| f'<meta name="twitter:image" content="{esc(img)}">', |
| f'<meta name="twitter:image:alt" content="{esc(full)}">', |
| *[f'<script type="application/ld+json">{_ld(x)}</script>' for x in jsonld or []], |
| ]) |
| doc = template() |
| doc = re.sub(r"<title>.*?</title>", f"<title>{esc(full)}</title>", doc, count=1, flags=re.S) |
| doc = re.sub(r'<meta name="description" content="[^"]*">', f'<meta name="description" content="{esc(description)}">', doc, count=1) |
| doc = doc.replace("</head>", f"{head}\n</head>", 1) |
| if body: |
| doc = re.sub(r'<main id="view">(.*?)</main>', lambda m: f'<main id="view">{m.group(1)}<div class="wrap page ssr">{body}</div></main>', doc, count=1, flags=re.S) |
| return HTMLResponse(doc, status_code=status, headers={"Cache-Control": "no-cache", **({"Retry-After": "60"} if status == 503 else {})}) |
|
|
|
|
| def _ld(x: dict[str, Any]) -> str: |
| |
| return json.dumps({"@context": "https://schema.org", **x}, ensure_ascii=False).replace("</", "<\\/") |
|
|
|
|
| def crumbs(base: str, items: list[tuple[str, str]]) -> tuple[str, dict[str, Any]]: |
| """Breadcrumbs as HTML and as schema.org BreadcrumbList: [(name, path)].""" |
| html_ = '<nav class="crumbs" aria-label="Breadcrumb">' + " › ".join( |
| f'<a href="{esc(p)}">{esc(n)}</a>' if p else f"<span>{esc(n)}</span>" for n, p in items) + "</nav>" |
| ld = {"@type": "BreadcrumbList", "itemListElement": [{"@type": "ListItem", "position": i + 1, "name": n, **({"item": base + p} if p else {})} |
| for i, (n, p) in enumerate(items)]} |
| return html_, ld |
|
|
|
|
| |
| def home(request: Request) -> HTMLResponse: |
| base = base_url(request) |
| by, rows = listing() |
| checks = space_checks.inventory() |
| rows = [r for r in public_rows(rows) if discoverable(r, checks)] |
| top = sorted(rows, key=lambda r: (r.get("trending") or 0) * 1e9 + (r.get("downloads") or 0), reverse=True)[:120] |
| desc = DESCRIPTION |
| body = (f"<header class=\"tp-head\"><h1>{TAGLINE}</h1><p class=\"lede\">{esc(desc)}</p></header>" |
| + _env_list(top) + '<p><a href="/community">Community rollouts</a> · <a href="/d/XiaomiMiMo/MiMo-V2.6-RL-oss">MiMo-V2.6 RL</a></p>') |
| ld = [{"@type": "WebSite", "name": SITE, "alternateName": TAGLINE, "url": base + "/", "description": desc, |
| "potentialAction": {"@type": "SearchAction", "target": {"@type": "EntryPoint", "urlTemplate": base + "/?q={search_term_string}"}, |
| "query-input": "required name=search_term_string"}, |
| "publisher": {"@type": "Organization", "name": "FineEnvs", "url": "https://huggingface.co/FineEnvs"}}, |
| {"@type": "CollectionPage", "name": TAGLINE, "url": base + "/", "about": "Reinforcement learning environments", |
| "mainEntity": {"@type": "ItemList", "numberOfItems": len(top), "itemListElement": [ |
| {"@type": "ListItem", "position": i + 1, "url": base + _href(r), "name": r.get("heading") or r["id"]} for i, r in enumerate(top[:50])]}}] |
| return page(request, title=f"{SITE}: {TAGLINE}", description=desc, path="/", body=body, jsonld=ld) |
|
|
|
|
| def _href(r: dict[str, Any]) -> str: |
| return f"/{'s' if r['kind'] == 'space' else 'd'}/{enc(r['id'])}" |
|
|
|
|
| def _env_list(rows: list[dict[str, Any]]) -> str: |
| return "<ul>" + "".join(f'<li><a href="{_href(r)}">{esc(r.get("heading") or r["id"])}</a> <span>({esc(r["id"])}, {esc(kind_of(r, r["id"]))})</span>' |
| + (f" <span>{esc(clip(r.get('brief') or '', 160))}</span>" if r.get("brief") else "") + "</li>" for r in rows) + "</ul>" |
|
|
|
|
| def environment(request: Request, spec: str) -> HTMLResponse: |
| base = base_url(request) |
| by, _ = listing() |
| r = next((r for r in public_rows(list(by.values())) if r["key"] == spec), None) |
| name = display_name(r, spec) |
| kind = kind_of(r, spec) |
| tasks = index_rows(spec) if r else [] |
| n = len(tasks) or ((r or {}).get("indexed") or {}).get("tasks") |
| brief = clip((r or {}).get("brief") or "", 300) |
| desc = clip(f"{name}: {kind} on Hugging Face{f' with {n:,} tasks' if n else ''}. " |
| f"{brief or 'See what each task asks, how it is graded and what it runs in, then run an agent on it.'}", 300) |
| path = f"/d/{enc(spec)}" |
| cr, cld = crumbs(base, [("Environments", "/"), (name, "")]) |
| body = (cr + f"<header class=\"tp-head\"><h1>{esc(spec)}</h1><p class=\"lede\">{esc(desc)}</p></header>" |
| + (f"<h2>Tasks</h2><ul>" + "".join(f'<li><a href="/t/{enc(spec)}/{enc(t["path"])}">{esc(clip(t.get("title") or t["path"], 140))}</a></li>' |
| for t in tasks[:150]) + "</ul>" + (f"<p>{len(tasks) - 150:,} more tasks.</p>" if len(tasks) > 150 else "") if tasks else "") |
| + f'<p><a href="https://huggingface.co/datasets/{enc(spec)}" rel="noopener">{esc(spec)} on the Hugging Face Hub</a></p>') |
| ld = [{"@type": "Dataset", "name": name, "alternateName": spec, "description": desc, "url": base + path, |
| "sameAs": f"https://huggingface.co/datasets/{spec}", "identifier": spec, "isAccessibleForFree": True, |
| "creator": {"@type": "Organization", "name": spec.split("/")[0], "url": f"https://huggingface.co/{spec.split('/')[0]}"}, |
| "keywords": [k for k in ["reinforcement learning", "RL environment", kind, *((r or {}).get("tags") or [])[:12]] if k], |
| **({"dateModified": r["updated"]} if r and r.get("updated") else {}), **({"dateCreated": r["created"]} if r and r.get("created") else {}), |
| "includedInDataCatalog": {"@type": "DataCatalog", "name": SITE, "url": base + "/"}, |
| "distribution": [{"@type": "DataDownload", "encodingFormat": "application/octet-stream", "contentUrl": f"https://huggingface.co/datasets/{spec}"}]}, |
| cld] |
| return page(request, title=f"{name} · {kind}", description=desc, path=path, body=body, jsonld=ld, image=f"/og{path}.png" if r else None, noindex=not r) |
|
|
|
|
| def task(request: Request, spec: str, ref: str) -> HTMLResponse: |
| base = base_url(request) |
| by, _ = listing() |
| r = next((r for r in public_rows(list(by.values())) if r["key"] == spec), None) |
| env_name = display_name(r, spec) |
| known = r is not None |
| row = next((t for t in index_rows(spec) if t["path"] == ref), None) if known else None |
| status = 200 |
| if known and not row and re.fullmatch(r"(?:[^/]+/){1,2}\d{1,10}", ref): |
| try: |
| row = seo_tasks.row(spec, ref) |
| except seo_tasks.Unavailable as exc: |
| status = exc.status |
| elif known and not row: |
| status = 404 |
| split, _, n = ref.rpartition("/") |
| |
| fallback = f"Row {n} ({split})" if split and n.isdigit() and "/" not in split else ref.rsplit("/", 1)[-1] |
| title = clip((row or {}).get("title") or fallback or env_name, 110) |
| what = clip((row or {}).get("brief") or "", 260) |
| desc = clip(f"{title}: a task in {env_name} ({kind_of(r, spec)}). {what or 'What the agent is asked, how it is graded, what it runs in, and its files.'}", 300) |
| path = f"/t/{enc(spec)}/{enc(ref)}" |
| cr, cld = crumbs(base, [("Environments", "/"), (env_name, f"/d/{enc(spec)}"), (title, "")]) |
| body = (cr + f"<header class=\"tp-head\"><h1>{esc(title)}</h1><p class=\"lede\">{esc(desc)}</p></header>" |
| + (f'<section><h2>The task</h2><p class="seo-prompt">{esc(row.get("brief") or "")}</p></section>' if row else "") |
| + f'<p>Part of <a href="/d/{enc(spec)}">{esc(spec)}</a>.</p>') |
| if row and type(row.get("total")) is int and n.isdigit(): |
| body += '<nav aria-label="Other tasks">' + " · ".join( |
| f'<a href="/t/{enc(spec)}/{enc(split)}/{i}">{label}</a>' |
| for i, label in ((int(n) - 1, "Previous task"), (int(n) + 1, "Next task")) if 0 <= i < row["total"]) + '</nav>' |
| ld = [{"@type": "CreativeWork", "name": title, "description": desc, "url": base + path, "learningResourceType": "RL environment task", |
| "isPartOf": {"@type": "Dataset", "name": env_name, "url": f"{base}/d/{enc(spec)}", "sameAs": f"https://huggingface.co/datasets/{spec}"}, |
| "about": "reinforcement learning", **({"genre": row["category"]} if row and row.get("category") else {})}, cld] |
| return page(request, title=f"{title} · {env_name}", description=desc, path=path, body=body, jsonld=ld, image=f"/og{path}.png" if known else None, |
| noindex=not row and status != 503, og_type="article", status=status) |
|
|
|
|
| def space(request: Request, spec: str) -> HTMLResponse: |
| base = base_url(request) |
| by, _ = listing() |
| r = next((r for r in public_rows(list(by.values())) if r["key"] == f"space:{spec}"), None) |
| if "task" in request.query_params: |
| return space_task(request, spec, r) |
| name = display_name(r, spec) |
| kind = kind_of(r, spec) if r else "environment Space" |
| desc = clip(f"{name}: an {kind} on Hugging Face. {clip((r or {}).get('brief') or '', 220) or 'See it live: its app, a playground with rewards, its tasks, and MCP for coding agents.'}", 300) |
| path = f"/s/{enc(spec)}" |
| cr, cld = crumbs(base, [("Environments", "/"), ("Spaces", "/?k=openenv"), (name, "")]) |
| body = (cr + f"<header class=\"tp-head\"><h1>{esc(spec)}</h1><p class=\"lede\">{esc(desc)}</p></header>" |
| + f'<p><a href="https://huggingface.co/spaces/{enc(spec)}" rel="noopener">{esc(spec)} on the Hugging Face Hub</a></p>') |
| ld = [{"@type": "SoftwareApplication", "name": name, "alternateName": spec, "description": desc, "url": base + path, |
| "sameAs": f"https://huggingface.co/spaces/{spec}", "applicationCategory": "DeveloperApplication", "operatingSystem": "Web", |
| "offers": {"@type": "Offer", "price": "0", "priceCurrency": "USD"}, |
| "author": {"@type": "Organization", "name": spec.split("/")[0], "url": f"https://huggingface.co/{spec.split('/')[0]}"}, |
| "keywords": ["reinforcement learning", "RL environment", *(["OpenEnv"] if (r or {}).get("openenv") else []), *((r or {}).get("tags") or [])[:10]]}, cld] |
| if r and discoverable(r): |
| ranges = seo_tasks.ranges((spaces_live.last_seen(spec) or {}).get("task_api")) |
| body += "".join(f'<section><h2>{esc(split)} tasks</h2><ul>' + "".join( |
| f'<li><a href="{esc(seo_tasks.space_path(spec, env, split, i))}">Task {i + 1}</a></li>' for i in range(min(n, 12))) |
| + "</ul></section>" for env, split, n in ranges) |
| return page(request, title=f"{name} · {kind}", description=desc, path=path, body=body, jsonld=ld, image=f"/og{path}.png" if r else None, noindex=not r or not discoverable(r)) |
|
|
|
|
| def space_task(request: Request, spec: str, environment: dict | None) -> HTMLResponse: |
| q = request.query_params |
| env, split, raw = q.get("env", ""), q.get("split", ""), q.get("task", "") |
| if not environment or not re.fullmatch(r"\d{1,10}", raw) or not env or not split or len(env) > 80 or len(split) > 200: |
| return not_found(request) |
| index = int(raw) |
| path = seo_tasks.space_path(spec, env, split, index) |
| status, row = 200, None |
| try: |
| row = seo_tasks.space(spec, env, split, index) |
| except seo_tasks.Unavailable as exc: |
| status = exc.status |
| except Exception: |
| status = 503 |
| if status == 404: |
| return not_found(request) |
| title = clip((row or {}).get("title") or f"Task {index + 1}", 110) |
| desc = clip(f"{title}: {split} task in {spec}, an RL environment on the Hugging Face Hub. {(row or {}).get('brief') or ''}", 300) |
| cr, cld = crumbs(base_url(request), [("Environments", "/"), (spec, f"/s/{enc(spec)}"), (title, "")]) |
| body = cr + f'<header class="tp-head"><h1>{esc(title)}</h1><p class="lede">{esc(desc)}</p></header>' |
| if row: |
| body += f'<section><h2>The task</h2><p class="seo-prompt">{esc(row["brief"])}</p></section>' |
| if row.get("fields"): |
| body += '<h2>Task details</h2><dl>' + "".join( |
| f'<dt>{esc(k.replace("_", " "))}</dt><dd>{esc(str(v)[:1000])}</dd>' for k, v in row["fields"].items()) + '</dl>' |
| body += '<nav aria-label="Other tasks">' + " · ".join( |
| f'<a href="{esc(seo_tasks.space_path(spec, env, split, i))}">{label}</a>' |
| for i, label in ((index - 1, "Previous task"), (index + 1, "Next task")) if 0 <= i < row["total"]) + '</nav>' |
| else: |
| body += '<p>The task server is temporarily unavailable. Please try again shortly.</p>' |
| ld = [{"@type": "CreativeWork", "name": title, "description": desc, "url": base_url(request) + path, |
| "isPartOf": {"@type": "SoftwareApplication", "name": spec, "url": base_url(request) + f"/s/{enc(spec)}"}}, cld] |
| return page(request, title=f"{title} · {spec}", description=desc, path=path, body=body, jsonld=ld, |
| image=f"/og/s/{enc(spec)}.png?" + urlencode({"env": env, "split": split, "task": index}), |
| og_type="article", status=status) |
|
|
|
|
| def simple(request: Request, path: str) -> HTMLResponse: |
| """The app's other pages: community (indexed), a rollout, your rollouts, a comparison (not indexed: they change, or are personal).""" |
| if path.startswith("/community"): |
| return page(request, title="Community rollouts", path="/community", |
| description="Public, graded RL rollouts across Harbor, MiMo, NeMo Gym and other supported runners. Explore traces and model results grouped by environment and scoring metric.", |
| body="<header class=\"tp-head\"><h1>Community rollouts</h1><p class=\"lede\">Public rollouts from everyone, shown without who ran them.</p></header>") |
| titles = {"/runs": "My rollouts"} |
| return page(request, title=titles.get(path, "Rollout" if path.startswith("/run/") else "Compare rollouts"), path=path, |
| description=DESCRIPTION, noindex=True) |
|
|
|
|
| def not_found(request: Request) -> HTMLResponse: |
| return page(request, title="Page not found", path=request.url.path, description=DESCRIPTION, noindex=True, status=404, |
| body='<header class="tp-head"><h1>Page not found</h1><p class="lede">There is nothing at this address. <a href="/">Explore RL environments</a>.</p></header>') |
|
|
|
|
| |
| def robots(request: Request) -> PlainTextResponse: |
| base = base_url(request) |
| return PlainTextResponse("\n".join([ |
| "User-agent: *", "Allow: /", "Disallow: /api/", "Disallow: /mcp/", "Disallow: /capture/", "Disallow: /run/", "Disallow: /runs", |
| "Allow: /api/env/", "Allow: /api/spaces/", "Allow: /api/search", "Allow: /api/environments", |
| "Disallow: /api/environments/mine", "Disallow: /compare/", "Disallow: /login", "Disallow: /logout", "", f"Sitemap: {base}/sitemap.xml", ""])) |
|
|
|
|
| _sitemap: dict[str, Any] = {"at": 0.0, "pages": [], "tasks": []} |
|
|
|
|
| def _entries() -> tuple[list[tuple[str, str | None]], list[tuple[str, str | None]]]: |
| with _lock: |
| if time.time() - _sitemap["at"] < 300 and _sitemap["pages"]: |
| return _sitemap["pages"], _sitemap["tasks"] |
| by, rows = listing() |
| rows = public_rows(rows) |
| checks = space_checks.inventory() |
| pages: list[tuple[str, str | None]] = [("/", None), ("/community", None)] |
| |
| pages += [(_href(r), (r.get("updated") or "")[:10] or None) for r in rows |
| if discoverable(r, checks)] |
| tasks: list[tuple[str, str | None]] = [] |
| specs = list(dict.fromkeys(r["id"] for r in rows if r["kind"] == "dataset")) |
| covered = set() |
| try: |
| |
| |
| with snapshot.use() as (_, conn): |
| refs = conn.execute("SELECT env, ref FROM tasks WHERE env IN (SELECT value FROM json_each(?)) ORDER BY env, ref", |
| (json.dumps(specs),)).fetchall() |
| for r in refs: |
| covered.add(r["env"]) |
| tasks.append((f"/t/{enc(r['env'])}/{enc(r['ref'])}", None)) |
| except snapshot.SnapshotError: |
| pass |
| for spec in dict.fromkeys(r["id"] for r in rows if r["kind"] == "dataset" and r["id"] not in covered |
| and (r["id"] == MIMO or (r.get("indexed") or {}).get("tasks"))): |
| try: |
| tasks += [(f"/t/{enc(spec)}/{enc(t['path'])}", None) for t in index_rows(spec)] |
| except Exception: |
| continue |
| with _lock: |
| tasks = list(dict.fromkeys(tasks)) |
| _sitemap.update(at=time.time(), pages=pages, tasks=tasks) |
| return pages, tasks |
|
|
|
|
| def _space_ranges(): |
| """Represent task coordinates as ranges; do not allocate millions of URLs or fetch any tasks.""" |
| records = space_checks.inventory() |
| _, rows = listing() |
| out = [] |
| for r in sorted(public_rows(rows), key=lambda r: r["key"]): |
| if r.get("kind") != "space" or not space_checks.browseable(records.get(r["id"])): |
| continue |
| rec = records[r["id"]] |
| if not (rec.get("task_catalog") or {}).get("tasks") and not rec.get("task_splits"): |
| continue |
| ranges = rec.get("task_splits") or seo_tasks.ranges((spaces_live.last_seen(r["id"]) or {}).get("task_api")) |
| for env, split, n in ranges: |
| if type(n) is int and 0 < n <= 1_000_000_000: |
| out.append((r["id"], env, split, n)) |
| return out |
|
|
|
|
| def _urlset(base: str, items: list[tuple[str, str | None]]) -> Response: |
| xml = ['<?xml version="1.0" encoding="UTF-8"?>', '<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">'] |
| xml += [f"<url><loc>{esc(base + p)}</loc>{f'<lastmod>{esc(m)}</lastmod>' if m else ''}</url>" for p, m in items if len((base + p).encode()) <= 2048] |
| xml.append("</urlset>") |
| return Response("\n".join(xml), media_type="application/xml") |
|
|
|
|
| def sitemap_index(request: Request) -> Response: |
| base = base_url(request) |
| _, tasks = _entries() |
| parts = ["/sitemap-pages.xml", *[f"/sitemap-tasks-{i + 1}.xml" for i in range((len(tasks) + SITEMAP_CHUNK - 1) // SITEMAP_CHUNK)]] |
| total = sum(r[3] for r in _space_ranges()) |
| parts += [f"/sitemap-space-tasks-{i + 1}.xml" for i in range(min(49000, (total + SITEMAP_CHUNK - 1) // SITEMAP_CHUNK))] |
| xml = ['<?xml version="1.0" encoding="UTF-8"?>', '<sitemapindex xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">'] |
| xml += [f"<sitemap><loc>{esc(base + p)}</loc></sitemap>" for p in parts] |
| xml.append("</sitemapindex>") |
| return Response("\n".join(xml), media_type="application/xml") |
|
|
|
|
| def sitemap_pages(request: Request) -> Response: |
| pages, _ = _entries() |
| return _urlset(base_url(request), pages) |
|
|
|
|
| def sitemap_tasks(request: Request, n: int) -> Response: |
| _, tasks = _entries() |
| chunk = tasks[(n - 1) * SITEMAP_CHUNK: n * SITEMAP_CHUNK] |
| if n < 1 or not chunk: |
| return Response("not found", status_code=404) |
| return _urlset(base_url(request), chunk) |
|
|
|
|
| def sitemap_space_tasks(request: Request, n: int) -> Response: |
| if not 1 <= n <= 49000: |
| return Response("not found", status_code=404) |
| offset, remaining, urls = (n - 1) * SITEMAP_CHUNK, SITEMAP_CHUNK, [] |
| for spec, env, split, count in _space_ranges(): |
| if offset >= count: |
| offset -= count |
| continue |
| stop = min(count, offset + remaining) |
| urls.extend((seo_tasks.space_path(spec, env, split, i), None) for i in range(offset, stop)) |
| remaining -= stop - offset |
| offset = 0 |
| if not remaining: |
| break |
| return _urlset(base_url(request), urls) if urls else Response("not found", status_code=404) |
|
|
|
|
| |
| |
| W, H = 1200, 630 |
|
|
|
|
| @lru_cache(maxsize=512) |
| def card_png(kind: str, title: str, sub: str, facts: str) -> bytes: |
| from PIL import Image, ImageDraw, ImageFont |
|
|
| img = Image.new("RGB", (W, H), "#fffdf7") |
| d = ImageDraw.Draw(img) |
| font = lambda s: ImageFont.load_default(size=s) |
| d.rectangle([0, 0, W, 12], fill="#ffcd36") |
| d.rounded_rectangle([56, 48, 198, 90], radius=10, fill="#ffdc62") |
| d.text((73, 58), "FINEENVS", font=font(23), fill="#352c0e") |
| d.text((220, 57), SITE, font=font(26), fill="#54504a") |
| d.text((56, 142), _ellipsize(d, kind.upper(), font(23), 1088), font=font(23), fill="#827256") |
| lines = _wrap(d, title, font(65), 1088) |
| size = 65 if len(lines) <= 3 else 54 |
| lines = _wrap(d, title, font(size), 1088)[:3] |
| for i, line in enumerate(lines): |
| d.text((56, 194 + i * 77), _ellipsize(d, line, font(size), 1088), font=font(size), fill="#1c1b19") |
| if sub: |
| d.text((56, 452), _ellipsize(d, sub, font(27), 1088), font=font(27), fill="#665e51") |
| d.line([56, 516, 1144, 516], fill="#e5dfd1", width=2) |
| d.text((56, 550), _ellipsize(d, facts or "Discover tasks. Inspect rewards. Run agents.", font(24), 680), font=font(24), fill="#524d42") |
| d.text((1144, 550), "Hugging Face Hub", font=font(24), fill="#827256", anchor="ra") |
| out = io.BytesIO() |
| img.save(out, "PNG", optimize=True) |
| return out.getvalue() |
|
|
|
|
| def _ellipsize(d, text, font, width): |
| text = str(text) |
| if d.textlength(text, font=font) <= width: |
| return text |
| while text and d.textlength(text + "…", font=font) > width: |
| text = text[:-1] |
| return text + "…" |
|
|
|
|
| def _wrap(d, text: str, font, width: int) -> list[str]: |
| if "\n" in text: |
| return [line for part in text.splitlines() for line in _wrap(d, part, font, width)] |
| words, lines, cur = str(text).split(), [], "" |
| for w in words: |
| nxt = f"{cur} {w}".strip() |
| if d.textlength(nxt, font=font) <= width: |
| cur = nxt |
| else: |
| if cur: |
| lines.append(cur) |
| cur = w |
| if cur: |
| lines.append(cur) |
| if len(lines) > 3: |
| lines = lines[:3] |
| lines[2] = lines[2].rstrip(".,;:") + "…" |
| return lines or [""] |
|
|
|
|
| def og_image(path: str, query=None) -> Response: |
| """The preview image of a page, from its path (`/d/org/name`, `/t/org/name/ref`, `/s/org/name`, or the site's).""" |
| if path == "/": |
| return Response(card_png("Reinforcement learning", "Explore RL environments\non the Hugging Face Hub", |
| "OpenEnv · Harbor · MiMo · NeMo Gym · Verifiers", ""), |
| media_type="image/png", headers={"Cache-Control": "public, max-age=86400"}) |
| by, _ = listing() |
| by = {r["key"]: r for r in public_rows(list(by.values()))} |
| parts = [p for p in path.strip("/").split("/") if p] |
| if parts and not (len(parts) >= 3 and parts[0] in ("d", "s", "t")): |
| return Response(status_code=404) |
| kind, title, sub, facts = "Reinforcement learning environments", TAGLINE, "Across OpenEnv, Harbor, Verifiers, NeMo Gym and more", "" |
| if len(parts) >= 3 and parts[0] in ("d", "s", "t"): |
| spec = f"{parts[1]}/{parts[2]}" |
| r = by.get(spec if parts[0] != "s" else f"space:{spec}") |
| if r is None: |
| return Response(status_code=404) |
| name = display_name(r, spec) |
| k = kind_of(r, spec) |
| if parts[0] == "t" and (r or spec == MIMO): |
| ref = "/".join(parts[3:]) |
| row = next((t for t in index_rows(spec) if t["path"] == ref), None) |
| if not row and re.fullmatch(r"(?:[^/]+/){1,2}\d{1,10}", ref): |
| try: |
| row = seo_tasks.row(spec, ref) |
| except seo_tasks.Unavailable as exc: |
| return Response(status_code=exc.status) |
| if not row: |
| return Response(status_code=404) |
| kind, title, sub = f"A task in {name}", clip((row or {}).get("title") or ref, 140), spec |
| elif parts[0] == "s" and query and "task" in query: |
| raw, env, split = query.get("task", ""), query.get("env", ""), query.get("split", "") |
| if not re.fullmatch(r"\d{1,10}", raw) or not env or not split or len(env) > 80 or len(split) > 200: |
| return Response(status_code=404) |
| try: |
| row = seo_tasks.space(spec, env, split, int(raw)) |
| except Exception: |
| return Response(status_code=503, headers={"Retry-After": "60"}) |
| kind, title, sub = f"{split} task", clip(row["title"], 140), spec |
| else: |
| n = ((r or {}).get("indexed") or {}).get("tasks") or (len(index_rows(spec)) if spec == MIMO else None) |
| kind, title, sub = k, name, spec |
| facts = " · ".join(x for x in [f"{n:,} tasks" if n else "", f"{(r or {}).get('downloads', 0):,} downloads" if r and r.get("downloads") else "", |
| f"{r['likes']:,} likes" if r and r.get("likes") else ""] if x) |
| return Response(card_png(kind, title, sub, facts), media_type="image/png", headers={"Cache-Control": "public, max-age=86400"}) |
|
|
