Auto-enrich new contacts

npx supersonic-cli objects add-field \
  --object-type-slug "contacts" \
  --name "LinkedIn" \
  --field-type "text"

npx supersonic-cli objects add-field \
  --object-type-slug "contacts" \
  --name "Bio" \
  --field-type "text"

#!/usr/bin/env python3
"""
Auto-enrich new contacts with web search data.

Env vars: SUPERSONIC_API_KEY, TAVILY_API_KEY
State file: ~/.supersonic_enrich_last_run
"""

import os
import json
from datetime import datetime, timezone
from pathlib import Path
import httpx

API_URL = "https://mcp.supersonic.cv/api/developers/mcp/call/"
API_KEY = os.environ["SUPERSONIC_API_KEY"]
TAVILY_API_KEY = os.environ["TAVILY_API_KEY"]
OBJECT_TYPE_SLUG = "contacts"
LAST_RUN_FILE = Path.home() / ".supersonic_enrich_last_run"


def api_call(tool: str, params: dict) -> dict:
    resp = httpx.post(
        API_URL,
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        json={"tool": tool, "params": params},
        timeout=15.0,
    )
    resp.raise_for_status()
    return resp.json()


def get_last_run() -> str | None:
    if LAST_RUN_FILE.exists():
        return LAST_RUN_FILE.read_text().strip()
    return None


def save_last_run(timestamp: str):
    LAST_RUN_FILE.write_text(timestamp)


def get_new_contacts(since: str | None) -> list[dict]:
    params = {
        "object_type_slug": OBJECT_TYPE_SLUG,
        "sort": "-created_at",
        "limit": 50,
    }
    data = api_call("records.list", params)
    records = data.get("records", [])

    if since:
        records = [r for r in records if r.get("created_at", "") > since]

    return records


def search_person(name: str, company: str | None) -> dict:
    """Search the web for a person and extract enrichment data."""
    query = name
    if company:
        query = f"{name} {company}"

    resp = httpx.post(
        "https://api.tavily.com/search",
        json={
            "api_key": TAVILY_API_KEY,
            "query": query,
            "search_depth": "basic",
            "max_results": 5,
        },
        timeout=15.0,
    )
    resp.raise_for_status()
    results = resp.json().get("results", [])

    enrichment = {}

    for result in results:
        url = result.get("url", "")
        content = result.get("content", "")

        # Extract LinkedIn URL
        if "linkedin.com/in/" in url and "LinkedIn" not in enrichment:
            enrichment["LinkedIn"] = url

        # Use snippet as bio if it mentions the person
        if name.split()[0].lower() in content.lower() and "Bio" not in enrichment:
            enrichment["Bio"] = content[:500]

    return enrichment


def update_record(record_id: str, data: dict):
    api_call("records.update", {
        "object_type_slug": OBJECT_TYPE_SLUG,
        "record_id": record_id,
        "data": data,
    })


def main():
    last_run = get_last_run()
    now = datetime.now(timezone.utc).isoformat()

    contacts = get_new_contacts(last_run)
    print(f"Found {len(contacts)} new contacts to enrich.")

    for contact in contacts:
        record_id = contact["id"]
        data = contact.get("data", {})
        name = data.get("Name", "")
        company = data.get("Company")

        if not name:
            continue

        # Skip if already enriched
        if data.get("LinkedIn") or data.get("Bio"):
            print(f"  Skipping {name} (already enriched)")
            continue

        print(f"  Enriching {name}...")
        try:
            enrichment = search_person(name, company)
            if enrichment:
                update_record(record_id, enrichment)
                print(f"    Updated: {list(enrichment.keys())}")
            else:
                print(f"    No data found")
        except Exception as e:
            print(f"    Error: {e}")

    save_last_run(now)
    print("Done.")


if __name__ == "__main__":
    main()

def search_person(name: str, company: str | None) -> dict:
    query = f"{name} {company}" if company else name

    resp = httpx.get(
        "https://serpapi.com/search",
        params={
            "api_key": os.environ["SERPAPI_KEY"],
            "q": query,
            "num": 5,
        },
        timeout=15.0,
    )
    resp.raise_for_status()
    results = resp.json().get("organic_results", [])

    enrichment = {}
    for result in results:
        link = result.get("link", "")
        snippet = result.get("snippet", "")

        if "linkedin.com/in/" in link and "LinkedIn" not in enrichment:
            enrichment["LinkedIn"] = link
        if name.split()[0].lower() in snippet.lower() and "Bio" not in enrichment:
            enrichment["Bio"] = snippet[:500]

    return enrichment