Spaces:
Sleeping
Sleeping
Z User
fix(enrich): actor inputs per deployed input_schema.json (queries/maxResults; companies/maxEmployees)
bedd72b | """Enrichment pipeline: Apify actors → normalized accounts/contacts → | |
| OpenSearch + ICP score + vector memory + optional Neo4j mirror.""" | |
| import logging | |
| import uuid | |
| from datetime import datetime, timezone | |
| from typing import Any, Dict, List, Optional | |
| from ..config import settings | |
| from ..stores import opensearch | |
| from ..stores.apify import apify_client as apify, normalize_account, normalize_contact | |
| from ..stores.cache import cache | |
| from ..stores.graph import graph | |
| from . import accounts as accounts_svc | |
| from . import signals as signals_svc | |
| logger = logging.getLogger("abm.enrich") | |
| def _now() -> str: | |
| return datetime.now(timezone.utc).isoformat() | |
| SOURCE_ACTORS = { | |
| "google_maps": lambda: settings.apify_actor_google_maps, | |
| "linkedin": lambda: settings.apify_actor_linkedin, | |
| } | |
| def build_actor_input(source: str, queries: List[str], limit: int) -> dict: | |
| """Actor inputs per each actor's real input_schema.json (verified against | |
| the deployed source, not guessed): the Google Maps master scraper takes | |
| `queries`/`maxResults`/`scrapeContacts`; the LinkedIn company-email | |
| scraper takes required `companies` + `maxEmployees`/`findEmails`.""" | |
| if source == "linkedin": | |
| return { | |
| "companies": queries, | |
| "maxEmployees": limit, | |
| "findEmails": True, | |
| "verifyEmails": False, | |
| "decisionMakersOnly": False, | |
| } | |
| return { | |
| "queries": queries, | |
| "maxResults": limit, | |
| "scrapeDetails": False, # contact enrichment runs inline without detail pages | |
| "scrapeContacts": True, | |
| "scrapeReviews": False, | |
| "scrapePhotos": False, | |
| "language": "en", | |
| "country": "us", | |
| } | |
| async def enrich( | |
| source: str, | |
| queries: List[str], | |
| limit: int = 20, | |
| account_limit: int = 100, | |
| ) -> Dict[str, Any]: | |
| """Run one enrichment batch end-to-end. Returns a summary with counts.""" | |
| if source not in SOURCE_ACTORS: | |
| raise ValueError(f"unknown source '{source}' (use google_maps or linkedin)") | |
| if not apify.configured: | |
| raise RuntimeError("Apify is not configured (APIFY_TOKEN missing)") | |
| actor = SOURCE_ACTORS[source]() | |
| run_input = build_actor_input(source, queries, limit) | |
| cache_key = f"enrich:{source}:{hash(tuple(sorted(queries)))}:{limit}" | |
| cached = await cache.get(cache_key) | |
| if cached: | |
| return {**cached, "cached": True} | |
| started = _now() | |
| items = await apify.run_sync(actor, run_input) | |
| summary: Dict[str, Any] = { | |
| "source": source, | |
| "actor": actor, | |
| "queries": queries, | |
| "items_raw": len(items), | |
| "accounts": 0, | |
| "contacts": 0, | |
| "skipped": 0, | |
| "account_ids": [], | |
| "started_at": started, | |
| "finished_at": None, | |
| } | |
| if source == "linkedin": | |
| # LinkedIn items are people; group them under a company account when | |
| # the item names one, otherwise under the query itself. | |
| for raw in items[:account_limit * 5]: | |
| contact = normalize_contact(raw) | |
| if not contact: | |
| summary["skipped"] += 1 | |
| continue | |
| company_name = contact.pop("company", None) or (queries[0] if queries else "Unknown") | |
| domain = contact.get("email", "").split("@")[-1] if contact.get("email") else None | |
| account = await accounts_svc.upsert_account({ | |
| "name": company_name, "domain": domain, "source": "linkedin", | |
| }) | |
| saved = await accounts_svc.upsert_contact(account["account_id"], contact) | |
| if saved: | |
| summary["contacts"] += 1 | |
| await graph.upsert_contact(account["account_id"], saved or {}) | |
| if account["account_id"] not in summary["account_ids"]: | |
| summary["account_ids"].append(account["account_id"]) | |
| summary["accounts"] = len(summary["account_ids"]) | |
| else: | |
| for raw in items[:account_limit]: | |
| account_fields = normalize_account(raw) | |
| if not account_fields: | |
| summary["skipped"] += 1 | |
| continue | |
| account = await accounts_svc.upsert_account(account_fields) | |
| summary["account_ids"].append(account["account_id"]) | |
| summary["accounts"] += 1 | |
| # Map-scraper items sometimes carry contact rows alongside. | |
| raw_contacts = raw.get("emails") or raw.get("contacts") or [] | |
| if isinstance(raw_contacts, list): | |
| for c in raw_contacts: | |
| if isinstance(c, str): | |
| c = {"email": c} | |
| contact = normalize_contact(c) | |
| if contact: | |
| saved = await accounts_svc.upsert_contact(account["account_id"], contact) | |
| if saved: | |
| summary["contacts"] += 1 | |
| await _finalize_account(account) | |
| await graph.upsert_account(account) | |
| summary["finished_at"] = _now() | |
| await cache.set(cache_key, summary, ttl_s=900) | |
| await opensearch.store.refresh("accounts") | |
| await opensearch.store.refresh("contacts") | |
| return summary | |
| async def _finalize_account(account: dict) -> None: | |
| """Score + embed + persist enrichment events for one account.""" | |
| contacts = await opensearch.store.search("contacts", { | |
| "query": {"term": {"account_id": account["account_id"]}}, | |
| "size": 1, | |
| }) | |
| # count without pulling rows | |
| from ..stores.opensearch import store | |
| try: | |
| res = await store.client().post( | |
| f"/{store._index('contacts')}/_count", | |
| json={"query": {"term": {"account_id": account["account_id"]}}}, | |
| ) | |
| contact_count = int(res.json().get("count", 0)) if res.status_code == 200 else 0 | |
| except Exception: | |
| contact_count = len(contacts) | |
| scored = signals_svc.score_account(account, contact_count) | |
| doc = {**account, **scored, "embedding": signals_svc.embed_account(account)} | |
| await store.upsert_doc("accounts", doc, account["account_id"]) | |
| await accounts_svc.record_event(account["account_id"], "enrichment_completed", { | |
| "icp_score": scored["icp_score"], "icp_label": scored["icp_label"], | |
| "contacts": contact_count, "at": _now(), | |
| }) | |