Spaces:
Sleeping
Sleeping
File size: 7,703 Bytes
1f5f31a a32af27 1f5f31a a32af27 1f5f31a b7c46f4 1f5f31a e740385 1f5f31a e469aff 1f5f31a e469aff 1f5f31a e469aff 1f5f31a e740385 1f5f31a a32af27 1f5f31a a32af27 1f5f31a e740385 1f5f31a b7c46f4 1f5f31a b7c46f4 1f5f31a b7c46f4 e740385 1f5f31a a32af27 1f5f31a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 | """
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}
|