Sync from GitHub · 2026-10-20 19:36 UTC

news/classify.py

@@ -12,7 +12,7 @@ from datetime import date
 from utils.groq_client import call_groq_json, LIGHT_MODEL
 from news.business_types import labels_for_prompt, valid_subset
 
-IMPACTS = ("threat", "opportunity", "neutral")
+IMPACTS = ("opportunity", "neutral")
 CATEGORIES = ("policy", "weather", "infrastructure", "funding", "security", "other")
 
 _SYSTEM_PROMPT = f"""You classify a single news/bulletin item for VenThrive, a Nigerian small-business platform.
@@ -28,27 +28,28 @@ Read the title and text given, then output ONLY a JSON object with these exact f
 Business type taxonomy (use these ids exactly, do not invent new ones):
 {labels_for_prompt()}
 
-"threat" and "opportunity" are reserved for items with a CONCRETE, near-term consequence for how
-a small business actually operates — a cost that goes up, a new restriction or compliance
-requirement, a market or supply disruption, a scheme/grant/program it could apply to, a new
-customer segment or channel opening up. If the only effect is on government finances, a
-company's share price, politics, or the general public, and there's no specific operational
-change a small-business owner would need to act on, that is "neutral" even if the headline sounds
-dramatic. A single incident (an accident, a crime, an arrest) is "neutral" unless the text itself
-states a resulting regulatory or operational change businesses must respond to. An aspirational
-government target or long-term roadmap with no specific program, deadline, or eligibility a
-business could act on today is "neutral", not "opportunity" — reserve "opportunity" for something
-a business owner could actually apply to, access, or benefit from now.
+There is no "threat" value. Reporting that something bad already happened has no value to a
+business owner who can no longer act on it — a cost increase, a flood, or a security incident is
+only useful here if it is a live, ongoing condition the owner can still respond to today (e.g. a
+flood warning still in effect, a tax change not yet in force). Forward-looking, predictive risk
+("what's likely to happen") is a separate system, not this classifier's job — don't try to infer
+or imply it.
+
+"opportunity" is reserved for something a business owner could actually apply to, access, or
+benefit from now: a scheme/grant/program with open eligibility, a new customer segment or channel
+opening up, a still-pending deadline. Everything else — including a cost going up, a disruption,
+an incident, an aspirational government target, or a roadmap with no program to act on today — is
+"neutral". Err toward "neutral" whenever in doubt; "opportunity" is the narrow, specific case.
 
 Examples:
-- A new CBN forex directive raising import costs -> impact=threat, category=policy, business_types=[relevant import/trade subsectors], location=Nationwide
-- A flood warning for Lagos -> impact=threat, category=weather, business_types=[subsectors with physical premises/stock in the area], location=Lagos
 - A state SME grant with an application deadline of 2026-12-01 -> impact=opportunity, category=funding, deadline=2026-12-01
 - A tax holiday for new small businesses with no stated end date -> impact=opportunity, category=policy, deadline=null
+- A new CBN forex directive raising import costs -> impact=neutral, category=policy, business_types=[relevant import/trade subsectors], location=Nationwide (nothing a business can apply to or act on, just a cost change)
+- A flood warning for Lagos -> impact=neutral, category=weather, business_types=[subsectors with physical premises/stock in the area], location=Lagos
 - A plane makes an emergency landing with no stated regulatory fallout -> impact=neutral, category=other (an isolated incident, not a business-operational change)
 - A senator publicly criticising a regulator's pace, with no rule actually changed yet -> impact=neutral, category=policy (nothing has changed for a business yet)
 - A 10-year national capital-market target or roadmap with no program a business can apply to today -> impact=neutral, category=policy (aspirational, not actionable)
-- A stock exchange index falling on a single trading day -> impact=neutral, category=other unless the text names a specific operational consequence for non-listed small businesses
+- A stock exchange index falling on a single trading day -> impact=neutral, category=other
 
 If the item is not genuinely business-relevant, or you cannot determine a category, business
 type, or location with reasonable confidence, leave the relevant field null/empty rather than

news/pipeline.py

@@ -1,7 +1,10 @@
 """
 Orchestrates one admin-triggered poll (ADR-0015): scrape every active source,
