Spaces:
Running
Running
| """Vesuvius Radar: a public Space that keeps collecting new work that could help read the Herculaneum scrolls, and serves | |
| it as a page, a JSON API, an RSS feed and an MCP server. | |
| A background collector checks each source on its own schedule (radar.json, every_min) and keeps every item from the | |
| last 30 days in one store, so a lookback from 1 minute to 30 days is a filter, not a new scan. Anyone may ask for a | |
| check now (POST /api/check), at most once every few minutes for the whole Space. The Space holds no keys and needs | |
| no login: every source is public and read logged out. | |
| GET / the page | |
| GET /api/items items in a lookback window, filtered and sorted (since=1m..30d, view, q, ...) | |
| GET /api/item?id=ID one item in full, with the items closest to it in meaning | |
| GET /api/overview counts behind the charts for a window and view | |
| GET /api/briefing the strongest items per topic for a window and view | |
| GET /api/sources the collector: each source's last check, next check and recent runs | |
| POST /api/check check every source now (one at a time for the whole Space, with a cooldown) | |
| GET /api/views, /api/config the reader views and the scoring settings | |
| GET /api/export.csv a window as CSV | |
| GET /feed.xml a window as RSS | |
| POST /mcp the same as an MCP server (radar_mcp.py) | |
| GET /docs the API's OpenAPI page | |
| """ | |
| from __future__ import annotations | |
| import csv | |
| import datetime as dt | |
| import io | |
| import json | |
| import os | |
| import re | |
| import sys | |
| import threading | |
| import time | |
| from pathlib import Path | |
| from xml.sax.saxutils import escape as xml_escape | |
| from fastapi import Depends, FastAPI, HTTPException, Query | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, PlainTextResponse, Response | |
| from fastapi.staticfiles import StaticFiles | |
| import scanner | |
| HERE = Path(__file__).resolve().parent | |
| SEED = HERE / "data" / "seed-store.json" | |
| DATA_DIR = Path(os.environ.get("RADAR_DATA") or ("/data/radar" if os.access("/data", os.W_OK) else "/tmp/radar")) | |
| DATA_DIR.mkdir(parents=True, exist_ok=True) | |
| STORE_FILE = DATA_DIR / "store.json" | |
| VECS_FILE = DATA_DIR / "vectors.npz" | |
| COLLECT = os.environ.get("RADAR_COLLECT", "1") != "0" # tests set 0: no background collector | |
| SEMANTIC = os.environ.get("RADAR_SEMANTIC", "1") != "0" | |
| PUBLIC_URL = "https://sharpmeow-vesuvius-radar.hf.space" | |
| VERSION = "1.2.2" | |
| CFG = scanner.load_config() | |
| SCORER = scanner.Scorer(CFG) | |
| SEM = scanner.Semantic(CFG) | |
| HEAD = scanner.Head() | |
| LOCK = threading.RLock() | |
| STORE: dict = {"saved": None, "items": {}, "sources": {}, "runs": [], "state": {}} | |
| CHECK = {"last": 0.0, "requested": None} | |
| BOOT = scanner.iso(scanner.utcnow()) | |
| # ---------------------------------------------------------------- views (who is reading) | |
| VIEWS = { | |
| "all": {"label": "Everyone", "about": "Everything the radar kept, from every source.", "any": []}, | |
| "research": {"label": "Researchers", | |
| "about": "New papers and methods: imaging, sheet and surface segmentation, ink detection, fibers, " | |
| "registration, 3D encoders and reading damaged text. arXiv, journals, Hugging Face daily " | |
| "papers and Zenodo publications.", | |
| "any": [{"sources": ["arxiv", "journals"]}, | |
| {"kinds": ["daily-paper", "publication", "preprint", "journal article", "conference paper", | |
| "book chapter", "report", "thesis", "presentation", "poster"]}]}, | |
| "team": {"label": "Challenge team", | |
| "about": "What the community shipped and said about the scrolls: villa pull requests, issues and releases, " | |
| "new repositories, Hugging Face uploads and Zenodo records that name the scrolls, and the " | |
| "Challenge blog.", | |
| "any": [{"sources": ["github"]}, {"sources": ["blog"], "kinds": ["post"]}, | |
| {"sources": ["hf", "zenodo", "journals", "arxiv"], "topics": ["Scroll work"]}]}, | |
| "builders": {"label": "Contributors", | |
| "about": "Things you can use today: new and updated repositories, releases, models, datasets and " | |
| "Spaces, package versions, and papers that link their code.", | |
| "any": [{"kinds": ["repo", "release", "model", "dataset", "space", "package", "software"]}, | |
| {"has_code": True}]}, | |
| } | |
| def in_view(i: dict, view: str) -> bool: | |
| v = VIEWS.get(view or "all") | |
| if v is None: | |
| raise KeyError(f"unknown view {view}; one of " + ", ".join(VIEWS)) | |
| if not v["any"]: | |
| return True | |
| for c in v["any"]: | |
| if "sources" in c and i["source"] not in c["sources"]: | |
| continue | |
| if "kinds" in c and i["kind"] not in c["kinds"]: | |
| continue | |
| if "topics" in c and not any(t in i["topics"] for t in c["topics"]): | |
| continue | |
| if c.get("has_code") and not i.get("code"): | |
| continue | |
| return True | |
| return False | |
| # ---------------------------------------------------------------- time windows | |
| UNITS = {"m": 60, "min": 60, "h": 3600, "d": 86400, "w": 604800} | |
| LOOKBACKS = ["1m", "5m", "15m", "30m", "1h", "3h", "6h", "12h", "1d", "2d", "3d", "7d", "14d", "30d"] | |
| def parse_since(s: str | None, default: str = "1d") -> dt.datetime: | |
| """A lookback such as 1m, 15m, 6h, 1d or 30d (or an ISO time) as the moment the window starts, at most 30 days | |
| ago. ValueError on anything else.""" | |
| s = (s or default).strip().lower() | |
| now = scanner.utcnow() | |
| m = re.fullmatch(r"(\d+(?:\.\d+)?)\s*(m|min|h|d|w)", s) | |
| if m: | |
| secs = float(m.group(1)) * UNITS[m.group(2)] | |
| if secs <= 0: | |
| raise ValueError("lookback must be above zero") | |
| t = now - dt.timedelta(seconds=secs) | |
| else: | |
| t = scanner.parse_time(s) | |
| if not t: | |
| raise ValueError(f"lookback '{s}' is not like 15m, 6h, 1d or an ISO time") | |
| return max(t, now - dt.timedelta(days=CFG["retention_days"])) | |
| # ---------------------------------------------------------------- store | |
| def _load(p: Path) -> dict | None: | |
| try: | |
| d = json.loads(p.read_text()) | |
| return d if isinstance(d.get("items"), dict) else None | |
| except (OSError, ValueError, AttributeError): | |
| return None | |
| def load_store(): | |
| """The newest of the saved store and the one built into the image.""" | |
| cands = [d for d in (_load(STORE_FILE), _load(SEED)) if d] | |
| if cands: | |
| best = max(cands, key=lambda d: d.get("saved") or "") | |
| with LOCK: | |
| STORE.update({k: best.get(k, STORE[k]) for k in STORE}) | |
| STORE.setdefault("state", {}) | |
| prune() | |
| def save_store(): | |
| with LOCK: | |
| STORE["saved"] = scanner.iso(scanner.utcnow()) | |
| text = json.dumps(STORE, ensure_ascii=False, separators=(",", ":")) | |
| tmp = STORE_FILE.with_suffix(".tmp") | |
| tmp.write_text(text) | |
| tmp.replace(STORE_FILE) | |
| def prune(): | |
| cut = scanner.iso(scanner.utcnow() - dt.timedelta(days=CFG["retention_days"] + 1)) | |
| with LOCK: | |
| for k in [k for k, i in STORE["items"].items() if i["date"] < cut and i.get("last_seen", "") < cut]: | |
| del STORE["items"][k] | |
| STORE["runs"] = STORE["runs"][-400:] | |
| def merge(items: list[dict]) -> int: | |
| """Add or update items; returns how many were new to the store.""" | |
| now = scanner.iso(scanner.utcnow()) | |
| new = 0 | |
| with LOCK: | |
| for i in items: | |
| old = STORE["items"].get(i["id"]) | |
| if old: | |
| i["first_seen"] = old.get("first_seen", now) | |
| if old.get("date") != i["date"]: | |
| i["updated"] = True | |
| else: | |
| i["first_seen"] = now | |
| new += 1 | |
| i["last_seen"] = now | |
| STORE["items"][i["id"]] = i | |
| return new | |
| # ---------------------------------------------------------------- collector | |
| RUNNING: dict[str, dict] = {} | |
| def _due(src: str, now: float) -> bool: | |
| s = STORE["sources"].get(src) or {} | |
| if src in RUNNING: | |
| return False | |
| if CHECK["requested"] and (s.get("finished") or "") < CHECK["requested"]: | |
| return True | |
| last = scanner.parse_time(s.get("finished")) | |
| return not last or now - last.timestamp() >= 60 * CFG["sources"][src]["every_min"] | |
| def run_source(src: str): | |
| sc = CFG["sources"][src] | |
| now = scanner.utcnow() | |
| prev = STORE["sources"].get(src) or {} | |
| last_ok = scanner.parse_time(prev.get("ok_since_started")) | |
| floor = now - dt.timedelta(days=sc["backfill_days"]) | |
| since = max(floor, last_ok - dt.timedelta(hours=6)) if last_ok else floor # overlap so late listings are not missed | |
| state = STORE["state"].setdefault(src, {}) | |
| RUNNING[src] = {"source": src, "label": sc["label"], "state": "running", "started": scanner.iso(now), "step": "starting"} | |
| def prog(row): | |
| RUNNING[src] = dict(row) | |
| try: | |
| items, row = scanner.collect(src, since, CFG, SCORER, SEM if SEM.ready else None, state, prog, HEAD) | |
| row["new"] = merge(items) | |
| with LOCK: | |
| if row["state"] != "error": | |
| row["ok_since_started"] = row["started"] | |
| else: | |
| row["ok_since_started"] = prev.get("ok_since_started") | |
| row["every_min"] = sc["every_min"] | |
| STORE["sources"][src] = row | |
| STORE["runs"].append({k: row[k] for k in ("source", "started", "finished", "since", "state", "read", "kept", "new", "ms")} | |
| | {"notes": row["notes"][:3]}) | |
| prune() | |
| save_store() | |
| finally: | |
| RUNNING.pop(src, None) | |
| warm_vectors() | |
| def collector_loop(): | |
| if SEMANTIC: | |
| SEM.load() | |
| SEM.load_cache(VECS_FILE) | |
| warm_vectors() | |
| while True: | |
| now = time.time() | |
| for src in scanner.SOURCES: | |
| if _due(src, now): | |
| threading.Thread(target=run_source, args=(src,), daemon=True, name=f"collect-{src}").start() | |
| if CHECK["requested"] and not RUNNING and all( | |
| (STORE["sources"].get(s) or {}).get("finished", "") >= CHECK["requested"] for s in scanner.SOURCES): | |
| CHECK["requested"] = None | |
| time.sleep(5) | |
| def request_check() -> tuple[int, dict]: | |
| with LOCK: | |
| wait = CFG["check_cooldown_s"] - (time.time() - CHECK["last"]) | |
| if CHECK["requested"] or RUNNING: | |
| return 409, {"ok": False, "reason": "a check is already running", "sources": sources_status()} | |
| if wait > 0: | |
| return 429, {"ok": False, "reason": f"the last check started under {CFG['check_cooldown_s'] // 60} minutes ago; " | |
| f"try again in {int(wait) + 1} s", "retry_after": int(wait) + 1} | |
| CHECK["last"] = time.time() | |
| CHECK["requested"] = scanner.iso(scanner.utcnow()) | |
| return 200, {"ok": True, "requested": CHECK["requested"], "next": "GET /api/sources until every source has finished"} | |
| def sources_status() -> dict: | |
| now = scanner.utcnow() | |
| out = {} | |
| for src in scanner.SOURCES: | |
| s = dict(STORE["sources"].get(src) or {}) | |
| last = scanner.parse_time(s.get("finished")) | |
| nxt = (last + dt.timedelta(minutes=CFG["sources"][src]["every_min"])) if last else now | |
| out[src] = {"label": scanner.LABELS[src], "every_min": CFG["sources"][src]["every_min"], | |
| "state": "running" if src in RUNNING else s.get("state", "waiting"), | |
| "step": (RUNNING.get(src) or {}).get("step", ""), | |
| "last_check": s.get("finished"), "next_check": scanner.iso(max(nxt, now)), | |
| "read": s.get("read", 0), "kept": s.get("kept", 0), "new": s.get("new", 0), | |
| "seconds": round(s.get("ms", 0) / 1000, 1), "notes": s.get("notes", []), | |
| "items_stored": sum(1 for i in STORE["items"].values() if i["source"] == src)} | |
| return out | |
| # ---------------------------------------------------------------- queries | |
| def _vec_text(i: dict) -> str: | |
| return scanner.embed_text(CFG, i["title"], i.get("summary", "")) | |
| def warm_vectors(): | |
| """Embed every stored item in the background, so similarity and meaning search answer at once.""" | |
| if not SEM.ready: | |
| return | |
| with LOCK: | |
| texts = [_vec_text(i) for i in STORE["items"].values()] | |
| for k in range(0, len(texts), 256): | |
| SEM.embed(texts[k:k + 256]) | |
| SEM.save_cache(VECS_FILE) | |
| def public(i: dict) -> dict: | |
| return named({k: v for k, v in i.items() if not k.startswith("_")}) | |
| class Filters: | |
| """One set of filters, shared by the page, the API and MCP. Lists are any-of; everything else must hold.""" | |
| FIELDS = ("since", "by", "view", "q", "by_meaning", "sources", "kinds", "topics", "bands", "min_score", "has_code", "ids") | |
| def __init__(self, since="1d", by="date", view="all", q="", by_meaning=False, sources=None, kinds=None, topics=None, | |
| bands=None, min_score=0, has_code=False, ids=None): | |
| if view not in VIEWS: | |
| raise KeyError(f"unknown view {view}; one of " + ", ".join(VIEWS)) | |
| if by not in ("date", "found"): | |
| raise ValueError("by must be date (published or updated) or found (first seen by the radar)") | |
| self.start = parse_since(since) | |
| self.since, self.by, self.view, self.q = since, by, view, (q or "").strip() | |
| self.by_meaning = bool(by_meaning) and bool(self.q) | |
| self.sources, self.kinds, self.topics, self.bands = sources or [], kinds or [], topics or [], bands or [] | |
| self.min_score, self.has_code, self.ids = int(min_score), bool(has_code), set(ids or []) | |
| def when(self, i: dict) -> str: | |
| return i.get("first_seen", i["date"]) if self.by == "found" else i["date"] | |
| def keep(self, i: dict, t0: str, words: list[str]) -> bool: | |
| if self.when(i) < t0 or i["score"] < self.min_score or not in_view(i, self.view): | |
| return False | |
| if self.ids and i["id"] not in self.ids: | |
| return False | |
| if self.sources and i["source"] not in self.sources: | |
| return False | |
| if self.kinds and i["kind"] not in self.kinds: | |
| return False | |
| if self.topics and not any(t in i["topics"] for t in self.topics): | |
| return False | |
| if self.bands and i["band"] not in self.bands: | |
| return False | |
| if self.has_code and not i.get("code"): | |
| return False | |
| if words: | |
| act = i.get("activity") or {} | |
| hay = " ".join([i["title"], i.get("summary", ""), " ".join(i.get("matched", [])), i.get("who", ""), | |
| " ".join(i["topics"]), i["id"], i.get("category", ""), act.get("who", ""), | |
| act.get("text", ""), " ".join(person(w) for w in i.get("who", "").split(",")), | |
| person(act.get("who", ""))]).lower() | |
| if not all(w in hay for w in words): | |
| return False | |
| return True | |
| def query(f: Filters, sort: str = "score") -> list[dict]: | |
| """Items that pass the filters. sort: score, date, popularity; a meaning search ranks by meaning instead and adds | |
| match (cosine similarity to the query) to each item.""" | |
| t0 = scanner.iso(f.start) | |
| meaning = f.by_meaning and SEM.ready | |
| words = [] if meaning else [w for w in f.q.lower().split() if w] | |
| with LOCK: | |
| out = [i for i in STORE["items"].values() if f.keep(i, t0, words)] | |
| if meaning and out: | |
| qv = SEM.embed_query(f.q) | |
| sims = SEM.embed([_vec_text(i) for i in out]) @ qv | |
| out = [dict(i, match=round(float(s), 3)) for i, s in zip(out, sims)] | |
| out.sort(key=lambda i: -i["match"]) | |
| return out | |
| if sort == "date": | |
| out.sort(key=lambda i: f.when(i), reverse=True) | |
| elif sort == "popularity": | |
| out.sort(key=lambda i: (-((i.get("stars") or 0) + (i.get("likes") or 0) + (i.get("upvotes") or 0)), -i["score"])) | |
| else: | |
| out.sort(key=lambda i: f.when(i), reverse=True) | |
| out.sort(key=lambda i: i["score"], reverse=True) | |
| return out | |
| def similar(item_id: str, limit: int = 8) -> list[dict]: | |
| with LOCK: | |
| it = STORE["items"].get(item_id) | |
| pool = [i for i in STORE["items"].values() if i["id"] != item_id] | |
| if not it: | |
| raise KeyError(f"no item {item_id}") | |
| if not SEM.ready or not pool: | |
| return [] | |
| import numpy as np | |
| v = SEM.embed([_vec_text(it)])[0] | |
| sims = SEM.embed([_vec_text(i) for i in pool]) @ v | |
| order = np.argsort(-sims)[:limit] | |
| return [{"id": pool[k]["id"], "title": pool[k]["title"], "url": pool[k]["url"], "source": pool[k]["source"], | |
| "kind": pool[k]["kind"], "date": pool[k]["date"], "score": pool[k]["score"], "band": pool[k]["band"], | |
| "similarity": round(float(sims[k]), 3)} for k in order] | |
| def tally(items, keys) -> dict: | |
| out: dict = {} | |
| for i in items: | |
| for k in keys(i): | |
| if k: | |
| out[k] = out.get(k, 0) + 1 | |
| return dict(sorted(out.items(), key=lambda kv: -kv[1])) | |
| BIN_STEPS = [(900, 60), (3600, 300), (6 * 3600, 900), (12 * 3600, 1800), (2 * 86400, 3600), (3 * 86400, 3 * 3600), | |
| (7 * 86400, 6 * 3600), (14 * 86400, 12 * 3600)] | |
| def timeline(f: Filters, items: list[dict]) -> dict: | |
| """Items per time bin, per source. The bin width grows with the window (1 minute up to 1 day).""" | |
| now = scanner.utcnow() | |
| span = (now - f.start).total_seconds() | |
| step = next((b for lim, b in BIN_STEPS if span <= lim + 1), 86400) # +1 s: the request itself takes time | |
| first = int(f.start.timestamp() // step * step) | |
| n = int((now.timestamp() - first) // step) + 1 | |
| bins = [{"t": scanner.iso(dt.datetime.fromtimestamp(first + k * step, dt.timezone.utc)), "n": {}} for k in range(n)] | |
| for i in items: | |
| t = scanner.parse_time(f.when(i)) | |
| k = int((t.timestamp() - first) // step) if t else -1 | |
| if 0 <= k < n: | |
| b = bins[k]["n"] | |
| b[i["source"]] = b.get(i["source"], 0) + 1 | |
| return {"bin_seconds": step, "bins": bins} | |
| WORK_SIM = 0.80 # two items by one person whose texts are this close in meaning are one piece of work | |
| # One name per person: an account or nickname listed under "people" in radar.json counts and shows as that person. | |
| PEOPLE = {a.strip().lower(): name for name, accounts in CFG.get("people", {}).items() for a in [name, *accounts]} | |
| def person(name: str) -> str: | |
| return PEOPLE.get(name.strip().lower(), name.strip()) | |
| def named(i: dict) -> dict: | |
| """A copy of the item with who_name (and activity.who_name) when its author is listed under another name; who | |
| keeps the account as the source gave it.""" | |
| out = dict(i) | |
| w = author(out) if "source" in out else "" | |
| if w and person(w) != w: | |
| out["who_name"] = person(w) | |
| a = out.get("activity") | |
| if a and a.get("who") and person(a["who"]) != a["who"]: | |
| out["activity"] = {**a, "who_name": person(a["who"])} | |
| return out | |
| def author(i: dict) -> str: | |
| return i.get("who", "").split(",")[0].strip() if i["source"] not in ("pypi", "blog") else "" | |
| def top_people(items: list[dict], limit: int = 10) -> list[dict]: | |
| """People and organizations ranked by distinct pieces of work, not raw item counts. A person's items are grouped | |
| when their texts are close in meaning (cosine >= WORK_SIM, same source), so twenty small related pull requests | |
| count as one piece of work; each piece is worth its best score, and the person's total is the sum. Without the | |
| meaning model, items from one source on the same day form one piece.""" | |
| by: dict[str, list[dict]] = {} | |
| names: dict[str, str] = {} | |
| for i in items: | |
| w = person(author(i)) | |
| if w: | |
| by.setdefault(w.lower(), []).append(i) | |
| names.setdefault(w.lower(), w) | |
| rows = [] | |
| for key, its in by.items(): | |
| n = len(its) | |
| parent = list(range(n)) | |
| def root(a): | |
| while parent[a] != a: | |
| parent[a] = parent[parent[a]] | |
| a = parent[a] | |
| return a | |
| if SEM.ready and n > 1: | |
| v = SEM.embed([_vec_text(i) for i in its]) | |
| sims = v @ v.T | |
| for a in range(n): | |
| for b in range(a + 1, n): | |
| if sims[a, b] >= WORK_SIM and its[a]["source"] == its[b]["source"]: | |
| parent[root(a)] = root(b) | |
| elif n > 1: | |
| first: dict = {} | |
| for a, i in enumerate(its): | |
| parent[a] = first.setdefault((i["source"], i["date"][:10]), a) | |
| groups: dict[int, list[dict]] = {} | |
| for a in range(n): | |
| groups.setdefault(root(a), []).append(its[a]) | |
| value = sum(max(i["score"] for i in g) for g in groups.values()) | |
| rows.append({"name": names[key], "value": value, "works": len(groups), "items": n}) | |
| rows.sort(key=lambda r: (-r["value"], -r["works"], r["name"].lower())) | |
| return rows[:limit] | |
| def overview(f: Filters) -> dict: | |
| items = query(f) | |
| people = top_people(items) | |
| hist = [0] * 20 | |
| for i in items: | |
| hist[min(19, i["score"] // 5)] += 1 | |
| matrix: dict = {} | |
| for i in items: | |
| for t in i["topics"]: | |
| row = matrix.setdefault(t, {}) | |
| row[i["source"]] = row.get(i["source"], 0) + 1 | |
| return {"since": scanner.iso(f.start), "until": scanner.iso(scanner.utcnow()), "view": f.view, "by": f.by, | |
| "items": len(items), "high": sum(1 for i in items if i["band"] == "high"), | |
| "medium": sum(1 for i in items if i["band"] == "medium"), | |
| "by_source": tally(items, lambda i: [i["source"]]), "by_band": tally(items, lambda i: [i["band"]]), | |
| "by_topic": tally(items, lambda i: i["topics"]), "by_kind": tally(items, lambda i: [i["kind"]]), | |
| "topic_source": matrix, "score_hist": hist, "top_people": {p["name"]: p["value"] for p in people}, | |
| "people": people, | |
| "community": {"villa_merged": sum(1 for i in items if i["kind"] == "pr" and i.get("state") == "merged"), | |
| "new_scroll_repos": sum(1 for i in items if i.get("new_repo") and "Scroll work" in i["topics"])}, | |
| "with_code": sum(1 for i in items if i.get("code")), | |
| "new_repos": sum(1 for i in items if i.get("new_repo")), | |
| "lifted_by_meaning": sum(1 for i in items if i.get("semantic") is not None and not i.get("learned") | |
| and i["score"] > i["keyword"]), | |
| "learned": {"ready": HEAD.ready, "items": sum(1 for i in items if i.get("learned")), | |
| "trained_on": HEAD.info.get("items"), | |
| "spearman": (HEAD.info.get("evaluation") or {}).get("full", {}).get("spearman")}, | |
| "timeline": timeline(f, items), | |
| "semantic": {"model": CFG["semantic"]["model"], "ready": SEM.ready, "error": SEM.error}, | |
| "store": {"items": len(STORE["items"]), "saved": STORE.get("saved"), "booted": BOOT}} | |
| def update_line(i: dict) -> str: | |
| """The latest update in words, for text outputs: 'Comment by X: what it said. Opened by Y. ' or ''.""" | |
| a = i.get("activity") or {} | |
| if not a.get("who"): | |
| return "" | |
| said = f": {a['text']}" if a.get("text") else "" | |
| end = "" if said[-1:] in (".", "!", "?") else "." | |
| return f"{a['what'][:1].upper()}{a['what'][1:]} by {a['who']}{said}{end} Opened by {i.get('who') or 'unknown'}. " | |
| def briefing(f: Filters, per_topic: int = 3) -> dict: | |
| items = query(f) | |
| used, topics = set(), {} | |
| for t in tally(items, lambda i: i["topics"]): | |
| pick = [i for i in items if t in i["topics"] and i["id"] not in used][:per_topic] | |
| if pick: | |
| topics[t] = [{k: i.get(k) for k in ("id", "title", "url", "source", "kind", "date", "score", "band", "who", | |
| "who_name", "activity", "summary", "code", "keyword", "semantic") | |
| if i.get(k) not in (None, "", [])} for i in map(named, pick)] | |
| used.update(i["id"] for i in pick) | |
| return {"since": scanner.iso(f.start), "view": f.view, "label": VIEWS[f.view]["label"], "about": VIEWS[f.view]["about"], | |
| "items": len(items), "topics": topics} | |
| # ---------------------------------------------------------------- app | |
| app = FastAPI(title="Vesuvius Radar API", version=VERSION, docs_url="/docs", redoc_url=None, | |
| description="New papers, code, models and data that could help read the Herculaneum scrolls, collected " | |
| "from public sources and scored for relevance. No key or login needed. See /CONNECT.md.") | |
| app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["GET", "POST"], allow_headers=["*"]) | |
| app.mount("/static", StaticFiles(directory=HERE / "static"), name="static") | |
| def _list(s: str | None) -> list[str]: | |
| return [x.strip() for x in (s or "").split(",") if x.strip()] | |
| def index(): | |
| return FileResponse(HERE / "index.html", headers={"Cache-Control": "no-cache"}) | |
| def connect_json(): | |
| return FileResponse(HERE / "connect.json") | |
| def connect_html(): | |
| return FileResponse(HERE / "connect.html") | |
| def connect_md(): | |
| return (HERE / "CONNECT.md").read_text() | |
| def healthz(): | |
| return "ok" | |
| def filters(since: str = Query("1d", description="Lookback: 1m, 5m, 15m, 30m, 1h, 3h, 6h, 12h, 1d, 2d, 3d, 7d, 14d, 30d " | |
| "(any number with m, h, d or w, at most 30 days) or an ISO time"), | |
| by: str = Query("date", description="date: when the item was published or last updated; found: when the radar " | |
| "first saw it"), | |
| view: str = Query("all", description="all, research, team or builders (see /api/views)"), | |
| q: str = Query("", description="Words that must all appear, or with by_meaning a phrase ranked by meaning"), | |
| by_meaning: bool = Query(False, description="Rank by closeness in meaning to q (sentence embeddings)"), | |
| sources: str = Query("", description="Comma list of " + ", ".join(scanner.SOURCES)), | |
| kinds: str = Query("", description="Comma list, such as paper,repo,pr,issue,release,model,dataset,space,package"), | |
| topics: str = Query("", description="Comma list of topics, such as Scroll work,Ink and imaging"), | |
| bands: str = Query("", description="Comma list of high, medium, low"), | |
| min_score: int = Query(0, ge=0, le=100), has_code: bool = Query(False, description="Only items that link code"), | |
| ids: str = Query("", description="Comma list of item ids (for example, your starred items)")) -> Filters: | |
| try: | |
| return Filters(since, by, view, q, by_meaning, _list(sources), _list(kinds), _list(topics), _list(bands), min_score, | |
| has_code, _list(ids)) | |
| except (KeyError, ValueError) as e: | |
| raise HTTPException(400, str(e).strip("'\"")) | |
| def api_items(f: Filters = Depends(filters), sort: str = Query("score", pattern="^(score|date|popularity)$"), | |
| limit: int = Query(50, ge=1, le=500), offset: int = Query(0, ge=0), | |
| compact: bool = Query(False, description="Drop the score breakdown and cut summaries to 400 characters")): | |
| items = query(f, sort) | |
| page = [public(i) for i in items[offset:offset + limit]] | |
| if compact: | |
| page = [{**{k: v for k, v in i.items() if k != "parts"}, "summary": i.get("summary", "")[:400]} for i in page] | |
| return JSONResponse({"since": scanner.iso(f.start), "until": scanner.iso(scanner.utcnow()), "view": f.view, "by": f.by, | |
| "matched": len(items), "offset": offset, "returned": len(page), | |
| "by_meaning": f.by_meaning and SEM.ready, "items": page}, headers={"Cache-Control": "no-cache"}) | |
| def api_counts(f: Filters = Depends(filters)): | |
| items = query(f.__class__("30d", f.by, f.view, f.q if not f.by_meaning else "", False, f.sources, f.kinds, f.topics, | |
| f.bands, f.min_score, f.has_code, f.ids)) | |
| now = scanner.utcnow() | |
| whens = [f.when(i) for i in items] | |
| out = {} | |
| for lb in LOOKBACKS: | |
| t0 = scanner.iso(parse_since(lb)) | |
| out[lb] = sum(1 for w in whens if w >= t0) | |
| return JSONResponse({"lookbacks": out, "at": scanner.iso(now)}, headers={"Cache-Control": "no-cache"}) | |
| def api_item(id: str): | |
| with LOCK: | |
| it = STORE["items"].get(id) | |
| if not it: | |
| raise HTTPException(404, f"no item {id}") | |
| return {**public(it), "similar": similar(id)} | |
| def api_overview(f: Filters = Depends(filters)): | |
| return JSONResponse(overview(f), headers={"Cache-Control": "no-cache"}) | |
| def api_briefing(f: Filters = Depends(filters), per_topic: int = Query(3, ge=1, le=10)): | |
| if f.min_score == 0: | |
| f.min_score = 35 | |
| return briefing(f, per_topic) | |
| def api_sources(): | |
| return JSONResponse({"sources": sources_status(), "check_requested": CHECK["requested"], | |
| "check_cooldown_s": CFG["check_cooldown_s"], "runs": STORE["runs"][-60:][::-1], | |
| "semantic": {"model": CFG["semantic"]["model"], "ready": SEM.ready, "error": SEM.error}, | |
| "store": {"items": len(STORE["items"]), "saved": STORE.get("saved"), "booted": BOOT}}, | |
| headers={"Cache-Control": "no-store"}) | |
| def api_check(): | |
| code, body = request_check() | |
| return JSONResponse(body, status_code=code) | |
| def api_views(): | |
| return {k: {"label": v["label"], "about": v["about"], "rule": v["any"]} for k, v in VIEWS.items()} | |
| def head_summary() -> dict: | |
| """What the learned head is, for the page and agents: no coefficients, just how it was fitted and how it did.""" | |
| if not HEAD.ready: | |
| return {"ready": False} | |
| i = HEAD.info | |
| return {"ready": True, "trained": i.get("trained"), "items": i.get("items"), "target": i.get("target"), | |
| "calibration": i.get("calibration"), "meta_features": i.get("meta_features"), | |
| "evaluation": i.get("evaluation")} | |
| def api_config(): | |
| return {**CFG, "learned": head_summary()} | |
| def export_csv(f: Filters = Depends(filters), sort: str = Query("score", pattern="^(score|date|popularity)$")): | |
| items = query(f, sort) | |
| buf = io.StringIO() | |
| cols = ["score", "band", "keyword", "semantic", "rule", "source", "kind", "date", "first_seen", "title", "url", "who", "topics", | |
| "matched", "code", "summary", "update", "update_by", "update_said", "update_url"] | |
| w = csv.writer(buf) | |
| w.writerow(cols) | |
| for i in items: | |
| a = i.get("activity") or {} | |
| i = {**i, "update": a.get("what", ""), "update_by": a.get("who", ""), "update_said": a.get("text", ""), | |
| "update_url": a.get("url", "")} | |
| w.writerow([("; ".join(i.get(c) or []) if c in ("topics", "matched", "code") else i.get(c, "")) for c in cols]) | |
| name = f"vesuvius-radar-{f.view}-{re.sub(r'[^0-9a-z]', '', f.since.lower())[:12]}-{scanner.iso(scanner.utcnow())[:10]}.csv" | |
| return Response(buf.getvalue(), media_type="text/csv", headers={"Content-Disposition": f'attachment; filename="{name}"'}) | |
| def feed(f: Filters = Depends(filters)): | |
| if f.since == "1d" and f.min_score == 0: # the defaults of /api/items are too loose for a feed | |
| f = Filters("7d", f.by, f.view, f.q, f.by_meaning, f.sources, f.kinds, f.topics, f.bands, 60, f.has_code) | |
| items = query(f, "date") | |
| rows = [] | |
| for i in items[:200]: | |
| d = scanner.parse_time(f.when(i)) or scanner.utcnow() | |
| desc = f"Score {i['score']} ({i['band']}), {scanner.LABELS.get(i['source'], i['source'])} {i['kind']}. " + update_line(i) + i.get("summary", "") | |
| rows.append(f"<item><title>{xml_escape(i['title'])}</title><link>{xml_escape(i['url'])}</link>" | |
| f'<guid isPermaLink="false">{xml_escape(i["id"])}</guid>' | |
| f"<pubDate>{d.strftime('%a, %d %b %Y %H:%M:%S +0000')}</pubDate>" | |
| f"<category>{xml_escape(scanner.LABELS.get(i['source'], i['source']))}</category>" | |
| f"<description>{xml_escape(desc)}</description></item>") | |
| title = f"Vesuvius Radar: {VIEWS[f.view]['label']}, score {f.min_score} or more" | |
| body = ('<?xml version="1.0" encoding="UTF-8"?><rss version="2.0"><channel>' | |
| f"<title>{xml_escape(title)}</title><link>{PUBLIC_URL}/</link>" | |
| "<description>New work that could help read the Herculaneum scrolls, scored for relevance.</description>" | |
| + "".join(rows) + "</channel></rss>") | |
| return Response(body, media_type="application/rss+xml") | |
| import radar_mcp # noqa: E402 (needs the functions above) | |
| radar_mcp.bind(sys.modules[__name__]) | |
| app.include_router(radar_mcp.router) | |
| def _start(): | |
| load_store() | |
| if COLLECT: | |
| threading.Thread(target=collector_loop, daemon=True, name="collector").start() | |
| elif SEMANTIC: | |
| threading.Thread(target=lambda: (SEM.load(), warm_vectors()), daemon=True).start() | |