connor
9cba8db7bb
The NYT is unreadable without a subscription, but dropping it would throw away real signal -- knowing the NYT covered a story is corroboration, and it feeds importance. So sources carry a `paywalled` flag: they still cluster and corroborate, but are never chosen as the link for a notification or a headline when a readable report of the same story exists, and are marked in the UI. Deliberately NOT doing paywall circumvention. The fix is to prefer a source you can actually read, and to add outlets that do not have the problem. Five free New York outlets added, all verified live: The City, Gothamist, New York Focus, Documented NY and amNY -- independent and largely nonprofit, covering the NYC politics the NYT was being relied on for. Hell Gate was probed too and has no discoverable feed at any of the usual paths. Includes an additive migration for the new column, since CREATE TABLE IF NOT EXISTS will not add a column to a table that already exists.
442 lines
17 KiB
Python
442 lines
17 KiB
Python
"""Web surface: the digest page and the feedback endpoint.
|
|
|
|
The feedback endpoint is the single most important route in the system. It is
|
|
what the ntfy buttons call, and it is the only way the taste model ever learns
|
|
anything. The old system collected 32 labels in a week because this did not
|
|
work; everything else here is downstream of it working.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from datetime import UTC, date, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
from starlette.applications import Starlette
|
|
from starlette.responses import JSONResponse, PlainTextResponse
|
|
from starlette.routing import Route
|
|
from starlette.templating import Jinja2Templates
|
|
|
|
from . import store
|
|
from .config import Config
|
|
from .notify import sign, verify
|
|
|
|
TEMPLATES = Jinja2Templates(directory=str(Path(__file__).parent / "templates"))
|
|
|
|
|
|
def _con():
|
|
return store.connect(Config().db_path)
|
|
|
|
|
|
# Scoring a page load is cheap (matrix ops over ~1,300 clusters), but building
|
|
# ProfileVectors loads the embedding model. Cached so only the first request
|
|
# pays for it.
|
|
_VECTORS = None
|
|
|
|
|
|
def _scored(con, cfg):
|
|
global _VECTORS
|
|
from .embed import Embedder
|
|
from .score import ProfileVectors, score_clusters
|
|
|
|
if _VECTORS is None:
|
|
_VECTORS = ProfileVectors(con, cfg.profile(), Embedder.from_config(cfg))
|
|
return score_clusters(con, _VECTORS, cfg.scoring, cfg.clustering)
|
|
|
|
|
|
async def healthz(request):
|
|
return PlainTextResponse("ok")
|
|
|
|
|
|
async def feedback(request):
|
|
"""POST /api/feedback/{cluster_id}/{kind}/{token}
|
|
|
|
Signed per (cluster, kind), so a leaked URL authorises one vote on one
|
|
story rather than acting as a bearer token for the whole model.
|
|
"""
|
|
secret = os.environ.get("NEWS_FEEDBACK_SECRET")
|
|
if not secret:
|
|
return JSONResponse({"error": "feedback not configured"}, status_code=503)
|
|
|
|
cluster_id = request.path_params["cluster_id"]
|
|
kind = request.path_params["kind"]
|
|
token = request.path_params["token"]
|
|
if kind not in ("interested", "not_interested"):
|
|
return JSONResponse({"error": "bad kind"}, status_code=400)
|
|
if not verify(secret, cluster_id, kind, token):
|
|
return JSONResponse({"error": "bad signature"}, status_code=403)
|
|
|
|
con = _con()
|
|
con.execute(
|
|
"""INSERT INTO feedback (cluster_id, kind, source, created_at)
|
|
VALUES (?,?,?,?)""",
|
|
(cluster_id, kind, request.query_params.get("src", "ntfy"),
|
|
datetime.now(UTC).isoformat()),
|
|
)
|
|
return JSONResponse({"ok": True, "cluster_id": cluster_id, "kind": kind})
|
|
|
|
|
|
async def triage(request):
|
|
"""The labelling machine.
|
|
|
|
This is where feedback volume comes from, so it is built to be fast to skim
|
|
and label: ranked, keyboard-driven, one keystroke per judgement. The old
|
|
system's UI could only be clicked, and its clicks did not persist.
|
|
"""
|
|
from datetime import UTC, datetime
|
|
|
|
from .route import route
|
|
|
|
cfg = Config()
|
|
con = _con()
|
|
scores = _scored(con, cfg)
|
|
decisions = {d.cluster_id: d for d in
|
|
route(con, scores, cfg.routing, cfg.delivery,
|
|
datetime.now(UTC), cfg.block_groups())}
|
|
|
|
secret = os.environ.get("NEWS_FEEDBACK_SECRET")
|
|
labelled = {r["cluster_id"] for r in con.execute("SELECT DISTINCT cluster_id FROM feedback")}
|
|
show = request.query_params.get("show", "unlabelled")
|
|
limit = min(int(request.query_params.get("limit", 60)), 200)
|
|
|
|
# Sort orders. "best" is the default and mixes both axes; "interest" is the
|
|
# one that answers "what is mine, ignoring whether the world thinks it
|
|
# matters", which the two-axis design otherwise hides.
|
|
SORTS = {
|
|
"best": lambda s: -max(s.interest_score, s.importance_score),
|
|
"interest": lambda s: -s.interest_score,
|
|
"important": lambda s: -s.importance_score,
|
|
"new": lambda s: -(s.last_seen.timestamp() if s.last_seen else 0),
|
|
"covered": lambda s: (-s.corroboration.origins, -s.interest_score),
|
|
}
|
|
sort = request.query_params.get("sort", "best")
|
|
if sort not in SORTS:
|
|
sort = "best"
|
|
ranked = sorted(scores, key=SORTS[sort])
|
|
rows = []
|
|
for s in ranked:
|
|
if s.veto:
|
|
continue
|
|
if show == "unlabelled" and s.cluster_id in labelled:
|
|
continue
|
|
d = decisions.get(s.cluster_id)
|
|
# Every member, not a slice: the largest cluster on this corpus is 23
|
|
# items and multi-item clusters hold 249 between them, so rendering the
|
|
# lot inline is cheaper than a lazy-loading endpoint would be to build.
|
|
members = con.execute(
|
|
"""SELECT i.title, i.url, i.published_at, so.name, so.paywalled
|
|
FROM cluster_members cm
|
|
JOIN items i ON i.id = cm.item_id
|
|
JOIN sources so ON so.id = i.source_id
|
|
WHERE cm.cluster_id = ?
|
|
-- Readable sources first, so the headline link actually opens.
|
|
ORDER BY so.paywalled ASC, i.published_at DESC""", (s.cluster_id,)).fetchall()
|
|
rows.append({
|
|
"id": s.cluster_id, "headline": s.lead_title,
|
|
"topic": s.interest.block, "importance_topic": s.importance.block,
|
|
"interest": s.interest_score, "importance": s.importance_score,
|
|
"origins": s.corroboration.origins, "demote": s.demote,
|
|
"lane": d.lane if d else "?", "reason": d.reason if d else "",
|
|
"labelled": s.cluster_id in labelled,
|
|
# Signed here so the browser posts to the SAME public endpoint the
|
|
# phone uses. The alternative -- an Authelia-gated path -- fails
|
|
# silently when a session expires: fetch() follows the redirect to
|
|
# the login page, gets a 200, and the UI reports success while
|
|
# nothing was recorded.
|
|
"tok_yes": sign(secret, s.cluster_id, "interested") if secret else "",
|
|
"tok_no": sign(secret, s.cluster_id, "not_interested") if secret else "",
|
|
"sources": [{"name": m["name"], "title": m["title"], "url": m["url"],
|
|
"when": (m["published_at"] or "")[11:16],
|
|
"paywalled": bool(m["paywalled"])}
|
|
for m in members],
|
|
})
|
|
if len(rows) >= limit:
|
|
break
|
|
|
|
return TEMPLATES.TemplateResponse(
|
|
request, "triage.html",
|
|
{"rows": rows, "show": show, "total": len(scores), "labelled": len(labelled),
|
|
"signed": bool(secret), "current": "triage", "sort": sort,
|
|
"sort_help": {
|
|
"best": "Highest of the two scores — mixes what you like with what matters.",
|
|
"interest": "Purely your profile. Ignores importance entirely, so a quiet "
|
|
"story that is very much yours outranks a big one that is not.",
|
|
"important": "Purely the importance axis — burst velocity, independent "
|
|
"corroboration, editorial prominence. Largely world news.",
|
|
"new": "Most recently seen first, regardless of score.",
|
|
"covered": "Most independent newsrooms first, syndicated copy collapsed.",
|
|
}[sort],
|
|
"sorts": [("best", "Best match"), ("interest", "My interests"),
|
|
("important", "Most important"), ("new", "Newest first"),
|
|
("covered", "Most covered")]})
|
|
|
|
|
|
async def tune(request):
|
|
"""Threshold tuning against real history.
|
|
|
|
The old system's replay harness hardcoded its parameters, so it could never
|
|
sweep the thresholds it existed to tune. Here the sliders drive the same
|
|
`route()` the live service calls, so what you see is what it would have sent.
|
|
"""
|
|
from datetime import UTC, datetime
|
|
|
|
from .route import DIGEST, DROP, IMPORTANT, INTEREST, route
|
|
|
|
cfg = Config()
|
|
con = _con()
|
|
scores = _scored(con, cfg)
|
|
|
|
routing = dict(cfg.routing)
|
|
for key in ("interest_threshold", "important_threshold", "digest_floor"):
|
|
if (raw := request.query_params.get(key)) is not None:
|
|
try:
|
|
routing[key] = float(raw)
|
|
except ValueError:
|
|
pass
|
|
for key in ("interest_per_day", "important_per_day", "digest_max_items",
|
|
"max_alerts_per_group_per_day"):
|
|
if (raw := request.query_params.get(key)) is not None:
|
|
try:
|
|
routing[key] = int(raw)
|
|
except ValueError:
|
|
pass
|
|
|
|
now = datetime.now(UTC)
|
|
decisions = route(con, scores, routing, cfg.delivery, now, cfg.block_groups())
|
|
by_id = {s.cluster_id: s for s in scores}
|
|
lanes: dict[str, list] = {}
|
|
for d in decisions:
|
|
lanes.setdefault(d.lane, []).append(d)
|
|
|
|
days = routing["max_age_hours"] / 24.0
|
|
counts = {lane: len(lanes.get(lane, [])) for lane in (IMPORTANT, INTEREST, DIGEST, DROP)}
|
|
alerts = [
|
|
{"lane": d.lane, "reason": d.reason, "score": d.score,
|
|
"headline": by_id[d.cluster_id].lead_title,
|
|
"topic": by_id[d.cluster_id].interest.block}
|
|
for d in sorted(lanes.get(IMPORTANT, []) + lanes.get(INTEREST, []),
|
|
key=lambda d: -d.score)
|
|
]
|
|
digest = [
|
|
{"reason": d.reason, "score": d.score,
|
|
"headline": by_id[d.cluster_id].lead_title,
|
|
"topic": by_id[d.cluster_id].interest.block}
|
|
for d in sorted(lanes.get(DIGEST, []), key=lambda d: -d.score)[:25]
|
|
]
|
|
|
|
return TEMPLATES.TemplateResponse(
|
|
request, "tune.html",
|
|
{"routing": routing, "counts": counts, "alerts": alerts, "digest": digest,
|
|
"per_day": (counts[IMPORTANT] + counts[INTEREST]) / days, "days": days,
|
|
"defaults": cfg.routing, "current": "tune"})
|
|
|
|
|
|
def _centroids(con):
|
|
rows = con.execute(
|
|
"""SELECT c.id, c.centroid, (SELECT title FROM items WHERE id=c.lead_item_id) AS lead
|
|
FROM clusters c ORDER BY c.last_seen_at DESC LIMIT 2000""").fetchall()
|
|
if not rows:
|
|
return [], np.zeros((0, 1), dtype=np.float32)
|
|
mat = np.stack([store.from_blob(r["centroid"]) for r in rows])
|
|
mat /= np.maximum(np.linalg.norm(mat, axis=1, keepdims=True), 1e-12)
|
|
return rows, mat
|
|
|
|
|
|
async def config_page(request):
|
|
"""What each block currently matches, and a box to try a new filter.
|
|
|
|
The point is to make narrowing down cheap. The old system's config view was
|
|
read-only and showed no consequences, so tuning meant editing a file,
|
|
restarting, and waiting a poll cycle to find out what changed.
|
|
"""
|
|
cfg = Config()
|
|
con = _con()
|
|
profile = cfg.profile()
|
|
rows, mat = _centroids(con)
|
|
|
|
from .embed import Embedder
|
|
from .score import ProfileVectors
|
|
|
|
global _VECTORS
|
|
if _VECTORS is None:
|
|
_VECTORS = ProfileVectors(con, profile, Embedder.from_config(cfg))
|
|
|
|
def summarise(kind: str, key: str, default: float):
|
|
out = []
|
|
for block in profile.get(kind, []):
|
|
sims = None
|
|
for bid, probes, _ in _VECTORS.blocks[kind]:
|
|
if bid == block["id"]:
|
|
sims = (mat @ probes.T).max(axis=1) if len(mat) else np.zeros(0)
|
|
thr = float(block.get(key, default))
|
|
hits = int((sims >= thr).sum()) if sims is not None else 0
|
|
top = []
|
|
if sims is not None and len(sims):
|
|
for i in np.argsort(-sims)[:3]:
|
|
top.append({"title": rows[i]["lead"], "score": float(sims[i])})
|
|
out.append({"id": block["id"], "weight": block.get("weight", 1.0),
|
|
"group": block.get("group", block["id"]),
|
|
"probes": len(block["probes"]), "threshold": thr,
|
|
"penalty": block.get("penalty"), "hits": hits, "top": top,
|
|
"requires_any": block.get("requires_any", [])})
|
|
return out
|
|
|
|
return TEMPLATES.TemplateResponse(
|
|
request, "config.html",
|
|
{"interests": summarise("interest", "___", 0.60),
|
|
"vetoes": summarise("veto", "threshold", 0.60),
|
|
"demotes": summarise("demote", "___", 0.55),
|
|
"clusters": len(rows), "scoring": cfg.scoring, "routing": cfg.routing,
|
|
"current": "config"})
|
|
|
|
|
|
async def config_preview(request):
|
|
"""Embed an arbitrary sentence and show what it would catch, right now."""
|
|
body = await request.json()
|
|
text = (body.get("text") or "").strip()
|
|
threshold = float(body.get("threshold", 0.60))
|
|
if not text:
|
|
return JSONResponse({"matches": [], "total": 0})
|
|
|
|
cfg = Config()
|
|
con = _con()
|
|
rows, mat = _centroids(con)
|
|
if not len(mat):
|
|
return JSONResponse({"matches": [], "total": 0})
|
|
|
|
from .embed import Embedder
|
|
|
|
vec = Embedder.from_config(cfg).encode([text])[0]
|
|
sims = mat @ vec
|
|
order = np.argsort(-sims)
|
|
matches = [{"title": rows[i]["lead"], "score": round(float(sims[i]), 3)}
|
|
for i in order[:25] if sims[i] >= threshold]
|
|
return JSONResponse({
|
|
"matches": matches,
|
|
"total": int((sims >= threshold).sum()),
|
|
"clusters": len(rows),
|
|
# The boundary is where judgement is needed, so show what sits just
|
|
# below the line as well as what is caught.
|
|
"near_miss": [{"title": rows[i]["lead"], "score": round(float(sims[i]), 3)}
|
|
for i in order if threshold - 0.06 <= sims[i] < threshold][:5],
|
|
})
|
|
|
|
|
|
async def config_save(request):
|
|
"""Write a change back to profile.toml, preserving comments.
|
|
|
|
tomlkit rather than tomllib+dump: this file carries the reasoning behind
|
|
every constant, and a round-trip that strips comments would quietly destroy
|
|
the record of why each number is what it is.
|
|
"""
|
|
import tomlkit
|
|
|
|
body = await request.json()
|
|
cfg = Config()
|
|
path = cfg.profile_path
|
|
doc = tomlkit.parse(path.read_text())
|
|
|
|
action = body.get("action")
|
|
if action == "weight":
|
|
for block in doc.get(body["kind"], []):
|
|
if block["id"] == body["id"]:
|
|
block["weight"] = float(body["value"])
|
|
break
|
|
else:
|
|
return JSONResponse({"error": "block not found"}, status_code=404)
|
|
elif action == "add":
|
|
kind = body["kind"]
|
|
if kind not in ("veto", "demote"):
|
|
return JSONResponse({"error": "kind must be veto or demote"}, status_code=400)
|
|
table = tomlkit.table()
|
|
table["id"] = body["id"]
|
|
if kind == "veto":
|
|
table["threshold"] = float(body.get("threshold", 0.60))
|
|
else:
|
|
table["penalty"] = float(body.get("penalty", 1.0))
|
|
arr = tomlkit.array().multiline(True)
|
|
for probe in body["probes"]:
|
|
arr.append(probe)
|
|
table["probes"] = arr
|
|
doc.setdefault(kind, tomlkit.aot()).append(table)
|
|
else:
|
|
return JSONResponse({"error": "unknown action"}, status_code=400)
|
|
|
|
try:
|
|
path.write_text(tomlkit.dumps(doc))
|
|
except OSError as exc:
|
|
return JSONResponse(
|
|
{"error": f"cannot write {path}: {exc}. Is /config mounted read-only?"},
|
|
status_code=500)
|
|
|
|
global _VECTORS
|
|
_VECTORS = None # probes changed; re-embed on the next request
|
|
return JSONResponse({"ok": True})
|
|
|
|
|
|
async def digest_page(request):
|
|
day = request.path_params.get("day")
|
|
try:
|
|
when = date.fromisoformat(day) if day else datetime.now(UTC).date()
|
|
except ValueError:
|
|
return PlainTextResponse("bad date", status_code=400)
|
|
|
|
con = _con()
|
|
start = datetime.combine(when, datetime.min.time(), tzinfo=UTC)
|
|
rows = con.execute(
|
|
"""SELECT n.cluster_id, n.lane, n.score, n.created_at,
|
|
(SELECT title FROM items WHERE id = c.lead_item_id) AS headline
|
|
FROM notification_log n
|
|
JOIN clusters c ON c.id = n.cluster_id
|
|
WHERE n.created_at >= ? AND n.created_at < ?
|
|
ORDER BY n.score DESC""",
|
|
(start.isoformat(), (start + timedelta(days=1)).isoformat()),
|
|
).fetchall()
|
|
|
|
stories = []
|
|
for r in rows:
|
|
members = con.execute(
|
|
"""SELECT i.title, i.url, i.published_at, s.name, s.paywalled
|
|
FROM cluster_members cm
|
|
JOIN items i ON i.id = cm.item_id
|
|
JOIN sources s ON s.id = i.source_id
|
|
WHERE cm.cluster_id = ?
|
|
ORDER BY s.paywalled ASC, i.published_at DESC LIMIT 6""",
|
|
(r["cluster_id"],),
|
|
).fetchall()
|
|
stories.append({
|
|
"id": r["cluster_id"], "headline": r["headline"], "lane": r["lane"],
|
|
"sources": [{"name": m["name"], "title": m["title"], "url": m["url"],
|
|
"when": (m["published_at"] or "")[11:16],
|
|
"paywalled": bool(m["paywalled"])}
|
|
for m in members],
|
|
})
|
|
|
|
prose = con.execute(
|
|
"SELECT briefing FROM digest_prose WHERE day = ?", (when.isoformat(),)
|
|
).fetchone()
|
|
|
|
return TEMPLATES.TemplateResponse(
|
|
request, "digest.html",
|
|
{"day": when, "stories": stories,
|
|
"briefing": prose["briefing"] if prose else None,
|
|
"prev": when - timedelta(days=1), "next": when + timedelta(days=1),
|
|
"current": "digest"},
|
|
)
|
|
|
|
|
|
routes = [
|
|
Route("/", triage),
|
|
Route("/tune", tune),
|
|
Route("/config", config_page),
|
|
Route("/config/preview", config_preview, methods=["POST"]),
|
|
Route("/config/save", config_save, methods=["POST"]),
|
|
Route("/healthz", healthz),
|
|
Route("/api/feedback/{cluster_id}/{kind}/{token}", feedback, methods=["POST"]),
|
|
Route("/digest", digest_page),
|
|
Route("/digest/{day}", digest_page),
|
|
]
|
|
|
|
app = Starlette(routes=routes)
|