-dedup by URL hash, classify everything that survives dedup, apply the
-completeness bar, and record a run summary in news_polls.
+dedup by URL hash, keyword-prefilter (news/relevance.py) what survives dedup,
+classify the business-relevant remainder through Groq, apply the completeness
+bar, and record a run summary in news_polls. The keyword prefilter exists
+because Groq is the expensive step — an item with zero business/economy signal
+is auto-rejected before it ever reaches classify_item().
 
 A single source failing to fetch (network error, bad feed, site down) must not
 fail the whole poll — it's logged and skipped, same as DataWiz's approach to
@@ -16,6 +19,7 @@ from datetime import datetime, timedelta, timezone
 from news.scrape import fetch_source
 from news.dedup import url_hash
 from news.classify import classify_item
+from news.relevance import score as relevance_score
 from news.windows import compute_expiry
 from utils.supabase_client import get_supabase
 
@@ -66,6 +70,8 @@ def execute_poll(poll_id: str, since_days: int) -> dict:
         sources = db.table("news_sources").select("*").eq("active", True).execute().data
 
         for source in sources:
+            db.table("news_polls").update({"current_source": source["name"], **counts}).eq("id", poll_id).execute()
+
             try:
                 raw_items = fetch_source(source)
             except Exception as e:
@@ -83,6 +89,28 @@ def execute_poll(poll_id: str, since_days: int) -> dict:
                     continue
 
                 counts["items_scraped"] += 1
+                scraped_at = datetime.now(timezone.utc)
+
+                # Keyword pre-filter (DataWiz's relevance-gate pattern): a scraped item with
+                # no business/economy/regulatory signal at all never reaches Groq — it's
+                # auto-rejected for free instead of burning an AI call on sports, celebrity
+                # or isolated-incident filler that a feed mixes in alongside real items.
+                relevance = relevance_score(f"{item.title} {item.text}")
+                if not relevance["is_candidate"]:
+                    db.table("news_items").insert({
+                        "source_id":     source["id"],
+                        "url":           item.url,
+                        "url_hash":      h,
+                        "title":         item.title,
+                        "raw_text":      item.text,
+                        "published_at":  item.published_at.isoformat() if item.published_at else None,
+                        "scraped_at":    scraped_at.isoformat(),
+                        "status":        "rejected",
+                        "review_reason": f"Pre-filter: {relevance['reason']}",
+                        "expires_at":    scraped_at.isoformat(),
+                    }).execute()
+                    counts["items_rejected"] += 1
+                    continue
 
                 try:
                     result = classify_item(item.title, item.text)
@@ -90,7 +118,6 @@ def execute_poll(poll_id: str, since_days: int) -> dict:
                     print(f"[news] classification failed for {item.url}: {e}")
                     continue
 
-                scraped_at = datetime.now(timezone.utc)
                 expires_at = compute_expiry(result.category, scraped_at, result.deadline)
                 status = _route(result)
 
@@ -126,6 +153,7 @@ def execute_poll(poll_id: str, since_days: int) -> dict:
 
     db.table("news_polls").update({
         "finished_at": datetime.now(timezone.utc).isoformat(),
+        "current_source": None,
         "error": error,
         **counts,
     }).eq("id", poll_id).execute()

news/relevance.py

