Spaces:
Running
Running
| """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 | |
| 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", | |
| } | |