ProspectingAgentLoop/sales-agent/src/connectors/exa.py

"""Exa.ai news/web harvester (real API) with hard request ceilings.

Free-plan protection (the whole point of this module):
  * A persistent MONTHLY counter in output/state/exa_usage.json, reset each
    calendar month. Once EXA_MONTHLY_REQUEST_CAP is hit, no more calls are made.
  * A PER-RUN counter (process-local) capped at EXA_PER_RUN_REQUEST_CAP.
A request is only counted when an HTTP response actually comes back from Exa
(i.e. a billable call). Network failures that never reach Exa are not counted.

Returns:
  list[RawSignal]  on a successful call (may be empty if Exa found no news)
  None             when Exa is disabled, a cap is reached, or the call errored
                   -> signals the news harvester to fall back to fixtures.
"""
from __future__ import annotations

import json
import urllib.request
import urllib.error
from datetime import timedelta

from .. import config
from ..types import RawSignal, now_utc, iso

_ENDPOINT = "https://api.exa.ai/search"

# Process-local counter -> enforces the per-run cap (resets every CLI invocation).
_run_count = 0


def _usage_path():
    return config.STATE_DIR / "exa_usage.json"


def _current_month() -> str:
    return now_utc().strftime("%Y-%m")


def _load_usage() -> dict:
    config.ensure_dirs()
    p = _usage_path()
    try:
        if p.exists():
            data = json.loads(p.read_text())
            if data.get("month") == _current_month() and isinstance(data.get("count"), int):
                return data
    except (OSError, ValueError):
        pass  # unreadable/corrupt file -> reset below
    # New file, new month, or malformed -> reset the counter.
    return {"month": _current_month(), "count": 0}


def _save_usage(data: dict) -> None:
    config.ensure_dirs()
    _usage_path().write_text(json.dumps(data, indent=2))


def usage_status() -> dict:
    """Snapshot of current budget, for the CLI / startup banner."""
    u = _load_usage()
    return {
        "enabled": config.EXA_ENABLED,
        "month": u["month"],
        "used_this_month": u["count"],
        "monthly_cap": config.EXA_MONTHLY_REQUEST_CAP,
        "monthly_remaining": max(0, config.EXA_MONTHLY_REQUEST_CAP - u["count"]),
        "per_run_used": _run_count,
        "per_run_cap": config.EXA_PER_RUN_REQUEST_CAP,
    }


def _within_caps() -> tuple[bool, str]:
    if _run_count >= config.EXA_PER_RUN_REQUEST_CAP:
        return False, f"per-run cap reached ({config.EXA_PER_RUN_REQUEST_CAP})"
    u = _load_usage()
    if u["count"] >= config.EXA_MONTHLY_REQUEST_CAP:
        return False, f"monthly cap reached ({config.EXA_MONTHLY_REQUEST_CAP})"
    return True, ""


def _count_request() -> None:
    global _run_count
    _run_count += 1
    u = _load_usage()
    u["count"] += 1
    _save_usage(u)


def _build_query(account: dict, domain_scoped: bool) -> str:
    # Event-focused: target datable business events, NOT evergreen pages. We avoid
    # generic words like "hiring"/"careers" that pull recruiting landing pages.
    name = account["accountName"]
    events = (
        "acquisition, merger, earnings results, product launch, partnership, "
        "expansion, restructuring, layoffs, leadership change, executive appointment, "
        "data center, cloud, modernization, regulatory, strategy"
    )
    if domain_scoped:
        return f"{name} news and announcements: {events}"
    # No domain restriction -> anchor on the exact company name to limit drift.
    return f'"{name}" news: {events}'


def harvest_news(account: dict) -> list[RawSignal] | None:
    if not config.EXA_ENABLED:
        return None

    ok, reason = _within_caps()
    if not ok:
        print(f"   [exa] skipped — {reason}; staying within free plan (using fixtures)")
        return None

    start = iso(now_utc() - timedelta(days=config.EXA_LOOKBACK_DAYS))
    domain = (account.get("domain") or "").strip()
    domain_scoped = config.EXA_DOMAIN_FILTER and bool(domain)
    payload = {
        "query": _build_query(account, domain_scoped),
        "type": "auto",
        "numResults": config.EXA_RESULTS_PER_QUERY,
        "startPublishedDate": start,
        "contents": {"text": {"maxCharacters": 1000}},
    }
    if config.EXA_CATEGORY:
        payload["category"] = config.EXA_CATEGORY
    if domain_scoped:
        payload["includeDomains"] = [domain]
    if config.EXA_EXCLUDE_DOMAINS:
        payload["excludeDomains"] = config.EXA_EXCLUDE_DOMAINS
    req = urllib.request.Request(
        _ENDPOINT,
        data=json.dumps(payload).encode("utf-8"),
        headers={"content-type": "application/json", "x-api-key": config.EXA_API_KEY},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            body = json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        # The call reached Exa (billable) but returned an error status.
        _count_request()
        detail = e.read().decode("utf-8")[:200] if hasattr(e, "read") else ""
        print(f"   [exa] HTTP {e.code}: {detail} — falling back to fixtures")
        return None
    except urllib.error.URLError as e:
        # Never reached Exa -> not counted.
        print(f"   [exa] network error ({e.reason}) — falling back to fixtures")
        return None

    _count_request()  # successful billable request

    results = body.get("results", []) or []
    signals: list[RawSignal] = []
    for r in results:
        title = (r.get("title") or "").strip()
        text = (r.get("text") or "").strip()
        if not (title or text):
            continue
        content = f"{title}. {text}" if title else text
        signals.append(
            RawSignal(
                account_id=account["accountId"],
                source="news",
                signal_type="news_article",
                content=content[:1200],
                published_at=r.get("publishedDate") or iso(now_utc()),
                url=r.get("url", ""),
                extra={"title": title, "exa_score": r.get("score"), "via": "exa",
                       "domain_scoped": domain_scoped},
            )
        )
    remaining = usage_status()["monthly_remaining"]
    cat = config.EXA_CATEGORY or "any"
    scope = f"domain={domain}" if domain_scoped else "broad web"
    print(f"   [exa] {len(signals)} {cat} article(s) for {account['accountName']} ({scope}) "
          f"(monthly budget left: {remaining}/{config.EXA_MONTHLY_REQUEST_CAP})")
    return signals