"""Typed contracts shared by connectors, the pipeline and the intelligence layer. FetchResult (fetch.py) → Extraction (this module) → snapshot row + blocks object → BlockDiff / StructuredDelta → changes row ↓ events service reads changes.structured_delta Everything is plain dataclasses / pydantic so it serialises to JSON for the `snapshots.extracted` and `changes.structured_delta` columns. Never put raw HTML in these — raw bytes live in the object store, referenced by hash. """ from __future__ import annotations from dataclasses import asdict, dataclass, field from datetime import datetime from typing import Any, Literal from companyatlas.taxonomy import Surface BlockKind = Literal["header", "hero", "nav", "section", "heading", "paragraph", "list", "table", "product_card", "pricing_plan", "job_listing", "person", "location", "news_item", "faq", "footer", "code", "quote", "other"] @dataclass(slots=True) class Block: """A semantic block of a page (spec §19). `key` is the stable identity used for added/removed/modified/moved diffing: it is derived from the block kind + heading path + a fuzzy content fingerprint, never from DOM position alone.""" key: str kind: str text: str path: str = "" # heading breadcrumb, e.g. "Pricing > Pro" hash: str = "" # sha256 of normalized text (exact) simhash: int = 0 # 64-bit near-duplicate fingerprint weight: float = 1.0 # importance weight (hero/pricing/job > footer/nav) order: int = 0 attrs: dict[str, Any] = field(default_factory=dict) def to_json(self) -> dict[str, Any]: return asdict(self) @dataclass(slots=True) class ExtractedJob: title: str url: str | None = None external_id: str | None = None department: str | None = None team: str | None = None location_text: str | None = None city: str | None = None region: str | None = None country: str | None = None # ISO-2 when confidently derivable, else None (never guess) remote: bool | None = None employment_type: str | None = None seniority: str | None = None skills: list[str] = field(default_factory=list) salary_min: float | None = None salary_max: float | None = None salary_currency: str | None = None salary_period: str | None = None posted_at: datetime | None = None description_hash: str | None = None raw: dict[str, Any] = field(default_factory=dict) @dataclass(slots=True) class ExtractedPerson: name: str title: str | None = None role_category: str | None = None # ceo | cfo | cto | coo | founder | president | chair | board | vp | head | other is_executive: bool = False url: str | None = None @dataclass(slots=True) class ExtractedProduct: name: str url: str | None = None category: str | None = None description: str | None = None @dataclass(slots=True) class ExtractedPlan: plan_name: str price: float | None = None price_text: str | None = None currency: str | None = None billing_period: str | None = None # month | year | one_time | usage | contact unit: str | None = None features: list[str] = field(default_factory=list) contact_sales: bool = False @dataclass(slots=True) class ExtractedLocation: name: str kind: str = "office" # headquarters | office | store | factory | warehouse | lab | data_center | other city: str | None = None region: str | None = None country: str | None = None address_text: str | None = None # only if the page states it; never invented @dataclass(slots=True) class ExtractedNewsItem: title: str url: str published_at: datetime | None = None summary: str | None = None category: str | None = None # press | blog | changelog | research | ir | other language: str | None = None @dataclass(slots=True) class DiscoveredUrl: url: str surface: Surface confidence: float anchor: str | None = None method: str = "nav" # nav | sitemap | robots | pattern | ats | feed | link | jsonld @dataclass(slots=True) class Extraction: """Output of `Connector.extract`. `text` is the normalized main-content text (what gets diffed at the text level), `blocks` the semantic blocks (block-level diff), the typed lists feed entity tables and structured deltas.""" text: str blocks: list[Block] title: str | None = None language: str | None = None meta: dict[str, Any] = field(default_factory=dict) # description, og tags, canonical, generator, jsonld types… jobs: list[ExtractedJob] = field(default_factory=list) people: list[ExtractedPerson] = field(default_factory=list) products: list[ExtractedProduct] = field(default_factory=list) plans: list[ExtractedPlan] = field(default_factory=list) locations: list[ExtractedLocation] = field(default_factory=list) news: list[ExtractedNewsItem] = field(default_factory=list) discovered: list[DiscoveredUrl] = field(default_factory=list) structured_hash: str = "" # hash of the structured payload (typed lists) — set by the pipeline if empty normalized_hash: str = "" # hash of `text` — set by the pipeline if empty def summary(self) -> dict[str, Any]: return {"job_count": len(self.jobs), "people_count": len(self.people), "product_count": len(self.products), "plan_count": len(self.plans), "location_count": len(self.locations), "news_count": len(self.news), "block_count": len(self.blocks), "text_length": len(self.text), "discovered_count": len(self.discovered)} def structured_payload(self) -> dict[str, Any]: return {"jobs": [asdict(j) for j in self.jobs], "people": [asdict(p) for p in self.people], "products": [asdict(p) for p in self.products], "plans": [asdict(p) for p in self.plans], "locations": [asdict(loc) for loc in self.locations], "news": [asdict(n) for n in self.news], "meta": self.meta} @dataclass(slots=True) class BlockDelta: key: str kind: str path: str before: str | None after: str | None weight: float similarity: float | None = None # for modified blocks @dataclass(slots=True) class BlockDiff: """Result of comparing two block lists (spec §19–20). `significance` ∈ [0,1] is computed by the diff engine from the weighted share of changed content, page importance and the typed deltas; the pipeline maps it to a ChangeKind.""" added: list[BlockDelta] = field(default_factory=list) removed: list[BlockDelta] = field(default_factory=list) modified: list[BlockDelta] = field(default_factory=list) moved: list[str] = field(default_factory=list) text_delta_ratio: float = 0.0 similarity: float = 1.0 significance: float = 0.0 reasons: list[str] = field(default_factory=list) @property def is_empty(self) -> bool: return not (self.added or self.removed or self.modified) def to_json(self, *, limit: int = 40) -> dict[str, Any]: def _cut(items: list[BlockDelta]) -> list[dict[str, Any]]: out = [] for d in items[:limit]: out.append({"key": d.key, "kind": d.kind, "path": d.path, "before": (d.before or "")[:600] or None, "after": (d.after or "")[:600] or None, "weight": d.weight, "similarity": d.similarity}) return out return {"added": _cut(self.added), "removed": _cut(self.removed), "modified": _cut(self.modified), "moved": self.moved[:limit], "counts": {"added": len(self.added), "removed": len(self.removed), "modified": len(self.modified), "moved": len(self.moved)}, "text_delta_ratio": round(self.text_delta_ratio, 4), "similarity": round(self.similarity, 4), "reasons": self.reasons} # StructuredDelta — JSON stored in `changes.structured_delta`, produced by the pipeline when it reconciles typed extractions # with the entity tables. Shape (all keys optional, lists bounded to 200 items): # { # "jobs": {"added": [ {title, url, location_text, country, remote, department, is_ai} ], "removed": [...], "open_before": n, "open_after": n}, # "people": {"added": [ {name, title, role_category} ], "removed": [...], "title_changed": [ {name, before, after} ]}, # "products": {"added": [ {name, url} ], "removed": [...]}, # "plans": {"added": [ {plan_name, price, currency, billing_period} ], "removed": [...], # "price_changed": [ {plan_name, before, after, currency, billing_period, pct} ]}, # "locations": {"added": [ {name, city, country, kind} ], "removed": [...], "new_countries": ["JP"]}, # "news": {"added": [ {title, url, published_at, category} ]}, # "meta": {"title_changed": {before, after}, "description_changed": bool, "language": "en"} # } StructuredDelta = dict[str, Any] __all__ = ["Block", "BlockDelta", "BlockDiff", "BlockKind", "DiscoveredUrl", "ExtractedJob", "ExtractedLocation", "ExtractedNewsItem", "ExtractedPerson", "ExtractedPlan", "ExtractedProduct", "Extraction", "StructuredDelta"]