# ui.py
import html, threading
from typing import Any, Dict, List
from fastapi import APIRouter, HTTPException
from fastapi.responses import HTMLResponse, JSONResponse
from data import (
get_pages, get_state, ensure_page_loaded, build_all_pages, is_complete,
model_id, license_label_and_url, search_models, flatten_models
)
router = APIRouter()
def _render_table(models: List[Dict[str, Any]]) -> str:
rows = []
for m in models:
mid = html.escape(model_id(m))
desc = html.escape((m.get("description") or "")[:200])
thumb = m.get("cover_image_url") or m.get("cover_image") or ""
vis = html.escape(m.get("visibility") or "")
tags = m.get("tags") or m.get("categories") or []
tags_html = " ".join(f"{html.escape(str(t))}" for t in tags[:8])
lic_label, lic_url = license_label_and_url(m)
lic_label_h = html.escape(lic_label)
lic_html = f"{lic_label_h}" if lic_url else lic_label_h
src_links = []
if m.get("github_url"):
src_links.append(f"GitHub")
if m.get("paper_url"):
src_links.append(f"Paper")
if lic_url:
src_links.append(f"License")
sources_html = " | ".join(src_links)
run_ct = m.get("run_count")
run_html = f"{run_ct:,} runs" if isinstance(run_ct, int) else ""
thumb_html = f"})
" if thumb else ""
rows.append(
f""
f"| {thumb_html} | "
f"{mid}
{run_html} | "
f"{desc} | "
f"{vis} | "
f"{lic_html} | "
f"{sources_html} | "
f"{tags_html} | "
f"
"
)
return (
""
""
"| Thumbnail | ID / Runs | Description | Visibility | "
"License | Sources | Tags | "
"
"
+ "\n".join(rows) +
"
"
)
def _render_pagination(current_page: int) -> str:
pages = len(get_pages())
links = []
for i in range(1, pages + 1):
if i == current_page:
links.append(f"{i}")
else:
links.append(f"{i}")
state = get_state()
if state["more"]:
links.append(f"Load {pages + 1}")
return ""
def _shell(page_index_0: int, html_table: str, *, search_query: str = "") -> str:
page_num = page_index_0 + 1
state = get_state()
search_disabled = "" if state["complete"] else "disabled"
search_opacity = "1.0" if state["complete"] else "0.45"
import html as _h
return f"""
Replicate Catalog — Page {page_num}
Replicate catalog
page {page_num} of {state['pages']} known
more pages available: {"yes" if state["more"] else "no"}
building: {"yes" if state["building"] else "no"}
complete: {"yes" if state["complete"] else "no"}
{html_table}
{_render_pagination(page_num)}
Commercial stats
Total–
Analyzed–
Unknown–
Commercial-friendly–
Copyleft–
Non-commercial–
No-derivatives–
License log
Total models–
Known licenses–
Unknown licenses–
"""
@router.get("/page/{n}", response_class=HTMLResponse)
def page(n: int):
if n < 1:
raise HTTPException(status_code=400, detail="Page must be >= 1")
idx = n - 1
ensure_page_loaded(idx)
pages = get_pages()
if idx >= len(pages):
raise HTTPException(status_code=404, detail="No more pages.")
models = pages[idx]["results"]
return HTMLResponse(_shell(idx, _render_table(models)))
@router.post("/build_all")
def build_all():
if is_complete():
return JSONResponse({"started": False, "reason": "complete"}, status_code=200)
threading.Thread(target=build_all_pages, daemon=True).start()
return JSONResponse({"started": True}, status_code=202)
@router.get("/status")
def status():
return JSONResponse(get_state())
@router.get("/search", response_class=HTMLResponse)
def search(q: str = ""):
idx = 0
base = get_pages()[0]["results"] if get_pages() else []
if is_complete() and q.strip():
results = search_models(q)
content = _render_table(results)
return HTMLResponse(_shell(idx, content, search_query=q))
else:
content = _render_table(base)
return HTMLResponse(_shell(idx, content, search_query=q))