"""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\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, }