"""Apify enrichment client — runs the owner's own actors and fetches results. Actors (defaults are the account's own published actors, configurable via env): - sitcod3.lab/fdn-google-map-master-scraper (Google Maps → companies) - sitcod3.lab/linkedin-company-email-scraper (LinkedIn → employees + emails) Both sync (run-sync-get-dataset-items, capped) and async (run + poll) modes are supported. Output items are schema-flexible: the normalizer probes common Google Maps / LinkedIn export field names case-insensitively. """ import asyncio import logging import time from typing import Any, Dict, List, Optional import httpx from ..config import settings logger = logging.getLogger("abm.apify") API = "https://api.apify.com/v2" class ApifyClient: def __init__(self) -> None: self._http: Optional[httpx.AsyncClient] = None @property def configured(self) -> bool: return settings.apify_configured def client(self) -> httpx.AsyncClient: if self._http is None: self._http = httpx.AsyncClient( base_url=API, params={"token": settings.apify_token}, timeout=httpx.Timeout(60.0), ) return self._http async def close(self) -> None: if self._http is not None: await self._http.aclose() self._http = None async def ping(self) -> bool: if not self.configured: return False try: res = await self.client().get("/users/me") return res.status_code == 200 except Exception: return False async def run_sync(self, actor: str, run_input: dict, timeout_s: Optional[int] = None) -> List[dict]: """Run an actor and collect its default dataset items (capped wait).""" deadline = timeout_s or settings.apify_run_timeout_s # Apify actor references use "username~actorName" (tilde), not "/". actor_ref = actor.replace("/", "~", 1) if "~" not in actor else actor url = f"/acts/{actor_ref}/run-sync-get-dataset-items" params = {"timeout": int(deadline * 1000)} # run-sync holds the HTTP connection for the WHOLE actor runtime # (minutes of Playwright scraping) — the 60s client default times out. res = await self.client().post( url, params=params, json=run_input, timeout=httpx.Timeout(30.0, read=deadline + 60.0), ) if res.status_code in (402, 403): raise RuntimeError(f"Apify rejected the run (HTTP {res.status_code}): {res.text[:200]}") res.raise_for_status() try: items = res.json() return items if isinstance(items, list) else [] except Exception: return [] async def run_async(self, actor: str, run_input: dict) -> Dict[str, str]: """Start an actor run; returns {run_id, dataset_id} for polling.""" actor_ref = actor.replace("/", "~", 1) if "~" not in actor else actor res = await self.client().post(f"/acts/{actor_ref}/runs", json=run_input) res.raise_for_status() data = res.json().get("data", {}) return {"run_id": data.get("id"), "dataset_id": data.get("defaultDatasetId")} async def wait_run(self, run_id: str, poll_s: int = 10, timeout_s: Optional[int] = None) -> str: deadline = time.monotonic() + (timeout_s or settings.apify_run_timeout_s) while time.monotonic() < deadline: res = await self.client().get(f"/actor-runs/{run_id}") res.raise_for_status() data = res.json().get("data", {}) status = data.get("status") if status in ("SUCCEEDED", "FAILED", "ABORTED", "TIMED-OUT"): return status await asyncio.sleep(poll_s) return "TIMED-OUT" async def dataset_items(self, dataset_id: str, limit: int = 5000) -> List[dict]: res = await self.client().get(f"/datasets/{dataset_id}/items", params={"limit": limit}) res.raise_for_status() items = res.json() return items if isinstance(items, list) else [] apify_client = ApifyClient() # ── normalization: raw actor items → ABM records ───────────────────────────── def _pick(item: dict, *keys: str) -> Any: low = {k.lower(): v for k, v in item.items() if isinstance(k, str)} for k in keys: if k in low and low[k]: return low[k] return None def normalize_account(raw: dict) -> Optional[dict]: """Map a Google Maps-ish item to an ABM account (schema-flexible).""" if not isinstance(raw, dict): return None name = _pick(raw, "name", "title", "companyname", "company", "searchstring") if not name: return None website = _pick(raw, "website", "websiteurl", "url", "site", "domain") domain = _pick(raw, "domain", "website_domain") if not domain and website: domain = str(website).replace("https://", "").replace("http://", "").split("/")[0] if isinstance(domain, str): domain = domain.removeprefix("www.").lower() or None return { "name": str(name)[:200], "domain": domain, "website": website, "phone": _pick(raw, "phone", "phonenumber", "telephone"), "industry": _pick(raw, "category", "maincategory", "categories", "industry", "subcategories"), "city": _pick(raw, "city", "locality"), "hq_country": _pick(raw, "country", "countrycode"), "address": _pick(raw, "address", "fulladdress", "location", "street", "addressline"), "employee_count": _pick(raw, "employeecount", "total_employees", "employees"), "rating": _pick(raw, "rating", "totalScore", "reviews", "reviewCount"), "description": _pick(raw, "description", "about", "snippet"), "source": "google_maps", } def normalize_contact(raw: dict, account_id_hint: Optional[str] = None) -> Optional[dict]: """Map a LinkedIn-ish item to an ABM contact.""" if not isinstance(raw, dict): return None email = _pick(raw, "email", "emailaddress", "mail", "business_email") full_name = _pick(raw, "fullname", "name", "full_name", " firstname") or \ " ".join(x for x in [_pick(raw, "firstname", "first_name"), _pick(raw, "lastname", "last_name")] if x) if not email and not full_name: return None return { "email": str(email).strip().lower() if email else None, "full_name": str(full_name).strip()[:160] if full_name else None, "title": _pick(raw, "title", "position", "jobtitle", "role", "headline"), "linkedin_url": _pick(raw, "linkedinurl", "linkedin_url", "profile", "url"), "phone": _pick(raw, "phone", "phonenumber"), "company": _pick(raw, "company", "companyname", "organization", "current_company"), "source": "linkedin", }