"""Company profile enrichment (reference atlas): Wikidata entity → Wikipedia summary → homepage facts → optional grounded LLM text. Every accepted value carries provenance `{field, source, url, retrieved_at}` and a better source is never overwritten by a weaker one (`FIELD_RANKS`). Nothing is inferred: what a source does not state stays `null`. result = await enrich_company(company, fetcher=fetcher) # pure: network via `fetcher`, no writes (reads stored snapshots) await persist(conn, company, result) # companies.source_meta.profile + column back-fills + people + relationships await enrich_pending(limit=300, concurrency=4) # batch runner, also the periodic task `company-enrichment` Sources - Wikidata `wbgetentities` (batched, JSON, labels/descriptions/claims/sitelinks) + labels-only lookups + `wbgetclaims` for country ISO codes (P297) and currency codes (P498). Wikimedia asks API clients for a descriptive User-Agent and moderate rates, not robots.txt (which targets page crawlers) — the API calls use `respect_robots=False` at ≤ 5 req/s (`settings.enrich_wikidata_rate_per_min`). - Wikipedia REST `page/summary` (≤ 10 req/s): `extract` → description under CC BY-SA 4.0 with attribution URL. - Homepage: the latest stored homepage snapshot (raw object re-parsed: meta description, JSON-LD Organization, icons) — fetched only when no snapshot exists. Icon/logo URLs are http(s)-only and SSRF-validated. - LLM (`llm_jobs` kind `company_profile`, medium model, budgeted): only when no Wikipedia extract exists and ≥ 400 chars of first-party text are available; figures absent from the source text reject the output (`numbers_grounded`). Column back-fills (description, logo_url, hq_*, country, founded_year, employees, ticker/exchange, industries, legal_name, lei, sec_cik) happen when the column is null or the new source outranks the recorded one (`source_meta.provenance[column]`); the replaced value is kept under `previous`. """ from __future__ import annotations import asyncio import contextlib import json import logging import re import time from dataclasses import dataclass, field from datetime import UTC, date, datetime, timedelta from typing import Any from urllib.parse import quote, unquote, urlencode from selectolax.lexbor import LexborHTMLParser from companyatlas import archive from companyatlas.config import settings from companyatlas.connectors._util import country_code, norm_name from companyatlas.db import connection, execute, fetch_all, fetch_one, fetch_val, jsonb, transaction from companyatlas.fetch import Fetcher, FetchError, FetchResult, decode_text, validate_destination_async from companyatlas.ids import new_id from companyatlas.registry.industries import is_valid_slug, map_industry from companyatlas.sdk.normalize import extract_jsonld, normalize_whitespace from companyatlas.sdk.normalize import parse as parse_page from companyatlas.services.periodic import periodic from companyatlas.taxonomy import FORBIDDEN_WORDING from companyatlas.urls import absolutize, host_of log = logging.getLogger(__name__) PROFILE_VERSION = "profile-v1" USER_AGENT = settings.user_agent WIKIDATA_API = "https://www.wikidata.org/w/api.php" WIKIDATA_ITEM = "https://www.wikidata.org/wiki/{qid}" COMMONS_FILE = "https://commons.wikimedia.org/wiki/Special:FilePath/{name}" WIKIPEDIA_SUMMARY = "https://{lang}.wikipedia.org/api/rest_v1/page/summary/{title}" WIKIPEDIA_LICENSE = "CC BY-SA 4.0" LLM_ATTRIBUTION = "Generated from the company's public pages" HOMEPAGE_ATTRIBUTION = "Meta description of the company's homepage" REF_CACHE_KEY = "enrichment:wikidata_refs" SOURCES = ("wikidata", "wikipedia", "homepage", "llm") DEFAULT_RANKS = {"wikidata": 4, "homepage": 3, "registry": 2, "wikipedia": 1, "llm": 0} FIELD_RANKS: dict[str, dict[str, int]] = { "description": {"wikipedia": 4, "llm": 3, "homepage": 2, "wikidata": 1, "registry": 0}, "logo_url": {"wikidata": 4, "homepage": 3, "registry": 2, "wikipedia": 1}, "icon_url": {"homepage": 3}, } # profile field → companies column (only these are ever back-filled). COLUMN_FIELDS: dict[str, str] = { "description": "description", "logo_url": "logo_url", "hq_city": "hq_city", "hq_region": "hq_region", "country": "country", "founded_year": "founded_year", "employees": "employees", "ticker": "ticker", "exchange": "exchange", "legal_name": "legal_name", "lei": "lei", "sec_cik": "sec_cik", } # Wikidata property ids used below. P = { "official_name": "P1448", "inception": "P571", "legal_form": "P1454", "hq": "P159", "coords": "P625", "country": "P17", "employees": "P1128", "point_in_time": "P585", "revenue": "P2139", "net_income": "P2295", "total_assets": "P2403", "industry": "P452", "products": "P1056", "ticker": "P249", "exchange": "P414", "isin": "P946", "lei": "P1278", "cik": "P5531", "website": "P856", "logo": "P154", "linkedin": "P4264", "x": "P2002", "youtube": "P2397", "facebook": "P2013", "instagram": "P2003", "github": "P2037", "tiktok": "P7085", "crunchbase": "P2088", "ceo": "P169", "chair": "P488", "founder": "P112", "key_people": "P3320", "parent": "P749", "subsidiary": "P355", "owned_by": "P127", "owner_of": "P1830", "start": "P580", "end": "P582", "position": "P39", "role": "P2868", "region_in": "P131", "iso2": "P297", "iso4217": "P498", } SOCIAL_TEMPLATES = { "linkedin": ("P4264", "https://www.linkedin.com/company/{}"), "x": ("P2002", "https://x.com/{}"), "youtube": ("P2397", "https://www.youtube.com/channel/{}"), "facebook": ("P2013", "https://www.facebook.com/{}"), "instagram": ("P2003", "https://www.instagram.com/{}"), "github": ("P2037", "https://github.com/{}"), "tiktok": ("P7085", "https://www.tiktok.com/@{}"), "crunchbase": ("P2088", "https://www.crunchbase.com/organization/{}"), } SOCIAL_HOSTS = {"linkedin.com": "linkedin", "twitter.com": "x", "x.com": "x", "youtube.com": "youtube", "facebook.com": "facebook", "instagram.com": "instagram", "github.com": "github", "tiktok.com": "tiktok", "crunchbase.com": "crunchbase"} PEOPLE_ROLES = {"P169": ("Chief Executive Officer", "ceo", True), "P488": ("Chairperson", "chair", True), "P112": ("Founder", "founder", True), "P3320": ("Key person", "other", False)} RELATION_KINDS = {"P749": "SUBSIDIARY_OF", "P355": "PARENT_OF", "P127": "OWNED_BY", "P1830": "OWNER_OF"} INVERSE_KIND = {"PARENT_OF": "SUBSIDIARY_OF", "SUBSIDIARY_OF": "PARENT_OF", "OWNED_BY": "OWNER_OF", "OWNER_OF": "OWNED_BY"} # Country → Wikipedia language used when the entity has no English sitelink. COUNTRY_LANG = {"DE": "de", "AT": "de", "CH": "de", "FR": "fr", "BE": "fr", "LU": "fr", "JP": "ja", "CN": "zh", "TW": "zh", "HK": "zh", "IT": "it", "ES": "es", "MX": "es", "AR": "es", "CL": "es", "CO": "es", "PE": "es", "KR": "ko", "RU": "ru", "BR": "pt", "PT": "pt", "NL": "nl", "SE": "sv", "NO": "no", "DK": "da", "FI": "fi", "PL": "pl", "TR": "tr", "ID": "id", "VN": "vi", "TH": "th", "CZ": "cs", "HU": "hu", "GR": "el", "IL": "he", "SA": "ar", "AE": "ar", "EG": "ar", "UA": "uk", "RO": "ro", "IR": "fa", "IN": "en", "MY": "ms"} LABEL_LANGS = ["en", "fr", "de", "es", "it", "pt", "nl", "ja", "zh", "ko", "ru", "sv", "pl", "tr", "mul"] SITEFILTER = sorted({f"{lang}wiki" for lang in COUNTRY_LANG.values()} | {"enwiki"}) # Common currencies (Wikidata item → ISO 4217); anything else is resolved live through P498 and cached. CURRENCIES = {"Q4917": "USD", "Q4916": "EUR", "Q25224": "GBP", "Q8146": "JPY", "Q39099": "CNY", "Q25344": "CHF", "Q1104069": "CAD", "Q259502": "AUD", "Q80524": "INR", "Q202040": "KRW", "Q122922": "SEK", "Q31015": "HKD", "Q190951": "SGD", "Q173117": "BRL", "Q4730": "MXN", "Q181907": "ZAR", "Q41044": "RUB", "Q208526": "TWD", "Q132643": "NOK", "Q25417": "DKK", "Q123213": "PLN", "Q172872": "TRY", "Q41588": "IDR", "Q199109": "SAR", "Q200294": "AED", "Q1472704": "NZD", "Q177882": "THB", "Q163712": "MYR", "Q131309": "ILS", "Q131016": "CZK", "Q47190": "HUF", "Q17193": "PHP", "Q199462": "EGP", "Q203567": "NGN", "Q200050": "CLP", "Q244819": "COP", "Q199578": "ARS", "Q188289": "PKR", "Q192090": "VND", "Q202714": "KES", "Q206386": "QAR", "Q319176": "KWD"} _NUMBER_RE = re.compile(r"\d[\d,.   ]*\d|\d") _CITATION_RE = re.compile(r"\[\d+\]|\[[a-z]\]|\[citation needed\]", re.IGNORECASE) _PAREN_PRON_RE = re.compile(r"\s*\((?:[^()]*?(?:pronounced|listen|ⓘ|/[^/()]+/)[^()]*)\)") def _now() -> datetime: return datetime.now(UTC).replace(microsecond=0) def _iso(dt: datetime | None = None) -> str: return (dt or _now()).isoformat() # ================================================================================================================ profile builder def empty_profile() -> dict[str, Any]: return { "description": None, "description_source": None, "description_url": None, "description_license": None, "description_attribution": None, "logo_url": None, "icon_url": None, "founded_year": None, "legal_form": None, "legal_name": None, "employees": None, "employees_year": None, "revenue": None, "net_income": None, "total_assets": None, "hq": {"city": None, "region": None, "country": None, "address": None, "lat": None, "lon": None}, "ticker": None, "exchange": None, "isin": None, "lei": None, "sec_cik": None, "public_company": False, "wikipedia_url": None, "wikidata_url": None, "official_website": None, "phone": None, "products": [], "industries": [], "industry_labels": [], "socials": {}, "enriched_at": None, "sources": [], "version": PROFILE_VERSION, } def rank_of(field_name: str, source: str) -> int: return FIELD_RANKS.get(field_name, DEFAULT_RANKS).get(source, DEFAULT_RANKS.get(source, 0)) HQ_FIELDS = {"hq_city": "city", "hq_region": "region", "country": "country", "hq_address": "address", "hq_lat": "lat", "hq_lon": "lon"} class ProfileBuilder: """Rank-aware assignment: a field is (re)assigned only when the new source outranks the one that set it. One provenance row per field.""" def __init__(self, profile: dict[str, Any] | None = None) -> None: self.profile = profile or empty_profile() self.provenance: dict[str, dict[str, Any]] = {} def set(self, field_name: str, value: Any, *, source: str, url: str | None, retrieved_at: str | None = None) -> bool: if value in (None, "", [], {}): return False current = self.provenance.get(field_name) if current is not None and rank_of(field_name, source) <= rank_of(field_name, current["source"]): return False if field_name in HQ_FIELDS: self.profile["hq"][HQ_FIELDS[field_name]] = value else: self.profile[field_name] = value self.provenance[field_name] = {"field": field_name, "source": source, "url": url, "retrieved_at": retrieved_at or _iso()} return True def get(self, field_name: str) -> Any: if field_name in HQ_FIELDS: return self.profile["hq"].get(HQ_FIELDS[field_name]) return self.profile.get(field_name) def source_of(self, field_name: str) -> str | None: p = self.provenance.get(field_name) return p["source"] if p else None def finish(self) -> dict[str, Any]: self.profile["public_company"] = bool(self.profile.get("public_company") or self.profile.get("ticker") or self.profile.get("isin")) self.profile["sources"] = sorted(self.provenance.values(), key=lambda p: p["field"]) self.profile["enriched_at"] = _iso() return self.profile def seed_from_company(builder: ProfileBuilder, company: dict[str, Any]) -> None: """Existing columns form the base layer (source = recorded provenance or `registry`); every later source may outrank them.""" prov = _dict(company.get("source_meta")).get("provenance") or {} base_at = _dict(company.get("source_meta")).get("seeded_at") or _iso(company.get("created_at") if isinstance(company.get("created_at"), datetime) else None) def src(col: str) -> str: return (prov.get(col) or {}).get("source") or "registry" for col in ("description", "logo_url", "hq_city", "hq_region", "country", "founded_year", "employees", "ticker", "exchange", "legal_name", "lei", "sec_cik"): builder.set(col, company.get(col), source=src(col), url=(prov.get(col) or {}).get("url"), retrieved_at=base_at) if company.get("industries"): builder.set("industries", list(company["industries"]), source=src("industries"), url=None, retrieved_at=base_at) labels = _dict(company.get("source_meta")).get("industry_labels") if labels: builder.set("industry_labels", list(labels)[:12], source="registry", url=None, retrieved_at=base_at) if company.get("public_company"): builder.profile["public_company"] = True if company.get("wikidata_id"): builder.set("wikidata_url", WIKIDATA_ITEM.format(qid=company["wikidata_id"]), source="registry", url=None, retrieved_at=base_at) if company.get("description"): builder.profile["description_source"] = "wikidata" if src("description") == "registry" else src("description") # ================================================================================================================ Wikidata def _dict(value: Any) -> dict[str, Any]: if isinstance(value, str): try: value = json.loads(value) except ValueError: return {} return value if isinstance(value, dict) else {} def _snak_value(snak: dict[str, Any] | None) -> Any: if not snak or snak.get("snaktype") != "value": return None return (snak.get("datavalue") or {}).get("value") def statement_value(st: dict[str, Any]) -> Any: return _snak_value(st.get("mainsnak")) def statement_qid(st: dict[str, Any]) -> str | None: v = statement_value(st) return v.get("id") if isinstance(v, dict) and v.get("entity-type") == "item" else None def qualifier(st: dict[str, Any], prop: str) -> Any: snaks = (st.get("qualifiers") or {}).get(prop) or [] return _snak_value(snaks[0]) if snaks else None def wd_time(value: Any) -> tuple[int | None, date | None, int]: """Wikidata time → (year, date-or-None, precision). Day precision (11) gives a full date, month (10) the 1st, year (9) Jan 1.""" if not isinstance(value, dict) or not value.get("time"): return None, None, 0 t = str(value["time"]) precision = int(value.get("precision") or 9) m = re.match(r"^([+-])(\d+)-(\d\d)-(\d\d)T", t) if not m: return None, None, precision sign, y, mo, d = m.groups() year = int(y) * (-1 if sign == "-" else 1) if year <= 0 or precision < 9: return (year if year > 0 else None), None, precision try: dt = date(year, int(mo) if precision >= 10 and int(mo) else 1, int(d) if precision >= 11 and int(d) else 1) except ValueError: dt = None return year, dt, precision def wd_quantity(value: Any) -> tuple[float | None, str | None]: if not isinstance(value, dict) or value.get("amount") is None: return None, None try: amount = float(str(value["amount"]).replace("+", "")) except ValueError: return None, None unit = str(value.get("unit") or "1") return amount, (unit.rsplit("/", 1)[-1] if unit.startswith("http") else None) def _end_date(st: dict[str, Any]) -> date | None: _y, d, _p = wd_time(qualifier(st, P["end"])) return d def _start_date(st: dict[str, Any]) -> date | None: _y, d, _p = wd_time(qualifier(st, P["start"])) return d def is_current(st: dict[str, Any], today: date | None = None) -> bool: end = _end_date(st) return end is None or end > (today or _now().date()) def claims(entity: dict[str, Any], prop: str, *, current_only: bool = False) -> list[dict[str, Any]]: """Statements for `prop`: deprecated dropped, preferred first, then current (no end date) before ended; Wikidata's order otherwise.""" out = [s for s in (entity.get("claims") or {}).get(prop) or [] if s.get("rank") != "deprecated" and statement_value(s) is not None] if current_only: out = [s for s in out if is_current(s)] return sorted(out, key=lambda s: (s.get("rank") != "preferred", not is_current(s))) def latest_quantity(entity: dict[str, Any], prop: str) -> tuple[float, str | None, int | None] | None: """(amount, unit qid, year) for the observation with the most recent point in time (P585); preferred rank wins ties.""" best: tuple[tuple[int, int], float, str | None, int | None] | None = None for st in claims(entity, prop): amount, unit = wd_quantity(statement_value(st)) if amount is None: continue year, _d, _p = wd_time(qualifier(st, P["point_in_time"])) key = (year or 0, 1 if st.get("rank") == "preferred" else 0) if best is None or key > best[0]: best = (key, amount, unit, year) return (best[1], best[2], best[3]) if best else None def commons_url(filename: str) -> str: return COMMONS_FILE.format(name=quote(filename.strip().replace(" ", "_"), safe="")) def wikipedia_url(lang: str, title: str) -> str: return f"https://{lang}.wikipedia.org/wiki/{quote(title.replace(' ', '_'), safe=':()/,')}" class WikidataClient: """Batched read access to the Wikidata API (labels cached for the process; country/currency codes cached in `settings_kv`).""" def __init__(self, fetcher: Any, *, refs: dict[str, Any] | None = None) -> None: self.fetcher = fetcher self.labels_cache: dict[str, str | None] = {} self.descriptions_cache: dict[str, str | None] = {} self.refs: dict[str, Any] = refs if refs is not None else {} # qid → {"iso2": …} / {"iso4217": …} self.requests = 0 # ---------------------------------------------------------------------------------------------- transport @staticmethod def url(params: dict[str, str]) -> str: return WIKIDATA_API + "?" + urlencode({"format": "json", **params}) async def _get(self, url: str, *, max_bytes: int | None = None) -> dict[str, Any] | None: self.requests += 1 try: res: FetchResult = await self.fetcher.get(url, accept="application/json", rate_per_min=settings.enrich_wikidata_rate_per_min, respect_robots=False, max_bytes=max_bytes or settings.enrich_wikidata_max_bytes) except FetchError as exc: log.warning("wikidata request failed", extra={"url": url[:200], "error": str(exc)[:200]}) return None try: data = res.json() except ValueError: return None return data if isinstance(data, dict) else None # ---------------------------------------------------------------------------------------------- entities / labels @staticmethod def entities_url(qids: list[str]) -> str: return WikidataClient.url({"action": "wbgetentities", "ids": "|".join(qids), "props": "labels|descriptions|claims|sitelinks", "languages": "|".join(LABEL_LANGS), "sitefilter": "|".join(SITEFILTER)}) @staticmethod def labels_url(qids: list[str]) -> str: return WikidataClient.url({"action": "wbgetentities", "ids": "|".join(qids), "props": "labels|descriptions", "languages": "|".join(LABEL_LANGS)}) @staticmethod def claims_url(qid: str, prop: str) -> str: return WikidataClient.url({"action": "wbgetclaims", "entity": qid, "property": prop, "props": ""}) async def entities(self, qids: list[str]) -> dict[str, dict[str, Any]]: out: dict[str, dict[str, Any]] = {} ids = list(dict.fromkeys(q for q in qids if q)) size = max(1, settings.enrich_wikidata_entity_batch) for i in range(0, len(ids), size): chunk = ids[i:i + size] data = await self._get(self.entities_url(chunk)) if data is None and len(chunk) > 1: # too large / transient → one by one for q in chunk: single = await self._get(self.entities_url([q])) out.update({k: v for k, v in ((single or {}).get("entities") or {}).items() if "missing" not in v}) continue out.update({k: v for k, v in ((data or {}).get("entities") or {}).items() if "missing" not in v}) for ent in out.values(): lab = _pick_label(ent.get("labels") or {}) if lab: self.labels_cache[ent["id"]] = lab self.descriptions_cache.setdefault(ent["id"], ((ent.get("descriptions") or {}).get("en") or {}).get("value")) return out async def labels(self, qids: list[str]) -> dict[str, str]: missing = list(dict.fromkeys(q for q in qids if q and q not in self.labels_cache)) size = max(1, min(50, settings.enrich_wikidata_label_batch)) for i in range(0, len(missing), size): chunk = missing[i:i + size] data = await self._get(self.labels_url(chunk), max_bytes=settings.max_body_bytes) for q in chunk: ent = ((data or {}).get("entities") or {}).get(q) or {} self.labels_cache[q] = _pick_label(ent.get("labels") or {}) self.descriptions_cache[q] = ((ent.get("descriptions") or {}).get("en") or {}).get("value") return {q: lab for q in qids if (lab := self.labels_cache.get(q))} def description_of(self, qid: str) -> str | None: return self.descriptions_cache.get(qid) async def claim_strings(self, qid: str, prop: str) -> list[str]: data = await self._get(self.claims_url(qid, prop), max_bytes=settings.max_body_bytes) out: list[str] = [] for st in ((data or {}).get("claims") or {}).get(prop) or []: v = statement_value(st) if isinstance(v, str) and st.get("rank") != "deprecated": out.append(v) return out async def country_iso(self, qid: str | None) -> str | None: if not qid: return None ref = self.refs.get(qid) if ref and "iso2" in ref: return ref["iso2"] codes = [c for c in await self.claim_strings(qid, P["iso2"]) if len(c) == 2 and c.isalpha()] code = codes[0].upper() if codes else None self.refs[qid] = {**(ref or {}), "iso2": code} return code async def currency_code(self, qid: str | None) -> str | None: if not qid: return None if qid in CURRENCIES: return CURRENCIES[qid] ref = self.refs.get(qid) if ref and "iso4217" in ref: return ref["iso4217"] codes = [c for c in await self.claim_strings(qid, P["iso4217"]) if len(c) == 3 and c.isalpha()] code = codes[0].upper() if codes else None self.refs[qid] = {**(ref or {}), "iso4217": code} return code LATIN_LANGS = ("en", "mul", "fr", "de", "es", "it", "pt", "nl", "sv", "pl", "tr") def _pick_label(labels: dict[str, Any]) -> str | None: """English first, then the multilingual label, then Latin-script languages, then anything (a Japanese-only item keeps its Japanese name).""" for lang in (*LATIN_LANGS, *LABEL_LANGS): v = labels.get(lang) if isinstance(v, dict) and v.get("value"): return str(v["value"]) for v in labels.values(): if isinstance(v, dict) and v.get("value"): return str(v["value"]) return None # Relationship targets (P355 subsidiaries, P1830 "owner of") are kept only when their English description reads like an organisation; # Wikidata lists domains, buildings, fonts and apps under "owner of". COMPANY_WORDS = re.compile(r"\b(compan(y|ies)|corporation|subsidiar(y|ies)|business|enterprise|manufacturer|bank|airline|firm|holding|startup|start-up|" r"developer|publisher|studio|retailer|provider|provides|services|operator|conglomerate|group|agency|label|network|carrier|brewery|" r"insurer|utility|railway|shipyard|automaker|chain|organi[sz]ation|joint venture|division|venture|fund|institution|cooperative|" r"maker|producer|distributor|supplier|vendor|consultancy|consulting|contractor|lender|broker|marketplace|team)\b", re.IGNORECASE) NON_COMPANY_WORDS = re.compile(r"\b(domain|top-level|building|skyscraper|font|typeface|software|website|web service|application|app|programming language|" r"file format|protocol|operating system|video game|film|album|song|book|magazine|television|product|device|smartphone|laptop|" r"car model|aircraft|satellite|rocket|street|campus|headquarters|stadium|hotel|data center|datacenter|patent|trademark|logo|" r"mascot|character|person|human|browser|search engine|brand of|line of|series of|technology|feature)\b", re.IGNORECASE) def looks_like_organisation(description: str | None, *, default: bool) -> bool: if not description: return default if COMPANY_WORDS.search(description): return True if NON_COMPANY_WORDS.search(description): return False return default @dataclass class PersonFact: name: str title: str role_category: str is_executive: bool status: str # listed | no_longer_listed valid_from: date | None valid_to: date | None source_url: str qid: str | None = None @dataclass class RelationshipFact: kind: str # PARENT_OF | SUBSIDIARY_OF | OWNED_BY | OWNER_OF to_qid: str to_name: str | None valid_from: date | None valid_to: date | None property: str source_url: str @dataclass class EnrichmentResult: company_id: str profile: dict[str, Any] provenance: dict[str, dict[str, Any]] people: list[PersonFact] = field(default_factory=list) relationships: list[RelationshipFact] = field(default_factory=list) column_updates: dict[str, Any] = field(default_factory=dict) industries: list[str] = field(default_factory=list) sources_used: list[str] = field(default_factory=list) errors: list[str] = field(default_factory=list) llm_job_id: str | None = None def summary(self) -> dict[str, Any]: return {"company_id": self.company_id, "sources": self.sources_used, "people": len(self.people), "relationships": len(self.relationships), "columns": sorted(self.column_updates), "description_source": self.profile.get("description_source"), "errors": self.errors} def _role_for(title: str) -> tuple[str, bool]: try: from companyatlas.connectors.generic_html import role_category return role_category(title) except Exception: # noqa: BLE001 — the connector module may be mid-edit; the profile must not depend on it return "other", False async def apply_wikidata(builder: ProfileBuilder, entity: dict[str, Any], wd: WikidataClient, *, country_hint: str | None) -> tuple[list[PersonFact], list[RelationshipFact]]: """Map a Wikidata entity onto the profile; returns people and relationship facts (names resolved through label lookups).""" qid = entity["id"] url = WIKIDATA_ITEM.format(qid=qid) at = _iso() src = "wikidata" builder.set("wikidata_url", url, source=src, url=url, retrieved_at=at) labels_wanted: list[str] = [] def want(q: str | None) -> None: if q: labels_wanted.append(q) # scalar facts ------------------------------------------------------------------------------------------- names = [v for st in claims(entity, P["official_name"], current_only=True) if isinstance(v := statement_value(st), dict) and v.get("text")] official = next((v for v in names if v.get("language") in ("en", "mul")), names[0] if names else None) if official: builder.set("legal_name", normalize_whitespace(official["text"])[:200], source=src, url=url, retrieved_at=at) for st in claims(entity, P["inception"]): year, _d, _p = wd_time(statement_value(st)) if year: builder.set("founded_year", year, source=src, url=url, retrieved_at=at) break legal_form = next((statement_qid(s) for s in claims(entity, P["legal_form"], current_only=True)), None) want(legal_form) hq_st = next(iter(claims(entity, P["hq"], current_only=True)), None) hq_qid = statement_qid(hq_st) if hq_st else None want(hq_qid) hq_country = qualifier(hq_st, P["country"]) if hq_st else None hq_country_qid = hq_country.get("id") if isinstance(hq_country, dict) else None region_qid = None if hq_st and isinstance(qualifier(hq_st, P["region_in"]), dict): region_qid = qualifier(hq_st, P["region_in"]).get("id") want(region_qid) coords = qualifier(hq_st, P["coords"]) if hq_st else None if isinstance(coords, dict) and coords.get("latitude") is not None: builder.set("hq_lat", round(float(coords["latitude"]), 5), source=src, url=url, retrieved_at=at) builder.set("hq_lon", round(float(coords["longitude"]), 5), source=src, url=url, retrieved_at=at) country_qid = next((statement_qid(s) for s in claims(entity, P["country"], current_only=True)), None) or hq_country_qid emp = latest_quantity(entity, P["employees"]) if emp and emp[0] > 0: builder.set("employees", round(emp[0]), source=src, url=url, retrieved_at=at) if emp[2]: builder.set("employees_year", emp[2], source=src, url=url, retrieved_at=at) money: dict[str, tuple[float, str | None, int | None]] = {} for key in ("revenue", "net_income", "total_assets"): q = latest_quantity(entity, P[key]) if q: money[key] = q industry_qids = [statement_qid(s) for s in claims(entity, P["industry"]) if statement_qid(s)] product_qids = [statement_qid(s) for s in claims(entity, P["products"]) if statement_qid(s)][: settings.enrich_max_products] for q in industry_qids + product_qids: want(q) exchange_st = next(iter(claims(entity, P["exchange"], current_only=True)), None) exchange_qid = statement_qid(exchange_st) if exchange_st else None want(exchange_qid) ticker = next((statement_value(s) for s in claims(entity, P["ticker"], current_only=True) if isinstance(statement_value(s), str)), None) if not ticker and exchange_st and isinstance(qualifier(exchange_st, P["ticker"]), str): ticker = qualifier(exchange_st, P["ticker"]) if ticker: builder.set("ticker", ticker.strip()[:20], source=src, url=url, retrieved_at=at) for key, prop in (("isin", "isin"), ("lei", "lei"), ("sec_cik", "cik")): v = next((statement_value(s) for s in claims(entity, P[prop], current_only=True) if isinstance(statement_value(s), str)), None) if v: builder.set(key, v.strip()[:40], source=src, url=url, retrieved_at=at) site = next((statement_value(s) for s in claims(entity, P["website"], current_only=True) if isinstance(statement_value(s), str)), None) if site and site.startswith(("http://", "https://")): builder.set("official_website", site.strip()[:300], source=src, url=url, retrieved_at=at) logo = next((statement_value(s) for s in claims(entity, P["logo"], current_only=True) if isinstance(statement_value(s), str)), None) if logo: builder.set("logo_url", commons_url(logo), source=src, url=url, retrieved_at=at) socials: dict[str, str] = {} for key, (prop, template) in SOCIAL_TEMPLATES.items(): handle = next((statement_value(s) for s in claims(entity, prop, current_only=True) if isinstance(statement_value(s), str)), None) if handle: socials[key] = template.format(quote(handle.strip(), safe="@/")) if socials: builder.set("socials", socials, source=src, url=url, retrieved_at=at) desc = ((entity.get("descriptions") or {}).get("en") or {}).get("value") if desc and builder.get("description") is None: builder.set("description", normalize_whitespace(desc)[:300], source=src, url=url, retrieved_at=at) builder.profile["description_source"] = "wikidata" builder.profile["description_url"] = url sitelinks = entity.get("sitelinks") or {} lang = "en" if "enwiki" in sitelinks else COUNTRY_LANG.get(country_hint or "", None) if lang and f"{lang}wiki" in sitelinks: builder.set("wikipedia_url", wikipedia_url(lang, sitelinks[f"{lang}wiki"]["title"]), source=src, url=url, retrieved_at=at) elif sitelinks: first_site = next((s for s in ("enwiki", *SITEFILTER) if s in sitelinks), None) if first_site: builder.set("wikipedia_url", wikipedia_url(first_site.removesuffix("wiki"), sitelinks[first_site]["title"]), source=src, url=url, retrieved_at=at) # people / relationships (QIDs now, labels below) --------------------------------------------------------- raw_people: list[tuple[str, str, dict[str, Any]]] = [] # (qid, prop, statement) for prop in PEOPLE_ROLES: for st in claims(entity, prop): pq = statement_qid(st) if pq: raw_people.append((pq, prop, st)) want(pq) role_q = qualifier(st, P["position"]) or qualifier(st, P["role"]) if isinstance(role_q, dict): want(role_q.get("id")) raw_rel: list[tuple[str, str, dict[str, Any]]] = [] cap = settings.enrich_max_relationships_per_property for prop in RELATION_KINDS: for st in claims(entity, prop)[:cap]: rq = statement_qid(st) if rq and rq != qid: raw_rel.append((rq, prop, st)) want(rq) # resolve labels + reference codes ------------------------------------------------------------------------ labels = await wd.labels(labels_wanted) if legal_form and labels.get(legal_form): builder.set("legal_form", labels[legal_form][:120], source=src, url=url, retrieved_at=at) if hq_qid and labels.get(hq_qid): builder.set("hq_city", labels[hq_qid][:120], source=src, url=url, retrieved_at=at) if region_qid and labels.get(region_qid): builder.set("hq_region", labels[region_qid][:120], source=src, url=url, retrieved_at=at) iso2 = await wd.country_iso(country_qid) if iso2: builder.set("country", iso2, source=src, url=url, retrieved_at=at) for key, (amount, unit_qid, year) in money.items(): currency = await wd.currency_code(unit_qid) if currency and year: builder.set(key, {"value": amount, "currency": currency, "year": year}, source=src, url=url, retrieved_at=at) ind_labels = [labels[q] for q in industry_qids if labels.get(q)] if ind_labels: builder.set("industry_labels", ind_labels[:12], source=src, url=url, retrieved_at=at) slugs = [s for s in map_industry(ind_labels, limit=6) if is_valid_slug(s)] if slugs: builder.set("industries", slugs, source=src, url=url, retrieved_at=at) prods = [labels[q] for q in product_qids if labels.get(q)] if prods: builder.set("products", prods, source=src, url=url, retrieved_at=at) if exchange_qid and labels.get(exchange_qid): builder.set("exchange", labels[exchange_qid][:80], source=src, url=url, retrieved_at=at) today = _now().date() people: dict[str, PersonFact] = {} for pq, prop, st in raw_people: name = labels.get(pq) if not name: continue title, cat, is_exec = PEOPLE_ROLES[prop] if prop == P["key_people"]: role_q = qualifier(st, P["position"]) or qualifier(st, P["role"]) role_label = labels.get(role_q.get("id")) if isinstance(role_q, dict) else None if role_label: title = role_label[:160] cat, is_exec = _role_for(role_label) start, end = _start_date(st), _end_date(st) status = "no_longer_listed" if end is not None and end <= today else "listed" fact = PersonFact(name=name[:200], title=title, role_category=cat, is_executive=is_exec, status=status, valid_from=start, valid_to=end, source_url=url, qid=pq) prev = people.get(pq) if prev is None or (prev.status != "listed" and status == "listed") or (prev.status == status and prev.role_category == "other" and cat != "other"): people[pq] = fact relationships: list[RelationshipFact] = [] for rq, prop, st in raw_rel: if prop in (P["subsidiary"], P["owner_of"]) and not looks_like_organisation(wd.description_of(rq), default=prop == P["subsidiary"]): continue relationships.append(RelationshipFact(kind=RELATION_KINDS[prop], to_qid=rq, to_name=(labels.get(rq) or None), valid_from=_start_date(st), valid_to=_end_date(st), property=prop, source_url=url)) return list(people.values()), relationships # ================================================================================================================ Wikipedia def clean_extract(text: str, *, max_chars: int | None = None) -> str | None: limit = max_chars or settings.enrich_description_max_chars t = _CITATION_RE.sub("", text or "") t = _PAREN_PRON_RE.sub("", t) paragraphs = [normalize_whitespace(p) for p in re.split(r"\n{1,}", t) if normalize_whitespace(p)] out = " ".join(paragraphs[:3]) if len(out) > limit: cut = out[:limit] end = max(cut.rfind(". "), cut.rfind("! "), cut.rfind("? ")) out = (cut[: end + 1] if end > limit // 2 else cut.rstrip() + "…") return out or None async def fetch_wikipedia_summary(fetcher: Any, lang: str, title: str) -> dict[str, Any] | None: url = WIKIPEDIA_SUMMARY.format(lang=lang, title=quote(title.replace(" ", "_"), safe="")) try: res = await fetcher.get(url, accept="application/json", rate_per_min=settings.enrich_wikipedia_rate_per_min, respect_robots=False) data = res.json() except (FetchError, ValueError) as exc: log.info("wikipedia summary unavailable", extra={"url": url, "error": str(exc)[:160]}) return None return data if isinstance(data, dict) and data.get("type") not in ("disambiguation",) else None def apply_wikipedia(builder: ProfileBuilder, summary: dict[str, Any]) -> bool: extract = clean_extract(summary.get("extract") or "") page_url = ((summary.get("content_urls") or {}).get("desktop") or {}).get("page") or builder.get("wikipedia_url") at = _iso() changed = False if extract and len(extract) >= 40 and builder.set("description", extract, source="wikipedia", url=page_url, retrieved_at=at): builder.profile.update({"description_source": "wikipedia", "description_url": page_url, "description_license": WIKIPEDIA_LICENSE, "description_attribution": f"Text from Wikipedia ({summary.get('lang') or 'en'}), {WIKIPEDIA_LICENSE}"}) changed = True if page_url: builder.set("wikipedia_url", page_url, source="wikipedia", url=page_url, retrieved_at=at) thumb = (summary.get("thumbnail") or {}).get("source") if thumb and str(thumb).startswith("https://") and builder.get("logo_url") is None: builder.set("logo_url", str(thumb).split("?", 1)[0], source="wikipedia", url=page_url, retrieved_at=at) return changed # ================================================================================================================ homepage @dataclass class HomepageFacts: url: str description: str | None = None legal_name: str | None = None name: str | None = None logo: str | None = None icon: str | None = None founded_year: int | None = None employees: int | None = None address: str | None = None city: str | None = None region: str | None = None country: str | None = None phone: str | None = None socials: dict[str, str] = field(default_factory=dict) def _first_str(value: Any) -> str | None: if isinstance(value, list): value = value[0] if value else None if isinstance(value, dict): value = value.get("url") or value.get("contentUrl") or value.get("@id") or value.get("name") return normalize_whitespace(str(value)) if isinstance(value, str | int | float) and str(value).strip() else None def _icon_size(node: Any) -> int: sizes = (node.attributes.get("sizes") or "").lower() m = re.search(r"(\d+)x(\d+)", sizes) return int(m.group(1)) if m else (180 if "apple" in (node.attributes.get("rel") or "").lower() else 32) def parse_homepage(html: str, url: str) -> HomepageFacts: """Meta description / og:description, JSON-LD Organization (name, legalName, logo, foundingDate, numberOfEmployees, address, telephone, sameAs), og:image → icon candidates (apple-touch-icon > largest icon > og:image).""" facts = HomepageFacts(url=url) tree = LexborHTMLParser(html) metas: dict[str, str] = {} for m in tree.css("meta"): name = (m.attributes.get("name") or m.attributes.get("property") or "").lower().strip() content = (m.attributes.get("content") or "").strip() if name and content and name not in metas: metas[name] = content desc = metas.get("description") or metas.get("og:description") or metas.get("twitter:description") if desc and len(normalize_whitespace(desc)) >= 40: facts.description = normalize_whitespace(desc)[:600] icons: list[tuple[int, str]] = [] for ln in tree.css("link[rel]"): rel = (ln.attributes.get("rel") or "").lower() href = ln.attributes.get("href") or "" if "icon" not in rel or not href: continue absu = absolutize(url, href) if absu: icons.append((_icon_size(ln) + (1000 if "apple" in rel else 0), absu)) og_image = absolutize(url, metas.get("og:image") or "") if metas.get("og:image") else None if icons: facts.icon = max(icons)[1] elif og_image: facts.icon = og_image domain = host_of(url).removeprefix("www.") orgs = extract_jsonld(tree).get("organizations") or [] org = next((o for o in orgs if domain and domain in str(o.get("url") or "").lower()), orgs[0] if orgs else None) if org: facts.name = _first_str(org.get("name")) facts.legal_name = _first_str(org.get("legalName")) logo = _first_str(org.get("logo")) or _first_str(org.get("image")) facts.logo = absolutize(url, logo) if logo else None fd = _first_str(org.get("foundingDate")) if fd and re.match(r"^\d{4}", fd): facts.founded_year = int(fd[:4]) emp = org.get("numberOfEmployees") if isinstance(emp, dict): emp = emp.get("value") if isinstance(emp, int | float) or (isinstance(emp, str) and emp.replace(",", "").strip().isdigit()): n = int(float(str(emp).replace(",", ""))) facts.employees = n if n > 0 else None addr = org.get("address") if isinstance(addr, list): addr = addr[0] if addr else None if isinstance(addr, dict): parts = [_first_str(addr.get(k)) for k in ("streetAddress", "postalCode", "addressLocality", "addressRegion", "addressCountry")] facts.city = parts[2] facts.region = parts[3] facts.country = country_code(parts[4]) if parts[4] else None facts.address = ", ".join(p for p in parts if p)[:300] or None elif isinstance(addr, str): facts.address = normalize_whitespace(addr)[:300] tel = _first_str(org.get("telephone")) if tel and re.search(r"\d{3}", tel): facts.phone = tel[:40] same_as = org.get("sameAs") or [] for link in (same_as if isinstance(same_as, list) else [same_as]): if not isinstance(link, str): continue host = host_of(link) key = next((k for h, k in SOCIAL_HOSTS.items() if host == h or host.endswith("." + h)), None) if key and key not in facts.socials and link.startswith(("http://", "https://")): facts.socials[key] = link.strip()[:300] if not facts.icon and facts.logo: facts.icon = facts.logo return facts async def _safe_url(url: str | None) -> str | None: if not url or not url.startswith(("http://", "https://")): return None try: await validate_destination_async(url) except Exception: # noqa: BLE001 — blocked destination or resolution failure: drop the URL return None return url[:500] async def apply_homepage(builder: ProfileBuilder, facts: HomepageFacts, *, retrieved_at: str | None = None) -> None: at = retrieved_at or _iso() src, url = "homepage", facts.url if facts.description and builder.set("description", facts.description, source=src, url=url, retrieved_at=at): builder.profile.update({"description_source": "homepage", "description_url": url, "description_license": None, "description_attribution": HOMEPAGE_ATTRIBUTION}) builder.set("legal_name", facts.legal_name, source=src, url=url, retrieved_at=at) builder.set("logo_url", await _safe_url(facts.logo), source=src, url=url, retrieved_at=at) builder.set("icon_url", await _safe_url(facts.icon), source=src, url=url, retrieved_at=at) builder.set("founded_year", facts.founded_year, source=src, url=url, retrieved_at=at) builder.set("employees", facts.employees, source=src, url=url, retrieved_at=at) builder.set("hq_address", facts.address, source=src, url=url, retrieved_at=at) builder.set("hq_city", facts.city, source=src, url=url, retrieved_at=at) builder.set("hq_region", facts.region, source=src, url=url, retrieved_at=at) builder.set("country", facts.country, source=src, url=url, retrieved_at=at) builder.set("phone", facts.phone, source=src, url=url, retrieved_at=at) if facts.socials: merged = {**facts.socials, **(builder.profile.get("socials") or {})} # Wikidata handles win on conflicts if builder.source_of("socials") in (None, "homepage"): builder.set("socials", merged, source=src, url=url, retrieved_at=at) else: builder.profile["socials"] = merged async def latest_snapshot(conn: Any, company_id: str, surface: str) -> dict[str, Any] | None: return await fetch_one(conn, """select s.id, s.object_key, s.text_key, s.fetched_at, s.extracted, se.url from snapshots s join sensors se on se.id = s.sensor_id where se.company_id = :c and se.surface = :surface order by s.fetched_at desc limit 1""", c=company_id, surface=surface) def _object_text(key: str | None) -> str | None: if not key: return None try: return decode_text(archive.get_bytes(key)) except (FileNotFoundError, OSError, ValueError): return None # ================================================================================================================ LLM def numbers_grounded(text: str, source: str) -> bool: """Every digit group of `text` must occur (as a normalised digit string) in `source`.""" src_digits = {re.sub(r"\D", "", m) for m in _NUMBER_RE.findall(source or "")} src_blob = re.sub(r"\D", "", source or "") for m in _NUMBER_RE.findall(text or ""): digits = re.sub(r"\D", "", m) if digits and digits not in src_digits and digits not in src_blob: return False return True async def llm_budget_left(conn: Any) -> int: used = await fetch_val(conn, "select count(*) from llm_jobs where finished_at >= date_trunc('day', now() at time zone 'utc') and status in ('done', 'failed')") return max(0, settings.llm_daily_budget - int(used or 0)) _llm_breaker = {"disabled_until": 0.0} def llm_available() -> bool: return settings.llm_configured and time.monotonic() >= _llm_breaker["disabled_until"] async def llm_profile_text(company: dict[str, Any], text: str, *, source_url: str) -> tuple[str | None, str | None, dict[str, Any]]: """Grounded description through `llm_jobs` (kind `company_profile`). Returns (description, job_id, info). Never raises. A transport failure or timeout opens a circuit breaker for `settings.enrich_llm_cooldown_s` so one slow model server cannot stall a batch.""" from companyatlas.services.llm.gateway import LLMError, get_provider from companyatlas.services.llm.prompts import load_prompt from companyatlas.services.llm.schemas import SCHEMA_VERSIONS, CompanyProfileText if not llm_available(): return None, None, {"skipped": "not configured" if not settings.llm_configured else "cooldown"} job_id = new_id("llm_job") async with transaction() as conn: if await llm_budget_left(conn) <= 0: return None, None, {"skipped": "budget"} await execute(conn, "insert into llm_jobs (id, kind, ref_id, company_id, status, attempts, started_at) values (:id, 'company_profile', :r, :c, 'running', 1, now())", id=job_id, r=company["id"], c=company["id"]) prompt = load_prompt("company-profile") context = {"company": {"name": company.get("display_name"), "domain": company.get("canonical_domain"), "country": company.get("country")}, "source_url": source_url, "text": text[: settings.enrich_llm_max_text_chars]} status, error, result, model, req, resp, latency = "failed", None, None, None, 0, 0, 0 description: str | None = None try: res = await asyncio.wait_for(get_provider().complete_json("medium", prompt.system, jsonb(context), CompanyProfileText, max_tokens=500), timeout=settings.enrich_llm_timeout_s) model, req, resp, latency = res.model, res.request_tokens, res.response_tokens, res.latency_ms d: CompanyProfileText = res.data if not numbers_grounded(d.description, text): error = "ungrounded figures in description" elif d.confidence < 0.3 or d.description.startswith("The company's public pages do not describe"): error = "insufficient source text" elif any(bad in d.description.lower() for bad in FORBIDDEN_WORDING): error = "forbidden wording" else: description, status = d.description, "done" result = {"schema_version": SCHEMA_VERSIONS["CompanyProfileText"], "description": d.description, "confidence": d.confidence, "language": d.language, "accepted": description is not None, "repaired": res.repaired} except LLMError as exc: error = str(exc)[:500] if exc.retryable or exc.status in (401, 403): # server down / swapping models, or a bad key: no point retrying per company _llm_breaker["disabled_until"] = time.monotonic() + settings.enrich_llm_cooldown_s log.warning("llm profile: server unavailable, pausing LLM enrichment", extra={"cooldown_s": settings.enrich_llm_cooldown_s, "error": error[:160]}) except TimeoutError: error = f"timeout after {settings.enrich_llm_timeout_s:.0f}s" _llm_breaker["disabled_until"] = time.monotonic() + settings.enrich_llm_cooldown_s except Exception as exc: # noqa: BLE001 error = f"{exc.__class__.__name__}: {exc}"[:500] async with transaction() as conn: await execute(conn, """update llm_jobs set status = :status, model = :model, prompt_version = :pv, result = cast(:result as jsonb), error = :error, request_tokens = :req, response_tokens = :resp, latency_ms = :latency, finished_at = now() where id = :id""", status=status, model=model, pv=prompt.ref, result=jsonb(result) if result is not None else None, error=error, req=req, resp=resp, latency=latency, id=job_id) if model and (req or resp): await execute(conn, """insert into cost_ledger (day, dimension, key, units, cost_estimate) values (current_date, 'llm', :key, :units, 0) on conflict (day, dimension, key) do update set units = cost_ledger.units + excluded.units""", key=model, units=float(req + resp)) return description, job_id, {"status": status, "error": error, "model": model, "prompt_version": prompt.ref} # ================================================================================================================ orchestration def plan_column_updates(company: dict[str, Any], builder: ProfileBuilder) -> dict[str, Any]: """Columns to back-fill: null, or the profile's source for that field outranks the column's recorded source.""" prov = _dict(company.get("source_meta")).get("provenance") or {} updates: dict[str, Any] = {} for field_name, col in COLUMN_FIELDS.items(): new = builder.get(field_name) src = builder.source_of(field_name) if new in (None, "") or src in (None, "registry"): continue current = company.get(col) current_src = (prov.get(col) or {}).get("source") or "registry" if current in (None, "") or (rank_of(field_name, src) > rank_of(field_name, current_src) and current != new): updates[col] = new if updates.get("country") and len(str(updates["country"])) != 2: updates.pop("country") if (builder.get("ticker") or builder.get("isin")) and not company.get("public_company"): updates["public_company"] = True return updates def _merge_industries(company: dict[str, Any], builder: ProfileBuilder) -> list[str]: new = [s for s in (builder.get("industries") or []) if is_valid_slug(s)] if builder.source_of("industries") in (None, "registry") or not new: return [] existing = [s for s in (company.get("industries") or []) if s] merged = existing + [s for s in new if s not in existing] return merged if merged != existing else [] async def enrich_company(company: dict[str, Any], *, fetcher: Any, wikidata: WikidataClient | None = None, entity: dict[str, Any] | None = None, sources: tuple[str, ...] | list[str] = SOURCES, use_db: bool = True, llm: bool = True) -> EnrichmentResult: """Build the profile for one company row (dict with the `companies` columns). No writes; stored snapshots are read when `use_db`.""" wd = wikidata or WikidataClient(fetcher) builder = ProfileBuilder() seed_from_company(builder, company) result = EnrichmentResult(company_id=company["id"], profile=builder.profile, provenance=builder.provenance) country_hint = company.get("country") qid = company.get("wikidata_id") # 1. Wikidata ------------------------------------------------------------------------------------------------- if "wikidata" in sources and qid: try: ent = entity if entity is not None else (await wd.entities([qid])).get(qid) if ent: people, rels = await apply_wikidata(builder, ent, wd, country_hint=country_hint) result.people, result.relationships = people, rels result.sources_used.append("wikidata") country_hint = builder.get("country") or country_hint else: result.errors.append("wikidata: entity unavailable") except Exception as exc: log.exception("wikidata enrichment failed", extra={"company": company.get("slug")}) result.errors.append(f"wikidata: {exc.__class__.__name__}: {exc}"[:200]) # 2. Wikipedia ------------------------------------------------------------------------------------------------- wiki_ok = False wp_url = builder.get("wikipedia_url") if "wikipedia" in sources and wp_url: try: m = re.match(r"^https://([a-z\-]+)\.wikipedia\.org/wiki/(.+)$", wp_url) if m: summary = await fetch_wikipedia_summary(fetcher, m.group(1), unquote(m.group(2).split("#", 1)[0]).replace("_", " ")) if summary: wiki_ok = apply_wikipedia(builder, summary) result.sources_used.append("wikipedia") except Exception as exc: log.exception("wikipedia enrichment failed", extra={"company": company.get("slug")}) result.errors.append(f"wikipedia: {exc.__class__.__name__}: {exc}"[:200]) # 3. Homepage (stored snapshot first, live fetch otherwise) ----------------------------------------------------- page_text: str | None = None about_url: str | None = None page_url = company.get("website") or "" if "homepage" in sources or ("llm" in sources and llm): try: html: str | None = None fetched_at: str | None = None if use_db: async with connection() as conn: snap = await latest_snapshot(conn, company["id"], "homepage") about = await latest_snapshot(conn, company["id"], "about") if snap: html = _object_text(snap.get("object_key")) page_url = snap.get("url") or page_url fetched_at = _iso(snap["fetched_at"]) if isinstance(snap.get("fetched_at"), datetime) else None page_text = _object_text(snap.get("text_key")) if about and about.get("text_key"): about_text = _object_text(about.get("text_key")) if about_text and len(about_text) >= settings.enrich_llm_min_text_chars: page_text, about_url = about_text, about.get("url") if html is None and "homepage" in sources and page_url: try: res = await fetcher.get(page_url, min_bytes=200, retries=0) html, page_url, fetched_at = res.text, res.final_url, _iso(res.fetched_at) except FetchError as exc: result.errors.append(f"homepage: {exc.failure}"[:120]) if html and "homepage" in sources: facts = parse_homepage(html, page_url) await apply_homepage(builder, facts, retrieved_at=fetched_at) result.sources_used.append("homepage") if page_text is None: with contextlib.suppress(Exception): page_text = parse_page(html, url=page_url, surface="homepage").text except Exception as exc: log.exception("homepage enrichment failed", extra={"company": company.get("slug")}) result.errors.append(f"homepage: {exc.__class__.__name__}: {exc}"[:200]) # 4. LLM (only without a Wikipedia extract, with enough first-party text) --------------------------------------- if "llm" in sources and llm and not wiki_ok and builder.source_of("description") != "wikipedia" and page_text \ and len(page_text) >= settings.enrich_llm_min_text_chars and llm_available(): try: src_url = about_url or page_url text, job_id, info = await llm_profile_text(company, page_text, source_url=src_url) result.llm_job_id = job_id if text and builder.set("description", text, source="llm", url=src_url): builder.profile.update({"description_source": "llm", "description_url": src_url, "description_license": None, "description_attribution": LLM_ATTRIBUTION}) result.sources_used.append("llm") elif info.get("error"): result.errors.append(f"llm: {info['error']}"[:160]) except Exception as exc: log.exception("llm profile failed", extra={"company": company.get("slug")}) result.errors.append(f"llm: {exc.__class__.__name__}: {exc}"[:200]) builder.finish() if builder.source_of("description") in (None, "registry"): builder.profile["description_source"] = "wikidata" if builder.get("description") else None result.column_updates = plan_column_updates(company, builder) result.industries = _merge_industries(company, builder) return result # ================================================================================================================ persistence async def _upsert_people(conn: Any, company_id: str, people: list[PersonFact]) -> int: n = 0 now = _now() for p in people: nn = norm_name(p.name) if not nn: continue removed = datetime.combine(p.valid_to, datetime.min.time(), tzinfo=UTC) if p.status == "no_longer_listed" and p.valid_to else None await execute(conn, """ insert into people (id, company_id, name, name_norm, title, role_category, is_executive, first_seen_at, last_seen_at, removed_at, status, source_url) values (:id, :c, :name, :nn, :title, :rc, :ex, :now, :now, :removed, :status, :url) on conflict (company_id, name_norm) do update set title = coalesce(people.title, excluded.title), role_category = case when people.role_category is null or people.role_category = 'other' then excluded.role_category else people.role_category end, is_executive = people.is_executive or excluded.is_executive, last_seen_at = case when people.source_url like 'https://www.wikidata.org/%' then excluded.last_seen_at else people.last_seen_at end, status = case when people.source_url like 'https://www.wikidata.org/%' then excluded.status else people.status end, removed_at = case when people.source_url like 'https://www.wikidata.org/%' then excluded.removed_at else people.removed_at end""", id=new_id("person"), c=company_id, name=p.name, nn=nn[:200], title=p.title[:160], rc=p.role_category, ex=p.is_executive, now=now, removed=removed, status=p.status, url=p.source_url) n += 1 return n async def _upsert_relationships(conn: Any, company_id: str, rels: list[RelationshipFact]) -> dict[str, int]: if not rels: return {"new": 0, "seen": 0} targets = await fetch_all(conn, "select id, wikidata_id from companies where wikidata_id = any(cast(:q as text[]))", q=sorted({r.to_qid for r in rels})) by_qid = {t["wikidata_id"]: t["id"] for t in targets} existing = await fetch_all(conn, """select id, from_company_id, to_company_id, lower(to_name) as to_name, kind, provenance from company_relationships where from_company_id = :c or to_company_id = :c""", c=company_id) index: dict[tuple[str, str, str], str] = {} for r in existing: if r["to_company_id"]: target = r["to_company_id"] elif r["to_name"]: target = f"name:{r['to_name']}" else: target = f"qid:{_dict(r['provenance']).get('qid')}" index[(r["from_company_id"], r["kind"], target)] = r["id"] counters = {"new": 0, "seen": 0} now = _now() async def upsert(frm: str, kind: str, to_id: str | None, to_name: str | None, rel: RelationshipFact) -> None: key_target = to_id or (f"name:{to_name.lower()}" if to_name else f"qid:{rel.to_qid}") prov = {"source": "wikidata", "property": rel.property, "qid": rel.to_qid, "retrieved_at": _iso(now)} rid = index.get((frm, kind, key_target)) if rid: # existing keys (e.g. the seed loader's `source`) win; `retrieved_at` is always refreshed await execute(conn, """update company_relationships set last_seen_at = :now, valid_from = coalesce(valid_from, :vf), valid_to = coalesce(valid_to, :vt), to_company_id = coalesce(to_company_id, :to_id), to_name = coalesce(to_name, :to_name), source_url = coalesce(source_url, :url), provenance = (cast(:prov as jsonb) || provenance) || jsonb_build_object('retrieved_at', cast(:at as text)) where id = :id""", now=now, vf=rel.valid_from, vt=rel.valid_to, to_id=to_id, to_name=to_name, url=rel.source_url, prov=jsonb(prov), at=_iso(now), id=rid) counters["seen"] += 1 return rid = new_id("relationship") await execute(conn, """insert into company_relationships (id, from_company_id, to_company_id, to_name, kind, valid_from, valid_to, first_seen_at, last_seen_at, source_url, confidence, provenance) values (:id, :frm, :to_id, :to_name, :kind, :vf, :vt, :now, :now, :url, 0.85, cast(:prov as jsonb))""", id=rid, frm=frm, to_id=to_id, to_name=to_name, kind=kind, vf=rel.valid_from, vt=rel.valid_to, now=now, url=rel.source_url, prov=jsonb(prov)) index[(frm, kind, key_target)] = rid counters["new"] += 1 seen_keys: set[tuple[str, str]] = set() for rel in rels: if (rel.kind, rel.to_qid) in seen_keys: continue seen_keys.add((rel.kind, rel.to_qid)) to_id = by_qid.get(rel.to_qid) if to_id == company_id: continue await upsert(company_id, rel.kind, to_id, rel.to_name, rel) if to_id: await upsert(to_id, INVERSE_KIND[rel.kind], company_id, None, rel) return counters async def persist(conn: Any, company: dict[str, Any], result: EnrichmentResult) -> dict[str, Any]: """Write `source_meta.profile` (+ provenance, enriched_at), back-fill columns, upsert people and relationships. One transaction (caller's).""" meta = _dict(company.get("source_meta")) prov = dict(meta.get("provenance") or {}) at = _iso() for col, value in result.column_updates.items(): field_name = next((f for f, c in COLUMN_FIELDS.items() if c == col), col) p = result.provenance.get(field_name) or {"source": "wikidata", "url": None} entry = {"source": p["source"], "url": p.get("url"), "retrieved_at": at} if company.get(col) not in (None, "", False) and company.get(col) != value: entry["previous"] = company.get(col) prov[col] = entry if result.industries: prov["industries"] = {"source": result.provenance.get("industries", {}).get("source", "wikidata"), "url": result.provenance.get("industries", {}).get("url"), "retrieved_at": at, "previous": list(company.get("industries") or [])} patch: dict[str, Any] = {"profile": result.profile, "provenance": prov, "enriched_at": at, "enrichment": {"sources": result.sources_used, "errors": result.errors[:10], "llm_job_id": result.llm_job_id, "at": at}} sets = ["source_meta = source_meta || cast(:patch as jsonb)", "updated_at = now()"] params: dict[str, Any] = {"patch": jsonb(patch), "id": company["id"]} for col, value in result.column_updates.items(): if col not in set(COLUMN_FIELDS.values()) | {"public_company"}: continue cast = {"founded_year": "int", "employees": "int", "public_company": "boolean", "country": "char(2)"}.get(col, "text") sets.append(f"{col} = cast(:v_{col} as {cast})") params[f"v_{col}"] = value if result.industries: sets.append("industries = cast(:industries as text[])") params["industries"] = result.industries if not company.get("industry_primary"): sets.append("industry_primary = :industry_primary") params["industry_primary"] = result.industries[0] await execute(conn, f"update companies set {', '.join(sets)} where id = :id", **params) people_n = await _upsert_people(conn, company["id"], result.people) rel = await _upsert_relationships(conn, company["id"], result.relationships) return {"columns": sorted(result.column_updates), "industries": bool(result.industries), "people": people_n, "relationships_new": rel["new"], "relationships_seen": rel["seen"]} # ================================================================================================================ batch runner PENDING_SQL = """ select * from companies where status <> 'DISSOLVED' and ((source_meta->>'enriched_at') is null or cast(source_meta->>'enriched_at' as timestamptz) < cast(:cutoff as timestamptz)) order by (source_meta->>'enriched_at') is not null, onboarding_status <> 'active', importance desc, id limit :limit""" async def pending_companies(conn: Any, limit: int) -> list[dict[str, Any]]: cutoff = datetime.now(UTC) - timedelta(days=settings.enrich_refresh_days) return await fetch_all(conn, PENDING_SQL, cutoff=cutoff, limit=limit) async def load_company(conn: Any, key: str) -> dict[str, Any] | None: return await fetch_one(conn, "select * from companies where slug = :k or id = :k or wikidata_id = :k limit 1", k=key) async def _load_refs(conn: Any) -> dict[str, Any]: return _dict(await fetch_val(conn, "select value from settings_kv where key = :k", k=REF_CACHE_KEY)) async def _save_refs(conn: Any, refs: dict[str, Any]) -> None: await execute(conn, "insert into settings_kv (key, value, updated_at) values (:k, cast(:v as jsonb), now()) on conflict (key) do update set value = excluded.value, updated_at = now()", k=REF_CACHE_KEY, v=jsonb(refs)) async def enrich_and_persist(company: dict[str, Any], *, fetcher: Any, wikidata: WikidataClient, entity: dict[str, Any] | None = None, sources: tuple[str, ...] | list[str] = SOURCES, llm: bool = True) -> dict[str, Any]: t0 = time.monotonic() result = await enrich_company(company, fetcher=fetcher, wikidata=wikidata, entity=entity, sources=sources, llm=llm) async with transaction() as conn: stored = await persist(conn, company, result) out = {**result.summary(), **stored, "duration_ms": int((time.monotonic() - t0) * 1000)} log.info("company enriched", extra={"company": company.get("slug"), **{k: v for k, v in out.items() if k != "company_id"}}) return out async def enrich_pending(limit: int | None = None, concurrency: int | None = None, *, sources: tuple[str, ...] | list[str] = SOURCES, llm: bool = True, company_keys: list[str] | None = None, fetcher: Any | None = None) -> dict[str, Any]: """Enrich never-enriched companies first (active before pending), then profiles older than `enrich_refresh_days`.""" limit = limit or settings.enrich_batch conc = max(1, concurrency or settings.enrich_concurrency) stats: dict[str, Any] = {"companies": 0, "ok": 0, "failed": 0, "people": 0, "relationships_new": 0, "columns": 0, "by_source": {s: 0 for s in SOURCES}, "requests": 0} async with connection() as conn: if company_keys: rows = [c for k in company_keys if (c := await load_company(conn, k))] else: rows = await pending_companies(conn, limit) refs = await _load_refs(conn) if not rows: return stats own_fetcher = fetcher is None f = fetcher or Fetcher(timeout_s=settings.enrich_http_timeout_s, max_connections=max(8, conc * 2)) if own_fetcher: await f.open() wd = WikidataClient(f, refs=refs) sem = asyncio.Semaphore(conc) try: size = max(1, settings.enrich_wikidata_entity_batch) tasks: list[asyncio.Task[None]] = [] for i in range(0, len(rows), size): chunk = rows[i:i + size] # entities for the next chunk are fetched while the previous chunk's companies are still being processed (semaphore-bounded) entities = await wd.entities([c["wikidata_id"] for c in chunk if c.get("wikidata_id")]) if "wikidata" in sources else {} async def one(company: dict[str, Any], ents: dict[str, dict[str, Any]]) -> None: async with sem: try: out = await enrich_and_persist(company, fetcher=f, wikidata=wd, entity=ents.get(company.get("wikidata_id") or ""), sources=sources, llm=llm) except Exception as exc: log.exception("enrichment failed", extra={"company": company.get("slug")}) stats["failed"] += 1 with contextlib.suppress(Exception): async with transaction() as conn: await execute(conn, "update companies set source_meta = source_meta || cast(:p as jsonb) where id = :id", p=jsonb({"enriched_at": _iso(), "enrichment": {"error": f"{exc.__class__.__name__}: {exc}"[:300], "at": _iso()}}), id=company["id"]) return stats["ok"] += 1 stats["people"] += out.get("people", 0) stats["relationships_new"] += out.get("relationships_new", 0) stats["columns"] += len(out.get("columns") or []) for s in out.get("sources") or []: stats["by_source"][s] = stats["by_source"].get(s, 0) + 1 tasks.extend(asyncio.create_task(one(c, entities)) for c in chunk) stats["companies"] += len(chunk) await asyncio.gather(*tasks) finally: stats["requests"] = wd.requests with contextlib.suppress(Exception): async with transaction() as conn: await _save_refs(conn, wd.refs) if own_fetcher: await f.close() log.info("company-enrichment", extra=stats) return stats @periodic("company-enrichment", every_s=600, initial_delay_s=90) async def enrichment_task() -> None: await enrich_pending(limit=settings.enrich_batch, concurrency=settings.enrich_concurrency) # ================================================================================================================ read side (API / CLI) def profile_facts(profile: dict[str, Any] | None) -> list[dict[str, Any]]: """Key facts for a company page: `{key, label, value, raw, source, url, retrieved_at}` — only fields the profile actually has.""" if not profile: return [] src = {s["field"]: s for s in profile.get("sources") or []} def fact(key: str, label: str, value: str | None, raw: Any, field_name: str) -> dict[str, Any] | None: if value in (None, ""): return None s = src.get(field_name) or {} return {"key": key, "label": label, "value": value, "raw": raw, "source": s.get("source"), "url": s.get("url"), "retrieved_at": s.get("retrieved_at")} hq = profile.get("hq") or {} hq_text = ", ".join(x for x in (hq.get("city"), hq.get("region"), hq.get("country")) if x) or None emp = profile.get("employees") emp_text = f"{emp:,}" + (f" ({profile['employees_year']})" if profile.get("employees_year") else "") if isinstance(emp, int) else None facts = [ fact("founded", "Founded", str(profile["founded_year"]) if profile.get("founded_year") else None, profile.get("founded_year"), "founded_year"), fact("headquarters", "Headquarters", hq_text, hq, "hq_city" if src.get("hq_city") else "country"), fact("employees", "Employees", emp_text, emp, "employees"), fact("revenue", "Revenue", _money_text(profile.get("revenue")), profile.get("revenue"), "revenue"), fact("net_income", "Net income", _money_text(profile.get("net_income")), profile.get("net_income"), "net_income"), fact("total_assets", "Total assets", _money_text(profile.get("total_assets")), profile.get("total_assets"), "total_assets"), fact("legal_form", "Legal form", profile.get("legal_form"), profile.get("legal_form"), "legal_form"), fact("listing", "Listing", " · ".join(x for x in (profile.get("ticker"), profile.get("exchange")) if x) or None, {"ticker": profile.get("ticker"), "exchange": profile.get("exchange")}, "ticker" if src.get("ticker") else "exchange"), fact("isin", "ISIN", profile.get("isin"), profile.get("isin"), "isin"), fact("lei", "LEI", profile.get("lei"), profile.get("lei"), "lei"), fact("sec_cik", "SEC CIK", profile.get("sec_cik"), profile.get("sec_cik"), "sec_cik"), fact("website", "Website", profile.get("official_website"), profile.get("official_website"), "official_website"), fact("wikipedia", "Wikipedia", profile.get("wikipedia_url"), profile.get("wikipedia_url"), "wikipedia_url"), ] return [f for f in facts if f] def _money_text(m: Any) -> str | None: if not isinstance(m, dict) or m.get("value") is None: return None v = float(m["value"]) for unit, div in (("T", 1e12), ("B", 1e9), ("M", 1e6), ("K", 1e3)): if abs(v) >= div: num = f"{v / div:.1f}".rstrip("0").rstrip(".") + f" {unit}" break else: num = f"{v:,.0f}" return f"{m.get('currency') or ''} {num}".strip() + (f" ({m['year']})" if m.get("year") else "") def person_source(source_url: str | None) -> str: return "wikidata" if source_url and host_of(source_url).endswith("wikidata.org") else "page" __all__ = [ "COLUMN_FIELDS", "FIELD_RANKS", "PROFILE_VERSION", "SOURCES", "EnrichmentResult", "HomepageFacts", "PersonFact", "ProfileBuilder", "RelationshipFact", "WikidataClient", "apply_homepage", "apply_wikidata", "apply_wikipedia", "claims", "clean_extract", "commons_url", "empty_profile", "enrich_and_persist", "enrich_company", "enrich_pending", "fetch_wikipedia_summary", "latest_quantity", "llm_available", "llm_profile_text", "load_company", "looks_like_organisation", "numbers_grounded", "parse_homepage", "pending_companies", "persist", "person_source", "plan_column_updates", "profile_facts", "rank_of", "seed_from_company", "wd_quantity", "wd_time", ]