"""Canonical, site-agnostic data structures. Every site recipe normalises its raw extraction into `Lead`, so leads from any source share one funnel table and one dashboard view. """ from __future__ import annotations import json import time from dataclasses import dataclass, field, asdict from enum import Enum from typing import Any, Optional class FunnelStage(str, Enum): NEW = "NEW" # name/headline only ENRICHED = "ENRICHED" # company / title / location / about captured CONTACT_FOUND = "CONTACT_FOUND" # email / phone / website captured EXPORTED = "EXPORTED" # pushed downstream (set by the user) @dataclass class Lead: # identity full_name: str = "" first_name: str = "" last_name: str = "" headline: str = "" title: str = "" # organisation company: str = "" company_url: str = "" industry: str = "" location: str = "" about: str = "" # contact (often empty on the first pass) email: str = "" phone: str = "" website: str = "" social_handles: dict[str, str] = field(default_factory=dict) # provenance source_site: str = "" source_url: str = "" captured_at: float = field(default_factory=time.time) updated_at: float = field(default_factory=time.time) screenshot_refs: list[str] = field(default_factory=list) # quality confidence: float = 0.0 fields_found: list[str] = field(default_factory=list) needs_review: bool = False # funnel stage: str = FunnelStage.NEW.value def compute_stage(self) -> str: """Advance the funnel stage from data completeness (never regress EXPORTED).""" if self.stage == FunnelStage.EXPORTED.value: return self.stage if self.email or self.phone or self.website or self.social_handles: return FunnelStage.CONTACT_FOUND.value if self.company or self.title or self.location or self.about: return FunnelStage.ENRICHED.value return FunnelStage.NEW.value def finalize(self) -> "Lead": """Derive first/last name, fields_found, and funnel stage.""" if self.full_name and not (self.first_name or self.last_name): parts = self.full_name.split() if parts: self.first_name = parts[0] self.last_name = " ".join(parts[1:]) self.fields_found = [ k for k, v in asdict(self).items() if k not in ("fields_found", "stage", "needs_review", "captured_at", "updated_at", "confidence") and v ] self.stage = self.compute_stage() return self def to_dict(self) -> dict[str, Any]: return asdict(self) @dataclass class PageMatch: """Result of a recipe testing a page context.""" site_id: str page_type: str # e.g. "profile", "search", "feed", "other" is_target: bool # True only for a page we should capture @dataclass class PageContext: """What the detector knows about the current front tab.""" url: Optional[str] = None ai_site: Optional[str] = None # site guessed visually by the AI ai_page_type: Optional[str] = None # page type guessed visually ai_url_text: Optional[str] = None # address-bar text read from a screenshot @property def best_url(self) -> str: return (self.url or self.ai_url_text or "").strip() @dataclass class DetectionResult: ok: bool # a valid, enabled target page is in front reason: str = "" # human-readable explanation (shown on stop) recipe_id: Optional[str] = None page_type: Optional[str] = None url: str = "" method: str = "" # "url" | "visual" def dumps(obj: Any) -> str: return json.dumps(obj, ensure_ascii=False)