@@ -0,0 +1,100 @@
+"""Business-relevance pre-filter (DataWiz-Project's `pipeline/relevance.py` pattern,
+applied to MSME news instead of security incidents).
+
+Groq is the expensive step of this pipeline, not scraping: every scraped item that
+reaches classify_item() costs an API call, and a feed mixes in sports, entertainment,
+school-life and aviation-incident filler alongside the genuinely business-relevant
+items. This module is a cheap, zero-cost keyword check that runs first — an item
+with no match to any business/economy/regulatory signal never reaches Groq at all
+and is auto-rejected on the spot, the same way DataWiz discards a zero-match article
+before it ever reaches the NLP extractor.
+
+This is intentionally coarser than DataWiz's incident classifier: it only answers
+"is this worth spending a Groq call on", not what category/impact it is — that fine
+classification still happens in classify.py. A false positive here just costs one
+wasted Groq call (same as today); a false negative would silently drop a relevant
+item, so the keyword list stays broad on purpose.
+"""
+
+import re
+
+# Roots covering the five categories classify.py sorts into, plus general
+# economic/business-operational signal. Stored as roots, matched as `\b<root>\w*\b`
+# so one entry covers inflected forms (e.g. "levy" matches "levies").
+BUSINESS_KEYWORDS: dict[str, str] = {
+    # policy / regulatory
+    "tax": "policy", "vat": "policy", "levy": "policy", "tariff": "policy",
+    "duty": "policy", "customs": "policy", "excise": "policy", "regulat": "policy",
+    "cbn": "policy", "naira": "policy", "forex": "policy", "fx rate": "policy",
+    "exchange rate": "policy", "interest rate": "policy", "monetary polic": "policy",
+    "minimum wage": "policy", "subsidy": "policy", "fuel price": "policy",
+    "petrol price": "policy", "firs": "policy", "son ": "policy", "nafdac": "policy",
+    "ban on": "policy", "import ban": "policy", "export ban": "policy",
+    "license": "policy", "licence": "policy", "compliance": "policy",
+    "inflation": "policy", "devalu": "policy",
+    # weather
+    "flood": "weather", "rainfall": "weather", "drought": "weather",
+    "heatwave": "weather", "harmattan": "weather", "storm": "weather",
+    "nimet": "weather", "weather alert": "weather", "weather warning": "weather",
+    "windstorm": "weather", "erosion": "weather",
+    # infrastructure
+    "power outage": "infrastructure", "blackout": "infrastructure",
+    "electricity tariff": "infrastructure", "grid collapse": "infrastructure",
+    "fuel scarcity": "infrastructure", "port congestion": "infrastructure",
+    "road closure": "infrastructure", "bridge collaps": "infrastructure",
+    "network outage": "infrastructure", "internet shutdown": "infrastructure",
+    "telecom": "infrastructure", "logistics": "infrastructure",
+    "supply chain": "infrastructure", "diesel price": "infrastructure",
+    # funding / programs
+    "grant": "funding", "loan scheme": "funding", "funding": "funding",
+    "boi": "funding", "nirsan": "funding", "smedan": "funding",
+    "empowerment program": "funding", "empowerment scheme": "funding",
+    "accelerator": "funding", "incubat": "funding", "disburse": "funding",
+    "credit facility": "funding", "intervention fund": "funding",
+    "msme": "funding", "sme ": "funding", "startup fund": "funding",
+    # security (business-operational, not generic crime)
+    "kidnap": "security", "bandit": "security", "robbery": "security",
+    "protest": "security", "strike action": "security", "shutdown order": "security",
+    "curfew": "security", "unrest": "security", "insecurity": "security",
+    "looting": "security", "market burnt": "security", "market fire": "security",
+    # general business/economy signal not tied to one category
+    "business": "other", "trade": "other", "commerce": "other", "market": "other",
+    "manufactur": "other", "agribusiness": "other", "investment": "other",
+    "economy": "other", "economic": "other", "entrepreneur": "other",
+    "export": "other", "import": "other", "stock exchange": "other",
+    "naira exchange": "other", "gdp": "other",
+}
+
+_PATTERN_OVERRIDES = {
+    "son ": r"\bSON\b",
+    "sme ": r"\bSMEs?\b",
+}
+
+
+def _compile(root: str) -> re.Pattern:
+    if root in _PATTERN_OVERRIDES:
+        return re.compile(_PATTERN_OVERRIDES[root], re.IGNORECASE)
+    if " " in root:
+        return re.compile(r"\b" + re.escape(root.strip()) + r"\w*\b", re.IGNORECASE)
+    return re.compile(r"\b" + re.escape(root) + r"\w*\b", re.IGNORECASE)
+
+
+_PATTERNS = {root: _compile(root) for root in BUSINESS_KEYWORDS}
+
+
+def score(text: str) -> dict:
+    """Cheap verdict on whether `text` is worth a Groq call.
+
+    Zero matches -> not a candidate; classify_item() is skipped entirely and the
+    item is auto-rejected. Any match -> candidate; Groq still does the real
+    classification (category/impact/business_types/location), since a keyword
+    hit only proves the text is business-adjacent, not which category it is.
+    """
+    matched = [root.strip() for root, pat in _PATTERNS.items() if pat.search(text)]
+    is_candidate = bool(matched)
+    reason = None if is_candidate else "no business-relevant keyword matched"
+    return {
+        "is_candidate": is_candidate,
+        "matched_keywords": matched,
+        "reason": reason,
+    }

