Spaces:
Running
Running
| """ | |
| 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} | |