"""Deterministic event generation (spec §21–23, §158, §167–168). changes (status='pending') ──▶ derive_events(): typed rules over `structured_delta` + block `diff` + surface ──▶ events (+ event_sources, clusters, dedupe keys, review_queue) ──▶ llm_jobs when useful Design: - `derive_events()` is a pure function (no I/O) so rules are unit-testable from fixtures. `process_pending_changes()` is the DB layer. - Wording is careful by construction: "detected", "listed", "no longer listed", "observed at". Never "fired", "laid off", "shut down". - Idempotent: `events.dedupe_key = sha(company, subtype, entity key, sensor, detection day)` → re-running a change is a no-op. - Importance = subtype default × f(significance, magnitude); confidence = evidence quality (ATS JSON 0.95 · JSON-LD 0.9 · HTML 0.8 · text-diff-only 0.7), corroboration handled by `services/clustering.py`. - Noise / minor changes never produce events. LLM enrichment is queued only for meaningful+ changes, within the daily budget. """ from __future__ import annotations import logging import re from dataclasses import dataclass, field from datetime import UTC, date, datetime from typing import Any from companyatlas.config import settings from companyatlas.db import execute, fetch_all, fetch_one, fetch_val, jsonb, transaction from companyatlas.ids import new_id, stable_hash from companyatlas.services.clustering import attach_to_cluster, normalize_entity_key from companyatlas.services.periodic import periodic from companyatlas.taxonomy import ( AI_KEYWORDS, EVENT_SUBTYPES, EVIDENCE_CONFIDENCE, FORBIDDEN_WORDING, ChangeKind, EventType, Surface, confidence_label, ) log = logging.getLogger(__name__) RULES_VERSION = "rules-v1" SCHEMA_VERSION = "event-v1" MAX_ENTITY_ITEMS = 50 # bounded entity lists on aggregate events PER_JOB_EVENT_MAX = 5 # NEW_JOB per job only when ≤ 5 jobs added PER_PERSON_EVENT_MAX = 10 NEWS_ITEMS_MAX = 20 NEWS_ITEMS_PER_EVENT_MAX = 3 # more items than this in one observation → one aggregate communication event LEADERSHIP_AGGREGATE_MIN = 3 SURGE_MIN_JOBS = 10 # fallback thresholds when no baseline is available yet SURGE_MIN_RATIO = 0.5 FREEZE_MIN_RATIO = 0.5 BASELINE_Z = 2.0 # surge/freeze when the delta exceeds mean + 2σ of the company's weekly baseline IMPORTANCE_FLOOR = 0.05 LEGAL_SURFACES = {Surface.LEGAL_TERMS, Surface.LEGAL_PRIVACY, Surface.SECURITY} DEVELOPER_SURFACES = {Surface.DOCS, Surface.DEVELOPER, Surface.API, Surface.CHANGELOG} NEWS_SURFACES = {Surface.NEWSROOM, Surface.BLOG, Surface.FEED, Surface.INVESTOR_RELATIONS, Surface.CHANGELOG, Surface.RESEARCH} AMBIGUOUS_SURFACES = {Surface.OTHER, Surface.PARTNERS, Surface.CUSTOMERS, Surface.SOLUTIONS, Surface.SERVICES, Surface.INDUSTRIES, Surface.SUPPORT, Surface.CONTACT, Surface.SITEMAP} SUMMARY_SUBTYPES = {"TERMS_CHANGE", "PRIVACY_POLICY_CHANGE", "SECURITY_UPDATE", "WEBSITE_CHANGE", "HOMEPAGE_REDESIGN", "NEWS_RELEASE", "INVESTOR_UPDATE", "EARNINGS_RELEASE", "MESSAGING_CHANGE"} ATS_CONNECTOR_HINTS = ("greenhouse", "lever", "ashby", "smartrecruiters", "workday", "workable", "json", "ats") SURFACE_LABEL: dict[str, str] = { Surface.CAREERS: "careers page", Surface.JOBS_BOARD: "job board", Surface.PRICING: "pricing page", Surface.LEADERSHIP: "leadership page", Surface.ABOUT: "about page", Surface.PRODUCTS: "products page", Surface.SERVICES: "services page", Surface.SOLUTIONS: "solutions page", Surface.LOCATIONS: "locations page", Surface.CONTACT: "contact page", Surface.NEWSROOM: "newsroom", Surface.BLOG: "blog", Surface.FEED: "feed", Surface.DOCS: "documentation", Surface.DEVELOPER: "developer portal", Surface.API: "API reference", Surface.CHANGELOG: "changelog", Surface.INVESTOR_RELATIONS: "investor relations page", Surface.LEGAL_TERMS: "terms of service page", Surface.LEGAL_PRIVACY: "privacy policy", Surface.SECURITY: "security page", Surface.HOMEPAGE: "homepage", Surface.SUSTAINABILITY: "sustainability page", Surface.STATUS: "status page", Surface.PARTNERS: "partners page", Surface.CUSTOMERS: "customers page", Surface.INDUSTRIES: "industries page", Surface.RESEARCH: "research page", Surface.SUPPORT: "support page", Surface.SITEMAP: "sitemap", Surface.OTHER: "monitored page", } CURRENCY_SYMBOL = {"USD": "$", "EUR": "€", "GBP": "£", "CAD": "CA$", "AUD": "A$", "JPY": "¥", "CHF": "CHF ", "INR": "₹", "BRL": "R$"} PERIOD_LABEL = {"month": "per month", "year": "per year", "one_time": "one-time", "usage": "usage-based", "contact": "contact sales"} EXEC_ROLES = {"ceo", "cfo", "cto", "coo", "founder", "president", "chair", "board", "cmo", "cro", "cpo", "ciso", "cio", "chro", "gm"} _EARNINGS_RE = re.compile(r"\b(earnings|quarterly results|q[1-4]\s*(fy)?\s*20\d\d|fiscal (year|quarter)|financial results|results for the (quarter|year))\b", re.IGNORECASE) _INVESTOR_RE = re.compile(r"\b(investor|shareholder|annual report|annual meeting|dividend|10-k|10-q|8-k|proxy statement|guidance)\b", re.IGNORECASE) _PRESS_RE = re.compile(r"\b(announces|announced|unveils|introduces|launches|partners with|acquires|appoints|names|expands|opens)\b", re.IGNORECASE) _LAUNCH_RE = re.compile(r"\b(launch(es|ed|ing)?|introduc(es|ed|ing)|unveil(s|ed)|now available|general availability)\b", re.IGNORECASE) _ACQ_RE = re.compile(r"\b(acquires|acquired|acquisition of|to acquire|merger|merges with)\b", re.IGNORECASE) _FUNDING_RE = re.compile(r"\b(raises|raised|series [a-f]\b|seed round|funding round|closes \$|financing of)\b", re.IGNORECASE) # ============================================================================================================== drafts @dataclass(slots=True) class EventDraft: subtype: str title: str entity_key: str summary: str | None = None old_value: str | None = None new_value: str | None = None entities: dict[str, Any] = field(default_factory=dict) payload: dict[str, Any] = field(default_factory=dict) tags: list[str] = field(default_factory=list) magnitude: float = 0.0 # 0–1 rule-specific magnitude (relative job delta, |price pct| …) → importance bonus evidence: str = "html" # ats_json | jsonld | html | text_diff effective_at: datetime | None = None published_at: datetime | None = None review: str | None = None # review_queue kind when the rule itself wants a human look importance: float = 0.0 # filled by finalize() confidence: float = 0.0 @property def event_type(self) -> str: return str(EVENT_SUBTYPES.get(self.subtype, (EventType.OTHER, 0.3))[0]) @dataclass(slots=True) class Derived: events: list[EventDraft] needs_classification: bool = False classification_reason: str | None = None summarize: list[str] = field(default_factory=list) # subtypes whose events deserve an LLM summary # ============================================================================================================== helpers def safe_wording(text: str) -> str: """Defensive: rewrite forbidden phrasing (rules never produce it, LLM output might).""" out = text for bad in FORBIDDEN_WORDING: if bad in out.lower(): out = re.sub(re.escape(bad), "no longer listed", out, flags=re.IGNORECASE) return out def scale_importance(default: float, significance: float, magnitude: float = 0.0) -> float: """importance = default × (0.7 + 0.6·significance) × (1 + 0.3·magnitude), clamped to [0.05, 1].""" sig = min(1.0, max(0.0, float(significance or 0.0))) mag = min(1.0, max(0.0, float(magnitude or 0.0))) return round(min(1.0, max(IMPORTANCE_FLOOR, default * (0.7 + 0.6 * sig) * (1.0 + 0.3 * mag))), 4) def evidence_kind(sensor: dict[str, Any], delta: dict[str, Any]) -> str: connector = (sensor.get("connector_id") or "").lower() surface = sensor.get("surface") or "" meta = delta.get("meta") or {} hinted = meta.get("evidence") if hinted in EVIDENCE_CONFIDENCE: return str(hinted) if surface == Surface.JOBS_BOARD or any(h in connector for h in ATS_CONNECTOR_HINTS) or sensor.get("fetch_mode") == "json": return "ats_json" if meta.get("jsonld") or "jsonld" in connector or "feed" in connector or surface == Surface.FEED: return "jsonld" if any(delta.get(k) for k in ("jobs", "people", "products", "plans", "locations", "news")): return "html" return "text_diff" def _label(surface: str) -> str: return SURFACE_LABEL.get(surface, "monitored page") def _plural(n: int, one: str, many: str | None = None) -> str: return one if n == 1 else (many or one + "s") def _money(amount: Any, currency: str | None) -> str: try: value = float(amount) except (TypeError, ValueError): return str(amount) text = f"{value:,.0f}" if value.is_integer() else f"{value:,.2f}" cur = (currency or "").upper() sym = CURRENCY_SYMBOL.get(cur) if sym: return f"{sym}{text}" return f"{text} {cur}".strip() def _place(item: dict[str, Any]) -> str: parts = [p for p in (item.get("city"), item.get("region")) if p] country = item.get("country") if country: parts.append(str(country).upper()) if parts: return ", ".join(parts) return item.get("name") or item.get("location_text") or "" def _is_ai(text: str | None) -> bool: if not text: return False hay = f" {text.lower()} " return any(k in hay for k in AI_KEYWORDS) def _job_is_ai(job: dict[str, Any]) -> bool: return bool(job.get("is_ai")) or _is_ai(job.get("title")) def _job_label(job: dict[str, Any]) -> str: title = (job.get("title") or "position").strip() loc = job.get("location_text") or _place(job) if job.get("remote") and not loc: loc = "Remote" return f"{title} ({loc})" if loc else title def _sections(diff: dict[str, Any]) -> list[str]: seen: list[str] = [] for bucket in ("modified", "added", "removed"): for d in diff.get(bucket) or []: path = (d.get("path") or "").strip() name = path.split(">")[-1].strip() if path else "" if not name: text = (d.get("after") or d.get("before") or "").strip() name = text.split("\n")[0][:80] if text else "" if name and name not in seen: seen.append(name) return seen[:20] def _blocks_changed(diff: dict[str, Any], change: dict[str, Any]) -> int: counts = diff.get("counts") or {} n = int(counts.get("added") or 0) + int(counts.get("removed") or 0) + int(counts.get("modified") or 0) if n == 0: n = int(change.get("blocks_added") or 0) + int(change.get("blocks_removed") or 0) + int(change.get("blocks_modified") or 0) return n def _dt(value: Any) -> datetime | None: if value is None: return None if isinstance(value, datetime): return value if value.tzinfo else value.replace(tzinfo=UTC) try: return datetime.fromisoformat(str(value)) except ValueError: return None # ============================================================================================================== rule families def _hiring_rules(delta: dict[str, Any], surface: str, evidence: str, baseline: dict[str, Any] | None) -> list[EventDraft]: jobs = delta.get("jobs") or {} added: list[dict[str, Any]] = list(jobs.get("added") or []) removed: list[dict[str, Any]] = list(jobs.get("removed") or []) open_before = jobs.get("open_before") open_after = jobs.get("open_after") n_add, n_rem = len(added), len(removed) if not (n_add or n_rem): return [] if isinstance(open_before, int) and isinstance(open_after, int): net = open_after - open_before else: net = n_add - n_rem label = _label(surface) out: list[EventDraft] = [] counts_key = f"{open_before}>{open_after}" if open_before is not None else f"+{n_add}-{n_rem}" base_denominator = max(int(open_before or 0), 5) countries = sorted({str(j.get("country")).upper() for j in added if j.get("country")}) departments = sorted({str(j.get("department")) for j in added if j.get("department")})[:20] ai_added = [j for j in added if _job_is_ai(j)] common_payload = {"added": n_add, "removed": n_rem, "open_before": open_before, "open_after": open_after, "net": net, "ai_added": len(ai_added), "countries": countries, "departments": departments} if net > 0 and n_add: title = f"{n_add} new {_plural(n_add, 'position')} detected on {label}" if n_rem: title += f" ({n_rem} no longer visible)" out.append(EventDraft( subtype="JOB_COUNT_INCREASE", title=title, entity_key=f"jobs:{counts_key}", evidence=evidence, summary=_open_summary(open_before, open_after, n_add, n_rem), old_value=_s(open_before), new_value=_s(open_after), entities={"jobs": [_job_entity(j) for j in added[:MAX_ENTITY_ITEMS]]}, payload=common_payload, tags=_hiring_tags(countries, ai_added), magnitude=min(1.0, n_add / base_denominator))) elif net < 0 and n_rem: title = f"{n_rem} monitored job {_plural(n_rem, 'listing')} no longer visible on {label}" if n_add: title += f" ({n_add} new)" out.append(EventDraft( subtype="JOB_COUNT_DECREASE", title=title, entity_key=f"jobs:{counts_key}", evidence=evidence, summary=_open_summary(open_before, open_after, n_add, n_rem), old_value=_s(open_before), new_value=_s(open_after), entities={"jobs": [_job_entity(j) for j in removed[:MAX_ENTITY_ITEMS]]}, payload=common_payload, tags=["hiring"], magnitude=min(1.0, n_rem / base_denominator))) if ai_added: k = len(ai_added) out.append(EventDraft( subtype="AI_HIRING", title=f"{k} AI-related {_plural(k, 'position')} detected on {label}", entity_key=f"ai_jobs:{counts_key}", evidence=evidence, summary="AI-related roles identified from listing titles: " + "; ".join(_job_label(j) for j in ai_added[:5]), entities={"jobs": [_job_entity(j) for j in ai_added[:MAX_ENTITY_ITEMS]]}, payload={"ai_added": k, "added": n_add}, tags=["hiring", "ai"], magnitude=min(1.0, k / 5))) if 1 <= n_add <= PER_JOB_EVENT_MAX: for j in added: out.append(EventDraft( subtype="NEW_JOB", title=f"New position listed: {_job_label(j)}", entity_key="job:" + normalize_entity_key(_job_label(j)), evidence=evidence, new_value=j.get("title"), entities={"jobs": [_job_entity(j)]}, payload={"url": j.get("url"), "department": j.get("department"), "country": j.get("country"), "remote": j.get("remote")}, tags=["hiring"] + (["ai"] if _job_is_ai(j) else []), published_at=_dt(j.get("posted_at")))) surge, freeze = _surge_freeze(n_add, n_rem, open_before, open_after, baseline) if surge is not None: out.append(EventDraft( subtype="HIRING_SURGE", title=f"Hiring surge signal: {n_add} new positions detected in one observation" + surge, entity_key=f"surge:{counts_key}", evidence=evidence, payload={**common_payload, "baseline": baseline}, summary="Signal, not a fact: the number of new listings exceeds this company's usual weekly volume.", tags=["hiring", "signal"], magnitude=min(1.0, n_add / max(base_denominator, SURGE_MIN_JOBS)))) if freeze is not None: out.append(EventDraft( subtype="HIRING_FREEZE_SIGNAL", title=f"Hiring slowdown signal: {n_rem} of {open_before if open_before is not None else n_rem + (open_after or 0)} monitored listings no longer visible" + freeze, entity_key=f"freeze:{counts_key}", evidence=evidence, payload={**common_payload, "baseline": baseline}, summary="Signal, not a fact: listings disappearing from a public careers page can reflect closed roles, ATS migrations or page changes.", tags=["hiring", "signal"], review="unexpected_activity", magnitude=min(1.0, n_rem / base_denominator))) return out def _surge_freeze(n_add: int, n_rem: int, open_before: Any, open_after: Any, baseline: dict[str, Any] | None) -> tuple[str | None, str | None]: surge = freeze = None b = (baseline or {}).get("jobs_new_weekly") before = int(open_before) if isinstance(open_before, int) else None if b and b.get("samples", 0) >= 4 and b.get("stddev") is not None: threshold = float(b["mean"]) + BASELINE_Z * max(float(b["stddev"]), 1.0) if n_add >= max(5, threshold): surge = f" (baseline ≈ {float(b['mean']):.1f} new/week)" elif n_add >= SURGE_MIN_JOBS and (before is None or n_add >= SURGE_MIN_RATIO * before): surge = "" if before and n_rem >= SURGE_MIN_JOBS and n_rem >= FREEZE_MIN_RATIO * before and (open_after is None or int(open_after) <= (1 - FREEZE_MIN_RATIO) * before): freeze = "" return surge, freeze def _open_summary(before: Any, after: Any, n_add: int, n_rem: int) -> str: parts = [f"{n_add} added" if n_add else "", f"{n_rem} no longer visible" if n_rem else ""] s = ", ".join(p for p in parts if p) if before is not None and after is not None: s += f"; open listings observed: {before} → {after}" return s + "." def _hiring_tags(countries: list[str], ai_added: list[dict[str, Any]]) -> list[str]: tags = ["hiring"] + [f"country:{c}" for c in countries[:5]] if ai_added: tags.append("ai") return tags def _job_entity(j: dict[str, Any]) -> dict[str, Any]: return {k: j.get(k) for k in ("title", "url", "location_text", "country", "remote", "department", "is_ai") if j.get(k) is not None} def _s(v: Any) -> str | None: return None if v is None else str(v) def _pricing_rules(delta: dict[str, Any], surface: str, evidence: str, diff: dict[str, Any], change: dict[str, Any]) -> list[EventDraft]: plans = delta.get("plans") or {} out: list[EventDraft] = [] for p in plans.get("price_changed") or []: name = p.get("plan_name") or "Plan" before, after = p.get("before"), p.get("after") try: b, a = float(before), float(after) except (TypeError, ValueError): continue if a == b: continue pct = p.get("pct") if pct is None and b: pct = round((a - b) / b * 100, 1) cur = p.get("currency") period = PERIOD_LABEL.get(p.get("billing_period") or "", "") subtype = "PRICE_INCREASE" if a > b else "PRICE_DECREASE" title = f"{name} plan price observed at {_money(a, cur)} (was {_money(b, cur)})" summary = f"{'Increase' if a > b else 'Decrease'} of {abs(pct):.1f}%" if pct is not None else None if summary and period: summary += f", billed {period}" out.append(EventDraft( subtype=subtype, title=title, entity_key="plan:" + normalize_entity_key(name), evidence=evidence, summary=(summary + "." if summary else None), old_value=_money(b, cur), new_value=_money(a, cur), payload={"plan_name": name, "before": b, "after": a, "pct": pct, "currency": cur, "billing_period": p.get("billing_period")}, entities={"plans": [{"plan_name": name}]}, tags=["pricing"], magnitude=min(1.0, abs(float(pct or 0)) / 50.0))) for p in plans.get("added") or []: name = p.get("plan_name") or "New plan" price = p.get("price") cur = p.get("currency") if p.get("contact_sales") or (price is None and (p.get("billing_period") == "contact")): price_txt = "contact sales" tags = ["pricing", "enterprise"] elif price is not None: price_txt = _money(price, cur) + (f" {PERIOD_LABEL.get(p.get('billing_period') or '', '')}".rstrip()) tags = ["pricing"] else: price_txt = p.get("price_text") or "price not stated" tags = ["pricing"] out.append(EventDraft( subtype="NEW_PRICING_TIER", title=f"New pricing tier listed: {name} ({price_txt})", entity_key="plan:" + normalize_entity_key(name), evidence=evidence, new_value=price_txt, payload={"plan_name": name, "price": price, "currency": cur, "billing_period": p.get("billing_period"), "contact_sales": bool(p.get("contact_sales"))}, entities={"plans": [{"plan_name": name}]}, tags=tags, magnitude=0.3)) for p in plans.get("removed") or []: name = p.get("plan_name") or "Plan" out.append(EventDraft( subtype="PRICING_TIER_REMOVED", title=f"Pricing tier no longer listed: {name}", entity_key="plan:" + normalize_entity_key(name), evidence=evidence, old_value=name, payload={"plan_name": name, "price": p.get("price"), "currency": p.get("currency")}, entities={"plans": [{"plan_name": name}]}, tags=["pricing"], magnitude=0.3)) if not out and surface == Surface.PRICING: n = _blocks_changed(diff, change) sections = _sections(diff) out.append(EventDraft( subtype="PRICING_CHANGE", title=f"Pricing page materially updated ({n} {_plural(n, 'block')} changed)", entity_key="pricing_page", evidence="text_diff", payload={"blocks_changed": n, "sections": sections, "text_delta_ratio": diff.get("text_delta_ratio")}, summary=("Sections affected: " + ", ".join(sections[:6]) + ".") if sections else None, tags=["pricing"], magnitude=min(1.0, float(diff.get("text_delta_ratio") or 0) * 2))) return out def _leadership_rules(delta: dict[str, Any], surface: str, evidence: str) -> list[EventDraft]: people = delta.get("people") or {} added = list(people.get("added") or []) removed = list(people.get("removed") or []) changed = list(people.get("title_changed") or []) if not (added or removed or changed): return [] label = _label(surface) if surface in (Surface.LEADERSHIP, Surface.ABOUT) else "monitored leadership page" out: list[EventDraft] = [] def is_exec(p: dict[str, Any]) -> bool: return bool(p.get("is_executive")) or (p.get("role_category") or "").lower() in EXEC_ROLES for p in [x for x in added if is_exec(x)][:PER_PERSON_EVENT_MAX]: name, title = p.get("name") or "Unnamed", p.get("title") out.append(EventDraft( subtype="NEW_EXECUTIVE", title=f"{name} listed as {title} on {label}" if title else f"{name} newly listed on {label}", entity_key="person:" + normalize_entity_key(name), evidence=evidence, new_value=title, entities={"people": [{"name": name, "title": title, "role_category": p.get("role_category")}]}, payload={"role_category": p.get("role_category")}, tags=["leadership"], magnitude=0.5 if (p.get("role_category") or "").lower() in {"ceo", "cfo", "cto", "coo", "president"} else 0.2)) for p in [x for x in removed if is_exec(x)][:PER_PERSON_EVENT_MAX]: name, title = p.get("name") or "Unnamed", p.get("title") out.append(EventDraft( subtype="EXECUTIVE_NO_LONGER_LISTED", title=f"{name} no longer listed on {label}", entity_key="person:" + normalize_entity_key(name), evidence=evidence, old_value=title, summary=f"Previously listed as {title}. Disappearance from a public page is not evidence of departure." if title else None, entities={"people": [{"name": name, "title": title, "role_category": p.get("role_category")}]}, payload={"role_category": p.get("role_category")}, tags=["leadership"], magnitude=0.5 if (p.get("role_category") or "").lower() in {"ceo", "cfo", "cto", "coo", "president"} else 0.2)) for p in changed[:PER_PERSON_EVENT_MAX]: name = p.get("name") or "Unnamed" before, after = p.get("before"), p.get("after") out.append(EventDraft( subtype="EXECUTIVE_TITLE_CHANGE", title=f"{name} now listed as {after} (was {before})", entity_key="person:" + normalize_entity_key(name), evidence=evidence, old_value=before, new_value=after, entities={"people": [{"name": name, "title": after}]}, tags=["leadership"], magnitude=0.3)) total = len(added) + len(removed) + len(changed) if total >= LEADERSHIP_AGGREGATE_MIN or (total and not out): bits = [f"{len(added)} added" if added else "", f"{len(removed)} no longer listed" if removed else "", f"{len(changed)} title {_plural(len(changed), 'change')}" if changed else ""] out.append(EventDraft( subtype="LEADERSHIP_CHANGE", title=f"{label[0].upper()}{label[1:]} updated: " + ", ".join(b for b in bits if b), entity_key=f"leadership:{len(added)}:{len(removed)}:{len(changed)}", evidence=evidence, entities={"people": [{"name": p.get("name"), "title": p.get("title"), "status": s} for s, lst in (("listed", added), ("no_longer_listed", removed)) for p in lst][:MAX_ENTITY_ITEMS]}, payload={"added": len(added), "removed": len(removed), "title_changed": len(changed)}, tags=["leadership"], magnitude=min(1.0, total / 6))) return out def _product_rules(delta: dict[str, Any], evidence: str) -> list[EventDraft]: products = delta.get("products") or {} out: list[EventDraft] = [] for p in list(products.get("added") or [])[:MAX_ENTITY_ITEMS]: name = p.get("name") or "Unnamed product" out.append(EventDraft( subtype="NEW_PRODUCT", title=f"New product listed: {name}", entity_key="product:" + normalize_entity_key(name), evidence=evidence, new_value=name, entities={"products": [{"name": name, "url": p.get("url")}]}, payload={"url": p.get("url"), "category": p.get("category")}, tags=["product"] + (["ai"] if _is_ai(name) else []), magnitude=0.3)) for p in list(products.get("removed") or [])[:MAX_ENTITY_ITEMS]: name = p.get("name") or "Unnamed product" out.append(EventDraft( subtype="PRODUCT_REMOVED", title=f"Product no longer listed: {name}", entity_key="product:" + normalize_entity_key(name), evidence=evidence, old_value=name, entities={"products": [{"name": name, "url": p.get("url")}]}, payload={"url": p.get("url")}, tags=["product"], magnitude=0.3)) return out def _location_rules(delta: dict[str, Any], evidence: str, country_names: dict[str, str]) -> list[EventDraft]: locations = delta.get("locations") or {} out: list[EventDraft] = [] kind_label = {"headquarters": "headquarters", "office": "office", "store": "store", "factory": "factory", "warehouse": "warehouse", "lab": "lab", "data_center": "data center"} for loc in list(locations.get("added") or [])[:MAX_ENTITY_ITEMS]: place = _place(loc) or "unnamed location" kind = kind_label.get(loc.get("kind") or "", "location") out.append(EventDraft( subtype="NEW_LOCATION", title=f"New {kind} listed: {place}", entity_key="location:" + normalize_entity_key(place), evidence=evidence, new_value=place, entities={"locations": [{"name": loc.get("name"), "city": loc.get("city"), "country": loc.get("country"), "kind": loc.get("kind")}]}, payload={"kind": loc.get("kind"), "country": loc.get("country")}, tags=["location"] + ([f"country:{str(loc['country']).upper()}"] if loc.get("country") else []), magnitude=0.4 if loc.get("kind") == "headquarters" else 0.2)) for loc in list(locations.get("removed") or [])[:MAX_ENTITY_ITEMS]: place = _place(loc) or "unnamed location" kind = kind_label.get(loc.get("kind") or "", "location") out.append(EventDraft( subtype="OFFICE_REMOVED", title=f"{kind[0].upper()}{kind[1:]} no longer listed: {place}", entity_key="location:" + normalize_entity_key(place), evidence=evidence, old_value=place, entities={"locations": [{"name": loc.get("name"), "city": loc.get("city"), "country": loc.get("country"), "kind": loc.get("kind")}]}, payload={"kind": loc.get("kind"), "country": loc.get("country")}, tags=["location"], magnitude=0.2)) for code in locations.get("new_countries") or []: code = str(code).upper() name = country_names.get(code, code) cities = [loc.get("city") for loc in locations.get("added") or [] if str(loc.get("country") or "").upper() == code and loc.get("city")] detail = f" ({', '.join(cities[:3])})" if cities else "" out.append(EventDraft( subtype="COUNTRY_EXPANSION", title=f"New country presence listed: {name}{detail}", entity_key=f"country:{code}", evidence=evidence, new_value=code, entities={"locations": [{"country": code, "city": c} for c in cities[:10]] or [{"country": code}]}, payload={"country": code, "cities": cities[:10]}, tags=["location", "expansion", f"country:{code}"], magnitude=0.6)) return out def _news_subtype(item: dict[str, Any], surface: str) -> str: title = item.get("title") or "" category = (item.get("category") or "").lower() if _EARNINGS_RE.search(title): return "EARNINGS_RELEASE" if category == "ir" or surface == Surface.INVESTOR_RELATIONS or _INVESTOR_RE.search(title): return "INVESTOR_UPDATE" if category == "changelog" or surface == Surface.CHANGELOG: return "CHANGELOG_ENTRY" if category == "press" or surface == Surface.NEWSROOM: return "NEWS_RELEASE" if category in ("blog", "research") or surface in (Surface.BLOG, Surface.RESEARCH): return "BLOG_POST" return "NEWS_RELEASE" if _PRESS_RE.search(title) else "BLOG_POST" def _news_rules(delta: dict[str, Any], surface: str, evidence: str) -> list[EventDraft]: news = delta.get("news") or {} prefix = {"NEWS_RELEASE": "News release", "BLOG_POST": "Blog post", "CHANGELOG_ENTRY": "Changelog entry", "INVESTOR_UPDATE": "Investor update", "EARNINGS_RELEASE": "Earnings release"} out: list[EventDraft] = [] added = [i for i in list(news.get("added") or []) if (i.get("title") or "").strip()] if len(added) > NEWS_ITEMS_PER_EVENT_MAX: # A burst of items in one observation (catalogue re-listing, archive page, many posts at once) is one communication event, # not a flood: individual titles stay in `entities.news` for the evidence drawer. subtypes = [_news_subtype(i, surface) for i in added] subtype = max(set(subtypes), key=subtypes.count) label = {"NEWS_RELEASE": "news releases", "BLOG_POST": "blog posts", "CHANGELOG_ENTRY": "changelog entries", "INVESTOR_UPDATE": "investor updates", "EARNINGS_RELEASE": "earnings releases"}[subtype] titles = [(i.get("title") or "").strip() for i in added] tags = ["news", subtype.lower()] if any(_is_ai(t) for t in titles): tags.append("ai") if any(_LAUNCH_RE.search(t) for t in titles): tags.append("launch") out.append(EventDraft( subtype=subtype, title=f"{len(added)} new {label} published on {_label(surface)}", entity_key=f"news_batch:{normalize_entity_key(titles[0])}:{len(added)}", evidence=evidence, summary="Latest: " + " · ".join(t[:80] for t in titles[:3]) + (" …" if len(titles) > 3 else ""), entities={"news": [{"title": i.get("title"), "url": i.get("url"), "published_at": i.get("published_at")} for i in added[:NEWS_ITEMS_MAX]]}, payload={"count": len(added), "category": added[0].get("category")}, tags=tags, magnitude=min(1.0, len(added) / 20))) return out for item in added[:NEWS_ITEMS_MAX]: title = (item.get("title") or "").strip() subtype = _news_subtype(item, surface) tags = ["news", subtype.lower()] if _is_ai(title): tags.append("ai") if _LAUNCH_RE.search(title): tags.append("launch") if _ACQ_RE.search(title): tags.append("m&a-mention") if _FUNDING_RE.search(title): tags.append("financing-mention") published = _dt(item.get("published_at")) out.append(EventDraft( subtype=subtype, title=f"{prefix[subtype]}: {title[:110]}", entity_key="news:" + normalize_entity_key(title), evidence=evidence, summary=(item.get("summary") or None), new_value=item.get("url"), entities={"news": [{"title": title, "url": item.get("url"), "published_at": item.get("published_at")}]}, payload={"url": item.get("url"), "category": item.get("category"), "published_at": item.get("published_at")}, tags=tags, published_at=published, effective_at=published, magnitude=0.5 if "launch" in tags else 0.1)) return out def _text_diff_rules(surface: str, change: dict[str, Any], diff: dict[str, Any], delta: dict[str, Any]) -> list[EventDraft]: """Meaningful+ diffs on surfaces without typed extractions.""" n = _blocks_changed(diff, change) sections = _sections(diff) ratio = float(diff.get("text_delta_ratio") or change.get("text_delta_ratio") or 0) kind = change.get("kind") or "" sec_txt = f"{len(sections)} {_plural(len(sections), 'section')} changed" if sections else f"{n} {_plural(n, 'block')} changed" payload = {"blocks_changed": n, "sections": sections, "text_delta_ratio": round(ratio, 4), "similarity": diff.get("similarity")} summary = ("Sections affected: " + ", ".join(sections[:8]) + ".") if sections else None mag = min(1.0, ratio * 2) ek = f"page:{change.get('sensor_id')}" if surface == Surface.LEGAL_TERMS: return [EventDraft("TERMS_CHANGE", f"Terms of service page materially updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["legal", "terms"], magnitude=mag, evidence="text_diff", review="legal_sensitive")] if surface == Surface.LEGAL_PRIVACY: return [EventDraft("PRIVACY_POLICY_CHANGE", f"Privacy policy materially updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["legal", "privacy"], magnitude=mag, evidence="text_diff", review="legal_sensitive")] if surface == Surface.SECURITY: return [EventDraft("SECURITY_UPDATE", f"Security page updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["security"], magnitude=mag, evidence="text_diff")] if surface == Surface.HOMEPAGE: out: list[EventDraft] = [] meta = delta.get("meta") or {} tc = meta.get("title_changed") if isinstance(tc, dict) and tc.get("before") and tc.get("after") and tc["before"] != tc["after"]: out.append(EventDraft("MESSAGING_CHANGE", f"Homepage title observed as “{str(tc['after'])[:60]}” (was “{str(tc['before'])[:60]}”)", "homepage:title", old_value=tc["before"], new_value=tc["after"], payload={"field": "title"}, tags=["messaging"], magnitude=0.4, evidence="html")) if kind in (ChangeKind.MAJOR, ChangeKind.CRITICAL): out.append(EventDraft("HOMEPAGE_REDESIGN", f"Homepage materially redesigned ({n} {_plural(n, 'block')} changed, {ratio:.0%} of text)", ek, summary=summary, payload=payload, tags=["website", "homepage"], magnitude=mag, evidence="text_diff")) else: out.append(EventDraft("WEBSITE_CHANGE", f"Homepage content updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["website", "homepage"], magnitude=mag, evidence="text_diff")) return out if surface == Surface.ABOUT: return [EventDraft("WEBSITE_CHANGE", f"About page updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["website", "about"], magnitude=mag, evidence="text_diff")] if surface == Surface.API: return [EventDraft("API_CHANGE", f"API reference updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["developer", "api"], magnitude=mag, evidence="text_diff")] if surface in (Surface.DOCS, Surface.DEVELOPER): return [EventDraft("DOC_CHANGE", f"Documentation updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["developer", "docs"], magnitude=mag, evidence="text_diff")] if surface == Surface.CHANGELOG: first = next((d for d in diff.get("added") or [] if (d.get("after") or "").strip()), None) head = (first["after"].strip().split("\n")[0][:90]) if first else None title = f"Changelog entry detected: {head}" if head else f"Changelog updated ({sec_txt})" return [EventDraft("CHANGELOG_ENTRY", title, "changelog:" + normalize_entity_key(head or ek), summary=summary, payload=payload, tags=["developer", "changelog"], magnitude=mag, evidence="text_diff")] if surface == Surface.PRICING: return [] # handled by _pricing_rules (generic PRICING_CHANGE) if surface in (Surface.PRODUCTS, Surface.SERVICES, Surface.SOLUTIONS): return [EventDraft("PRODUCT_UPDATE", f"{_label(surface)[0].upper()}{_label(surface)[1:]} updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["product"], magnitude=mag, evidence="text_diff")] if surface == Surface.INVESTOR_RELATIONS: return [EventDraft("INVESTOR_UPDATE", f"Investor relations page updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["investor-relations"], magnitude=mag, evidence="text_diff")] if surface == Surface.SUSTAINABILITY: return [EventDraft("SUSTAINABILITY_UPDATE", f"Sustainability page updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["sustainability"], magnitude=mag, evidence="text_diff")] if surface == Surface.STATUS: return [EventDraft("OPERATIONS_UPDATE", f"Status page updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["operations"], magnitude=mag, evidence="text_diff")] if surface in (Surface.CAREERS, Surface.JOBS_BOARD): return [EventDraft("WEBSITE_CHANGE", f"Careers page updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["website", "careers"], magnitude=mag, evidence="text_diff")] if surface == Surface.LEADERSHIP: return [EventDraft("LEADERSHIP_CHANGE", f"Leadership page updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["leadership"], magnitude=mag, evidence="text_diff", review="low_confidence")] if surface == Surface.LOCATIONS: return [EventDraft("WEBSITE_CHANGE", f"Locations page updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["website", "locations"], magnitude=mag, evidence="text_diff")] if surface in (Surface.NEWSROOM, Surface.BLOG, Surface.FEED, Surface.RESEARCH): first = next((d for d in diff.get("added") or [] if (d.get("after") or "").strip()), None) head = (first["after"].strip().split("\n")[0][:100]) if first else None if head: subtype = "NEWS_RELEASE" if surface == Surface.NEWSROOM or _PRESS_RE.search(head) else "BLOG_POST" pre = "News release" if subtype == "NEWS_RELEASE" else "Blog post" return [EventDraft(subtype, f"{pre}: {head}", "news:" + normalize_entity_key(head), payload=payload, tags=["news"], magnitude=0.1, evidence="text_diff")] return [EventDraft("WEBSITE_CHANGE", f"{_label(surface)[0].upper()}{_label(surface)[1:]} updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["website"], magnitude=mag, evidence="text_diff")] return [] # ============================================================================================================== derive # Media / entertainment / gaming publishers: their "news" and "blog" surfaces ARE their product (articles, live tickers, videos) — a new # article is not a corporate communication event. Their items still land in `news_items` for the profile; only event generation is skipped. MEDIA_PUBLISHER_INDUSTRIES = frozenset({"media", "entertainment", "gaming", "publishing", "broadcasting", "news-media"}) def _is_media_publisher(company: dict[str, Any]) -> bool: inds = {str(i) for i in (company.get("industries") or [])} return bool(inds & MEDIA_PUBLISHER_INDUSTRIES) or str(company.get("industry_primary") or "") in MEDIA_PUBLISHER_INDUSTRIES def derive_events(change: dict[str, Any], company: dict[str, Any], sensor: dict[str, Any], *, baseline: dict[str, Any] | None = None, country_names: dict[str, str] | None = None) -> Derived: """Pure rule evaluation for one change. Returns drafts with importance/confidence finalised, plus LLM hints.""" kind = str(change.get("kind") or "") significance = float(change.get("significance") or 0.0) if kind in (ChangeKind.NOISE, ChangeKind.MINOR) or significance < settings.meaningful_threshold and kind not in (ChangeKind.MEANINGFUL, ChangeKind.MAJOR, ChangeKind.CRITICAL): return Derived(events=[]) delta: dict[str, Any] = dict(change.get("structured_delta") or {}) diff: dict[str, Any] = dict(change.get("diff") or {}) surface = str(change.get("surface") or sensor.get("surface") or Surface.OTHER) evidence = evidence_kind({**sensor, "surface": surface}, delta) names = country_names or {} drafts: list[EventDraft] = [] drafts += _hiring_rules(delta, surface, evidence, baseline) drafts += _pricing_rules(delta, surface, evidence, diff, change) drafts += _leadership_rules(delta, surface, evidence) drafts += _product_rules(delta, evidence) drafts += _location_rules(delta, evidence, names) if not _is_media_publisher(company): drafts += _news_rules(delta, surface, evidence) elif (delta.get("news") or {}).get("added"): change.setdefault("notes", []).append("editorial items of a media publisher are not corporate communication events") structured_hit = bool(drafts) if not structured_hit: drafts += _text_diff_rules(surface, change, diff, delta) drafts = [d for d in drafts if "\ufffd" not in d.title and "\ufffd" not in (d.summary or "") and re.search(r"[^\W\d_]", d.title)] for d in drafts: default = EVENT_SUBTYPES.get(d.subtype, (EventType.OTHER, 0.3))[1] d.importance = scale_importance(default, significance, d.magnitude) d.confidence = EVIDENCE_CONFIDENCE.get(d.evidence, EVIDENCE_CONFIDENCE["html"]) d.title = safe_wording(d.title)[:200] if d.summary: d.summary = safe_wording(d.summary)[:600] if kind == ChangeKind.CRITICAL and not d.review: d.review = "major_event" elif d.confidence < 0.5 and not d.review: d.review = "low_confidence" needs = False reason = None if not drafts: needs, reason = True, "no_deterministic_event" elif surface in AMBIGUOUS_SURFACES or (not structured_hit and surface in (Surface.HOMEPAGE, Surface.ABOUT, Surface.PRODUCTS, Surface.SERVICES, Surface.SOLUTIONS)): needs, reason = True, "ambiguous_surface" summarize = sorted({d.subtype for d in drafts if d.subtype in SUMMARY_SUBTYPES and (diff.get("added") or diff.get("modified") or d.summary)}) return Derived(events=drafts, needs_classification=needs, classification_reason=reason, summarize=summarize) def dedupe_key_for(company_id: str, subtype: str, entity_key: str, sensor_id: str | None, day: date) -> str: return stable_hash(company_id, subtype, normalize_entity_key(entity_key), sensor_id or "", day.isoformat(), length=40) # ============================================================================================================== persistence async def _country_names(conn) -> dict[str, str]: # type: ignore[no-untyped-def] rows = await fetch_all(conn, "select code, name from countries") return {str(r["code"]).upper(): r["name"] for r in rows} async def _baseline(conn, company_id: str) -> dict[str, Any]: # type: ignore[no-untyped-def] rows = await fetch_all(conn, "select metric, mean, stddev, samples from baselines where company_id = :c", c=company_id) return {r["metric"]: {"mean": r["mean"], "stddev": r["stddev"], "samples": r["samples"]} for r in rows} async def _llm_budget_left(conn) -> int: # type: ignore[no-untyped-def] used = await fetch_val(conn, "select count(*) from llm_jobs where created_at >= date_trunc('day', now() at time zone 'utc')") return max(0, int(settings.llm_daily_budget) - int(used or 0)) async def _enqueue_llm(conn, *, kind: str, ref_id: str, company_id: str, budget: dict[str, int]) -> bool: # type: ignore[no-untyped-def] if not settings.llm_configured or budget["left"] <= 0: return False exists = await fetch_val(conn, "select 1 from llm_jobs where kind = :k and ref_id = :r and status in ('pending', 'running', 'done')", k=kind, r=ref_id) if exists: return False await execute(conn, "insert into llm_jobs (id, kind, ref_id, company_id, status) values (:id, :k, :r, :c, 'pending')", id=new_id("llm_job"), k=kind, r=ref_id, c=company_id) budget["left"] -= 1 return True async def persist_change_events(conn, change: dict[str, Any], company: dict[str, Any], sensor: dict[str, Any], derived: Derived, # type: ignore[no-untyped-def] *, budget: dict[str, int] | None = None, enqueue_llm: bool = True) -> dict[str, Any]: """Insert the drafts for one change (idempotent), cluster them, queue reviews and LLM jobs. Caller owns the transaction.""" detected_at: datetime = change.get("detected_at") or datetime.now(UTC) if detected_at.tzinfo is None: detected_at = detected_at.replace(tzinfo=UTC) day = detected_at.astimezone(UTC).date() created: list[str] = [] duplicates = 0 budget = budget if budget is not None else {"left": await _llm_budget_left(conn)} for d in derived.events: key = dedupe_key_for(company["id"], d.subtype, d.entity_key, change.get("sensor_id"), day) event_id = new_id("event") payload = {**d.payload, "rules_version": RULES_VERSION, "evidence": d.evidence, "significance": change.get("significance"), "change_kind": change.get("kind"), "entity_key": d.entity_key} row = await fetch_one(conn, """ insert into events (id, company_id, sensor_id, change_id, surface, event_type, event_subtype, importance, confidence, confidence_label, title, summary, old_value, new_value, payload, entities, tags, detected_at, effective_at, published_at, source_url, snapshot_before, snapshot_after, language, origin, schema_version, status, dedupe_key) values (:id, :company_id, :sensor_id, :change_id, :surface, :event_type, :subtype, :importance, :confidence, :label, :title, :summary, :old_value, :new_value, cast(:payload as jsonb), cast(:entities as jsonb), cast(:tags as text[]), :detected_at, :effective_at, :published_at, :source_url, :snap_before, :snap_after, :language, 'deterministic', :schema_version, 'active', :dedupe_key) on conflict (dedupe_key) do nothing returning id""", id=event_id, company_id=company["id"], sensor_id=change.get("sensor_id"), change_id=change["id"], surface=change.get("surface"), event_type=d.event_type, subtype=d.subtype, importance=d.importance, confidence=d.confidence, label=confidence_label(d.confidence), title=d.title, summary=d.summary, old_value=(d.old_value[:400] if d.old_value else None), new_value=(d.new_value[:400] if d.new_value else None), payload=jsonb(payload), entities=jsonb(d.entities), tags=list(dict.fromkeys(d.tags)), detected_at=detected_at, effective_at=d.effective_at, published_at=d.published_at, source_url=sensor.get("url"), snap_before=change.get("snapshot_before"), snap_after=change.get("snapshot_after"), language=((change.get("structured_delta") or {}).get("meta") or {}).get("language"), schema_version=SCHEMA_VERSION, dedupe_key=key) if row is None: continue created.append(event_id) if sensor.get("url"): await execute(conn, """ insert into event_sources (event_id, sensor_id, source_url, snapshot_id, surface, detected_at, kind) values (:e, :s, :url, :snap, :surface, :at, 'primary') on conflict do nothing""", e=event_id, s=change.get("sensor_id"), url=sensor["url"], snap=change.get("snapshot_after"), surface=change.get("surface"), at=detected_at) ev = {"id": event_id, "company_id": company["id"], "sensor_id": change.get("sensor_id"), "surface": change.get("surface"), "event_type": d.event_type, "event_subtype": d.subtype, "title": d.title, "confidence": d.confidence, "detected_at": detected_at, "source_url": sensor.get("url"), "snapshot_after": change.get("snapshot_after")} _, dup = await attach_to_cluster(conn, ev, entity_key=d.entity_key) duplicates += int(dup) if d.review and not dup: await execute(conn, """ insert into review_queue (id, kind, ref_id, company_id, payload) values (:id, :kind, :ref, :c, cast(:p as jsonb))""", id=new_id("review"), kind=d.review, ref=event_id, c=company["id"], p=jsonb({"event_subtype": d.subtype, "title": d.title, "confidence": d.confidence, "significance": change.get("significance")})) if enqueue_llm and not dup and d.subtype in derived.summarize and float(change.get("significance") or 0) >= settings.llm_min_significance: await _enqueue_llm(conn, kind="summarize_event", ref_id=event_id, company_id=company["id"], budget=budget) if enqueue_llm and derived.needs_classification and float(change.get("significance") or 0) >= settings.llm_min_significance: await _enqueue_llm(conn, kind="classify_change", ref_id=change["id"], company_id=company["id"], budget=budget) await execute(conn, "update changes set status = 'processed', processed_at = now() where id = :id and status <> 'enriched'", id=change["id"]) if created: active = len(created) - duplicates if change.get("sensor_id"): await execute(conn, "update sensors set event_count = event_count + :n where id = :s", n=len(created), s=change["sensor_id"]) await execute(conn, "update companies set last_event_at = greatest(coalesce(last_event_at, cast(:at as timestamptz)), cast(:at as timestamptz)) where id = :c", at=detected_at, c=company["id"]) log.info("events created", extra={"change_id": change["id"], "company": company.get("slug"), "events": len(created), "duplicates": duplicates, "active": active}) return {"created": created, "duplicates": duplicates} _CHANGE_SQL = """ select c.*, s.url as sensor_url, s.connector_id, s.surface as sensor_surface, s.config as sensor_config, co.slug as company_slug, co.display_name, co.country as company_country, co.industries from changes c join sensors s on s.id = c.sensor_id join companies co on co.id = c.company_id where {where} order by c.detected_at limit :limit for update of c skip locked""" def _split(row: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: company = {"id": row["company_id"], "slug": row["company_slug"], "display_name": row["display_name"], "country": row["company_country"], "industries": row.get("industries") or []} sensor = {"id": row["sensor_id"], "url": row["sensor_url"], "connector_id": row["connector_id"], "surface": row["sensor_surface"], "fetch_mode": (row.get("sensor_config") or {}).get("fetch_mode")} change = {k: v for k, v in row.items() if k not in ("sensor_url", "connector_id", "sensor_surface", "sensor_config", "company_slug", "display_name", "company_country", "industries")} return change, company, sensor async def process_pending_changes(limit: int = 200) -> dict[str, int]: """Claim pending changes (SKIP LOCKED), derive events, persist. Returns counters. Noise/minor pending rows are archived.""" stats = {"changes": 0, "events": 0, "duplicates": 0, "archived": 0, "llm_jobs": 0} new_event_ids: list[str] = [] async with transaction() as conn: rows = await fetch_all(conn, _CHANGE_SQL.format(where="c.status = 'pending'"), limit=limit) if not rows: return stats names = await _country_names(conn) budget = {"left": await _llm_budget_left(conn)} start_budget = budget["left"] baselines: dict[str, dict[str, Any]] = {} for row in rows: change, company, sensor = _split(row) if change.get("kind") in (ChangeKind.NOISE, ChangeKind.MINOR): await execute(conn, "update changes set status = 'archived', processed_at = now() where id = :id", id=change["id"]) stats["archived"] += 1 continue if company["id"] not in baselines: baselines[company["id"]] = await _baseline(conn, company["id"]) derived = derive_events(change, company, sensor, baseline=baselines[company["id"]], country_names=names) res = await persist_change_events(conn, change, company, sensor, derived, budget=budget) stats["changes"] += 1 stats["events"] += len(res["created"]) stats["duplicates"] += res["duplicates"] new_event_ids += res["created"] stats["llm_jobs"] = start_budget - budget["left"] if new_event_ids: try: from companyatlas.services.alerts import evaluate_alerts await evaluate_alerts(new_event_ids) except Exception: log.exception("alert evaluation failed") return stats async def reprocess_events(since: datetime, *, limit: int = 5000, company_id: str | None = None) -> dict[str, int]: """Re-run the deterministic rules over already processed changes (no refetch). Dedupe keys make this idempotent; new rules add events.""" stats = {"changes": 0, "events": 0, "duplicates": 0} where = "c.status in ('processed', 'enriched') and c.detected_at >= :since and c.kind in ('meaningful', 'major', 'critical')" params: dict[str, Any] = {"since": since, "limit": limit} if company_id: where += " and c.company_id = :company_id" params["company_id"] = company_id async with transaction() as conn: rows = await fetch_all(conn, _CHANGE_SQL.format(where=where), **params) names = await _country_names(conn) for row in rows: change, company, sensor = _split(row) derived = derive_events(change, company, sensor, baseline=await _baseline(conn, company["id"]), country_names=names) res = await persist_change_events(conn, change, company, sensor, derived, enqueue_llm=False) stats["changes"] += 1 stats["events"] += len(res["created"]) stats["duplicates"] += res["duplicates"] return stats async def list_events(*, company: str | None = None, event_type: str | None = None, limit: int = 50) -> list[dict[str, Any]]: where = ["e.status in ('active', 'review')"] params: dict[str, Any] = {"limit": limit} if company: where.append("(co.slug = :company or co.id = :company)") params["company"] = company if event_type: where.append("(e.event_type = :t or e.event_subtype = :t)") params["t"] = event_type.upper() async with transaction() as conn: return await fetch_all(conn, f""" select e.id, co.slug, e.event_type, e.event_subtype, e.importance, e.confidence, e.confidence_label, e.title, e.detected_at, e.origin, e.surface, e.cluster_id, e.status from events e join companies co on co.id = e.company_id where {' and '.join(where)} order by e.detected_at desc limit :limit""", **params) @periodic("process-changes", every_s=20) async def process_changes_task() -> None: stats = await process_pending_changes(limit=200) if stats["changes"] or stats["archived"]: log.info("process-changes", extra=stats) __all__ = [ "RULES_VERSION", "Derived", "EventDraft", "dedupe_key_for", "derive_events", "evidence_kind", "list_events", "persist_change_events", "process_pending_changes", "reprocess_events", "safe_wording", "scale_importance", ]