app/stores/apify.py

"""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",
    }