#!/usr/bin/env python3 """ athena/scoring.py — Sprint 1: Pure Rule-Based Bucket Classifier + Scorer. DESIGN CONSTRAINT (founder directive, 2026-07-15): Pure rules only. No embeddings, no semantic similarity, no LLM classification. We are still discovering the editorial taxonomy. Deterministic systems are easier to debug; misclassifications are signal; we must be able to explain WHY every story landed where it did before we add intelligence. Pipeline position: ingestion/dedup (pipeline.py) -> [ATTACH HERE] -> rendering Call attach_scoring(conn) immediately after store_entries and before render. Schema migration (idempotent): bucket, shipping_score, utility_score, replication_score, enthusiast_score, novelty_score, hype_penalty, final_score, actionability_score, narrative_id, topic_id, relation_json Review report: human-readable, per-bucket, exposes every fired rule. Usage: python3 athena/scoring.py --migrate # add columns python3 athena/scoring.py --backfill # score all unscored rows python3 athena/scoring.py --review # write review report python3 athena/scoring.py --all # migrate + backfill + review """ import argparse import json import os import re import sqlite3 from datetime import datetime, timezone HERE = os.path.dirname(os.path.abspath(__file__)) DB_PATH = os.path.join(os.path.dirname(HERE), "oracle.db") # ~/oracle/oracle.db # --------------------------------------------------------------------------- # SCORE WEIGHTS (transparent, tunable in one place) # --------------------------------------------------------------------------- WEIGHTS = { "shipping": 0.20, # a working artifact exists "utility": 0.20, # clear practical use for an enthusiast "replication": 0.25, # can a reader reproduce/run it locally "enthusiast": 0.20, # signals a builder/DIY practitioner audience "novelty": 0.15, # new, specific, non-generic } HYPE_CAP = 0.45 # final_score is floored if hype penalty is high # --------------------------------------------------------------------------- # BUCKETS + PURE RULES # Each bucket: keyword hits and/or source constraints. First match wins, # evaluated in BUCKET_ORDER (most specific taxon first). # --------------------------------------------------------------------------- # Keyword sets (lowercased; matched against title + summary + tags text). KW_SHIPPING = [ "released", "launch", "v1.0", "v2.0", "v3.0", "shipping", "now available", "open source", "open-source", "open weights", "weights released", "live now", "beta", "public beta", "ga release", "general availability", "ships", "deployed", "production", "now in", "available today", "download", "gradio", "demo", "playground", ] KW_LOCAL_AI = [ "local llm", "local model", "local ai", "run locally", "run it locally", "on-device", "on device", "ollama", "llama.cpp", "llamacpp", "gguf", "ggml", "lm studio", "consumer hardware", "consumer gpu", "rtx", "your own gpu", "offline", "private ai", "local-only", "self-host", "self-hosted", "home server", "edge device", "edge inference", "quantized", "quantization", "q4", "q8", "int4", "fp16", "fine-tune at home", "train at home", "local inference", "local deployment", "no api", "no cloud", ] KW_PROBLEM_SOLVED = [ "how to", "how i", "solved", "fix", "fixed", "workaround", "benchmark", "improves", "improved", "speedup", "speed-up", "reduces", "reduce", "cut", "cuts", "boost", "optimize", "optimized", "optimisation", "faster", "3x", "10x", "2x", "latency", "throughput", "roi", "cost", "cheaper", "save", "saves", "eliminate", "eliminated", "from 117s to 30s", "p95", "memory usage", "vram", "token cost", "bottleneck", "case study", "results", "we measured", "we tested", "showdown", "comparison", ] KW_MODEL_RELEASE = [ "releases", "released", "unveils", "introduces", "new model", "new flagship", "gpt-", "claude", "gemini", "llama", "mistral", "qwen", "deepseek", "grok", "phi-", "command-r", "api access", "weights", "open model", "open-models", "frontier", "checkpoint", "fine-tune", "finetune", "rl-trained", "rl train", "trained", "post-training", "post training", "distilled", "distillation", ] KW_RESEARCH = [ "paper", "arxiv", "preprint", "study", "research", "we propose", "we present", "we introduce", "we show", "method", "framework", "theorem", "analysis of", "survey", "benchmark", "dataset", "neural", "transformer", "diffusion", "gradient", "ablation", "we find", "our approach", "novel", "state-of-the-art", "sota", "cs.lg", "cs.cl", "cs.cv", "cs.ai", ] KW_BUSINESS = [ "raises", "raised", "$", "valuation", "series a", "series b", "funding", "round", "ipo", "acquisition", "acquires", "merger", "deal", "revenue", "layoff", "hiring", "partnership", "invests", "investment", "market", "vc", "compute deal", "billion", "million", "forecast", "miss", "earnings", "stock", ] KW_INFRA = [ "gpu", "tpu", "data center", "datacenter", "data centre", "cluster", "cuda", "rocm", "vllm", "tensorrt", "inference server", "serving", "kubernetes", "docker", "pipeline", "mlops", "ci/cd", "rag", "vector db", "vector database", "agent", "agents", "orchestration", "observability", "evaluation", "eval", "guardrail", "safety", "red team", "jailbreak", "prompt injection", "fine-tuning stack", ] KW_CULTURE = [ "says", "argues", "opinion", "essay", "think", "thinks", "the real", "why we", "the future of", "dystopia", "utopia", "philosophy", "ethics", "regulation", "policy", "ban", "lawsuit", "eu", "senate", "congress", "interview", "podcast", "controversy", "controversial", "debate", "debate", "critic", "criticism", "creepy", "creeping", "not sexy", "vibe", "hot take", "unpopular", ] # Buckets evaluated in this order (specific -> generic). source_whitelist matches # raw `source` value exactly; if present, story must come from one of those sources. BUCKETS = { "SHIPPING": { "kw": KW_SHIPPING, "source_whitelist": None, "order": 0, }, "LOCAL AI": { "kw": KW_LOCAL_AI, "source_whitelist": None, "order": 1, }, "PROBLEM SOLVED": { "kw": KW_PROBLEM_SOLVED, "source_whitelist": None, "order": 2, }, "MODEL RELEASE": { "kw": KW_MODEL_RELEASE, "source_whitelist": None, "order": 3, }, "RESEARCH": { "kw": KW_RESEARCH, "source_whitelist": ["arxiv"], "order": 4, }, "BUSINESS": { "kw": KW_BUSINESS, "source_whitelist": None, "order": 5, }, "INFRASTRUCTURE": { "kw": KW_INFRA, "source_whitelist": None, "order": 6, }, "CULTURE": { "kw": KW_CULTURE, "source_whitelist": None, "order": 7, }, } BUCKET_ORDER = sorted(BUCKETS.keys(), key=lambda b: BUCKETS[b]["order"]) # Hype / low-signal penalty terms HYPE_TERMS = [ "revolutionary", "game-changing", "game changer", "breakthrough", "mind-blowing", "insane", "crazy", "unbelievable", "shocking", "the future is here", "omg", "you won't believe", "secret", "they don't want you to know", "leaked", "viral", "hype", "buzzword", "disrupt", "disrupting everything", "ai will replace", "will change everything", "paradigm shift", "godlike", "magic", "miracle", ] # Enthusiast-audience signals (builder / DIY / practitioner) ENTHUSIAST_SIGNALS = [ "github", "repo", "repository", "self-host", "local", "ollama", "llamacpp", "hugging face", "huggingface", "colab", "notebook", "pip install", "docker", "cli", "open source", "open-source", "diy", "build your own", "tutorial", "how to", "implementation", "agent", "agents", "fine-tune", "finetune", "quantiz", "vllm", "rtx", "gpu", "consumer", "homelab", "self-hosted", "machine-learning", "machine learning", "deep learning", "python", "rust", "benchmark", "reproduc", "weights", "gguf", ] # Source -> enthusiast affinity bonus SOURCE_ENTHUSIAST_BONUS = { "github": 0.20, "huggingface": 0.20, "arxiv": 0.10, "hackernews": 0.10, "reddit": 0.05, "rss": 0.0, } def _norm(text): if not text: return "" if isinstance(text, bytes): text = text.decode("utf-8", "replace") return " " + re.sub(r"\s+", " ", text.lower()) + " " def classify(entry): """Pure rule classification. Returns (bucket, matched_list) where matched_list is human-readable proof: ["kw:ollama", "kw:gguf", "src:huggingface", "tag:local"] """ title = _norm(entry.get("title") or "") summary = _norm(_summary_text(entry.get("summary"))) tags_raw = entry.get("category_tags") or "" try: tags = " ".join(json.loads(tags_raw)) if tags_raw else "" except Exception: tags = tags_raw tags = _norm(tags) source = (entry.get("source") or "").lower() haystack = title + " " + summary + " " + tags matched = ["source=%s" % source] best_bucket = "UNCATEGORIZED" best_hits = 0 for bucket in BUCKET_ORDER: spec = BUCKETS[bucket] whitelist = spec["source_whitelist"] if whitelist and source not in whitelist: continue hits = [] for kw in spec["kw"]: kwn = " " + kw.lower() + " " if kwn in haystack: hits.append(kw) if hits: # record proof (cap displayed hits to keep report readable) for h in hits[:8]: matched.append("kw:%s" % h) if len(hits) > best_hits: best_hits = len(hits) best_bucket = bucket if best_bucket == "UNCATEGORIZED": matched.append("(no rule fired)") return best_bucket, matched def _summary_text(raw): if not raw: return "" try: d = json.loads(raw) if isinstance(d, dict): return " ".join(str(v) for v in d.values() if isinstance(v, str)) except Exception: pass return raw def score_entry(bucket, matched, entry): """Return dict of component scores (0..1) + final (0..1). Components are deterministic functions of signals; see inline rationale. """ source = (entry.get("source") or "").lower() title = _norm(entry.get("title") or "") summary = _norm(_summary_text(entry.get("summary"))) tags_raw = entry.get("category_tags") or "" try: tags = " ".join(json.loads(tags_raw)) if tags_raw else "" except Exception: tags = tags_raw tags = _norm(tags) haystack = title + " " + summary + " " + tags # --- enthusiast score --- ent_hits = sum(1 for s in ENTHUSIAST_SIGNALS if (" " + s + " ") in haystack) ent_base = min(ent_hits / 5.0, 1.0) # 5+ signals = full ent_src = SOURCE_ENTHUSIAST_BONUS.get(source, 0.0) enthusiast = min(ent_base + ent_src, 1.0) # --- shipping score --- ship_kw = [k for k in KW_SHIPPING if (" " + k + " ") in haystack] shipping = 0.0 if bucket == "SHIPPING": shipping = 0.9 elif ship_kw: shipping = min(0.4 + 0.1 * len(ship_kw), 0.8) # source artifacts (github/hf) imply something shippable exists if source in ("github", "huggingface"): shipping = max(shipping, 0.7) # --- utility score --- util_kw = [k for k in KW_PROBLEM_SOLVED if (" " + k + " ") in haystack] utility = 0.0 if bucket == "PROBLEM SOLVED": utility = 0.85 elif util_kw: utility = min(0.4 + 0.1 * len(util_kw), 0.8) if "github" in haystack or "huggingface" in haystack or "demo" in haystack: utility = max(utility, 0.6) # --- replication score (can a reader reproduce/run locally) --- repl_kw = [k for k in KW_LOCAL_AI if (" " + k + " ") in haystack] replication = 0.0 if bucket == "LOCAL AI": replication = 1.0 elif repl_kw: replication = min(0.5 + 0.1 * len(repl_kw), 0.9) if source in ("github", "huggingface"): replication = max(replication, 0.7) if "open source" in haystack or "open-source" in haystack or "weights" in haystack: replication = max(replication, 0.6) # --- novelty score --- nov_kw = ["new", "novel", "first", "breakthrough-method", "we propose", "we introduce", "we present", "state-of-the-art", "sota", "unveils"] novelty = 0.0 if bucket in ("RESEARCH", "MODEL RELEASE"): novelty = 0.6 if any((" " + k + " ") in haystack for k in nov_kw): novelty = min(novelty + 0.2, 0.9) if bucket == "CULTURE": novelty = min(novelty, 0.3) # opinion pieces are rarely novel technically # --- hype penalty --- hype_hits = [t for t in HYPE_TERMS if (" " + t + " ") in haystack] hype_penalty = min(0.1 * len(hype_hits), 0.6) # --- final --- raw = ( WEIGHTS["shipping"] * shipping + WEIGHTS["utility"] * utility + WEIGHTS["replication"] * replication + WEIGHTS["enthusiast"] * enthusiast + WEIGHTS["novelty"] * novelty ) final = max(raw - hype_penalty, 0.0) final = min(final, 1.0) return { "shipping_score": round(shipping, 3), "utility_score": round(utility, 3), "replication_score": round(replication, 3), "enthusiast_score": round(enthusiast, 3), "novelty_score": round(novelty, 3), "hype_penalty": round(hype_penalty, 3), "final_score": round(final, 4), } # --------------------------------------------------------------------------- # DB OPERATIONS # --------------------------------------------------------------------------- NEW_COLUMNS = [ "bucket TEXT", "shipping_score REAL DEFAULT 0", "utility_score REAL DEFAULT 0", "replication_score REAL DEFAULT 0", "enthusiast_score REAL DEFAULT 0", "novelty_score REAL DEFAULT 0", "hype_penalty REAL DEFAULT 0", "final_score REAL DEFAULT 0", "actionability_score REAL DEFAULT 0", "narrative_id TEXT", "topic_id TEXT", "relation_json TEXT", ] def migrate(db_path=DB_PATH): """Idempotent schema migration — only adds missing columns.""" conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("PRAGMA table_info(entries)") existing = {row[1] for row in cur.fetchall()} added = [] for col in NEW_COLUMNS: name = col.split(" ")[0] if name not in existing: cur.execute("ALTER TABLE entries ADD COLUMN %s" % col) added.append(name) conn.commit() conn.close() print("[migrate] added columns: %s" % (", ".join(added) if added else "none (already present)")) return added def fetch_unscored(conn): cur = conn.cursor() cur.execute("""SELECT id, source, source_id, url, title, summary, category_tags, raw_metadata FROM entries WHERE bucket IS NULL OR bucket = ''""") cols = ["id", "source", "source_id", "url", "title", "summary", "category_tags", "raw_metadata"] return [dict(zip(cols, row)) for row in cur.fetchall()] def attach_scoring(db_path=DB_PATH, dry_run=False): """Score every unscored entry. Call after ingestion/dedup, before render.""" conn = sqlite3.connect(db_path) rows = fetch_unscored(conn) print("[attach] scoring %d unscored entries" % len(rows)) for e in rows: bucket, matched = classify(e) scores = score_entry(bucket, matched, e) if dry_run: continue conn.execute( """UPDATE entries SET bucket=?, shipping_score=?, utility_score=?, replication_score=?, enthusiast_score=?, novelty_score=?, hype_penalty=?, final_score=?, actionability_score=?, narrative_id=?, topic_id=?, relation_json=? WHERE id=?""", (bucket, scores["shipping_score"], scores["utility_score"], scores["replication_score"], scores["enthusiast_score"], scores["novelty_score"], scores["hype_penalty"], scores["final_score"], 0.0, # actionability_score: reserved, unused in Sprint 1 None, None, json.dumps({"matched_rules": matched}), e["id"]), ) if not dry_run: conn.commit() conn.close() print("[attach] done.") # --------------------------------------------------------------------------- # REVIEW REPORT # --------------------------------------------------------------------------- REVIEW_BUCKETS = ["SHIPPING", "LOCAL AI", "PROBLEM SOLVED", "MODEL RELEASE", "RESEARCH", "BUSINESS", "INFRASTRUCTURE", "CULTURE", "UNCATEGORIZED"] def generate_review(db_path=DB_PATH, limit=200, out_path=None): conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("""SELECT id, source, title, url, bucket, final_score, relation_json, shipping_score, utility_score, replication_score, enthusiast_score, novelty_score, hype_penalty FROM entries ORDER BY first_seen DESC LIMIT ?""", (limit,)) rows = cur.fetchall() conn.close() by_bucket = {b: [] for b in REVIEW_BUCKETS} for r in rows: (eid, src, title, url, bucket, final, rel_json, sh, ut, rp, en, no, hy) = r try: matched = json.loads(rel_json).get("matched_rules", []) if rel_json else [] except Exception: matched = [] by_bucket.setdefault(bucket or "UNCATEGORIZED", []).append( (eid, src, title, url, final, matched, (sh, ut, rp, en, no, hy)) ) lines = [] lines.append("=" * 70) lines.append("ATHENA SPRINT 1 — MANUAL REVIEW REPORT") lines.append("Generated: %s" % datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")) lines.append("Stories reviewed: %d (most recent %d)" % (len(rows), limit)) lines.append("=" * 70) lines.append("") lines.append("HOW TO READ: For each story, the fired rules are listed as proof.") lines.append("Bucket = first matching taxon (specific -> generic). Scores are") lines.append("deterministic. 'Publish?' is for HUMAN review only — not algorithmic.") lines.append("") total_pub = total_rej = total_border = 0 for b in REVIEW_BUCKETS: items = by_bucket.get(b, []) if not items: continue lines.append(b) lines.append("-" * len(b)) for (eid, src, title, url, final, matched, comps) in items: sh, ut, rp, en, no, hy = comps lines.append("") lines.append("Story: %s" % (title or "(untitled)")) lines.append(" id=%s source=%s final=%.3f" % (eid, src, final)) lines.append(" url: %s" % (url or "")) lines.append(" Bucket: %s" % b) lines.append(" Matched:") for m in matched: lines.append(" - %s" % m) lines.append(" Score Components:") lines.append(" Shipping: %.2f" % sh) lines.append(" Utility: %.2f" % ut) lines.append(" Replication: %.2f" % rp) lines.append(" Enthusiast: %.2f" % en) lines.append(" Novelty: %.2f" % no) lines.append(" Hype: %.2f" % hy) lines.append(" Final: %.3f" % final) lines.append(" Publish? [Y/N] <- human review only") lines.append("") lines.append("") # Summary block (the founder's key metric) lines.append("=" * 70) lines.append("BUCKET DISTRIBUTION") lines.append("=" * 70) for b in REVIEW_BUCKETS: n = len(by_bucket.get(b, [])) if n: lines.append(" %-16s %3d" % (b, n)) lines.append("") lines.append("HUMAN REVIEW TALLY (fill in after manual pass):") lines.append(" Published: %d" % total_pub) lines.append(" Rejected: %d" % total_rej) lines.append(" Borderline: %d" % total_border) lines.append("") lines.append("First question is not 'is the classifier accurate?'") lines.append("First question: 'Would we proudly publish these stories?'") lines.append("=" * 70) report = "\n".join(lines) + "\n" if out_path is None: out_path = os.path.join(os.path.dirname(HERE), "athena_review_report.txt") with open(out_path, "w") as f: f.write(report) print("[review] wrote %s (%d stories)" % (out_path, len(rows))) # also print to stdout for immediate visibility print(report) return report def main(): ap = argparse.ArgumentParser(description="Athena Sprint 1 scoring (pure rules)") ap.add_argument("--migrate", action="store_true", help="add new columns") ap.add_argument("--backfill", action="store_true", help="score all unscored rows") ap.add_argument("--review", action="store_true", help="write manual review report") ap.add_argument("--all", action="store_true", help="migrate + backfill + review") ap.add_argument("--limit", type=int, default=200, help="review story count") ap.add_argument("--dry-run", action="store_true", help="classify but don't write") ap.add_argument("--db", default=DB_PATH, help="db path override") args = ap.parse_args() if args.all: migrate(args.db) attach_scoring(args.db, dry_run=args.dry_run) generate_review(args.db, limit=args.limit) else: if args.migrate: migrate(args.db) if args.backfill: attach_scoring(args.db, dry_run=args.dry_run) if args.review: generate_review(args.db, limit=args.limit) if not (args.migrate or args.backfill or args.review): ap.print_help() if __name__ == "__main__": main()