spb/company-atlas
Public
Python 66.3%
TypeScript 22.7%
JavaScript 8.6%
HTML 1.4%
CSS 0.7%
1"""Company profile enrichment (reference atlas): Wikidata entity → Wikipedia summary → homepage facts → optional grounded LLM text.23Every accepted value carries provenance `{field, source, url, retrieved_at}` and a better source is never overwritten by a weaker one4(`FIELD_RANKS`). Nothing is inferred: what a source does not state stays `null`.56 result = await enrich_company(company, fetcher=fetcher) # pure: network via `fetcher`, no writes (reads stored snapshots)7 await persist(conn, company, result) # companies.source_meta.profile + column back-fills + people + relationships8 await enrich_pending(limit=300, concurrency=4) # batch runner, also the periodic task `company-enrichment`910Sources11- Wikidata `wbgetentities` (batched, JSON, labels/descriptions/claims/sitelinks) + labels-only lookups + `wbgetclaims` for country ISO12 codes (P297) and currency codes (P498). Wikimedia asks API clients for a descriptive User-Agent and moderate rates, not robots.txt13 (which targets page crawlers) — the API calls use `respect_robots=False` at ≤ 5 req/s (`settings.enrich_wikidata_rate_per_min`).14- Wikipedia REST `page/summary` (≤ 10 req/s): `extract` → description under CC BY-SA 4.0 with attribution URL.15- Homepage: the latest stored homepage snapshot (raw object re-parsed: meta description, JSON-LD Organization, icons) — fetched only16 when no snapshot exists. Icon/logo URLs are http(s)-only and SSRF-validated.17- LLM (`llm_jobs` kind `company_profile`, medium model, budgeted): only when no Wikipedia extract exists and ≥ 400 chars of first-party18 text are available; figures absent from the source text reject the output (`numbers_grounded`).1920Column back-fills (description, logo_url, hq_*, country, founded_year, employees, ticker/exchange, industries, legal_name, lei, sec_cik)21happen when the column is null or the new source outranks the recorded one (`source_meta.provenance[column]`); the replaced value is kept22under `previous`.23"""24from __future__ import annotations2526import asyncio27import contextlib28import json29import logging30import re31import time32from dataclasses import dataclass, field33from datetime import UTC, date, datetime, timedelta34from typing import Any35from urllib.parse import quote, unquote, urlencode3637from selectolax.lexbor import LexborHTMLParser3839from companyatlas import archive40from companyatlas.config import settings41from companyatlas.connectors._util import country_code, norm_name42from companyatlas.db import connection, execute, fetch_all, fetch_one, fetch_val, jsonb, transaction43from companyatlas.fetch import Fetcher, FetchError, FetchResult, decode_text, validate_destination_async44from companyatlas.ids import new_id45from companyatlas.registry.industries import is_valid_slug, map_industry46from companyatlas.sdk.normalize import extract_jsonld, normalize_whitespace47from companyatlas.sdk.normalize import parse as parse_page48from companyatlas.services.periodic import periodic49from companyatlas.taxonomy import FORBIDDEN_WORDING50from companyatlas.urls import absolutize, host_of5152log = logging.getLogger(__name__)5354PROFILE_VERSION = "profile-v1"55USER_AGENT = settings.user_agent56WIKIDATA_API = "https://www.wikidata.org/w/api.php"57WIKIDATA_ITEM = "https://www.wikidata.org/wiki/{qid}"58COMMONS_FILE = "https://commons.wikimedia.org/wiki/Special:FilePath/{name}"59WIKIPEDIA_SUMMARY = "https://{lang}.wikipedia.org/api/rest_v1/page/summary/{title}"60WIKIPEDIA_LICENSE = "CC BY-SA 4.0"61LLM_ATTRIBUTION = "Generated from the company's public pages"62HOMEPAGE_ATTRIBUTION = "Meta description of the company's homepage"63REF_CACHE_KEY = "enrichment:wikidata_refs"6465SOURCES = ("wikidata", "wikipedia", "homepage", "llm")66DEFAULT_RANKS = {"wikidata": 4, "homepage": 3, "registry": 2, "wikipedia": 1, "llm": 0}67FIELD_RANKS: dict[str, dict[str, int]] = {68 "description": {"wikipedia": 4, "llm": 3, "homepage": 2, "wikidata": 1, "registry": 0},69 "logo_url": {"wikidata": 4, "homepage": 3, "registry": 2, "wikipedia": 1},70 "icon_url": {"homepage": 3},71}72# profile field → companies column (only these are ever back-filled).73COLUMN_FIELDS: dict[str, str] = {74 "description": "description", "logo_url": "logo_url", "hq_city": "hq_city", "hq_region": "hq_region", "country": "country",75 "founded_year": "founded_year", "employees": "employees", "ticker": "ticker", "exchange": "exchange", "legal_name": "legal_name",76 "lei": "lei", "sec_cik": "sec_cik",77}7879# Wikidata property ids used below.80P = {81 "official_name": "P1448", "inception": "P571", "legal_form": "P1454", "hq": "P159", "coords": "P625", "country": "P17",82 "employees": "P1128", "point_in_time": "P585", "revenue": "P2139", "net_income": "P2295", "total_assets": "P2403",83 "industry": "P452", "products": "P1056", "ticker": "P249", "exchange": "P414", "isin": "P946", "lei": "P1278", "cik": "P5531",84 "website": "P856", "logo": "P154", "linkedin": "P4264", "x": "P2002", "youtube": "P2397", "facebook": "P2013", "instagram": "P2003",85 "github": "P2037", "tiktok": "P7085", "crunchbase": "P2088", "ceo": "P169", "chair": "P488", "founder": "P112", "key_people": "P3320",86 "parent": "P749", "subsidiary": "P355", "owned_by": "P127", "owner_of": "P1830", "start": "P580", "end": "P582",87 "position": "P39", "role": "P2868", "region_in": "P131", "iso2": "P297", "iso4217": "P498",88}89SOCIAL_TEMPLATES = {90 "linkedin": ("P4264", "https://www.linkedin.com/company/{}"), "x": ("P2002", "https://x.com/{}"), "youtube": ("P2397", "https://www.youtube.com/channel/{}"),91 "facebook": ("P2013", "https://www.facebook.com/{}"), "instagram": ("P2003", "https://www.instagram.com/{}"), "github": ("P2037", "https://github.com/{}"),92 "tiktok": ("P7085", "https://www.tiktok.com/@{}"), "crunchbase": ("P2088", "https://www.crunchbase.com/organization/{}"),93}94SOCIAL_HOSTS = {"linkedin.com": "linkedin", "twitter.com": "x", "x.com": "x", "youtube.com": "youtube", "facebook.com": "facebook",95 "instagram.com": "instagram", "github.com": "github", "tiktok.com": "tiktok", "crunchbase.com": "crunchbase"}96PEOPLE_ROLES = {"P169": ("Chief Executive Officer", "ceo", True), "P488": ("Chairperson", "chair", True), "P112": ("Founder", "founder", True),97 "P3320": ("Key person", "other", False)}98RELATION_KINDS = {"P749": "SUBSIDIARY_OF", "P355": "PARENT_OF", "P127": "OWNED_BY", "P1830": "OWNER_OF"}99INVERSE_KIND = {"PARENT_OF": "SUBSIDIARY_OF", "SUBSIDIARY_OF": "PARENT_OF", "OWNED_BY": "OWNER_OF", "OWNER_OF": "OWNED_BY"}100# Country → Wikipedia language used when the entity has no English sitelink.101COUNTRY_LANG = {"DE": "de", "AT": "de", "CH": "de", "FR": "fr", "BE": "fr", "LU": "fr", "JP": "ja", "CN": "zh", "TW": "zh", "HK": "zh", "IT": "it",102 "ES": "es", "MX": "es", "AR": "es", "CL": "es", "CO": "es", "PE": "es", "KR": "ko", "RU": "ru", "BR": "pt", "PT": "pt", "NL": "nl",103 "SE": "sv", "NO": "no", "DK": "da", "FI": "fi", "PL": "pl", "TR": "tr", "ID": "id", "VN": "vi", "TH": "th", "CZ": "cs", "HU": "hu",104 "GR": "el", "IL": "he", "SA": "ar", "AE": "ar", "EG": "ar", "UA": "uk", "RO": "ro", "IR": "fa", "IN": "en", "MY": "ms"}105LABEL_LANGS = ["en", "fr", "de", "es", "it", "pt", "nl", "ja", "zh", "ko", "ru", "sv", "pl", "tr", "mul"]106SITEFILTER = sorted({f"{lang}wiki" for lang in COUNTRY_LANG.values()} | {"enwiki"})107# Common currencies (Wikidata item → ISO 4217); anything else is resolved live through P498 and cached.108CURRENCIES = {"Q4917": "USD", "Q4916": "EUR", "Q25224": "GBP", "Q8146": "JPY", "Q39099": "CNY", "Q25344": "CHF", "Q1104069": "CAD", "Q259502": "AUD",109 "Q80524": "INR", "Q202040": "KRW", "Q122922": "SEK", "Q31015": "HKD", "Q190951": "SGD", "Q173117": "BRL", "Q4730": "MXN", "Q181907": "ZAR",110 "Q41044": "RUB", "Q208526": "TWD", "Q132643": "NOK", "Q25417": "DKK", "Q123213": "PLN", "Q172872": "TRY", "Q41588": "IDR", "Q199109": "SAR",111 "Q200294": "AED", "Q1472704": "NZD", "Q177882": "THB", "Q163712": "MYR", "Q131309": "ILS", "Q131016": "CZK", "Q47190": "HUF",112 "Q17193": "PHP", "Q199462": "EGP", "Q203567": "NGN", "Q200050": "CLP", "Q244819": "COP", "Q199578": "ARS", "Q188289": "PKR",113 "Q192090": "VND", "Q202714": "KES", "Q206386": "QAR", "Q319176": "KWD"}114_NUMBER_RE = re.compile(r"\d[\d,. ]*\d|\d")115_CITATION_RE = re.compile(r"\[\d+\]|\[[a-z]\]|\[citation needed\]", re.IGNORECASE)116_PAREN_PRON_RE = re.compile(r"\s*\((?:[^()]*?(?:pronounced|listen|ⓘ|/[^/()]+/)[^()]*)\)")117118119def _now() -> datetime:120 return datetime.now(UTC).replace(microsecond=0)121122123def _iso(dt: datetime | None = None) -> str:124 return (dt or _now()).isoformat()125126127# ================================================================================================================ profile builder128129130def empty_profile() -> dict[str, Any]:131 return {132 "description": None, "description_source": None, "description_url": None, "description_license": None, "description_attribution": None,133 "logo_url": None, "icon_url": None, "founded_year": None, "legal_form": None, "legal_name": None,134 "employees": None, "employees_year": None, "revenue": None, "net_income": None, "total_assets": None,135 "hq": {"city": None, "region": None, "country": None, "address": None, "lat": None, "lon": None},136 "ticker": None, "exchange": None, "isin": None, "lei": None, "sec_cik": None, "public_company": False,137 "wikipedia_url": None, "wikidata_url": None, "official_website": None, "phone": None,138 "products": [], "industries": [], "industry_labels": [], "socials": {},139 "enriched_at": None, "sources": [], "version": PROFILE_VERSION,140 }141142143def rank_of(field_name: str, source: str) -> int:144 return FIELD_RANKS.get(field_name, DEFAULT_RANKS).get(source, DEFAULT_RANKS.get(source, 0))145146147HQ_FIELDS = {"hq_city": "city", "hq_region": "region", "country": "country", "hq_address": "address", "hq_lat": "lat", "hq_lon": "lon"}148149150class ProfileBuilder:151 """Rank-aware assignment: a field is (re)assigned only when the new source outranks the one that set it. One provenance row per field."""152153 def __init__(self, profile: dict[str, Any] | None = None) -> None:154 self.profile = profile or empty_profile()155 self.provenance: dict[str, dict[str, Any]] = {}156157 def set(self, field_name: str, value: Any, *, source: str, url: str | None, retrieved_at: str | None = None) -> bool:158 if value in (None, "", [], {}):159 return False160 current = self.provenance.get(field_name)161 if current is not None and rank_of(field_name, source) <= rank_of(field_name, current["source"]):162 return False163 if field_name in HQ_FIELDS:164 self.profile["hq"][HQ_FIELDS[field_name]] = value165 else:166 self.profile[field_name] = value167 self.provenance[field_name] = {"field": field_name, "source": source, "url": url, "retrieved_at": retrieved_at or _iso()}168 return True169170 def get(self, field_name: str) -> Any:171 if field_name in HQ_FIELDS:172 return self.profile["hq"].get(HQ_FIELDS[field_name])173 return self.profile.get(field_name)174175 def source_of(self, field_name: str) -> str | None:176 p = self.provenance.get(field_name)177 return p["source"] if p else None178179 def finish(self) -> dict[str, Any]:180 self.profile["public_company"] = bool(self.profile.get("public_company") or self.profile.get("ticker") or self.profile.get("isin"))181 self.profile["sources"] = sorted(self.provenance.values(), key=lambda p: p["field"])182 self.profile["enriched_at"] = _iso()183 return self.profile184185186def seed_from_company(builder: ProfileBuilder, company: dict[str, Any]) -> None:187 """Existing columns form the base layer (source = recorded provenance or `registry`); every later source may outrank them."""188 prov = _dict(company.get("source_meta")).get("provenance") or {}189 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)190191 def src(col: str) -> str:192 return (prov.get(col) or {}).get("source") or "registry"193194 for col in ("description", "logo_url", "hq_city", "hq_region", "country", "founded_year", "employees", "ticker", "exchange", "legal_name", "lei", "sec_cik"):195 builder.set(col, company.get(col), source=src(col), url=(prov.get(col) or {}).get("url"), retrieved_at=base_at)196 if company.get("industries"):197 builder.set("industries", list(company["industries"]), source=src("industries"), url=None, retrieved_at=base_at)198 labels = _dict(company.get("source_meta")).get("industry_labels")199 if labels:200 builder.set("industry_labels", list(labels)[:12], source="registry", url=None, retrieved_at=base_at)201 if company.get("public_company"):202 builder.profile["public_company"] = True203 if company.get("wikidata_id"):204 builder.set("wikidata_url", WIKIDATA_ITEM.format(qid=company["wikidata_id"]), source="registry", url=None, retrieved_at=base_at)205 if company.get("description"):206 builder.profile["description_source"] = "wikidata" if src("description") == "registry" else src("description")207208209# ================================================================================================================ Wikidata210211212def _dict(value: Any) -> dict[str, Any]:213 if isinstance(value, str):214 try:215 value = json.loads(value)216 except ValueError:217 return {}218 return value if isinstance(value, dict) else {}219220221def _snak_value(snak: dict[str, Any] | None) -> Any:222 if not snak or snak.get("snaktype") != "value":223 return None224 return (snak.get("datavalue") or {}).get("value")225226227def statement_value(st: dict[str, Any]) -> Any:228 return _snak_value(st.get("mainsnak"))229230231def statement_qid(st: dict[str, Any]) -> str | None:232 v = statement_value(st)233 return v.get("id") if isinstance(v, dict) and v.get("entity-type") == "item" else None234235236def qualifier(st: dict[str, Any], prop: str) -> Any:237 snaks = (st.get("qualifiers") or {}).get(prop) or []238 return _snak_value(snaks[0]) if snaks else None239240241def wd_time(value: Any) -> tuple[int | None, date | None, int]:242 """Wikidata time → (year, date-or-None, precision). Day precision (11) gives a full date, month (10) the 1st, year (9) Jan 1."""243 if not isinstance(value, dict) or not value.get("time"):244 return None, None, 0245 t = str(value["time"])246 precision = int(value.get("precision") or 9)247 m = re.match(r"^([+-])(\d+)-(\d\d)-(\d\d)T", t)248 if not m:249 return None, None, precision250 sign, y, mo, d = m.groups()251 year = int(y) * (-1 if sign == "-" else 1)252 if year <= 0 or precision < 9:253 return (year if year > 0 else None), None, precision254 try:255 dt = date(year, int(mo) if precision >= 10 and int(mo) else 1, int(d) if precision >= 11 and int(d) else 1)256 except ValueError:257 dt = None258 return year, dt, precision259260261def wd_quantity(value: Any) -> tuple[float | None, str | None]:262 if not isinstance(value, dict) or value.get("amount") is None:263 return None, None264 try:265 amount = float(str(value["amount"]).replace("+", ""))266 except ValueError:267 return None, None268 unit = str(value.get("unit") or "1")269 return amount, (unit.rsplit("/", 1)[-1] if unit.startswith("http") else None)270271272def _end_date(st: dict[str, Any]) -> date | None:273 _y, d, _p = wd_time(qualifier(st, P["end"]))274 return d275276277def _start_date(st: dict[str, Any]) -> date | None:278 _y, d, _p = wd_time(qualifier(st, P["start"]))279 return d280281282def is_current(st: dict[str, Any], today: date | None = None) -> bool:283 end = _end_date(st)284 return end is None or end > (today or _now().date())285286287def claims(entity: dict[str, Any], prop: str, *, current_only: bool = False) -> list[dict[str, Any]]:288 """Statements for `prop`: deprecated dropped, preferred first, then current (no end date) before ended; Wikidata's order otherwise."""289 out = [s for s in (entity.get("claims") or {}).get(prop) or [] if s.get("rank") != "deprecated" and statement_value(s) is not None]290 if current_only:291 out = [s for s in out if is_current(s)]292 return sorted(out, key=lambda s: (s.get("rank") != "preferred", not is_current(s)))293294295def latest_quantity(entity: dict[str, Any], prop: str) -> tuple[float, str | None, int | None] | None:296 """(amount, unit qid, year) for the observation with the most recent point in time (P585); preferred rank wins ties."""297 best: tuple[tuple[int, int], float, str | None, int | None] | None = None298 for st in claims(entity, prop):299 amount, unit = wd_quantity(statement_value(st))300 if amount is None:301 continue302 year, _d, _p = wd_time(qualifier(st, P["point_in_time"]))303 key = (year or 0, 1 if st.get("rank") == "preferred" else 0)304 if best is None or key > best[0]:305 best = (key, amount, unit, year)306 return (best[1], best[2], best[3]) if best else None307308309def commons_url(filename: str) -> str:310 return COMMONS_FILE.format(name=quote(filename.strip().replace(" ", "_"), safe=""))311312313def wikipedia_url(lang: str, title: str) -> str:314 return f"https://{lang}.wikipedia.org/wiki/{quote(title.replace(' ', '_'), safe=':()/,')}"315316317class WikidataClient:318 """Batched read access to the Wikidata API (labels cached for the process; country/currency codes cached in `settings_kv`)."""319320 def __init__(self, fetcher: Any, *, refs: dict[str, Any] | None = None) -> None:321 self.fetcher = fetcher322 self.labels_cache: dict[str, str | None] = {}323 self.descriptions_cache: dict[str, str | None] = {}324 self.refs: dict[str, Any] = refs if refs is not None else {} # qid → {"iso2": …} / {"iso4217": …}325 self.requests = 0326327 # ---------------------------------------------------------------------------------------------- transport328 @staticmethod329 def url(params: dict[str, str]) -> str:330 return WIKIDATA_API + "?" + urlencode({"format": "json", **params})331332 async def _get(self, url: str, *, max_bytes: int | None = None) -> dict[str, Any] | None:333 self.requests += 1334 try:335 res: FetchResult = await self.fetcher.get(url, accept="application/json", rate_per_min=settings.enrich_wikidata_rate_per_min,336 respect_robots=False, max_bytes=max_bytes or settings.enrich_wikidata_max_bytes)337 except FetchError as exc:338 log.warning("wikidata request failed", extra={"url": url[:200], "error": str(exc)[:200]})339 return None340 try:341 data = res.json()342 except ValueError:343 return None344 return data if isinstance(data, dict) else None345346 # ---------------------------------------------------------------------------------------------- entities / labels347 @staticmethod348 def entities_url(qids: list[str]) -> str:349 return WikidataClient.url({"action": "wbgetentities", "ids": "|".join(qids), "props": "labels|descriptions|claims|sitelinks",350 "languages": "|".join(LABEL_LANGS), "sitefilter": "|".join(SITEFILTER)})351352 @staticmethod353 def labels_url(qids: list[str]) -> str:354 return WikidataClient.url({"action": "wbgetentities", "ids": "|".join(qids), "props": "labels|descriptions", "languages": "|".join(LABEL_LANGS)})355356 @staticmethod357 def claims_url(qid: str, prop: str) -> str:358 return WikidataClient.url({"action": "wbgetclaims", "entity": qid, "property": prop, "props": ""})359360 async def entities(self, qids: list[str]) -> dict[str, dict[str, Any]]:361 out: dict[str, dict[str, Any]] = {}362 ids = list(dict.fromkeys(q for q in qids if q))363 size = max(1, settings.enrich_wikidata_entity_batch)364 for i in range(0, len(ids), size):365 chunk = ids[i:i + size]366 data = await self._get(self.entities_url(chunk))367 if data is None and len(chunk) > 1: # too large / transient → one by one368 for q in chunk:369 single = await self._get(self.entities_url([q]))370 out.update({k: v for k, v in ((single or {}).get("entities") or {}).items() if "missing" not in v})371 continue372 out.update({k: v for k, v in ((data or {}).get("entities") or {}).items() if "missing" not in v})373 for ent in out.values():374 lab = _pick_label(ent.get("labels") or {})375 if lab:376 self.labels_cache[ent["id"]] = lab377 self.descriptions_cache.setdefault(ent["id"], ((ent.get("descriptions") or {}).get("en") or {}).get("value"))378 return out379380 async def labels(self, qids: list[str]) -> dict[str, str]:381 missing = list(dict.fromkeys(q for q in qids if q and q not in self.labels_cache))382 size = max(1, min(50, settings.enrich_wikidata_label_batch))383 for i in range(0, len(missing), size):384 chunk = missing[i:i + size]385 data = await self._get(self.labels_url(chunk), max_bytes=settings.max_body_bytes)386 for q in chunk:387 ent = ((data or {}).get("entities") or {}).get(q) or {}388 self.labels_cache[q] = _pick_label(ent.get("labels") or {})389 self.descriptions_cache[q] = ((ent.get("descriptions") or {}).get("en") or {}).get("value")390 return {q: lab for q in qids if (lab := self.labels_cache.get(q))}391392 def description_of(self, qid: str) -> str | None:393 return self.descriptions_cache.get(qid)394395 async def claim_strings(self, qid: str, prop: str) -> list[str]:396 data = await self._get(self.claims_url(qid, prop), max_bytes=settings.max_body_bytes)397 out: list[str] = []398 for st in ((data or {}).get("claims") or {}).get(prop) or []:399 v = statement_value(st)400 if isinstance(v, str) and st.get("rank") != "deprecated":401 out.append(v)402 return out403404 async def country_iso(self, qid: str | None) -> str | None:405 if not qid:406 return None407 ref = self.refs.get(qid)408 if ref and "iso2" in ref:409 return ref["iso2"]410 codes = [c for c in await self.claim_strings(qid, P["iso2"]) if len(c) == 2 and c.isalpha()]411 code = codes[0].upper() if codes else None412 self.refs[qid] = {**(ref or {}), "iso2": code}413 return code414415 async def currency_code(self, qid: str | None) -> str | None:416 if not qid:417 return None418 if qid in CURRENCIES:419 return CURRENCIES[qid]420 ref = self.refs.get(qid)421 if ref and "iso4217" in ref:422 return ref["iso4217"]423 codes = [c for c in await self.claim_strings(qid, P["iso4217"]) if len(c) == 3 and c.isalpha()]424 code = codes[0].upper() if codes else None425 self.refs[qid] = {**(ref or {}), "iso4217": code}426 return code427428429LATIN_LANGS = ("en", "mul", "fr", "de", "es", "it", "pt", "nl", "sv", "pl", "tr")430431432def _pick_label(labels: dict[str, Any]) -> str | None:433 """English first, then the multilingual label, then Latin-script languages, then anything (a Japanese-only item keeps its Japanese name)."""434 for lang in (*LATIN_LANGS, *LABEL_LANGS):435 v = labels.get(lang)436 if isinstance(v, dict) and v.get("value"):437 return str(v["value"])438 for v in labels.values():439 if isinstance(v, dict) and v.get("value"):440 return str(v["value"])441 return None442443444# Relationship targets (P355 subsidiaries, P1830 "owner of") are kept only when their English description reads like an organisation;445# Wikidata lists domains, buildings, fonts and apps under "owner of".446COMPANY_WORDS = re.compile(r"\b(compan(y|ies)|corporation|subsidiar(y|ies)|business|enterprise|manufacturer|bank|airline|firm|holding|startup|start-up|"447 r"developer|publisher|studio|retailer|provider|provides|services|operator|conglomerate|group|agency|label|network|carrier|brewery|"448 r"insurer|utility|railway|shipyard|automaker|chain|organi[sz]ation|joint venture|division|venture|fund|institution|cooperative|"449 r"maker|producer|distributor|supplier|vendor|consultancy|consulting|contractor|lender|broker|marketplace|team)\b", re.IGNORECASE)450NON_COMPANY_WORDS = re.compile(r"\b(domain|top-level|building|skyscraper|font|typeface|software|website|web service|application|app|programming language|"451 r"file format|protocol|operating system|video game|film|album|song|book|magazine|television|product|device|smartphone|laptop|"452 r"car model|aircraft|satellite|rocket|street|campus|headquarters|stadium|hotel|data center|datacenter|patent|trademark|logo|"453 r"mascot|character|person|human|browser|search engine|brand of|line of|series of|technology|feature)\b", re.IGNORECASE)454455456def looks_like_organisation(description: str | None, *, default: bool) -> bool:457 if not description:458 return default459 if COMPANY_WORDS.search(description):460 return True461 if NON_COMPANY_WORDS.search(description):462 return False463 return default464465466@dataclass467class PersonFact:468 name: str469 title: str470 role_category: str471 is_executive: bool472 status: str # listed | no_longer_listed473 valid_from: date | None474 valid_to: date | None475 source_url: str476 qid: str | None = None477478479@dataclass480class RelationshipFact:481 kind: str # PARENT_OF | SUBSIDIARY_OF | OWNED_BY | OWNER_OF482 to_qid: str483 to_name: str | None484 valid_from: date | None485 valid_to: date | None486 property: str487 source_url: str488489490@dataclass491class EnrichmentResult:492 company_id: str493 profile: dict[str, Any]494 provenance: dict[str, dict[str, Any]]495 people: list[PersonFact] = field(default_factory=list)496 relationships: list[RelationshipFact] = field(default_factory=list)497 column_updates: dict[str, Any] = field(default_factory=dict)498 industries: list[str] = field(default_factory=list)499 sources_used: list[str] = field(default_factory=list)500 errors: list[str] = field(default_factory=list)501 llm_job_id: str | None = None502503 def summary(self) -> dict[str, Any]:504 return {"company_id": self.company_id, "sources": self.sources_used, "people": len(self.people), "relationships": len(self.relationships),505 "columns": sorted(self.column_updates), "description_source": self.profile.get("description_source"), "errors": self.errors}506507508def _role_for(title: str) -> tuple[str, bool]:509 try:510 from companyatlas.connectors.generic_html import role_category511512 return role_category(title)513 except Exception: # noqa: BLE001 — the connector module may be mid-edit; the profile must not depend on it514 return "other", False515516517async def apply_wikidata(builder: ProfileBuilder, entity: dict[str, Any], wd: WikidataClient, *, country_hint: str | None) -> tuple[list[PersonFact], list[RelationshipFact]]:518 """Map a Wikidata entity onto the profile; returns people and relationship facts (names resolved through label lookups)."""519 qid = entity["id"]520 url = WIKIDATA_ITEM.format(qid=qid)521 at = _iso()522 src = "wikidata"523 builder.set("wikidata_url", url, source=src, url=url, retrieved_at=at)524 labels_wanted: list[str] = []525526 def want(q: str | None) -> None:527 if q:528 labels_wanted.append(q)529530 # scalar facts -------------------------------------------------------------------------------------------531 names = [v for st in claims(entity, P["official_name"], current_only=True) if isinstance(v := statement_value(st), dict) and v.get("text")]532 official = next((v for v in names if v.get("language") in ("en", "mul")), names[0] if names else None)533 if official:534 builder.set("legal_name", normalize_whitespace(official["text"])[:200], source=src, url=url, retrieved_at=at)535 for st in claims(entity, P["inception"]):536 year, _d, _p = wd_time(statement_value(st))537 if year:538 builder.set("founded_year", year, source=src, url=url, retrieved_at=at)539 break540 legal_form = next((statement_qid(s) for s in claims(entity, P["legal_form"], current_only=True)), None)541 want(legal_form)542 hq_st = next(iter(claims(entity, P["hq"], current_only=True)), None)543 hq_qid = statement_qid(hq_st) if hq_st else None544 want(hq_qid)545 hq_country = qualifier(hq_st, P["country"]) if hq_st else None546 hq_country_qid = hq_country.get("id") if isinstance(hq_country, dict) else None547 region_qid = None548 if hq_st and isinstance(qualifier(hq_st, P["region_in"]), dict):549 region_qid = qualifier(hq_st, P["region_in"]).get("id")550 want(region_qid)551 coords = qualifier(hq_st, P["coords"]) if hq_st else None552 if isinstance(coords, dict) and coords.get("latitude") is not None:553 builder.set("hq_lat", round(float(coords["latitude"]), 5), source=src, url=url, retrieved_at=at)554 builder.set("hq_lon", round(float(coords["longitude"]), 5), source=src, url=url, retrieved_at=at)555 country_qid = next((statement_qid(s) for s in claims(entity, P["country"], current_only=True)), None) or hq_country_qid556 emp = latest_quantity(entity, P["employees"])557 if emp and emp[0] > 0:558 builder.set("employees", round(emp[0]), source=src, url=url, retrieved_at=at)559 if emp[2]:560 builder.set("employees_year", emp[2], source=src, url=url, retrieved_at=at)561 money: dict[str, tuple[float, str | None, int | None]] = {}562 for key in ("revenue", "net_income", "total_assets"):563 q = latest_quantity(entity, P[key])564 if q:565 money[key] = q566 industry_qids = [statement_qid(s) for s in claims(entity, P["industry"]) if statement_qid(s)]567 product_qids = [statement_qid(s) for s in claims(entity, P["products"]) if statement_qid(s)][: settings.enrich_max_products]568 for q in industry_qids + product_qids:569 want(q)570 exchange_st = next(iter(claims(entity, P["exchange"], current_only=True)), None)571 exchange_qid = statement_qid(exchange_st) if exchange_st else None572 want(exchange_qid)573 ticker = next((statement_value(s) for s in claims(entity, P["ticker"], current_only=True) if isinstance(statement_value(s), str)), None)574 if not ticker and exchange_st and isinstance(qualifier(exchange_st, P["ticker"]), str):575 ticker = qualifier(exchange_st, P["ticker"])576 if ticker:577 builder.set("ticker", ticker.strip()[:20], source=src, url=url, retrieved_at=at)578 for key, prop in (("isin", "isin"), ("lei", "lei"), ("sec_cik", "cik")):579 v = next((statement_value(s) for s in claims(entity, P[prop], current_only=True) if isinstance(statement_value(s), str)), None)580 if v:581 builder.set(key, v.strip()[:40], source=src, url=url, retrieved_at=at)582 site = next((statement_value(s) for s in claims(entity, P["website"], current_only=True) if isinstance(statement_value(s), str)), None)583 if site and site.startswith(("http://", "https://")):584 builder.set("official_website", site.strip()[:300], source=src, url=url, retrieved_at=at)585 logo = next((statement_value(s) for s in claims(entity, P["logo"], current_only=True) if isinstance(statement_value(s), str)), None)586 if logo:587 builder.set("logo_url", commons_url(logo), source=src, url=url, retrieved_at=at)588 socials: dict[str, str] = {}589 for key, (prop, template) in SOCIAL_TEMPLATES.items():590 handle = next((statement_value(s) for s in claims(entity, prop, current_only=True) if isinstance(statement_value(s), str)), None)591 if handle:592 socials[key] = template.format(quote(handle.strip(), safe="@/"))593 if socials:594 builder.set("socials", socials, source=src, url=url, retrieved_at=at)595 desc = ((entity.get("descriptions") or {}).get("en") or {}).get("value")596 if desc and builder.get("description") is None:597 builder.set("description", normalize_whitespace(desc)[:300], source=src, url=url, retrieved_at=at)598 builder.profile["description_source"] = "wikidata"599 builder.profile["description_url"] = url600 sitelinks = entity.get("sitelinks") or {}601 lang = "en" if "enwiki" in sitelinks else COUNTRY_LANG.get(country_hint or "", None)602 if lang and f"{lang}wiki" in sitelinks:603 builder.set("wikipedia_url", wikipedia_url(lang, sitelinks[f"{lang}wiki"]["title"]), source=src, url=url, retrieved_at=at)604 elif sitelinks:605 first_site = next((s for s in ("enwiki", *SITEFILTER) if s in sitelinks), None)606 if first_site:607 builder.set("wikipedia_url", wikipedia_url(first_site.removesuffix("wiki"), sitelinks[first_site]["title"]), source=src, url=url, retrieved_at=at)608609 # people / relationships (QIDs now, labels below) ---------------------------------------------------------610 raw_people: list[tuple[str, str, dict[str, Any]]] = [] # (qid, prop, statement)611 for prop in PEOPLE_ROLES:612 for st in claims(entity, prop):613 pq = statement_qid(st)614 if pq:615 raw_people.append((pq, prop, st))616 want(pq)617 role_q = qualifier(st, P["position"]) or qualifier(st, P["role"])618 if isinstance(role_q, dict):619 want(role_q.get("id"))620 raw_rel: list[tuple[str, str, dict[str, Any]]] = []621 cap = settings.enrich_max_relationships_per_property622 for prop in RELATION_KINDS:623 for st in claims(entity, prop)[:cap]:624 rq = statement_qid(st)625 if rq and rq != qid:626 raw_rel.append((rq, prop, st))627 want(rq)628629 # resolve labels + reference codes ------------------------------------------------------------------------630 labels = await wd.labels(labels_wanted)631 if legal_form and labels.get(legal_form):632 builder.set("legal_form", labels[legal_form][:120], source=src, url=url, retrieved_at=at)633 if hq_qid and labels.get(hq_qid):634 builder.set("hq_city", labels[hq_qid][:120], source=src, url=url, retrieved_at=at)635 if region_qid and labels.get(region_qid):636 builder.set("hq_region", labels[region_qid][:120], source=src, url=url, retrieved_at=at)637 iso2 = await wd.country_iso(country_qid)638 if iso2:639 builder.set("country", iso2, source=src, url=url, retrieved_at=at)640 for key, (amount, unit_qid, year) in money.items():641 currency = await wd.currency_code(unit_qid)642 if currency and year:643 builder.set(key, {"value": amount, "currency": currency, "year": year}, source=src, url=url, retrieved_at=at)644 ind_labels = [labels[q] for q in industry_qids if labels.get(q)]645 if ind_labels:646 builder.set("industry_labels", ind_labels[:12], source=src, url=url, retrieved_at=at)647 slugs = [s for s in map_industry(ind_labels, limit=6) if is_valid_slug(s)]648 if slugs:649 builder.set("industries", slugs, source=src, url=url, retrieved_at=at)650 prods = [labels[q] for q in product_qids if labels.get(q)]651 if prods:652 builder.set("products", prods, source=src, url=url, retrieved_at=at)653 if exchange_qid and labels.get(exchange_qid):654 builder.set("exchange", labels[exchange_qid][:80], source=src, url=url, retrieved_at=at)655656 today = _now().date()657 people: dict[str, PersonFact] = {}658 for pq, prop, st in raw_people:659 name = labels.get(pq)660 if not name:661 continue662 title, cat, is_exec = PEOPLE_ROLES[prop]663 if prop == P["key_people"]:664 role_q = qualifier(st, P["position"]) or qualifier(st, P["role"])665 role_label = labels.get(role_q.get("id")) if isinstance(role_q, dict) else None666 if role_label:667 title = role_label[:160]668 cat, is_exec = _role_for(role_label)669 start, end = _start_date(st), _end_date(st)670 status = "no_longer_listed" if end is not None and end <= today else "listed"671 fact = PersonFact(name=name[:200], title=title, role_category=cat, is_executive=is_exec, status=status, valid_from=start, valid_to=end,672 source_url=url, qid=pq)673 prev = people.get(pq)674 if prev is None or (prev.status != "listed" and status == "listed") or (prev.status == status and prev.role_category == "other" and cat != "other"):675 people[pq] = fact676 relationships: list[RelationshipFact] = []677 for rq, prop, st in raw_rel:678 if prop in (P["subsidiary"], P["owner_of"]) and not looks_like_organisation(wd.description_of(rq), default=prop == P["subsidiary"]):679 continue680 relationships.append(RelationshipFact(kind=RELATION_KINDS[prop], to_qid=rq, to_name=(labels.get(rq) or None), valid_from=_start_date(st),681 valid_to=_end_date(st), property=prop, source_url=url))682 return list(people.values()), relationships683684685# ================================================================================================================ Wikipedia686687688def clean_extract(text: str, *, max_chars: int | None = None) -> str | None:689 limit = max_chars or settings.enrich_description_max_chars690 t = _CITATION_RE.sub("", text or "")691 t = _PAREN_PRON_RE.sub("", t)692 paragraphs = [normalize_whitespace(p) for p in re.split(r"\n{1,}", t) if normalize_whitespace(p)]693 out = " ".join(paragraphs[:3])694 if len(out) > limit:695 cut = out[:limit]696 end = max(cut.rfind(". "), cut.rfind("! "), cut.rfind("? "))697 out = (cut[: end + 1] if end > limit // 2 else cut.rstrip() + "…")698 return out or None699700701async def fetch_wikipedia_summary(fetcher: Any, lang: str, title: str) -> dict[str, Any] | None:702 url = WIKIPEDIA_SUMMARY.format(lang=lang, title=quote(title.replace(" ", "_"), safe=""))703 try:704 res = await fetcher.get(url, accept="application/json", rate_per_min=settings.enrich_wikipedia_rate_per_min, respect_robots=False)705 data = res.json()706 except (FetchError, ValueError) as exc:707 log.info("wikipedia summary unavailable", extra={"url": url, "error": str(exc)[:160]})708 return None709 return data if isinstance(data, dict) and data.get("type") not in ("disambiguation",) else None710711712def apply_wikipedia(builder: ProfileBuilder, summary: dict[str, Any]) -> bool:713 extract = clean_extract(summary.get("extract") or "")714 page_url = ((summary.get("content_urls") or {}).get("desktop") or {}).get("page") or builder.get("wikipedia_url")715 at = _iso()716 changed = False717 if extract and len(extract) >= 40 and builder.set("description", extract, source="wikipedia", url=page_url, retrieved_at=at):718 builder.profile.update({"description_source": "wikipedia", "description_url": page_url, "description_license": WIKIPEDIA_LICENSE,719 "description_attribution": f"Text from Wikipedia ({summary.get('lang') or 'en'}), {WIKIPEDIA_LICENSE}"})720 changed = True721 if page_url:722 builder.set("wikipedia_url", page_url, source="wikipedia", url=page_url, retrieved_at=at)723 thumb = (summary.get("thumbnail") or {}).get("source")724 if thumb and str(thumb).startswith("https://") and builder.get("logo_url") is None:725 builder.set("logo_url", str(thumb).split("?", 1)[0], source="wikipedia", url=page_url, retrieved_at=at)726 return changed727728729# ================================================================================================================ homepage730731732@dataclass733class HomepageFacts:734 url: str735 description: str | None = None736 legal_name: str | None = None737 name: str | None = None738 logo: str | None = None739 icon: str | None = None740 founded_year: int | None = None741 employees: int | None = None742 address: str | None = None743 city: str | None = None744 region: str | None = None745 country: str | None = None746 phone: str | None = None747 socials: dict[str, str] = field(default_factory=dict)748749750def _first_str(value: Any) -> str | None:751 if isinstance(value, list):752 value = value[0] if value else None753 if isinstance(value, dict):754 value = value.get("url") or value.get("contentUrl") or value.get("@id") or value.get("name")755 return normalize_whitespace(str(value)) if isinstance(value, str | int | float) and str(value).strip() else None756757758def _icon_size(node: Any) -> int:759 sizes = (node.attributes.get("sizes") or "").lower()760 m = re.search(r"(\d+)x(\d+)", sizes)761 return int(m.group(1)) if m else (180 if "apple" in (node.attributes.get("rel") or "").lower() else 32)762763764def parse_homepage(html: str, url: str) -> HomepageFacts:765 """Meta description / og:description, JSON-LD Organization (name, legalName, logo, foundingDate, numberOfEmployees, address,766 telephone, sameAs), og:image → icon candidates (apple-touch-icon > largest icon > og:image)."""767 facts = HomepageFacts(url=url)768 tree = LexborHTMLParser(html)769 metas: dict[str, str] = {}770 for m in tree.css("meta"):771 name = (m.attributes.get("name") or m.attributes.get("property") or "").lower().strip()772 content = (m.attributes.get("content") or "").strip()773 if name and content and name not in metas:774 metas[name] = content775 desc = metas.get("description") or metas.get("og:description") or metas.get("twitter:description")776 if desc and len(normalize_whitespace(desc)) >= 40:777 facts.description = normalize_whitespace(desc)[:600]778 icons: list[tuple[int, str]] = []779 for ln in tree.css("link[rel]"):780 rel = (ln.attributes.get("rel") or "").lower()781 href = ln.attributes.get("href") or ""782 if "icon" not in rel or not href:783 continue784 absu = absolutize(url, href)785 if absu:786 icons.append((_icon_size(ln) + (1000 if "apple" in rel else 0), absu))787 og_image = absolutize(url, metas.get("og:image") or "") if metas.get("og:image") else None788 if icons:789 facts.icon = max(icons)[1]790 elif og_image:791 facts.icon = og_image792 domain = host_of(url).removeprefix("www.")793 orgs = extract_jsonld(tree).get("organizations") or []794 org = next((o for o in orgs if domain and domain in str(o.get("url") or "").lower()), orgs[0] if orgs else None)795 if org:796 facts.name = _first_str(org.get("name"))797 facts.legal_name = _first_str(org.get("legalName"))798 logo = _first_str(org.get("logo")) or _first_str(org.get("image"))799 facts.logo = absolutize(url, logo) if logo else None800 fd = _first_str(org.get("foundingDate"))801 if fd and re.match(r"^\d{4}", fd):802 facts.founded_year = int(fd[:4])803 emp = org.get("numberOfEmployees")804 if isinstance(emp, dict):805 emp = emp.get("value")806 if isinstance(emp, int | float) or (isinstance(emp, str) and emp.replace(",", "").strip().isdigit()):807 n = int(float(str(emp).replace(",", "")))808 facts.employees = n if n > 0 else None809 addr = org.get("address")810 if isinstance(addr, list):811 addr = addr[0] if addr else None812 if isinstance(addr, dict):813 parts = [_first_str(addr.get(k)) for k in ("streetAddress", "postalCode", "addressLocality", "addressRegion", "addressCountry")]814 facts.city = parts[2]815 facts.region = parts[3]816 facts.country = country_code(parts[4]) if parts[4] else None817 facts.address = ", ".join(p for p in parts if p)[:300] or None818 elif isinstance(addr, str):819 facts.address = normalize_whitespace(addr)[:300]820 tel = _first_str(org.get("telephone"))821 if tel and re.search(r"\d{3}", tel):822 facts.phone = tel[:40]823 same_as = org.get("sameAs") or []824 for link in (same_as if isinstance(same_as, list) else [same_as]):825 if not isinstance(link, str):826 continue827 host = host_of(link)828 key = next((k for h, k in SOCIAL_HOSTS.items() if host == h or host.endswith("." + h)), None)829 if key and key not in facts.socials and link.startswith(("http://", "https://")):830 facts.socials[key] = link.strip()[:300]831 if not facts.icon and facts.logo:832 facts.icon = facts.logo833 return facts834835836async def _safe_url(url: str | None) -> str | None:837 if not url or not url.startswith(("http://", "https://")):838 return None839 try:840 await validate_destination_async(url)841 except Exception: # noqa: BLE001 — blocked destination or resolution failure: drop the URL842 return None843 return url[:500]844845846async def apply_homepage(builder: ProfileBuilder, facts: HomepageFacts, *, retrieved_at: str | None = None) -> None:847 at = retrieved_at or _iso()848 src, url = "homepage", facts.url849 if facts.description and builder.set("description", facts.description, source=src, url=url, retrieved_at=at):850 builder.profile.update({"description_source": "homepage", "description_url": url, "description_license": None, "description_attribution": HOMEPAGE_ATTRIBUTION})851 builder.set("legal_name", facts.legal_name, source=src, url=url, retrieved_at=at)852 builder.set("logo_url", await _safe_url(facts.logo), source=src, url=url, retrieved_at=at)853 builder.set("icon_url", await _safe_url(facts.icon), source=src, url=url, retrieved_at=at)854 builder.set("founded_year", facts.founded_year, source=src, url=url, retrieved_at=at)855 builder.set("employees", facts.employees, source=src, url=url, retrieved_at=at)856 builder.set("hq_address", facts.address, source=src, url=url, retrieved_at=at)857 builder.set("hq_city", facts.city, source=src, url=url, retrieved_at=at)858 builder.set("hq_region", facts.region, source=src, url=url, retrieved_at=at)859 builder.set("country", facts.country, source=src, url=url, retrieved_at=at)860 builder.set("phone", facts.phone, source=src, url=url, retrieved_at=at)861 if facts.socials:862 merged = {**facts.socials, **(builder.profile.get("socials") or {})} # Wikidata handles win on conflicts863 if builder.source_of("socials") in (None, "homepage"):864 builder.set("socials", merged, source=src, url=url, retrieved_at=at)865 else:866 builder.profile["socials"] = merged867868869async def latest_snapshot(conn: Any, company_id: str, surface: str) -> dict[str, Any] | None:870 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_id871 where se.company_id = :c and se.surface = :surface order by s.fetched_at desc limit 1""", c=company_id, surface=surface)872873874def _object_text(key: str | None) -> str | None:875 if not key:876 return None877 try:878 return decode_text(archive.get_bytes(key))879 except (FileNotFoundError, OSError, ValueError):880 return None881882883# ================================================================================================================ LLM884885886def numbers_grounded(text: str, source: str) -> bool:887 """Every digit group of `text` must occur (as a normalised digit string) in `source`."""888 src_digits = {re.sub(r"\D", "", m) for m in _NUMBER_RE.findall(source or "")}889 src_blob = re.sub(r"\D", "", source or "")890 for m in _NUMBER_RE.findall(text or ""):891 digits = re.sub(r"\D", "", m)892 if digits and digits not in src_digits and digits not in src_blob:893 return False894 return True895896897async def llm_budget_left(conn: Any) -> int:898 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')")899 return max(0, settings.llm_daily_budget - int(used or 0))900901902_llm_breaker = {"disabled_until": 0.0}903904905def llm_available() -> bool:906 return settings.llm_configured and time.monotonic() >= _llm_breaker["disabled_until"]907908909async def llm_profile_text(company: dict[str, Any], text: str, *, source_url: str) -> tuple[str | None, str | None, dict[str, Any]]:910 """Grounded description through `llm_jobs` (kind `company_profile`). Returns (description, job_id, info). Never raises. A transport911 failure or timeout opens a circuit breaker for `settings.enrich_llm_cooldown_s` so one slow model server cannot stall a batch."""912 from companyatlas.services.llm.gateway import LLMError, get_provider913 from companyatlas.services.llm.prompts import load_prompt914 from companyatlas.services.llm.schemas import SCHEMA_VERSIONS, CompanyProfileText915916 if not llm_available():917 return None, None, {"skipped": "not configured" if not settings.llm_configured else "cooldown"}918 job_id = new_id("llm_job")919 async with transaction() as conn:920 if await llm_budget_left(conn) <= 0:921 return None, None, {"skipped": "budget"}922 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())",923 id=job_id, r=company["id"], c=company["id"])924 prompt = load_prompt("company-profile")925 context = {"company": {"name": company.get("display_name"), "domain": company.get("canonical_domain"), "country": company.get("country")},926 "source_url": source_url, "text": text[: settings.enrich_llm_max_text_chars]}927 status, error, result, model, req, resp, latency = "failed", None, None, None, 0, 0, 0928 description: str | None = None929 try:930 res = await asyncio.wait_for(get_provider().complete_json("medium", prompt.system, jsonb(context), CompanyProfileText, max_tokens=500),931 timeout=settings.enrich_llm_timeout_s)932 model, req, resp, latency = res.model, res.request_tokens, res.response_tokens, res.latency_ms933 d: CompanyProfileText = res.data934 if not numbers_grounded(d.description, text):935 error = "ungrounded figures in description"936 elif d.confidence < 0.3 or d.description.startswith("The company's public pages do not describe"):937 error = "insufficient source text"938 elif any(bad in d.description.lower() for bad in FORBIDDEN_WORDING):939 error = "forbidden wording"940 else:941 description, status = d.description, "done"942 result = {"schema_version": SCHEMA_VERSIONS["CompanyProfileText"], "description": d.description, "confidence": d.confidence, "language": d.language,943 "accepted": description is not None, "repaired": res.repaired}944 except LLMError as exc:945 error = str(exc)[:500]946 if exc.retryable or exc.status in (401, 403): # server down / swapping models, or a bad key: no point retrying per company947 _llm_breaker["disabled_until"] = time.monotonic() + settings.enrich_llm_cooldown_s948 log.warning("llm profile: server unavailable, pausing LLM enrichment", extra={"cooldown_s": settings.enrich_llm_cooldown_s, "error": error[:160]})949 except TimeoutError:950 error = f"timeout after {settings.enrich_llm_timeout_s:.0f}s"951 _llm_breaker["disabled_until"] = time.monotonic() + settings.enrich_llm_cooldown_s952 except Exception as exc: # noqa: BLE001953 error = f"{exc.__class__.__name__}: {exc}"[:500]954 async with transaction() as conn:955 await execute(conn, """update llm_jobs set status = :status, model = :model, prompt_version = :pv, result = cast(:result as jsonb), error = :error,956 request_tokens = :req, response_tokens = :resp, latency_ms = :latency, finished_at = now() where id = :id""",957 status=status, model=model, pv=prompt.ref, result=jsonb(result) if result is not None else None, error=error, req=req, resp=resp,958 latency=latency, id=job_id)959 if model and (req or resp):960 await execute(conn, """insert into cost_ledger (day, dimension, key, units, cost_estimate) values (current_date, 'llm', :key, :units, 0)961 on conflict (day, dimension, key) do update set units = cost_ledger.units + excluded.units""", key=model, units=float(req + resp))962 return description, job_id, {"status": status, "error": error, "model": model, "prompt_version": prompt.ref}963964965# ================================================================================================================ orchestration966967968def plan_column_updates(company: dict[str, Any], builder: ProfileBuilder) -> dict[str, Any]:969 """Columns to back-fill: null, or the profile's source for that field outranks the column's recorded source."""970 prov = _dict(company.get("source_meta")).get("provenance") or {}971 updates: dict[str, Any] = {}972 for field_name, col in COLUMN_FIELDS.items():973 new = builder.get(field_name)974 src = builder.source_of(field_name)975 if new in (None, "") or src in (None, "registry"):976 continue977 current = company.get(col)978 current_src = (prov.get(col) or {}).get("source") or "registry"979 if current in (None, "") or (rank_of(field_name, src) > rank_of(field_name, current_src) and current != new):980 updates[col] = new981 if updates.get("country") and len(str(updates["country"])) != 2:982 updates.pop("country")983 if (builder.get("ticker") or builder.get("isin")) and not company.get("public_company"):984 updates["public_company"] = True985 return updates986987988def _merge_industries(company: dict[str, Any], builder: ProfileBuilder) -> list[str]:989 new = [s for s in (builder.get("industries") or []) if is_valid_slug(s)]990 if builder.source_of("industries") in (None, "registry") or not new:991 return []992 existing = [s for s in (company.get("industries") or []) if s]993 merged = existing + [s for s in new if s not in existing]994 return merged if merged != existing else []995996997async def enrich_company(company: dict[str, Any], *, fetcher: Any, wikidata: WikidataClient | None = None, entity: dict[str, Any] | None = None,998 sources: tuple[str, ...] | list[str] = SOURCES, use_db: bool = True, llm: bool = True) -> EnrichmentResult:999 """Build the profile for one company row (dict with the `companies` columns). No writes; stored snapshots are read when `use_db`."""1000 wd = wikidata or WikidataClient(fetcher)1001 builder = ProfileBuilder()1002 seed_from_company(builder, company)1003 result = EnrichmentResult(company_id=company["id"], profile=builder.profile, provenance=builder.provenance)1004 country_hint = company.get("country")1005 qid = company.get("wikidata_id")10061007 # 1. Wikidata -------------------------------------------------------------------------------------------------1008 if "wikidata" in sources and qid:1009 try:1010 ent = entity if entity is not None else (await wd.entities([qid])).get(qid)1011 if ent:1012 people, rels = await apply_wikidata(builder, ent, wd, country_hint=country_hint)1013 result.people, result.relationships = people, rels1014 result.sources_used.append("wikidata")1015 country_hint = builder.get("country") or country_hint1016 else:1017 result.errors.append("wikidata: entity unavailable")1018 except Exception as exc:1019 log.exception("wikidata enrichment failed", extra={"company": company.get("slug")})1020 result.errors.append(f"wikidata: {exc.__class__.__name__}: {exc}"[:200])10211022 # 2. Wikipedia -------------------------------------------------------------------------------------------------1023 wiki_ok = False1024 wp_url = builder.get("wikipedia_url")1025 if "wikipedia" in sources and wp_url:1026 try:1027 m = re.match(r"^https://([a-z\-]+)\.wikipedia\.org/wiki/(.+)$", wp_url)1028 if m:1029 summary = await fetch_wikipedia_summary(fetcher, m.group(1), unquote(m.group(2).split("#", 1)[0]).replace("_", " "))1030 if summary:1031 wiki_ok = apply_wikipedia(builder, summary)1032 result.sources_used.append("wikipedia")1033 except Exception as exc:1034 log.exception("wikipedia enrichment failed", extra={"company": company.get("slug")})1035 result.errors.append(f"wikipedia: {exc.__class__.__name__}: {exc}"[:200])10361037 # 3. Homepage (stored snapshot first, live fetch otherwise) -----------------------------------------------------1038 page_text: str | None = None1039 about_url: str | None = None1040 page_url = company.get("website") or ""1041 if "homepage" in sources or ("llm" in sources and llm):1042 try:1043 html: str | None = None1044 fetched_at: str | None = None1045 if use_db:1046 async with connection() as conn:1047 snap = await latest_snapshot(conn, company["id"], "homepage")1048 about = await latest_snapshot(conn, company["id"], "about")1049 if snap:1050 html = _object_text(snap.get("object_key"))1051 page_url = snap.get("url") or page_url1052 fetched_at = _iso(snap["fetched_at"]) if isinstance(snap.get("fetched_at"), datetime) else None1053 page_text = _object_text(snap.get("text_key"))1054 if about and about.get("text_key"):1055 about_text = _object_text(about.get("text_key"))1056 if about_text and len(about_text) >= settings.enrich_llm_min_text_chars:1057 page_text, about_url = about_text, about.get("url")1058 if html is None and "homepage" in sources and page_url:1059 try:1060 res = await fetcher.get(page_url, min_bytes=200, retries=0)1061 html, page_url, fetched_at = res.text, res.final_url, _iso(res.fetched_at)1062 except FetchError as exc:1063 result.errors.append(f"homepage: {exc.failure}"[:120])1064 if html and "homepage" in sources:1065 facts = parse_homepage(html, page_url)1066 await apply_homepage(builder, facts, retrieved_at=fetched_at)1067 result.sources_used.append("homepage")1068 if page_text is None:1069 with contextlib.suppress(Exception):1070 page_text = parse_page(html, url=page_url, surface="homepage").text1071 except Exception as exc:1072 log.exception("homepage enrichment failed", extra={"company": company.get("slug")})1073 result.errors.append(f"homepage: {exc.__class__.__name__}: {exc}"[:200])10741075 # 4. LLM (only without a Wikipedia extract, with enough first-party text) ---------------------------------------1076 if "llm" in sources and llm and not wiki_ok and builder.source_of("description") != "wikipedia" and page_text \1077 and len(page_text) >= settings.enrich_llm_min_text_chars and llm_available():1078 try:1079 src_url = about_url or page_url1080 text, job_id, info = await llm_profile_text(company, page_text, source_url=src_url)1081 result.llm_job_id = job_id1082 if text and builder.set("description", text, source="llm", url=src_url):1083 builder.profile.update({"description_source": "llm", "description_url": src_url, "description_license": None, "description_attribution": LLM_ATTRIBUTION})1084 result.sources_used.append("llm")1085 elif info.get("error"):1086 result.errors.append(f"llm: {info['error']}"[:160])1087 except Exception as exc:1088 log.exception("llm profile failed", extra={"company": company.get("slug")})1089 result.errors.append(f"llm: {exc.__class__.__name__}: {exc}"[:200])10901091 builder.finish()1092 if builder.source_of("description") in (None, "registry"):1093 builder.profile["description_source"] = "wikidata" if builder.get("description") else None1094 result.column_updates = plan_column_updates(company, builder)1095 result.industries = _merge_industries(company, builder)1096 return result109710981099# ================================================================================================================ persistence110011011102async def _upsert_people(conn: Any, company_id: str, people: list[PersonFact]) -> int:1103 n = 01104 now = _now()1105 for p in people:1106 nn = norm_name(p.name)1107 if not nn:1108 continue1109 removed = datetime.combine(p.valid_to, datetime.min.time(), tzinfo=UTC) if p.status == "no_longer_listed" and p.valid_to else None1110 await execute(conn, """1111 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)1112 values (:id, :c, :name, :nn, :title, :rc, :ex, :now, :now, :removed, :status, :url)1113 on conflict (company_id, name_norm) do update set1114 title = coalesce(people.title, excluded.title),1115 role_category = case when people.role_category is null or people.role_category = 'other' then excluded.role_category else people.role_category end,1116 is_executive = people.is_executive or excluded.is_executive,1117 last_seen_at = case when people.source_url like 'https://www.wikidata.org/%' then excluded.last_seen_at else people.last_seen_at end,1118 status = case when people.source_url like 'https://www.wikidata.org/%' then excluded.status else people.status end,1119 removed_at = case when people.source_url like 'https://www.wikidata.org/%' then excluded.removed_at else people.removed_at end""",1120 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,1121 removed=removed, status=p.status, url=p.source_url)1122 n += 11123 return n112411251126async def _upsert_relationships(conn: Any, company_id: str, rels: list[RelationshipFact]) -> dict[str, int]:1127 if not rels:1128 return {"new": 0, "seen": 0}1129 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}))1130 by_qid = {t["wikidata_id"]: t["id"] for t in targets}1131 existing = await fetch_all(conn, """select id, from_company_id, to_company_id, lower(to_name) as to_name, kind, provenance from company_relationships1132 where from_company_id = :c or to_company_id = :c""", c=company_id)1133 index: dict[tuple[str, str, str], str] = {}1134 for r in existing:1135 if r["to_company_id"]:1136 target = r["to_company_id"]1137 elif r["to_name"]:1138 target = f"name:{r['to_name']}"1139 else:1140 target = f"qid:{_dict(r['provenance']).get('qid')}"1141 index[(r["from_company_id"], r["kind"], target)] = r["id"]1142 counters = {"new": 0, "seen": 0}1143 now = _now()11441145 async def upsert(frm: str, kind: str, to_id: str | None, to_name: str | None, rel: RelationshipFact) -> None:1146 key_target = to_id or (f"name:{to_name.lower()}" if to_name else f"qid:{rel.to_qid}")1147 prov = {"source": "wikidata", "property": rel.property, "qid": rel.to_qid, "retrieved_at": _iso(now)}1148 rid = index.get((frm, kind, key_target))1149 if rid:1150 # existing keys (e.g. the seed loader's `source`) win; `retrieved_at` is always refreshed1151 await execute(conn, """update company_relationships set last_seen_at = :now, valid_from = coalesce(valid_from, :vf), valid_to = coalesce(valid_to, :vt),1152 to_company_id = coalesce(to_company_id, :to_id), to_name = coalesce(to_name, :to_name), source_url = coalesce(source_url, :url),1153 provenance = (cast(:prov as jsonb) || provenance) || jsonb_build_object('retrieved_at', cast(:at as text)) where id = :id""",1154 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)1155 counters["seen"] += 11156 return1157 rid = new_id("relationship")1158 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,1159 source_url, confidence, provenance) values (:id, :frm, :to_id, :to_name, :kind, :vf, :vt, :now, :now, :url, 0.85, cast(:prov as jsonb))""",1160 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))1161 index[(frm, kind, key_target)] = rid1162 counters["new"] += 111631164 seen_keys: set[tuple[str, str]] = set()1165 for rel in rels:1166 if (rel.kind, rel.to_qid) in seen_keys:1167 continue1168 seen_keys.add((rel.kind, rel.to_qid))1169 to_id = by_qid.get(rel.to_qid)1170 if to_id == company_id:1171 continue1172 await upsert(company_id, rel.kind, to_id, rel.to_name, rel)1173 if to_id:1174 await upsert(to_id, INVERSE_KIND[rel.kind], company_id, None, rel)1175 return counters117611771178async def persist(conn: Any, company: dict[str, Any], result: EnrichmentResult) -> dict[str, Any]:1179 """Write `source_meta.profile` (+ provenance, enriched_at), back-fill columns, upsert people and relationships. One transaction (caller's)."""1180 meta = _dict(company.get("source_meta"))1181 prov = dict(meta.get("provenance") or {})1182 at = _iso()1183 for col, value in result.column_updates.items():1184 field_name = next((f for f, c in COLUMN_FIELDS.items() if c == col), col)1185 p = result.provenance.get(field_name) or {"source": "wikidata", "url": None}1186 entry = {"source": p["source"], "url": p.get("url"), "retrieved_at": at}1187 if company.get(col) not in (None, "", False) and company.get(col) != value:1188 entry["previous"] = company.get(col)1189 prov[col] = entry1190 if result.industries:1191 prov["industries"] = {"source": result.provenance.get("industries", {}).get("source", "wikidata"), "url": result.provenance.get("industries", {}).get("url"),1192 "retrieved_at": at, "previous": list(company.get("industries") or [])}1193 patch: dict[str, Any] = {"profile": result.profile, "provenance": prov, "enriched_at": at, "enrichment": {"sources": result.sources_used, "errors": result.errors[:10],1194 "llm_job_id": result.llm_job_id, "at": at}}1195 sets = ["source_meta = source_meta || cast(:patch as jsonb)", "updated_at = now()"]1196 params: dict[str, Any] = {"patch": jsonb(patch), "id": company["id"]}1197 for col, value in result.column_updates.items():1198 if col not in set(COLUMN_FIELDS.values()) | {"public_company"}:1199 continue1200 cast = {"founded_year": "int", "employees": "int", "public_company": "boolean", "country": "char(2)"}.get(col, "text")1201 sets.append(f"{col} = cast(:v_{col} as {cast})")1202 params[f"v_{col}"] = value1203 if result.industries:1204 sets.append("industries = cast(:industries as text[])")1205 params["industries"] = result.industries1206 if not company.get("industry_primary"):1207 sets.append("industry_primary = :industry_primary")1208 params["industry_primary"] = result.industries[0]1209 await execute(conn, f"update companies set {', '.join(sets)} where id = :id", **params)1210 people_n = await _upsert_people(conn, company["id"], result.people)1211 rel = await _upsert_relationships(conn, company["id"], result.relationships)1212 return {"columns": sorted(result.column_updates), "industries": bool(result.industries), "people": people_n, "relationships_new": rel["new"],1213 "relationships_seen": rel["seen"]}121412151216# ================================================================================================================ batch runner121712181219PENDING_SQL = """1220select * from companies1221where status <> 'DISSOLVED'1222 and ((source_meta->>'enriched_at') is null or cast(source_meta->>'enriched_at' as timestamptz) < cast(:cutoff as timestamptz))1223order by (source_meta->>'enriched_at') is not null, onboarding_status <> 'active', importance desc, id1224limit :limit"""122512261227async def pending_companies(conn: Any, limit: int) -> list[dict[str, Any]]:1228 cutoff = datetime.now(UTC) - timedelta(days=settings.enrich_refresh_days)1229 return await fetch_all(conn, PENDING_SQL, cutoff=cutoff, limit=limit)123012311232async def load_company(conn: Any, key: str) -> dict[str, Any] | None:1233 return await fetch_one(conn, "select * from companies where slug = :k or id = :k or wikidata_id = :k limit 1", k=key)123412351236async def _load_refs(conn: Any) -> dict[str, Any]:1237 return _dict(await fetch_val(conn, "select value from settings_kv where key = :k", k=REF_CACHE_KEY))123812391240async def _save_refs(conn: Any, refs: dict[str, Any]) -> None:1241 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()",1242 k=REF_CACHE_KEY, v=jsonb(refs))124312441245async def enrich_and_persist(company: dict[str, Any], *, fetcher: Any, wikidata: WikidataClient, entity: dict[str, Any] | None = None,1246 sources: tuple[str, ...] | list[str] = SOURCES, llm: bool = True) -> dict[str, Any]:1247 t0 = time.monotonic()1248 result = await enrich_company(company, fetcher=fetcher, wikidata=wikidata, entity=entity, sources=sources, llm=llm)1249 async with transaction() as conn:1250 stored = await persist(conn, company, result)1251 out = {**result.summary(), **stored, "duration_ms": int((time.monotonic() - t0) * 1000)}1252 log.info("company enriched", extra={"company": company.get("slug"), **{k: v for k, v in out.items() if k != "company_id"}})1253 return out125412551256async def enrich_pending(limit: int | None = None, concurrency: int | None = None, *, sources: tuple[str, ...] | list[str] = SOURCES, llm: bool = True,1257 company_keys: list[str] | None = None, fetcher: Any | None = None) -> dict[str, Any]:1258 """Enrich never-enriched companies first (active before pending), then profiles older than `enrich_refresh_days`."""1259 limit = limit or settings.enrich_batch1260 conc = max(1, concurrency or settings.enrich_concurrency)1261 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}1262 async with connection() as conn:1263 if company_keys:1264 rows = [c for k in company_keys if (c := await load_company(conn, k))]1265 else:1266 rows = await pending_companies(conn, limit)1267 refs = await _load_refs(conn)1268 if not rows:1269 return stats1270 own_fetcher = fetcher is None1271 f = fetcher or Fetcher(timeout_s=settings.enrich_http_timeout_s, max_connections=max(8, conc * 2))1272 if own_fetcher:1273 await f.open()1274 wd = WikidataClient(f, refs=refs)1275 sem = asyncio.Semaphore(conc)1276 try:1277 size = max(1, settings.enrich_wikidata_entity_batch)1278 tasks: list[asyncio.Task[None]] = []1279 for i in range(0, len(rows), size):1280 chunk = rows[i:i + size]1281 # entities for the next chunk are fetched while the previous chunk's companies are still being processed (semaphore-bounded)1282 entities = await wd.entities([c["wikidata_id"] for c in chunk if c.get("wikidata_id")]) if "wikidata" in sources else {}12831284 async def one(company: dict[str, Any], ents: dict[str, dict[str, Any]]) -> None:1285 async with sem:1286 try:1287 out = await enrich_and_persist(company, fetcher=f, wikidata=wd, entity=ents.get(company.get("wikidata_id") or ""), sources=sources, llm=llm)1288 except Exception as exc:1289 log.exception("enrichment failed", extra={"company": company.get("slug")})1290 stats["failed"] += 11291 with contextlib.suppress(Exception):1292 async with transaction() as conn:1293 await execute(conn, "update companies set source_meta = source_meta || cast(:p as jsonb) where id = :id",1294 p=jsonb({"enriched_at": _iso(), "enrichment": {"error": f"{exc.__class__.__name__}: {exc}"[:300], "at": _iso()}}), id=company["id"])1295 return1296 stats["ok"] += 11297 stats["people"] += out.get("people", 0)1298 stats["relationships_new"] += out.get("relationships_new", 0)1299 stats["columns"] += len(out.get("columns") or [])1300 for s in out.get("sources") or []:1301 stats["by_source"][s] = stats["by_source"].get(s, 0) + 113021303 tasks.extend(asyncio.create_task(one(c, entities)) for c in chunk)1304 stats["companies"] += len(chunk)1305 await asyncio.gather(*tasks)1306 finally:1307 stats["requests"] = wd.requests1308 with contextlib.suppress(Exception):1309 async with transaction() as conn:1310 await _save_refs(conn, wd.refs)1311 if own_fetcher:1312 await f.close()1313 log.info("company-enrichment", extra=stats)1314 return stats131513161317@periodic("company-enrichment", every_s=600, initial_delay_s=90)1318async def enrichment_task() -> None:1319 await enrich_pending(limit=settings.enrich_batch, concurrency=settings.enrich_concurrency)132013211322# ================================================================================================================ read side (API / CLI)132313241325def profile_facts(profile: dict[str, Any] | None) -> list[dict[str, Any]]:1326 """Key facts for a company page: `{key, label, value, raw, source, url, retrieved_at}` — only fields the profile actually has."""1327 if not profile:1328 return []1329 src = {s["field"]: s for s in profile.get("sources") or []}13301331 def fact(key: str, label: str, value: str | None, raw: Any, field_name: str) -> dict[str, Any] | None:1332 if value in (None, ""):1333 return None1334 s = src.get(field_name) or {}1335 return {"key": key, "label": label, "value": value, "raw": raw, "source": s.get("source"), "url": s.get("url"), "retrieved_at": s.get("retrieved_at")}13361337 hq = profile.get("hq") or {}1338 hq_text = ", ".join(x for x in (hq.get("city"), hq.get("region"), hq.get("country")) if x) or None1339 emp = profile.get("employees")1340 emp_text = f"{emp:,}" + (f" ({profile['employees_year']})" if profile.get("employees_year") else "") if isinstance(emp, int) else None1341 facts = [1342 fact("founded", "Founded", str(profile["founded_year"]) if profile.get("founded_year") else None, profile.get("founded_year"), "founded_year"),1343 fact("headquarters", "Headquarters", hq_text, hq, "hq_city" if src.get("hq_city") else "country"),1344 fact("employees", "Employees", emp_text, emp, "employees"),1345 fact("revenue", "Revenue", _money_text(profile.get("revenue")), profile.get("revenue"), "revenue"),1346 fact("net_income", "Net income", _money_text(profile.get("net_income")), profile.get("net_income"), "net_income"),1347 fact("total_assets", "Total assets", _money_text(profile.get("total_assets")), profile.get("total_assets"), "total_assets"),1348 fact("legal_form", "Legal form", profile.get("legal_form"), profile.get("legal_form"), "legal_form"),1349 fact("listing", "Listing", " · ".join(x for x in (profile.get("ticker"), profile.get("exchange")) if x) or None,1350 {"ticker": profile.get("ticker"), "exchange": profile.get("exchange")}, "ticker" if src.get("ticker") else "exchange"),1351 fact("isin", "ISIN", profile.get("isin"), profile.get("isin"), "isin"),1352 fact("lei", "LEI", profile.get("lei"), profile.get("lei"), "lei"),1353 fact("sec_cik", "SEC CIK", profile.get("sec_cik"), profile.get("sec_cik"), "sec_cik"),1354 fact("website", "Website", profile.get("official_website"), profile.get("official_website"), "official_website"),1355 fact("wikipedia", "Wikipedia", profile.get("wikipedia_url"), profile.get("wikipedia_url"), "wikipedia_url"),1356 ]1357 return [f for f in facts if f]135813591360def _money_text(m: Any) -> str | None:1361 if not isinstance(m, dict) or m.get("value") is None:1362 return None1363 v = float(m["value"])1364 for unit, div in (("T", 1e12), ("B", 1e9), ("M", 1e6), ("K", 1e3)):1365 if abs(v) >= div:1366 num = f"{v / div:.1f}".rstrip("0").rstrip(".") + f" {unit}"1367 break1368 else:1369 num = f"{v:,.0f}"1370 return f"{m.get('currency') or ''} {num}".strip() + (f" ({m['year']})" if m.get("year") else "")137113721373def person_source(source_url: str | None) -> str:1374 return "wikidata" if source_url and host_of(source_url).endswith("wikidata.org") else "page"137513761377__all__ = [1378 "COLUMN_FIELDS",1379 "FIELD_RANKS",1380 "PROFILE_VERSION",1381 "SOURCES",1382 "EnrichmentResult",1383 "HomepageFacts",1384 "PersonFact",1385 "ProfileBuilder",1386 "RelationshipFact",1387 "WikidataClient",1388 "apply_homepage",1389 "apply_wikidata",1390 "apply_wikipedia",1391 "claims",1392 "clean_extract",1393 "commons_url",1394 "empty_profile",1395 "enrich_and_persist",1396 "enrich_company",1397 "enrich_pending",1398 "fetch_wikipedia_summary",1399 "latest_quantity",1400 "llm_available",1401 "llm_profile_text",1402 "load_company",1403 "looks_like_organisation",1404 "numbers_grounded",1405 "parse_homepage",1406 "pending_companies",1407 "persist",1408 "person_source",1409 "plan_column_updates",1410 "profile_facts",1411 "rank_of",1412 "seed_from_company",1413 "wd_quantity",1414 "wd_time",1415]1416