"""Deterministic natural-language parser for `/ask` (used when `companyatlas.services.llm.ask` is not available). "companies hiring AI engineers in Canada" → {event_types: [HIRING], ai: true, country: "CA", window: "30d"}. Only vocabulary from the taxonomy and the reference tables is recognised; anything else becomes free-text `terms` used for company search. Nothing is guessed: when no filter matches, the interpretation says so and the answer falls back to "no monitored evidence". """ from __future__ import annotations import re from dataclasses import asdict, dataclass, field from typing import Any from companyatlas.ids import normalize_alias WINDOW_PATTERNS: list[tuple[re.Pattern[str], str]] = [ (re.compile(r"\b(today|last 24 ?h(ours)?|past 24 ?h(ours)?)\b", re.IGNORECASE), "24h"), (re.compile(r"\b(yesterday)\b", re.IGNORECASE), "24h"), (re.compile(r"\b(this week|last (7|seven) days|past (7|seven) days|last week|weekly)\b", re.IGNORECASE), "7d"), (re.compile(r"\b(this month|last (30|thirty) days|past (30|thirty) days|last month|monthly|recently|recent)\b", re.IGNORECASE), "30d"), (re.compile(r"\b(this quarter|last (90|ninety) days|past (90|ninety) days|last quarter|quarterly|last (3|three) months)\b", re.IGNORECASE), "90d"), (re.compile(r"\b(this year|last (12|twelve) months|past year|last year|yearly|annual)\b", re.IGNORECASE), "1y"), ] EVENT_KEYWORDS: dict[str, tuple[str, ...]] = { "HIRING": ("hiring", "hire", "hires", "jobs", "job", "recruit", "recruiting", "openings", "positions", "careers", "vacanc", "headcount", "layoff", "layoffs"), "PRICING": ("pricing", "price", "prices", "plan", "plans", "tier", "tiers", "subscription", "cheaper", "expensive"), "PRODUCT": ("launch", "launched", "launches", "product", "products", "release", "released", "feature", "features", "shipped", "ship"), "LEADERSHIP": ("ceo", "cfo", "cto", "coo", "executive", "executives", "leadership", "appointed", "appoint", "board", "founder", "president", "chief"), "LOCATION": ("office", "offices", "location", "locations", "expansion", "expanding", "expand", "opened", "opening", "headquarters", "hq", "store", "stores", "factory"), "FINANCING": ("funding", "raised", "raise", "round", "series a", "series b", "investment", "investors", "ipo", "valuation"), "M&A": ("acquisition", "acquired", "acquire", "acquires", "merger", "merged", "buyout", "divest", "divestiture"), "PARTNERSHIP": ("partnership", "partner", "partners", "partnered", "alliance", "integration"), "DEVELOPER": ("api", "apis", "sdk", "developer", "developers", "docs", "documentation", "changelog"), "LEGAL": ("legal", "terms", "privacy", "policy", "compliance", "regulatory", "regulation", "gdpr"), "SECURITY": ("security", "breach", "incident", "vulnerability", "cve"), "TECHNOLOGY": ("technology", "tech stack", "adopted", "adoption", "platform"), "SUSTAINABILITY": ("sustainability", "esg", "climate", "carbon", "net zero"), "INVESTOR_RELATIONS": ("earnings", "investor", "quarterly results", "annual report", "guidance"), "COMMUNICATION": ("news", "press", "announcement", "announced", "blog", "published"), } AI_TERMS = ("ai", "a.i.", "artificial intelligence", "machine learning", "ml", "llm", "llms", "generative", "genai", "copilot", "agentic", "deep learning") RANKING_INTENT: list[tuple[re.Pattern[str], str]] = [ (re.compile(r"\b(most active|fastest moving|moving fastest|busiest)\b", re.IGNORECASE), "most_active"), (re.compile(r"\b(hiring (the )?(most|fastest)|fastest hiring|hiring growth|growing headcount)\b", re.IGNORECASE), "hiring_growth"), (re.compile(r"\b(hiring (decline|freeze|slowdown)|cutting|fewer jobs|shrinking)\b", re.IGNORECASE), "hiring_decline"), (re.compile(r"\b(product velocity|shipping (the )?most|most launches)\b", re.IGNORECASE), "product_velocity"), (re.compile(r"\b(most ai|ai[- ]active|ai adoption|ai leaders)\b", re.IGNORECASE), "ai_active"), (re.compile(r"\b(expanding (abroad|internationally|geographically)|geographic expansion|new countries)\b", re.IGNORECASE), "geo_expansion"), (re.compile(r"\b(developer momentum|developer[- ]focused)\b", re.IGNORECASE), "developer_momentum"), (re.compile(r"\b(pricing changes|changed (their )?pricing|price (increase|hike)s?)\b", re.IGNORECASE), "pricing_changes"), (re.compile(r"\b(unusual|anomal|abnormal|behaving (unusually|strangely))", re.IGNORECASE), "unusual_activity"), ] DEMONYMS: dict[str, str] = { "canadian": "CA", "american": "US", "us": "US", "usa": "US", "u.s.": "US", "british": "GB", "uk": "GB", "u.k.": "GB", "english": "GB", "french": "FR", "german": "DE", "japanese": "JP", "korean": "KR", "indian": "IN", "australian": "AU", "chinese": "CN", "brazilian": "BR", "mexican": "MX", "dutch": "NL", "swedish": "SE", "swiss": "CH", "spanish": "ES", "italian": "IT", "israeli": "IL", "singaporean": "SG", "irish": "IE", "european": None, # region, not a country } STOPWORDS = {"the", "a", "an", "of", "in", "on", "at", "for", "to", "and", "or", "with", "which", "what", "who", "are", "is", "was", "were", "has", "have", "had", "do", "does", "did", "show", "me", "list", "find", "companies", "company", "that", "this", "these", "those", "any", "all", "from", "by", "about", "their", "its", "new", "recently", "recent", "engineers", "engineer", "people", "roles", "many", "how", "much", "top", "best", "biggest", "largest", "most", "events", "event", "changes", "change", "signal", "signals", "atlas"} @dataclass class Interpretation: query: str window: str = "30d" event_types: list[str] = field(default_factory=list) country: str | None = None country_name: str | None = None industry: str | None = None industry_name: str | None = None ai: bool = False ranking_kind: str | None = None company_terms: list[str] = field(default_factory=list) terms: list[str] = field(default_factory=list) matched: list[str] = field(default_factory=list) def to_json(self) -> dict[str, Any]: d = asdict(self) d["filters"] = {k: v for k, v in {"event_types": self.event_types or None, "country": self.country, "industry": self.industry, "ai": self.ai or None, "window": self.window}.items() if v} return d def _contains(text: str, phrase: str) -> bool: return re.search(rf"(? Interpretation: """`countries`: [{code, name}], `industries`: [{slug, name, keywords}]. Deterministic, order-independent.""" text = " " + re.sub(r"\s+", " ", (query or "").strip().lower()) + " " it = Interpretation(query=query.strip()) consumed: list[str] = [] for pat, w in WINDOW_PATTERNS: m = pat.search(text) if m: it.window = w consumed.append(m.group(0)) it.matched.append(f"window:{w}") break for pat, kind in RANKING_INTENT: m = pat.search(text) if m: it.ranking_kind = kind consumed.append(m.group(0)) it.matched.append(f"ranking:{kind}") break if any(_contains(text, t) for t in AI_TERMS): it.ai = True it.matched.append("ai") consumed.extend(t for t in AI_TERMS if _contains(text, t)) for etype, words in EVENT_KEYWORDS.items(): hits = [w for w in words if _contains(text, w)] if hits: it.event_types.append(etype) consumed.extend(hits) it.matched.append(f"event_type:{etype}") by_len = sorted(countries, key=lambda c: -len(c.get("name") or "")) for c in by_len: name = (c.get("name") or "").lower() if name and _contains(text, name): it.country, it.country_name = c["code"], c["name"] consumed.append(name) it.matched.append(f"country:{c['code']}") break if it.country is None: codes = {c["code"].upper(): c for c in countries} for dem, code in DEMONYMS.items(): if code and _contains(text, dem) and code in codes: it.country, it.country_name = code, codes[code]["name"] consumed.append(dem) it.matched.append(f"country:{code}") break for ind in sorted(industries, key=lambda i: -len(i.get("name") or "")): candidates = [ind.get("name") or "", ind.get("slug", "").replace("-", " ")] + list(ind.get("keywords") or []) hit = next((c for c in candidates if c and len(c) >= 3 and _contains(text, c.lower())), None) if hit: it.industry, it.industry_name = ind["slug"], ind.get("name") or ind["slug"] consumed.append(hit.lower()) it.matched.append(f"industry:{ind['slug']}") break residual = text for phrase in sorted(set(consumed), key=len, reverse=True): residual = re.sub(rf"(? 1] it.terms = tokens[:8] it.company_terms = [t for t in tokens if len(t) >= 3][:4] return it def compose_answer(it: Interpretation, *, events_total: int, companies_count: int, sample_titles: list[str]) -> str: """Careful, non-fabricating summary sentence (spec §162–168).""" scope = [] if it.event_types: scope.append(" / ".join(it.event_types).lower() + " events") else: scope.append("events") if it.ai: scope.append("with observable AI signals") if it.industry_name: scope.append(f"in {it.industry_name}") if it.country_name: scope.append(f"in {it.country_name}") window = {"24h": "the last 24 hours", "7d": "the last 7 days", "30d": "the last 30 days", "90d": "the last 90 days", "1y": "the last 12 months"}[it.window] if events_total == 0 and companies_count == 0: return f"No monitored evidence matches this question yet ({' '.join(scope)}, {window}). Coverage grows as sensors observe more pages." parts = [f"Detected {events_total:,} {' '.join(scope)} across {companies_count:,} monitored compan{'y' if companies_count == 1 else 'ies'} in {window}."] if sample_titles: parts.append("Most recent: " + "; ".join(t[:90] for t in sample_titles[:3]) + ".") parts.append("Each event links to its public source; interpretations are signals, not verified facts.") return " ".join(parts) def alias_key(term: str) -> str: return normalize_alias(term) __all__ = ["EVENT_KEYWORDS", "Interpretation", "alias_key", "compose_answer", "interpret"]