Spaces:
Sleeping
Sleeping
Z User
feat: ABM Engine v1 — OpenSearch store + Apify enrichment + ICP scoring + k-NN memory + optional Neo4j mirror
fb0ec21 | """Account + contact services on top of OpenSearch.""" | |
| import uuid | |
| from datetime import datetime, timezone | |
| from typing import Any, Dict, List, Optional | |
| from ..stores import opensearch | |
| def _now() -> str: | |
| return datetime.now(timezone.utc).isoformat() | |
| async def upsert_account(payload: Dict[str, Any]) -> Dict[str, Any]: | |
| """Create or update an account keyed by domain (fallback: name hash).""" | |
| domain = (payload.get("domain") or "").lower().strip() or None | |
| name = (payload.get("name") or "").strip() or None | |
| if not domain and not name: | |
| raise ValueError("domain or name is required") | |
| existing = None | |
| if domain: | |
| hits = await opensearch.store.search("accounts", { | |
| "query": {"term": {"domain": domain}}, | |
| "size": 1, | |
| }) | |
| existing = hits[0] if hits else None | |
| account_id = (existing or {}).get("account_id") or str(uuid.uuid4()) | |
| doc = { | |
| "account_id": account_id, | |
| "domain": domain, | |
| "name": name or (existing or {}).get("name"), | |
| "industry": payload.get("industry") or (existing or {}).get("industry"), | |
| "employee_count": payload.get("employee_count") or (existing or {}).get("employee_count"), | |
| "hq_country": payload.get("hq_country") or (existing or {}).get("hq_country"), | |
| "city": payload.get("city") or (existing or {}).get("city"), | |
| "website": payload.get("website") or (existing or {}).get("website"), | |
| "phone": payload.get("phone") or (existing or {}).get("phone"), | |
| "address": payload.get("address") or (existing or {}).get("address"), | |
| "description": payload.get("description") or (existing or {}).get("description"), | |
| "rating": payload.get("rating") or (existing or {}).get("rating"), | |
| "source": payload.get("source") or (existing or {}).get("source") or "manual", | |
| "icp_score": (existing or {}).get("icp_score"), | |
| "icp_label": (existing or {}).get("icp_label"), | |
| "updated_at": _now(), | |
| } | |
| doc = {k: v for k, v in doc.items() if v is not None} | |
| await opensearch.store.upsert_doc("accounts", doc, account_id) | |
| return doc | |
| async def get_account(account_id: str) -> Optional[dict]: | |
| account = await opensearch.store.get_doc("accounts", account_id) | |
| if account is None: | |
| return None | |
| contacts = await opensearch.store.search("contacts", { | |
| "query": {"term": {"account_id": account_id}}, | |
| "size": 100, | |
| "sort": [{"_id": "asc"}], | |
| }) | |
| events = await opensearch.store.search("events", { | |
| "query": {"term": {"account_id": account_id}}, | |
| "size": 50, | |
| "sort": [{"created_at": {"order": "desc"}}], | |
| }) | |
| return {"account": account, "contacts": contacts, "events": events} | |
| async def search_accounts(q: str = "", industry: str = "", country: str = "", | |
| min_score: float = 0.0, limit: int = 25) -> List[dict]: | |
| filters: List[dict] = [] | |
| if industry: | |
| filters.append({"term": {"industry": industry.lower()}}) | |
| if country: | |
| filters.append({"term": {"hq_country": country.lower()}}) | |
| if min_score > 0: | |
| filters.append({"range": {"icp_score": {"gte": min_score}}}) | |
| must: List[dict] = [] | |
| if q: | |
| must.append({"multi_match": { | |
| "query": q, | |
| "fields": ["name^3", "name.kw^2", "domain^3", "description", "industry"], | |
| "fuzziness": "AUTO", | |
| }}) | |
| body = { | |
| "query": {"bool": {"must": must or [{"match_all": {}}], "filter": filters}}, | |
| "size": min(max(limit, 1), 100), | |
| } | |
| return await opensearch.store.search("accounts", body) | |
| async def upsert_contact(account_id: str, contact: Dict[str, Any]) -> Optional[dict]: | |
| """Dedupe contacts per account on email (fallback: linkedin_url).""" | |
| email = (contact.get("email") or "").lower().strip() or None | |
| li = contact.get("linkedin_url") | |
| existing = None | |
| if email: | |
| hits = await opensearch.store.search("contacts", { | |
| "query": {"bool": {"filter": [ | |
| {"term": {"account_id": account_id}}, {"term": {"email": email}} | |
| ]}}, | |
| "size": 1, | |
| }) | |
| existing = hits[0] if hits else None | |
| if existing is None and li: | |
| hits = await opensearch.store.search("contacts", { | |
| "query": {"bool": {"filter": [ | |
| {"term": {"account_id": account_id}}, {"term": {"linkedin_url": li}} | |
| ]}}, | |
| "size": 1, | |
| }) | |
| existing = hits[0] if hits else None | |
| contact_id = (existing or {}).get("contact_id") or str(uuid.uuid4()) | |
| doc = { | |
| "contact_id": contact_id, | |
| "account_id": account_id, | |
| "email": email, | |
| "full_name": contact.get("full_name"), | |
| "title": contact.get("title"), | |
| "linkedin_url": li, | |
| "phone": contact.get("phone"), | |
| "source": contact.get("source") or "manual", | |
| "updated_at": _now(), | |
| } | |
| doc = {k: v for k, v in doc.items() if v is not None} | |
| await opensearch.store.upsert_doc("contacts", doc, contact_id) | |
| return doc | |
| async def record_event(account_id: str, event_type: str, payload: dict) -> dict: | |
| doc = { | |
| "event_id": str(uuid.uuid4()), | |
| "account_id": account_id, | |
| "type": event_type, | |
| "payload": payload, | |
| "created_at": _now(), | |
| } | |
| await opensearch.store.index_doc("events", doc) | |
| return doc | |