""" Orchestrates one admin-triggered poll (ADR-0015): scrape every active source, 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 per-source scrape failures. A single item failing to classify (Groq error, malformed response) is likewise skipped rather than aborting the run; it stays unclassified and will be retried on the next poll since it's never marked 'rejected'. """ 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 from intel.events import record_news_item # Three-way routing by confidence (replaces the old complete/incomplete split). # A complete, high-confidence item is business-relevant enough to publish with # no human step. A low-confidence item — incomplete AND the model wasn't sure # about it either — is almost always noise (celebrity gossip, a scraped nav # menu) and gets discarded automatically rather than wasting a reviewer's time. # Everything in between still needs a human: either it's complete but the # model wasn't confident, or it's incomplete but confident enough that a # person might be able to fill the gap. AUTO_PUBLISH_CONFIDENCE = 0.8 AUTO_DISCARD_CONFIDENCE = 0.5 def _route(result) -> str: confidence = result.confidence if result.confidence is not None else 0.0 if result.is_complete and confidence >= AUTO_PUBLISH_CONFIDENCE: return "classified" if not result.is_complete and confidence < AUTO_DISCARD_CONFIDENCE: return "rejected" return "needs_review" def start_poll(since_days: int, triggered_by: str) -> str: """Records the poll row and hands back its id immediately — the scrape+classify work for even a handful of sources takes minutes (Groq calls per item), which is well past most gateway timeouts (HF Spaces included). The caller runs execute_poll(poll_id, since_days) as a background task rather than awaiting it inline, so the HTTP response never has to wait on it.""" db = get_supabase() poll = db.table("news_polls").insert({ "triggered_by": triggered_by, "since_days": since_days, "started_at": datetime.now(timezone.utc).isoformat(), }).execute() return poll.data[0]["id"] def execute_poll(poll_id: str, since_days: int) -> dict: db = get_supabase() cutoff = datetime.now(timezone.utc) - timedelta(days=since_days) counts = {"items_scraped": 0, "items_deduped_out": 0, "items_classified": 0, "items_flagged": 0, "items_rejected": 0} error: str | None = None try: 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: print(f"[news] source '{source['name']}' failed to fetch: {e}") continue for item in raw_items: if item.published_at and item.published_at < cutoff: continue h = url_hash(item.url) existing = db.table("news_items").select("id").eq("url_hash", h).limit(1).execute().data if existing: counts["items_deduped_out"] += 1 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) except Exception as e: print(f"[news] classification failed for {item.url}: {e}") continue expires_at = compute_expiry(result.category, scraped_at, result.deadline) status = _route(result) row = { "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": status, "impact": result.impact, "category": result.category, "business_types": result.business_types, "location": result.location, "deadline": result.deadline.isoformat() if result.deadline else None, "expires_at": expires_at.isoformat(), "confidence": result.confidence, "review_reason": result.review_reason(), "classified_at": scraped_at.isoformat(), } db.table("news_items").insert(row).execute() if status == "classified": counts["items_classified"] += 1 # Published items also become permanent event history for the # intel store (no-op when it's off) — the feed expires them, the # correlation step must not lose them. record_news_item(row) elif status == "rejected": counts["items_rejected"] += 1 else: counts["items_flagged"] += 1 except Exception as e: error = str(e) db.table("news_polls").update({ "finished_at": datetime.now(timezone.utc).isoformat(), "current_source": None, "error": error, **counts, }).eq("id", poll_id).execute() return {"poll_id": poll_id, "error": error, **counts}