spb/company-atlas
Public
Python 66.3%
TypeScript 22.7%
JavaScript 8.6%
HTML 1.4%
CSS 0.7%
1"""Deterministic event generation (spec §21–23, §158, §167–168).23 changes (status='pending') ──▶ derive_events(): typed rules over `structured_delta` + block `diff` + surface4 ──▶ events (+ event_sources, clusters, dedupe keys, review_queue) ──▶ llm_jobs when useful56Design:7- `derive_events()` is a pure function (no I/O) so rules are unit-testable from fixtures. `process_pending_changes()` is the DB layer.8- Wording is careful by construction: "detected", "listed", "no longer listed", "observed at". Never "fired", "laid off", "shut down".9- Idempotent: `events.dedupe_key = sha(company, subtype, entity key, sensor, detection day)` → re-running a change is a no-op.10- Importance = subtype default × f(significance, magnitude); confidence = evidence quality (ATS JSON 0.95 · JSON-LD 0.9 · HTML 0.8 ·11 text-diff-only 0.7), corroboration handled by `services/clustering.py`.12- Noise / minor changes never produce events. LLM enrichment is queued only for meaningful+ changes, within the daily budget.13"""14from __future__ import annotations1516import logging17import re18from dataclasses import dataclass, field19from datetime import UTC, date, datetime20from typing import Any2122from companyatlas.config import settings23from companyatlas.db import execute, fetch_all, fetch_one, fetch_val, jsonb, transaction24from companyatlas.ids import new_id, stable_hash25from companyatlas.services.clustering import attach_to_cluster, normalize_entity_key26from companyatlas.services.periodic import periodic27from companyatlas.taxonomy import (28 AI_KEYWORDS,29 EVENT_SUBTYPES,30 EVIDENCE_CONFIDENCE,31 FORBIDDEN_WORDING,32 ChangeKind,33 EventType,34 Surface,35 confidence_label,36)3738log = logging.getLogger(__name__)3940RULES_VERSION = "rules-v1"41SCHEMA_VERSION = "event-v1"42MAX_ENTITY_ITEMS = 50 # bounded entity lists on aggregate events43PER_JOB_EVENT_MAX = 5 # NEW_JOB per job only when ≤ 5 jobs added44PER_PERSON_EVENT_MAX = 1045NEWS_ITEMS_MAX = 2046NEWS_ITEMS_PER_EVENT_MAX = 3 # more items than this in one observation → one aggregate communication event47LEADERSHIP_AGGREGATE_MIN = 348SURGE_MIN_JOBS = 10 # fallback thresholds when no baseline is available yet49SURGE_MIN_RATIO = 0.550FREEZE_MIN_RATIO = 0.551BASELINE_Z = 2.0 # surge/freeze when the delta exceeds mean + 2σ of the company's weekly baseline52IMPORTANCE_FLOOR = 0.0553LEGAL_SURFACES = {Surface.LEGAL_TERMS, Surface.LEGAL_PRIVACY, Surface.SECURITY}54DEVELOPER_SURFACES = {Surface.DOCS, Surface.DEVELOPER, Surface.API, Surface.CHANGELOG}55NEWS_SURFACES = {Surface.NEWSROOM, Surface.BLOG, Surface.FEED, Surface.INVESTOR_RELATIONS, Surface.CHANGELOG, Surface.RESEARCH}56AMBIGUOUS_SURFACES = {Surface.OTHER, Surface.PARTNERS, Surface.CUSTOMERS, Surface.SOLUTIONS, Surface.SERVICES, Surface.INDUSTRIES,57 Surface.SUPPORT, Surface.CONTACT, Surface.SITEMAP}58SUMMARY_SUBTYPES = {"TERMS_CHANGE", "PRIVACY_POLICY_CHANGE", "SECURITY_UPDATE", "WEBSITE_CHANGE", "HOMEPAGE_REDESIGN", "NEWS_RELEASE",59 "INVESTOR_UPDATE", "EARNINGS_RELEASE", "MESSAGING_CHANGE"}60ATS_CONNECTOR_HINTS = ("greenhouse", "lever", "ashby", "smartrecruiters", "workday", "workable", "json", "ats")6162SURFACE_LABEL: dict[str, str] = {63 Surface.CAREERS: "careers page", Surface.JOBS_BOARD: "job board", Surface.PRICING: "pricing page", Surface.LEADERSHIP: "leadership page",64 Surface.ABOUT: "about page", Surface.PRODUCTS: "products page", Surface.SERVICES: "services page", Surface.SOLUTIONS: "solutions page",65 Surface.LOCATIONS: "locations page", Surface.CONTACT: "contact page", Surface.NEWSROOM: "newsroom", Surface.BLOG: "blog",66 Surface.FEED: "feed", Surface.DOCS: "documentation", Surface.DEVELOPER: "developer portal", Surface.API: "API reference",67 Surface.CHANGELOG: "changelog", Surface.INVESTOR_RELATIONS: "investor relations page", Surface.LEGAL_TERMS: "terms of service page",68 Surface.LEGAL_PRIVACY: "privacy policy", Surface.SECURITY: "security page", Surface.HOMEPAGE: "homepage",69 Surface.SUSTAINABILITY: "sustainability page", Surface.STATUS: "status page", Surface.PARTNERS: "partners page",70 Surface.CUSTOMERS: "customers page", Surface.INDUSTRIES: "industries page", Surface.RESEARCH: "research page",71 Surface.SUPPORT: "support page", Surface.SITEMAP: "sitemap", Surface.OTHER: "monitored page",72}73CURRENCY_SYMBOL = {"USD": "$", "EUR": "€", "GBP": "£", "CAD": "CA$", "AUD": "A$", "JPY": "¥", "CHF": "CHF ", "INR": "₹", "BRL": "R$"}74PERIOD_LABEL = {"month": "per month", "year": "per year", "one_time": "one-time", "usage": "usage-based", "contact": "contact sales"}75EXEC_ROLES = {"ceo", "cfo", "cto", "coo", "founder", "president", "chair", "board", "cmo", "cro", "cpo", "ciso", "cio", "chro", "gm"}76_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)77_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)78_PRESS_RE = re.compile(r"\b(announces|announced|unveils|introduces|launches|partners with|acquires|appoints|names|expands|opens)\b", re.IGNORECASE)79_LAUNCH_RE = re.compile(r"\b(launch(es|ed|ing)?|introduc(es|ed|ing)|unveil(s|ed)|now available|general availability)\b", re.IGNORECASE)80_ACQ_RE = re.compile(r"\b(acquires|acquired|acquisition of|to acquire|merger|merges with)\b", re.IGNORECASE)81_FUNDING_RE = re.compile(r"\b(raises|raised|series [a-f]\b|seed round|funding round|closes \$|financing of)\b", re.IGNORECASE)828384# ============================================================================================================== drafts858687@dataclass(slots=True)88class EventDraft:89 subtype: str90 title: str91 entity_key: str92 summary: str | None = None93 old_value: str | None = None94 new_value: str | None = None95 entities: dict[str, Any] = field(default_factory=dict)96 payload: dict[str, Any] = field(default_factory=dict)97 tags: list[str] = field(default_factory=list)98 magnitude: float = 0.0 # 0–1 rule-specific magnitude (relative job delta, |price pct| …) → importance bonus99 evidence: str = "html" # ats_json | jsonld | html | text_diff100 effective_at: datetime | None = None101 published_at: datetime | None = None102 review: str | None = None # review_queue kind when the rule itself wants a human look103 importance: float = 0.0 # filled by finalize()104 confidence: float = 0.0105106 @property107 def event_type(self) -> str:108 return str(EVENT_SUBTYPES.get(self.subtype, (EventType.OTHER, 0.3))[0])109110111@dataclass(slots=True)112class Derived:113 events: list[EventDraft]114 needs_classification: bool = False115 classification_reason: str | None = None116 summarize: list[str] = field(default_factory=list) # subtypes whose events deserve an LLM summary117118119# ============================================================================================================== helpers120121122def safe_wording(text: str) -> str:123 """Defensive: rewrite forbidden phrasing (rules never produce it, LLM output might)."""124 out = text125 for bad in FORBIDDEN_WORDING:126 if bad in out.lower():127 out = re.sub(re.escape(bad), "no longer listed", out, flags=re.IGNORECASE)128 return out129130131def scale_importance(default: float, significance: float, magnitude: float = 0.0) -> float:132 """importance = default × (0.7 + 0.6·significance) × (1 + 0.3·magnitude), clamped to [0.05, 1]."""133 sig = min(1.0, max(0.0, float(significance or 0.0)))134 mag = min(1.0, max(0.0, float(magnitude or 0.0)))135 return round(min(1.0, max(IMPORTANCE_FLOOR, default * (0.7 + 0.6 * sig) * (1.0 + 0.3 * mag))), 4)136137138def evidence_kind(sensor: dict[str, Any], delta: dict[str, Any]) -> str:139 connector = (sensor.get("connector_id") or "").lower()140 surface = sensor.get("surface") or ""141 meta = delta.get("meta") or {}142 hinted = meta.get("evidence")143 if hinted in EVIDENCE_CONFIDENCE:144 return str(hinted)145 if surface == Surface.JOBS_BOARD or any(h in connector for h in ATS_CONNECTOR_HINTS) or sensor.get("fetch_mode") == "json":146 return "ats_json"147 if meta.get("jsonld") or "jsonld" in connector or "feed" in connector or surface == Surface.FEED:148 return "jsonld"149 if any(delta.get(k) for k in ("jobs", "people", "products", "plans", "locations", "news")):150 return "html"151 return "text_diff"152153154def _label(surface: str) -> str:155 return SURFACE_LABEL.get(surface, "monitored page")156157158def _plural(n: int, one: str, many: str | None = None) -> str:159 return one if n == 1 else (many or one + "s")160161162def _money(amount: Any, currency: str | None) -> str:163 try:164 value = float(amount)165 except (TypeError, ValueError):166 return str(amount)167 text = f"{value:,.0f}" if value.is_integer() else f"{value:,.2f}"168 cur = (currency or "").upper()169 sym = CURRENCY_SYMBOL.get(cur)170 if sym:171 return f"{sym}{text}"172 return f"{text} {cur}".strip()173174175def _place(item: dict[str, Any]) -> str:176 parts = [p for p in (item.get("city"), item.get("region")) if p]177 country = item.get("country")178 if country:179 parts.append(str(country).upper())180 if parts:181 return ", ".join(parts)182 return item.get("name") or item.get("location_text") or ""183184185def _is_ai(text: str | None) -> bool:186 if not text:187 return False188 hay = f" {text.lower()} "189 return any(k in hay for k in AI_KEYWORDS)190191192def _job_is_ai(job: dict[str, Any]) -> bool:193 return bool(job.get("is_ai")) or _is_ai(job.get("title"))194195196def _job_label(job: dict[str, Any]) -> str:197 title = (job.get("title") or "position").strip()198 loc = job.get("location_text") or _place(job)199 if job.get("remote") and not loc:200 loc = "Remote"201 return f"{title} ({loc})" if loc else title202203204def _sections(diff: dict[str, Any]) -> list[str]:205 seen: list[str] = []206 for bucket in ("modified", "added", "removed"):207 for d in diff.get(bucket) or []:208 path = (d.get("path") or "").strip()209 name = path.split(">")[-1].strip() if path else ""210 if not name:211 text = (d.get("after") or d.get("before") or "").strip()212 name = text.split("\n")[0][:80] if text else ""213 if name and name not in seen:214 seen.append(name)215 return seen[:20]216217218def _blocks_changed(diff: dict[str, Any], change: dict[str, Any]) -> int:219 counts = diff.get("counts") or {}220 n = int(counts.get("added") or 0) + int(counts.get("removed") or 0) + int(counts.get("modified") or 0)221 if n == 0:222 n = int(change.get("blocks_added") or 0) + int(change.get("blocks_removed") or 0) + int(change.get("blocks_modified") or 0)223 return n224225226def _dt(value: Any) -> datetime | None:227 if value is None:228 return None229 if isinstance(value, datetime):230 return value if value.tzinfo else value.replace(tzinfo=UTC)231 try:232 return datetime.fromisoformat(str(value))233 except ValueError:234 return None235236237# ============================================================================================================== rule families238239240def _hiring_rules(delta: dict[str, Any], surface: str, evidence: str, baseline: dict[str, Any] | None) -> list[EventDraft]:241 jobs = delta.get("jobs") or {}242 added: list[dict[str, Any]] = list(jobs.get("added") or [])243 removed: list[dict[str, Any]] = list(jobs.get("removed") or [])244 open_before = jobs.get("open_before")245 open_after = jobs.get("open_after")246 n_add, n_rem = len(added), len(removed)247 if not (n_add or n_rem):248 return []249 if isinstance(open_before, int) and isinstance(open_after, int):250 net = open_after - open_before251 else:252 net = n_add - n_rem253 label = _label(surface)254 out: list[EventDraft] = []255 counts_key = f"{open_before}>{open_after}" if open_before is not None else f"+{n_add}-{n_rem}"256 base_denominator = max(int(open_before or 0), 5)257 countries = sorted({str(j.get("country")).upper() for j in added if j.get("country")})258 departments = sorted({str(j.get("department")) for j in added if j.get("department")})[:20]259 ai_added = [j for j in added if _job_is_ai(j)]260 common_payload = {"added": n_add, "removed": n_rem, "open_before": open_before, "open_after": open_after, "net": net,261 "ai_added": len(ai_added), "countries": countries, "departments": departments}262263 if net > 0 and n_add:264 title = f"{n_add} new {_plural(n_add, 'position')} detected on {label}"265 if n_rem:266 title += f" ({n_rem} no longer visible)"267 out.append(EventDraft(268 subtype="JOB_COUNT_INCREASE", title=title, entity_key=f"jobs:{counts_key}", evidence=evidence,269 summary=_open_summary(open_before, open_after, n_add, n_rem), old_value=_s(open_before), new_value=_s(open_after),270 entities={"jobs": [_job_entity(j) for j in added[:MAX_ENTITY_ITEMS]]}, payload=common_payload,271 tags=_hiring_tags(countries, ai_added), magnitude=min(1.0, n_add / base_denominator)))272 elif net < 0 and n_rem:273 title = f"{n_rem} monitored job {_plural(n_rem, 'listing')} no longer visible on {label}"274 if n_add:275 title += f" ({n_add} new)"276 out.append(EventDraft(277 subtype="JOB_COUNT_DECREASE", title=title, entity_key=f"jobs:{counts_key}", evidence=evidence,278 summary=_open_summary(open_before, open_after, n_add, n_rem), old_value=_s(open_before), new_value=_s(open_after),279 entities={"jobs": [_job_entity(j) for j in removed[:MAX_ENTITY_ITEMS]]}, payload=common_payload,280 tags=["hiring"], magnitude=min(1.0, n_rem / base_denominator)))281282 if ai_added:283 k = len(ai_added)284 out.append(EventDraft(285 subtype="AI_HIRING", title=f"{k} AI-related {_plural(k, 'position')} detected on {label}", entity_key=f"ai_jobs:{counts_key}",286 evidence=evidence, summary="AI-related roles identified from listing titles: " + "; ".join(_job_label(j) for j in ai_added[:5]),287 entities={"jobs": [_job_entity(j) for j in ai_added[:MAX_ENTITY_ITEMS]]}, payload={"ai_added": k, "added": n_add},288 tags=["hiring", "ai"], magnitude=min(1.0, k / 5)))289290 if 1 <= n_add <= PER_JOB_EVENT_MAX:291 for j in added:292 out.append(EventDraft(293 subtype="NEW_JOB", title=f"New position listed: {_job_label(j)}", entity_key="job:" + normalize_entity_key(_job_label(j)),294 evidence=evidence, new_value=j.get("title"), entities={"jobs": [_job_entity(j)]},295 payload={"url": j.get("url"), "department": j.get("department"), "country": j.get("country"), "remote": j.get("remote")},296 tags=["hiring"] + (["ai"] if _job_is_ai(j) else []), published_at=_dt(j.get("posted_at"))))297298 surge, freeze = _surge_freeze(n_add, n_rem, open_before, open_after, baseline)299 if surge is not None:300 out.append(EventDraft(301 subtype="HIRING_SURGE", title=f"Hiring surge signal: {n_add} new positions detected in one observation" + surge,302 entity_key=f"surge:{counts_key}", evidence=evidence, payload={**common_payload, "baseline": baseline},303 summary="Signal, not a fact: the number of new listings exceeds this company's usual weekly volume.",304 tags=["hiring", "signal"], magnitude=min(1.0, n_add / max(base_denominator, SURGE_MIN_JOBS))))305 if freeze is not None:306 out.append(EventDraft(307 subtype="HIRING_FREEZE_SIGNAL",308 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,309 entity_key=f"freeze:{counts_key}", evidence=evidence, payload={**common_payload, "baseline": baseline},310 summary="Signal, not a fact: listings disappearing from a public careers page can reflect closed roles, ATS migrations or page changes.",311 tags=["hiring", "signal"], review="unexpected_activity", magnitude=min(1.0, n_rem / base_denominator)))312 return out313314315def _surge_freeze(n_add: int, n_rem: int, open_before: Any, open_after: Any, baseline: dict[str, Any] | None) -> tuple[str | None, str | None]:316 surge = freeze = None317 b = (baseline or {}).get("jobs_new_weekly")318 before = int(open_before) if isinstance(open_before, int) else None319 if b and b.get("samples", 0) >= 4 and b.get("stddev") is not None:320 threshold = float(b["mean"]) + BASELINE_Z * max(float(b["stddev"]), 1.0)321 if n_add >= max(5, threshold):322 surge = f" (baseline ≈ {float(b['mean']):.1f} new/week)"323 elif n_add >= SURGE_MIN_JOBS and (before is None or n_add >= SURGE_MIN_RATIO * before):324 surge = ""325 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):326 freeze = ""327 return surge, freeze328329330def _open_summary(before: Any, after: Any, n_add: int, n_rem: int) -> str:331 parts = [f"{n_add} added" if n_add else "", f"{n_rem} no longer visible" if n_rem else ""]332 s = ", ".join(p for p in parts if p)333 if before is not None and after is not None:334 s += f"; open listings observed: {before} → {after}"335 return s + "."336337338def _hiring_tags(countries: list[str], ai_added: list[dict[str, Any]]) -> list[str]:339 tags = ["hiring"] + [f"country:{c}" for c in countries[:5]]340 if ai_added:341 tags.append("ai")342 return tags343344345def _job_entity(j: dict[str, Any]) -> dict[str, Any]:346 return {k: j.get(k) for k in ("title", "url", "location_text", "country", "remote", "department", "is_ai") if j.get(k) is not None}347348349def _s(v: Any) -> str | None:350 return None if v is None else str(v)351352353def _pricing_rules(delta: dict[str, Any], surface: str, evidence: str, diff: dict[str, Any], change: dict[str, Any]) -> list[EventDraft]:354 plans = delta.get("plans") or {}355 out: list[EventDraft] = []356 for p in plans.get("price_changed") or []:357 name = p.get("plan_name") or "Plan"358 before, after = p.get("before"), p.get("after")359 try:360 b, a = float(before), float(after)361 except (TypeError, ValueError):362 continue363 if a == b:364 continue365 pct = p.get("pct")366 if pct is None and b:367 pct = round((a - b) / b * 100, 1)368 cur = p.get("currency")369 period = PERIOD_LABEL.get(p.get("billing_period") or "", "")370 subtype = "PRICE_INCREASE" if a > b else "PRICE_DECREASE"371 title = f"{name} plan price observed at {_money(a, cur)} (was {_money(b, cur)})"372 summary = f"{'Increase' if a > b else 'Decrease'} of {abs(pct):.1f}%" if pct is not None else None373 if summary and period:374 summary += f", billed {period}"375 out.append(EventDraft(376 subtype=subtype, title=title, entity_key="plan:" + normalize_entity_key(name), evidence=evidence, summary=(summary + "." if summary else None),377 old_value=_money(b, cur), new_value=_money(a, cur),378 payload={"plan_name": name, "before": b, "after": a, "pct": pct, "currency": cur, "billing_period": p.get("billing_period")},379 entities={"plans": [{"plan_name": name}]}, tags=["pricing"], magnitude=min(1.0, abs(float(pct or 0)) / 50.0)))380 for p in plans.get("added") or []:381 name = p.get("plan_name") or "New plan"382 price = p.get("price")383 cur = p.get("currency")384 if p.get("contact_sales") or (price is None and (p.get("billing_period") == "contact")):385 price_txt = "contact sales"386 tags = ["pricing", "enterprise"]387 elif price is not None:388 price_txt = _money(price, cur) + (f" {PERIOD_LABEL.get(p.get('billing_period') or '', '')}".rstrip())389 tags = ["pricing"]390 else:391 price_txt = p.get("price_text") or "price not stated"392 tags = ["pricing"]393 out.append(EventDraft(394 subtype="NEW_PRICING_TIER", title=f"New pricing tier listed: {name} ({price_txt})", entity_key="plan:" + normalize_entity_key(name),395 evidence=evidence, new_value=price_txt, payload={"plan_name": name, "price": price, "currency": cur, "billing_period": p.get("billing_period"),396 "contact_sales": bool(p.get("contact_sales"))},397 entities={"plans": [{"plan_name": name}]}, tags=tags, magnitude=0.3))398 for p in plans.get("removed") or []:399 name = p.get("plan_name") or "Plan"400 out.append(EventDraft(401 subtype="PRICING_TIER_REMOVED", title=f"Pricing tier no longer listed: {name}", entity_key="plan:" + normalize_entity_key(name),402 evidence=evidence, old_value=name, payload={"plan_name": name, "price": p.get("price"), "currency": p.get("currency")},403 entities={"plans": [{"plan_name": name}]}, tags=["pricing"], magnitude=0.3))404 if not out and surface == Surface.PRICING:405 n = _blocks_changed(diff, change)406 sections = _sections(diff)407 out.append(EventDraft(408 subtype="PRICING_CHANGE", title=f"Pricing page materially updated ({n} {_plural(n, 'block')} changed)", entity_key="pricing_page",409 evidence="text_diff", payload={"blocks_changed": n, "sections": sections, "text_delta_ratio": diff.get("text_delta_ratio")},410 summary=("Sections affected: " + ", ".join(sections[:6]) + ".") if sections else None, tags=["pricing"],411 magnitude=min(1.0, float(diff.get("text_delta_ratio") or 0) * 2)))412 return out413414415def _leadership_rules(delta: dict[str, Any], surface: str, evidence: str) -> list[EventDraft]:416 people = delta.get("people") or {}417 added = list(people.get("added") or [])418 removed = list(people.get("removed") or [])419 changed = list(people.get("title_changed") or [])420 if not (added or removed or changed):421 return []422 label = _label(surface) if surface in (Surface.LEADERSHIP, Surface.ABOUT) else "monitored leadership page"423 out: list[EventDraft] = []424425 def is_exec(p: dict[str, Any]) -> bool:426 return bool(p.get("is_executive")) or (p.get("role_category") or "").lower() in EXEC_ROLES427428 for p in [x for x in added if is_exec(x)][:PER_PERSON_EVENT_MAX]:429 name, title = p.get("name") or "Unnamed", p.get("title")430 out.append(EventDraft(431 subtype="NEW_EXECUTIVE", title=f"{name} listed as {title} on {label}" if title else f"{name} newly listed on {label}",432 entity_key="person:" + normalize_entity_key(name), evidence=evidence, new_value=title,433 entities={"people": [{"name": name, "title": title, "role_category": p.get("role_category")}]},434 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))435 for p in [x for x in removed if is_exec(x)][:PER_PERSON_EVENT_MAX]:436 name, title = p.get("name") or "Unnamed", p.get("title")437 out.append(EventDraft(438 subtype="EXECUTIVE_NO_LONGER_LISTED", title=f"{name} no longer listed on {label}", entity_key="person:" + normalize_entity_key(name),439 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,440 entities={"people": [{"name": name, "title": title, "role_category": p.get("role_category")}]},441 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))442 for p in changed[:PER_PERSON_EVENT_MAX]:443 name = p.get("name") or "Unnamed"444 before, after = p.get("before"), p.get("after")445 out.append(EventDraft(446 subtype="EXECUTIVE_TITLE_CHANGE", title=f"{name} now listed as {after} (was {before})", entity_key="person:" + normalize_entity_key(name),447 evidence=evidence, old_value=before, new_value=after, entities={"people": [{"name": name, "title": after}]}, tags=["leadership"], magnitude=0.3))448 total = len(added) + len(removed) + len(changed)449 if total >= LEADERSHIP_AGGREGATE_MIN or (total and not out):450 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 ""]451 out.append(EventDraft(452 subtype="LEADERSHIP_CHANGE", title=f"{label[0].upper()}{label[1:]} updated: " + ", ".join(b for b in bits if b),453 entity_key=f"leadership:{len(added)}:{len(removed)}:{len(changed)}", evidence=evidence,454 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]},455 payload={"added": len(added), "removed": len(removed), "title_changed": len(changed)}, tags=["leadership"], magnitude=min(1.0, total / 6)))456 return out457458459def _product_rules(delta: dict[str, Any], evidence: str) -> list[EventDraft]:460 products = delta.get("products") or {}461 out: list[EventDraft] = []462 for p in list(products.get("added") or [])[:MAX_ENTITY_ITEMS]:463 name = p.get("name") or "Unnamed product"464 out.append(EventDraft(465 subtype="NEW_PRODUCT", title=f"New product listed: {name}", entity_key="product:" + normalize_entity_key(name), evidence=evidence,466 new_value=name, entities={"products": [{"name": name, "url": p.get("url")}]}, payload={"url": p.get("url"), "category": p.get("category")},467 tags=["product"] + (["ai"] if _is_ai(name) else []), magnitude=0.3))468 for p in list(products.get("removed") or [])[:MAX_ENTITY_ITEMS]:469 name = p.get("name") or "Unnamed product"470 out.append(EventDraft(471 subtype="PRODUCT_REMOVED", title=f"Product no longer listed: {name}", entity_key="product:" + normalize_entity_key(name), evidence=evidence,472 old_value=name, entities={"products": [{"name": name, "url": p.get("url")}]}, payload={"url": p.get("url")}, tags=["product"], magnitude=0.3))473 return out474475476def _location_rules(delta: dict[str, Any], evidence: str, country_names: dict[str, str]) -> list[EventDraft]:477 locations = delta.get("locations") or {}478 out: list[EventDraft] = []479 kind_label = {"headquarters": "headquarters", "office": "office", "store": "store", "factory": "factory", "warehouse": "warehouse",480 "lab": "lab", "data_center": "data center"}481 for loc in list(locations.get("added") or [])[:MAX_ENTITY_ITEMS]:482 place = _place(loc) or "unnamed location"483 kind = kind_label.get(loc.get("kind") or "", "location")484 out.append(EventDraft(485 subtype="NEW_LOCATION", title=f"New {kind} listed: {place}", entity_key="location:" + normalize_entity_key(place), evidence=evidence,486 new_value=place, entities={"locations": [{"name": loc.get("name"), "city": loc.get("city"), "country": loc.get("country"), "kind": loc.get("kind")}]},487 payload={"kind": loc.get("kind"), "country": loc.get("country")}, tags=["location"] + ([f"country:{str(loc['country']).upper()}"] if loc.get("country") else []),488 magnitude=0.4 if loc.get("kind") == "headquarters" else 0.2))489 for loc in list(locations.get("removed") or [])[:MAX_ENTITY_ITEMS]:490 place = _place(loc) or "unnamed location"491 kind = kind_label.get(loc.get("kind") or "", "location")492 out.append(EventDraft(493 subtype="OFFICE_REMOVED", title=f"{kind[0].upper()}{kind[1:]} no longer listed: {place}", entity_key="location:" + normalize_entity_key(place),494 evidence=evidence, old_value=place, entities={"locations": [{"name": loc.get("name"), "city": loc.get("city"), "country": loc.get("country"), "kind": loc.get("kind")}]},495 payload={"kind": loc.get("kind"), "country": loc.get("country")}, tags=["location"], magnitude=0.2))496 for code in locations.get("new_countries") or []:497 code = str(code).upper()498 name = country_names.get(code, code)499 cities = [loc.get("city") for loc in locations.get("added") or [] if str(loc.get("country") or "").upper() == code and loc.get("city")]500 detail = f" ({', '.join(cities[:3])})" if cities else ""501 out.append(EventDraft(502 subtype="COUNTRY_EXPANSION", title=f"New country presence listed: {name}{detail}", entity_key=f"country:{code}", evidence=evidence,503 new_value=code, entities={"locations": [{"country": code, "city": c} for c in cities[:10]] or [{"country": code}]},504 payload={"country": code, "cities": cities[:10]}, tags=["location", "expansion", f"country:{code}"], magnitude=0.6))505 return out506507508def _news_subtype(item: dict[str, Any], surface: str) -> str:509 title = item.get("title") or ""510 category = (item.get("category") or "").lower()511 if _EARNINGS_RE.search(title):512 return "EARNINGS_RELEASE"513 if category == "ir" or surface == Surface.INVESTOR_RELATIONS or _INVESTOR_RE.search(title):514 return "INVESTOR_UPDATE"515 if category == "changelog" or surface == Surface.CHANGELOG:516 return "CHANGELOG_ENTRY"517 if category == "press" or surface == Surface.NEWSROOM:518 return "NEWS_RELEASE"519 if category in ("blog", "research") or surface in (Surface.BLOG, Surface.RESEARCH):520 return "BLOG_POST"521 return "NEWS_RELEASE" if _PRESS_RE.search(title) else "BLOG_POST"522523524def _news_rules(delta: dict[str, Any], surface: str, evidence: str) -> list[EventDraft]:525 news = delta.get("news") or {}526 prefix = {"NEWS_RELEASE": "News release", "BLOG_POST": "Blog post", "CHANGELOG_ENTRY": "Changelog entry", "INVESTOR_UPDATE": "Investor update",527 "EARNINGS_RELEASE": "Earnings release"}528 out: list[EventDraft] = []529 added = [i for i in list(news.get("added") or []) if (i.get("title") or "").strip()]530 if len(added) > NEWS_ITEMS_PER_EVENT_MAX:531 # A burst of items in one observation (catalogue re-listing, archive page, many posts at once) is one communication event,532 # not a flood: individual titles stay in `entities.news` for the evidence drawer.533 subtypes = [_news_subtype(i, surface) for i in added]534 subtype = max(set(subtypes), key=subtypes.count)535 label = {"NEWS_RELEASE": "news releases", "BLOG_POST": "blog posts", "CHANGELOG_ENTRY": "changelog entries", "INVESTOR_UPDATE": "investor updates",536 "EARNINGS_RELEASE": "earnings releases"}[subtype]537 titles = [(i.get("title") or "").strip() for i in added]538 tags = ["news", subtype.lower()]539 if any(_is_ai(t) for t in titles):540 tags.append("ai")541 if any(_LAUNCH_RE.search(t) for t in titles):542 tags.append("launch")543 out.append(EventDraft(544 subtype=subtype, title=f"{len(added)} new {label} published on {_label(surface)}",545 entity_key=f"news_batch:{normalize_entity_key(titles[0])}:{len(added)}", evidence=evidence,546 summary="Latest: " + " · ".join(t[:80] for t in titles[:3]) + (" …" if len(titles) > 3 else ""),547 entities={"news": [{"title": i.get("title"), "url": i.get("url"), "published_at": i.get("published_at")} for i in added[:NEWS_ITEMS_MAX]]},548 payload={"count": len(added), "category": added[0].get("category")}, tags=tags, magnitude=min(1.0, len(added) / 20)))549 return out550 for item in added[:NEWS_ITEMS_MAX]:551 title = (item.get("title") or "").strip()552 subtype = _news_subtype(item, surface)553 tags = ["news", subtype.lower()]554 if _is_ai(title):555 tags.append("ai")556 if _LAUNCH_RE.search(title):557 tags.append("launch")558 if _ACQ_RE.search(title):559 tags.append("m&a-mention")560 if _FUNDING_RE.search(title):561 tags.append("financing-mention")562 published = _dt(item.get("published_at"))563 out.append(EventDraft(564 subtype=subtype, title=f"{prefix[subtype]}: {title[:110]}", entity_key="news:" + normalize_entity_key(title), evidence=evidence,565 summary=(item.get("summary") or None), new_value=item.get("url"),566 entities={"news": [{"title": title, "url": item.get("url"), "published_at": item.get("published_at")}]},567 payload={"url": item.get("url"), "category": item.get("category"), "published_at": item.get("published_at")}, tags=tags,568 published_at=published, effective_at=published, magnitude=0.5 if "launch" in tags else 0.1))569 return out570571572def _text_diff_rules(surface: str, change: dict[str, Any], diff: dict[str, Any], delta: dict[str, Any]) -> list[EventDraft]:573 """Meaningful+ diffs on surfaces without typed extractions."""574 n = _blocks_changed(diff, change)575 sections = _sections(diff)576 ratio = float(diff.get("text_delta_ratio") or change.get("text_delta_ratio") or 0)577 kind = change.get("kind") or ""578 sec_txt = f"{len(sections)} {_plural(len(sections), 'section')} changed" if sections else f"{n} {_plural(n, 'block')} changed"579 payload = {"blocks_changed": n, "sections": sections, "text_delta_ratio": round(ratio, 4), "similarity": diff.get("similarity")}580 summary = ("Sections affected: " + ", ".join(sections[:8]) + ".") if sections else None581 mag = min(1.0, ratio * 2)582 ek = f"page:{change.get('sensor_id')}"583 if surface == Surface.LEGAL_TERMS:584 return [EventDraft("TERMS_CHANGE", f"Terms of service page materially updated ({sec_txt})", ek, summary=summary, payload=payload,585 tags=["legal", "terms"], magnitude=mag, evidence="text_diff", review="legal_sensitive")]586 if surface == Surface.LEGAL_PRIVACY:587 return [EventDraft("PRIVACY_POLICY_CHANGE", f"Privacy policy materially updated ({sec_txt})", ek, summary=summary, payload=payload,588 tags=["legal", "privacy"], magnitude=mag, evidence="text_diff", review="legal_sensitive")]589 if surface == Surface.SECURITY:590 return [EventDraft("SECURITY_UPDATE", f"Security page updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["security"],591 magnitude=mag, evidence="text_diff")]592 if surface == Surface.HOMEPAGE:593 out: list[EventDraft] = []594 meta = delta.get("meta") or {}595 tc = meta.get("title_changed")596 if isinstance(tc, dict) and tc.get("before") and tc.get("after") and tc["before"] != tc["after"]:597 out.append(EventDraft("MESSAGING_CHANGE", f"Homepage title observed as “{str(tc['after'])[:60]}” (was “{str(tc['before'])[:60]}”)",598 "homepage:title", old_value=tc["before"], new_value=tc["after"], payload={"field": "title"}, tags=["messaging"],599 magnitude=0.4, evidence="html"))600 if kind in (ChangeKind.MAJOR, ChangeKind.CRITICAL):601 out.append(EventDraft("HOMEPAGE_REDESIGN", f"Homepage materially redesigned ({n} {_plural(n, 'block')} changed, {ratio:.0%} of text)", ek,602 summary=summary, payload=payload, tags=["website", "homepage"], magnitude=mag, evidence="text_diff"))603 else:604 out.append(EventDraft("WEBSITE_CHANGE", f"Homepage content updated ({sec_txt})", ek, summary=summary, payload=payload,605 tags=["website", "homepage"], magnitude=mag, evidence="text_diff"))606 return out607 if surface == Surface.ABOUT:608 return [EventDraft("WEBSITE_CHANGE", f"About page updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["website", "about"],609 magnitude=mag, evidence="text_diff")]610 if surface == Surface.API:611 return [EventDraft("API_CHANGE", f"API reference updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["developer", "api"],612 magnitude=mag, evidence="text_diff")]613 if surface in (Surface.DOCS, Surface.DEVELOPER):614 return [EventDraft("DOC_CHANGE", f"Documentation updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["developer", "docs"],615 magnitude=mag, evidence="text_diff")]616 if surface == Surface.CHANGELOG:617 first = next((d for d in diff.get("added") or [] if (d.get("after") or "").strip()), None)618 head = (first["after"].strip().split("\n")[0][:90]) if first else None619 title = f"Changelog entry detected: {head}" if head else f"Changelog updated ({sec_txt})"620 return [EventDraft("CHANGELOG_ENTRY", title, "changelog:" + normalize_entity_key(head or ek), summary=summary, payload=payload,621 tags=["developer", "changelog"], magnitude=mag, evidence="text_diff")]622 if surface == Surface.PRICING:623 return [] # handled by _pricing_rules (generic PRICING_CHANGE)624 if surface in (Surface.PRODUCTS, Surface.SERVICES, Surface.SOLUTIONS):625 return [EventDraft("PRODUCT_UPDATE", f"{_label(surface)[0].upper()}{_label(surface)[1:]} updated ({sec_txt})", ek, summary=summary,626 payload=payload, tags=["product"], magnitude=mag, evidence="text_diff")]627 if surface == Surface.INVESTOR_RELATIONS:628 return [EventDraft("INVESTOR_UPDATE", f"Investor relations page updated ({sec_txt})", ek, summary=summary, payload=payload,629 tags=["investor-relations"], magnitude=mag, evidence="text_diff")]630 if surface == Surface.SUSTAINABILITY:631 return [EventDraft("SUSTAINABILITY_UPDATE", f"Sustainability page updated ({sec_txt})", ek, summary=summary, payload=payload,632 tags=["sustainability"], magnitude=mag, evidence="text_diff")]633 if surface == Surface.STATUS:634 return [EventDraft("OPERATIONS_UPDATE", f"Status page updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["operations"],635 magnitude=mag, evidence="text_diff")]636 if surface in (Surface.CAREERS, Surface.JOBS_BOARD):637 return [EventDraft("WEBSITE_CHANGE", f"Careers page updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["website", "careers"],638 magnitude=mag, evidence="text_diff")]639 if surface == Surface.LEADERSHIP:640 return [EventDraft("LEADERSHIP_CHANGE", f"Leadership page updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["leadership"],641 magnitude=mag, evidence="text_diff", review="low_confidence")]642 if surface == Surface.LOCATIONS:643 return [EventDraft("WEBSITE_CHANGE", f"Locations page updated ({sec_txt})", ek, summary=summary, payload=payload, tags=["website", "locations"],644 magnitude=mag, evidence="text_diff")]645 if surface in (Surface.NEWSROOM, Surface.BLOG, Surface.FEED, Surface.RESEARCH):646 first = next((d for d in diff.get("added") or [] if (d.get("after") or "").strip()), None)647 head = (first["after"].strip().split("\n")[0][:100]) if first else None648 if head:649 subtype = "NEWS_RELEASE" if surface == Surface.NEWSROOM or _PRESS_RE.search(head) else "BLOG_POST"650 pre = "News release" if subtype == "NEWS_RELEASE" else "Blog post"651 return [EventDraft(subtype, f"{pre}: {head}", "news:" + normalize_entity_key(head), payload=payload, tags=["news"], magnitude=0.1,652 evidence="text_diff")]653 return [EventDraft("WEBSITE_CHANGE", f"{_label(surface)[0].upper()}{_label(surface)[1:]} updated ({sec_txt})", ek, summary=summary,654 payload=payload, tags=["website"], magnitude=mag, evidence="text_diff")]655 return []656657658# ============================================================================================================== derive659660661662# Media / entertainment / gaming publishers: their "news" and "blog" surfaces ARE their product (articles, live tickers, videos) — a new663# article is not a corporate communication event. Their items still land in `news_items` for the profile; only event generation is skipped.664MEDIA_PUBLISHER_INDUSTRIES = frozenset({"media", "entertainment", "gaming", "publishing", "broadcasting", "news-media"})665666667def _is_media_publisher(company: dict[str, Any]) -> bool:668 inds = {str(i) for i in (company.get("industries") or [])}669 return bool(inds & MEDIA_PUBLISHER_INDUSTRIES) or str(company.get("industry_primary") or "") in MEDIA_PUBLISHER_INDUSTRIES670671672def derive_events(change: dict[str, Any], company: dict[str, Any], sensor: dict[str, Any], *, baseline: dict[str, Any] | None = None,673 country_names: dict[str, str] | None = None) -> Derived:674 """Pure rule evaluation for one change. Returns drafts with importance/confidence finalised, plus LLM hints."""675 kind = str(change.get("kind") or "")676 significance = float(change.get("significance") or 0.0)677 if kind in (ChangeKind.NOISE, ChangeKind.MINOR) or significance < settings.meaningful_threshold and kind not in (ChangeKind.MEANINGFUL, ChangeKind.MAJOR, ChangeKind.CRITICAL):678 return Derived(events=[])679 delta: dict[str, Any] = dict(change.get("structured_delta") or {})680 diff: dict[str, Any] = dict(change.get("diff") or {})681 surface = str(change.get("surface") or sensor.get("surface") or Surface.OTHER)682 evidence = evidence_kind({**sensor, "surface": surface}, delta)683 names = country_names or {}684685 drafts: list[EventDraft] = []686 drafts += _hiring_rules(delta, surface, evidence, baseline)687 drafts += _pricing_rules(delta, surface, evidence, diff, change)688 drafts += _leadership_rules(delta, surface, evidence)689 drafts += _product_rules(delta, evidence)690 drafts += _location_rules(delta, evidence, names)691 if not _is_media_publisher(company):692 drafts += _news_rules(delta, surface, evidence)693 elif (delta.get("news") or {}).get("added"):694 change.setdefault("notes", []).append("editorial items of a media publisher are not corporate communication events")695 structured_hit = bool(drafts)696 if not structured_hit:697 drafts += _text_diff_rules(surface, change, diff, delta)698699 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)]700 for d in drafts:701 default = EVENT_SUBTYPES.get(d.subtype, (EventType.OTHER, 0.3))[1]702 d.importance = scale_importance(default, significance, d.magnitude)703 d.confidence = EVIDENCE_CONFIDENCE.get(d.evidence, EVIDENCE_CONFIDENCE["html"])704 d.title = safe_wording(d.title)[:200]705 if d.summary:706 d.summary = safe_wording(d.summary)[:600]707 if kind == ChangeKind.CRITICAL and not d.review:708 d.review = "major_event"709 elif d.confidence < 0.5 and not d.review:710 d.review = "low_confidence"711712 needs = False713 reason = None714 if not drafts:715 needs, reason = True, "no_deterministic_event"716 elif surface in AMBIGUOUS_SURFACES or (not structured_hit and surface in (Surface.HOMEPAGE, Surface.ABOUT, Surface.PRODUCTS, Surface.SERVICES, Surface.SOLUTIONS)):717 needs, reason = True, "ambiguous_surface"718 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)})719 return Derived(events=drafts, needs_classification=needs, classification_reason=reason, summarize=summarize)720721722def dedupe_key_for(company_id: str, subtype: str, entity_key: str, sensor_id: str | None, day: date) -> str:723 return stable_hash(company_id, subtype, normalize_entity_key(entity_key), sensor_id or "", day.isoformat(), length=40)724725726# ============================================================================================================== persistence727728729async def _country_names(conn) -> dict[str, str]: # type: ignore[no-untyped-def]730 rows = await fetch_all(conn, "select code, name from countries")731 return {str(r["code"]).upper(): r["name"] for r in rows}732733734async def _baseline(conn, company_id: str) -> dict[str, Any]: # type: ignore[no-untyped-def]735 rows = await fetch_all(conn, "select metric, mean, stddev, samples from baselines where company_id = :c", c=company_id)736 return {r["metric"]: {"mean": r["mean"], "stddev": r["stddev"], "samples": r["samples"]} for r in rows}737738739async def _llm_budget_left(conn) -> int: # type: ignore[no-untyped-def]740 used = await fetch_val(conn, "select count(*) from llm_jobs where created_at >= date_trunc('day', now() at time zone 'utc')")741 return max(0, int(settings.llm_daily_budget) - int(used or 0))742743744async def _enqueue_llm(conn, *, kind: str, ref_id: str, company_id: str, budget: dict[str, int]) -> bool: # type: ignore[no-untyped-def]745 if not settings.llm_configured or budget["left"] <= 0:746 return False747 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)748 if exists:749 return False750 await execute(conn, "insert into llm_jobs (id, kind, ref_id, company_id, status) values (:id, :k, :r, :c, 'pending')",751 id=new_id("llm_job"), k=kind, r=ref_id, c=company_id)752 budget["left"] -= 1753 return True754755756async def persist_change_events(conn, change: dict[str, Any], company: dict[str, Any], sensor: dict[str, Any], derived: Derived, # type: ignore[no-untyped-def]757 *, budget: dict[str, int] | None = None, enqueue_llm: bool = True) -> dict[str, Any]:758 """Insert the drafts for one change (idempotent), cluster them, queue reviews and LLM jobs. Caller owns the transaction."""759 detected_at: datetime = change.get("detected_at") or datetime.now(UTC)760 if detected_at.tzinfo is None:761 detected_at = detected_at.replace(tzinfo=UTC)762 day = detected_at.astimezone(UTC).date()763 created: list[str] = []764 duplicates = 0765 budget = budget if budget is not None else {"left": await _llm_budget_left(conn)}766 for d in derived.events:767 key = dedupe_key_for(company["id"], d.subtype, d.entity_key, change.get("sensor_id"), day)768 event_id = new_id("event")769 payload = {**d.payload, "rules_version": RULES_VERSION, "evidence": d.evidence, "significance": change.get("significance"),770 "change_kind": change.get("kind"), "entity_key": d.entity_key}771 row = await fetch_one(conn, """772 insert into events (id, company_id, sensor_id, change_id, surface, event_type, event_subtype, importance, confidence, confidence_label,773 title, summary, old_value, new_value, payload, entities, tags, detected_at, effective_at, published_at, source_url,774 snapshot_before, snapshot_after, language, origin, schema_version, status, dedupe_key)775 values (:id, :company_id, :sensor_id, :change_id, :surface, :event_type, :subtype, :importance, :confidence, :label, :title, :summary,776 :old_value, :new_value, cast(:payload as jsonb), cast(:entities as jsonb), cast(:tags as text[]), :detected_at, :effective_at,777 :published_at, :source_url, :snap_before, :snap_after, :language, 'deterministic', :schema_version, 'active', :dedupe_key)778 on conflict (dedupe_key) do nothing779 returning id""",780 id=event_id, company_id=company["id"], sensor_id=change.get("sensor_id"), change_id=change["id"], surface=change.get("surface"),781 event_type=d.event_type, subtype=d.subtype, importance=d.importance, confidence=d.confidence, label=confidence_label(d.confidence),782 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),783 payload=jsonb(payload), entities=jsonb(d.entities), tags=list(dict.fromkeys(d.tags)), detected_at=detected_at, effective_at=d.effective_at,784 published_at=d.published_at, source_url=sensor.get("url"), snap_before=change.get("snapshot_before"), snap_after=change.get("snapshot_after"),785 language=((change.get("structured_delta") or {}).get("meta") or {}).get("language"), schema_version=SCHEMA_VERSION, dedupe_key=key)786 if row is None:787 continue788 created.append(event_id)789 if sensor.get("url"):790 await execute(conn, """791 insert into event_sources (event_id, sensor_id, source_url, snapshot_id, surface, detected_at, kind)792 values (:e, :s, :url, :snap, :surface, :at, 'primary') on conflict do nothing""",793 e=event_id, s=change.get("sensor_id"), url=sensor["url"], snap=change.get("snapshot_after"), surface=change.get("surface"), at=detected_at)794 ev = {"id": event_id, "company_id": company["id"], "sensor_id": change.get("sensor_id"), "surface": change.get("surface"), "event_type": d.event_type,795 "event_subtype": d.subtype, "title": d.title, "confidence": d.confidence, "detected_at": detected_at, "source_url": sensor.get("url"),796 "snapshot_after": change.get("snapshot_after")}797 _, dup = await attach_to_cluster(conn, ev, entity_key=d.entity_key)798 duplicates += int(dup)799 if d.review and not dup:800 await execute(conn, """801 insert into review_queue (id, kind, ref_id, company_id, payload) values (:id, :kind, :ref, :c, cast(:p as jsonb))""",802 id=new_id("review"), kind=d.review, ref=event_id, c=company["id"],803 p=jsonb({"event_subtype": d.subtype, "title": d.title, "confidence": d.confidence, "significance": change.get("significance")}))804 if enqueue_llm and not dup and d.subtype in derived.summarize and float(change.get("significance") or 0) >= settings.llm_min_significance:805 await _enqueue_llm(conn, kind="summarize_event", ref_id=event_id, company_id=company["id"], budget=budget)806807 if enqueue_llm and derived.needs_classification and float(change.get("significance") or 0) >= settings.llm_min_significance:808 await _enqueue_llm(conn, kind="classify_change", ref_id=change["id"], company_id=company["id"], budget=budget)809810 await execute(conn, "update changes set status = 'processed', processed_at = now() where id = :id and status <> 'enriched'", id=change["id"])811 if created:812 active = len(created) - duplicates813 if change.get("sensor_id"):814 await execute(conn, "update sensors set event_count = event_count + :n where id = :s", n=len(created), s=change["sensor_id"])815 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",816 at=detected_at, c=company["id"])817 log.info("events created", extra={"change_id": change["id"], "company": company.get("slug"), "events": len(created), "duplicates": duplicates,818 "active": active})819 return {"created": created, "duplicates": duplicates}820821822_CHANGE_SQL = """823 select c.*, s.url as sensor_url, s.connector_id, s.surface as sensor_surface, s.config as sensor_config,824 co.slug as company_slug, co.display_name, co.country as company_country, co.industries825 from changes c826 join sensors s on s.id = c.sensor_id827 join companies co on co.id = c.company_id828 where {where}829 order by c.detected_at830 limit :limit831 for update of c skip locked"""832833834def _split(row: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:835 company = {"id": row["company_id"], "slug": row["company_slug"], "display_name": row["display_name"], "country": row["company_country"],836 "industries": row.get("industries") or []}837 sensor = {"id": row["sensor_id"], "url": row["sensor_url"], "connector_id": row["connector_id"], "surface": row["sensor_surface"],838 "fetch_mode": (row.get("sensor_config") or {}).get("fetch_mode")}839 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",840 "company_country", "industries")}841 return change, company, sensor842843844async def process_pending_changes(limit: int = 200) -> dict[str, int]:845 """Claim pending changes (SKIP LOCKED), derive events, persist. Returns counters. Noise/minor pending rows are archived."""846 stats = {"changes": 0, "events": 0, "duplicates": 0, "archived": 0, "llm_jobs": 0}847 new_event_ids: list[str] = []848 async with transaction() as conn:849 rows = await fetch_all(conn, _CHANGE_SQL.format(where="c.status = 'pending'"), limit=limit)850 if not rows:851 return stats852 names = await _country_names(conn)853 budget = {"left": await _llm_budget_left(conn)}854 start_budget = budget["left"]855 baselines: dict[str, dict[str, Any]] = {}856 for row in rows:857 change, company, sensor = _split(row)858 if change.get("kind") in (ChangeKind.NOISE, ChangeKind.MINOR):859 await execute(conn, "update changes set status = 'archived', processed_at = now() where id = :id", id=change["id"])860 stats["archived"] += 1861 continue862 if company["id"] not in baselines:863 baselines[company["id"]] = await _baseline(conn, company["id"])864 derived = derive_events(change, company, sensor, baseline=baselines[company["id"]], country_names=names)865 res = await persist_change_events(conn, change, company, sensor, derived, budget=budget)866 stats["changes"] += 1867 stats["events"] += len(res["created"])868 stats["duplicates"] += res["duplicates"]869 new_event_ids += res["created"]870 stats["llm_jobs"] = start_budget - budget["left"]871 if new_event_ids:872 try:873 from companyatlas.services.alerts import evaluate_alerts874875 await evaluate_alerts(new_event_ids)876 except Exception:877 log.exception("alert evaluation failed")878 return stats879880881async def reprocess_events(since: datetime, *, limit: int = 5000, company_id: str | None = None) -> dict[str, int]:882 """Re-run the deterministic rules over already processed changes (no refetch). Dedupe keys make this idempotent; new rules add events."""883 stats = {"changes": 0, "events": 0, "duplicates": 0}884 where = "c.status in ('processed', 'enriched') and c.detected_at >= :since and c.kind in ('meaningful', 'major', 'critical')"885 params: dict[str, Any] = {"since": since, "limit": limit}886 if company_id:887 where += " and c.company_id = :company_id"888 params["company_id"] = company_id889 async with transaction() as conn:890 rows = await fetch_all(conn, _CHANGE_SQL.format(where=where), **params)891 names = await _country_names(conn)892 for row in rows:893 change, company, sensor = _split(row)894 derived = derive_events(change, company, sensor, baseline=await _baseline(conn, company["id"]), country_names=names)895 res = await persist_change_events(conn, change, company, sensor, derived, enqueue_llm=False)896 stats["changes"] += 1897 stats["events"] += len(res["created"])898 stats["duplicates"] += res["duplicates"]899 return stats900901902async def list_events(*, company: str | None = None, event_type: str | None = None, limit: int = 50) -> list[dict[str, Any]]:903 where = ["e.status in ('active', 'review')"]904 params: dict[str, Any] = {"limit": limit}905 if company:906 where.append("(co.slug = :company or co.id = :company)")907 params["company"] = company908 if event_type:909 where.append("(e.event_type = :t or e.event_subtype = :t)")910 params["t"] = event_type.upper()911 async with transaction() as conn:912 return await fetch_all(conn, f"""913 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,914 e.surface, e.cluster_id, e.status915 from events e join companies co on co.id = e.company_id916 where {' and '.join(where)} order by e.detected_at desc limit :limit""", **params)917918919@periodic("process-changes", every_s=20)920async def process_changes_task() -> None:921 stats = await process_pending_changes(limit=200)922 if stats["changes"] or stats["archived"]:923 log.info("process-changes", extra=stats)924925926__all__ = [927 "RULES_VERSION",928 "Derived",929 "EventDraft",930 "dedupe_key_for",931 "derive_events",932 "evidence_kind",933 "list_events",934 "persist_change_events",935 "process_pending_changes",936 "reprocess_events",937 "safe_wording",938 "scale_importance",939]940