routes/admin.py

@@ -9,6 +9,7 @@ Routes:
   PUT  /api/admin/users/{uid}     → update locked fields (fullName, phone, role)
   DELETE /api/admin/users/{uid}   → delete user account + Firestore data
   GET  /api/admin/stats           → platform overview stats
+  GET  /api/admin/audit           → audit trail of manual admin actions
 """
 
 from fastapi import APIRouter, HTTPException, Depends
@@ -16,6 +17,7 @@ from pydantic import BaseModel
 from firebase_admin import auth as fb_auth, firestore
 from auth import verify_firebase_token
 from routes.account import delete_all_user_data
+from utils.supabase_client import get_supabase
 
 router = APIRouter()
 
@@ -116,6 +118,18 @@ def delete_user(
     return {"success": True, "deleted": target_uid}
 
 
+@router.get("/audit")
+def audit_trail(kind: str | None = None, limit: int = 200, user: dict = Depends(verify_firebase_token)):
+    """Every manual admin action (review decisions, source toggles, run starts),
+    newest first — see backend/sql/admin_audit_schema.sql."""
+    _verify_admin(user)
+    db = get_supabase()
+    query = db.table("admin_audit_log").select("*").order("at", desc=True).limit(limit)
+    if kind:
+        query = query.eq("kind", kind)
+    return {"entries": query.execute().data}
+
+
 @router.get("/stats")
 def platform_stats(user: dict = Depends(verify_firebase_token)):
     _verify_admin(user)

routes/news.py

@@ -24,6 +24,7 @@ from news.pipeline import start_poll, execute_poll
 from news.classify import IMPACTS, CATEGORIES
 from news.business_types import valid_subset
 from utils.supabase_client import get_supabase
+from utils.audit import log_action
 
 router = APIRouter()
 
@@ -49,6 +50,7 @@ def poll_news(body: PollRequest, background_tasks: BackgroundTasks, user: dict =
     admin_uid = _verify_admin(user)
     poll_id = start_poll(since_days=body.since_days, triggered_by=admin_uid)
     background_tasks.add_task(execute_poll, poll_id, body.since_days)
+    log_action(admin_uid, "run", "started", f"Looking back {body.since_days} day(s)", target=poll_id)
     return {"poll_id": poll_id, "started": True}
 
 
@@ -95,9 +97,11 @@ class SourceUpdate(BaseModel):
 
 @router.patch("/admin/sources/{source_id}")
 def update_source(source_id: str, body: SourceUpdate, user: dict = Depends(verify_firebase_token)):
-    _verify_admin(user)
+    admin_uid = _verify_admin(user)
     db = get_supabase()
-    db.table("news_sources").update({"active": body.active}).eq("id", source_id).execute()
+    updated = db.table("news_sources").update({"active": body.active}).eq("id", source_id).execute().data
+    name = updated[0]["name"] if updated else source_id
+    log_action(admin_uid, "source", "enabled" if body.active else "disabled", name, target=source_id)
     return {"success": True}
 
 
@@ -107,7 +111,7 @@ def list_polls(user: dict = Depends(verify_firebase_token)):
     db = get_supabase()
     polls = (
         db.table("news_polls")
-        .select("id, triggered_by, since_days, started_at, finished_at, items_scraped, items_deduped_out, items_classified, items_flagged, items_rejected, error")
+        .select("id, triggered_by, since_days, started_at, finished_at, items_scraped, items_deduped_out, items_classified, items_flagged, items_rejected, current_source, error")
         .order("started_at", desc=True)
         .limit(10)
         .execute()
@@ -178,11 +182,13 @@ class ReviewFix(BaseModel):
 
 @router.patch("/admin/review/{item_id}")
 def fix_review_item(item_id: str, body: ReviewFix, user: dict = Depends(verify_firebase_token)):
-    _verify_admin(user)
+    admin_uid = _verify_admin(user)
     db = get_supabase()
 
     if body.status == "rejected":
-        db.table("news_items").update({"status": "rejected"}).eq("id", item_id).execute()
+        updated = db.table("news_items").update({"status": "rejected"}).eq("id", item_id).execute().data
+        title = updated[0]["title"] if updated else item_id
+        log_action(admin_uid, "review", "discarded", title, target=item_id)
         return {"success": True, "status": "rejected"}
 
     # Completing by hand still has to satisfy the same bar automated classification does.
@@ -195,7 +201,7 @@ def fix_review_item(item_id: str, body: ReviewFix, user: dict = Depends(verify_f
     if not (body.category and body.location and business_types):
         raise HTTPException(status_code=400, detail="category, location and at least one valid business_type are required to mark this classified.")
 
-    db.table("news_items").update({
+    updated = db.table("news_items").update({
         "status":         "classified",
         "impact":         body.impact,
         "category":       body.category,
@@ -203,6 +209,8 @@ def fix_review_item(item_id: str, body: ReviewFix, user: dict = Depends(verify_f
         "location":       body.location,
         "review_reason":  None,
         "classified_at":  datetime.now(timezone.utc).isoformat(),
-    }).eq("id", item_id).execute()
+    }).eq("id", item_id).execute().data
 
+    title = updated[0]["title"] if updated else item_id
+    log_action(admin_uid, "review", "published", title, target=item_id)
     return {"success": True, "status": "classified"}

sql/admin_audit_schema.sql

@@ -0,0 +1,19 @@
+-- Admin audit trail — every manual admin action (review decisions, source
+-- toggles, poll runs), newest first. Mirrors DataWiz-Project's /admin/audit.
+-- Automatic pipeline decisions (auto-publish, auto-discard) are NOT logged
+-- here — they're already visible per-item in news_items.status, and logging
+-- every one of them would make this table useless as a human-action trail.
+
+create extension if not exists pgcrypto;
+
+create table if not exists admin_audit_log (
+  id         uuid primary key default gen_random_uuid(),
+  at         timestamptz not null default now(),
+  actor      text not null,   -- admin's Firebase UID
+  kind       text not null check (kind in ('review', 'source', 'run')),
+  action     text not null,   -- e.g. 'published', 'discarded', 'enabled', 'disabled', 'started'
+  target     text,            -- the item/source/poll id this action was on
+  summary    text not null    -- short human-readable description, e.g. the item title
+);
+
+create index if not exists admin_audit_log_at_idx on admin_audit_log (at desc);

sql/news_schema.sql

@@ -47,8 +47,10 @@ create table if not exists news_items (
   status          text not null default 'pending'
                     check (status in ('pending', 'classified', 'needs_review', 'rejected')),
 
-  -- Classification output (ADR-0015) — all null until classified.
-  impact          text check (impact in ('threat', 'opportunity', 'neutral')),
+  -- Classification output (ADR-0015) — all null until classified. No 'threat' value:
+  -- reporting something bad that already happened has no value once it can't be acted
+  -- on; a predictive risk system is a separate, forward-looking feature.
+  impact          text check (impact in ('opportunity', 'neutral')),
   category        text check (category in ('policy', 'weather', 'infrastructure', 'funding', 'security', 'other')),
   business_types  text[],          -- subsector ids from data/sectors.ts, see backend/news/subsectors.json
   location        text,
@@ -86,7 +88,16 @@ create table if not exists news_polls (
   items_classified   int not null default 0,
   items_flagged      int not null default 0,
   items_rejected     int not null default 0,  -- auto-discarded: incomplete and the model wasn't confident either
+  current_source     text,  -- name of the source being scraped right now, for a live run console
   error              text
 );
 
 alter table news_polls add column if not exists items_rejected int not null default 0;
+alter table news_polls add column if not exists current_source text;
+
+-- 'threat' removed from impact (ADR-0015 addendum): a reactive "this already happened"
+-- label has no value to a business owner who can't act on it anymore. Existing rows
+-- are reclassified neutral; a predictive risk/forecast system is planned separately.
+update news_items set impact = 'neutral' where impact = 'threat';
+alter table news_items drop constraint if exists news_items_impact_check;
+alter table news_items add constraint news_items_impact_check check (impact in ('opportunity', 'neutral'));

tests/fixtures/groq/news_classify_complete.json

@@ -1,5 +1,5 @@
 {
-  "impact": "threat",
+  "impact": "neutral",
   "category": "weather",
   "business_types": ["catering_food_service", "food_production_processing"],
   "location": "Lagos",

tests/test_news.py

@@ -87,7 +87,7 @@ def test_classify_item_complete_result_is_complete(classify_module):
     result = classify_module.classify_item("Lagos flood warning", "text", fixture="news_classify_complete")
     assert result.is_complete
     assert result.review_reason() is None
-    assert result.impact == "threat"
+    assert result.impact == "neutral"
     assert result.category == "weather"
 
 
@@ -139,7 +139,7 @@ def test_classification_result_malformed_deadline_is_ignored():
 _SOME_VALID_ID = next(iter(VALID_IDS))
 
 
-def _result(impact="threat", category="weather", business_types=None, location="Lagos", confidence=None):
+def _result(impact="neutral", category="weather", business_types=None, location="Lagos", confidence=None):
     business_types = [_SOME_VALID_ID] if business_types is None else business_types
     return ClassificationResult({
         "impact": impact, "category": category,

tests/test_relevance.py

@@ -0,0 +1,35 @@
+"""Unit tests for the keyword pre-filter (news/relevance.py) that gates Groq calls."""
+
+from news.relevance import score
+
+
+def test_rejects_text_with_no_business_signal():
+    result = score("Bunk mates fall out over missing snacks at boarding school dormitory")
+    assert result["is_candidate"] is False
+    assert result["matched_keywords"] == []
+
+
+def test_rejects_isolated_incident_with_no_business_angle():
+    result = score("Aircraft loses contact mid-flight before making emergency landing")
+    assert result["is_candidate"] is False
+
+
+def test_accepts_policy_item():
+    result = score("CBN raises interest rate, forex dealers react")
+    assert result["is_candidate"] is True
+    assert "cbn" in result["matched_keywords"]
+
+
+def test_accepts_funding_item():
+    result = score("BOI opens new loan scheme for small manufacturers")
+    assert result["is_candidate"] is True
+
+
+def test_accepts_weather_item():
+    result = score("NiMet issues flood warning for Lagos ahead of heavy rainfall")
+    assert result["is_candidate"] is True
+
+
+def test_accepts_security_item_with_business_operational_angle():
+    result = score("Bandits attack causes market shutdown order in Zamfara")
+    assert result["is_candidate"] is True

utils/audit.py

@@ -0,0 +1,15 @@
+"""
+Writes to the admin audit trail (backend/sql/admin_audit_schema.sql). Never
+raises — a failed audit write must not break the action it's recording.
+"""
+
+from utils.supabase_client import get_supabase
+
+
+def log_action(actor: str, kind: str, action: str, summary: str, target: str | None = None) -> None:
+    try:
+        get_supabase().table("admin_audit_log").insert({
+            "actor": actor, "kind": kind, "action": action, "target": target, "summary": summary,
+        }).execute()
+    except Exception as e:
+        print(f"[audit] failed to log {kind}/{action}: {e}")