feat: integrate QuickEnrich API for suggested people functionality

services/cards-api/.env.example

@@ -30,3 +30,8 @@ FRONTEND_URL=http://localhost:3002
 INDEXNOW_KEY=
 BING_API_KEY=
 BING_SITE_URL=https://zynd.ai
+
+# ── QuickEnrich (optional; free people search for suggested-people outsiders) ──
+QUICKENRICH_BASE_URL=
+QUICKENRICH_API_KEY=
+QUICKENRICH_AUTH_HEADER=Authorization

services/cards-api/.env.prod.example

@@ -32,3 +32,8 @@ FRONTEND_URL=https://www.zynd.ai
 INDEXNOW_KEY=
 BING_API_KEY=
 BING_SITE_URL=https://www.zynd.ai
+
+# ── QuickEnrich (optional; free people search for suggested-people outsiders) ──
+QUICKENRICH_BASE_URL=
+QUICKENRICH_API_KEY=
+QUICKENRICH_AUTH_HEADER=Authorization

services/cards-api/config.py

@@ -88,6 +88,12 @@
 MEMORY_SERVICE_TOKEN: str = os.getenv("MEMORY_SERVICE_TOKEN", "")
 MEMORY_REFRESH_INTERVAL_HOURS: int = int(os.getenv("MEMORY_REFRESH_INTERVAL_HOURS", "6"))
 
+# ── QuickEnrich (free contact-finder only; cards-api owns this client) ──
+QUICKENRICH_BASE_URL: str = os.getenv("QUICKENRICH_BASE_URL", "")
+QUICKENRICH_API_KEY: str = os.getenv("QUICKENRICH_API_KEY", "")
+QUICKENRICH_AUTH_HEADER: str = os.getenv("QUICKENRICH_AUTH_HEADER", "Authorization")
+QUICKENRICH_TIMEOUT: float = float(os.getenv("QUICKENRICH_TIMEOUT", "30"))
+
 
 def _get_supabase():
     global _sb_service

services/cards-api/services/suggested_people.py

@@ -1,10 +1,17 @@
 """Recommend published ZYND cards with overlapping profile interests."""
 from __future__ import annotations
 
+import logging
+from urllib.parse import urlparse
+
+import httpx
+
 import config
 from models.card import AgentProfileCard
 from services import cards as cards_service
 
+logger = logging.getLogger(__name__)
+
 FIELDS = ("working_on", "can_help_with", "connect_with", "love_talking_about")
 
 KEYWORDS = {
@@ -95,6 +102,148 @@ def _card_interests(card: AgentProfileCard) -> dict[str, set[str]]:
     }
 
 
+_TITLE_FOR = {
+    "Founders": "Founder",
+    "Investors": "Investor",
+    "Engineers": "Software Engineer",
+    "ML Researchers": "Machine Learning Engineer",
+    "Product Managers": "Product Manager",
+    "Designers": "Designer",
+    "Operators": "COO",
+    "Scientists": "Scientist",
+    "Recruiters": "Recruiter",
+    "Mentors": "Mentor",
+    "Potential co-founders": "Co-Founder",
+    "Students": "Student",
+    "Writers": "Writer",
+    "DevRel": "Developer Advocate",
+    "Researchers in my field": "Researcher",
+    "Healthcare builders": "Healthcare",
+    "Robotics people": "Robotics Engineer",
+}
+
+_KEYWORD_FOR = {
+    "Building with AI": "AI",
+    "Building a startup": "startup",
+    "Open source": "open source",
+    "B2B SaaS": "SaaS",
+    "Consumer apps": "consumer",
+    "Deep tech / research": "deep tech",
+    "Indie hacking": "indie hacker",
+    "ML / AI": "AI",
+    "AI / ML": "AI",
+    "Agentic AI": "AI agents",
+    "Startups": "startup",
+    "Developer tools": "developer tools",
+    "Web3 / Crypto": "crypto",
+    "Climate tech": "climate",
+    "Robotics": "robotics",
+    "SaaS": "SaaS",
+}
+
+
+def _linkedin_key(url: str) -> str:
+    raw = (url or "").strip().lower().rstrip("/")
+    if not raw:
+        return ""
+    parsed = urlparse(raw if "://" in raw else f"https://{raw}")
+    host = (parsed.hostname or "").removeprefix("www.")
+    path = parsed.path.rstrip("/")
+    return f"{host}{path}" if host else ""
+
+
+def _india_body() -> dict:
+    return {
+        "page": 1,
+        "per_page": 4,
+        "country_code": {"include": ["IN"], "exclude": []},
+    }
+
+
+def outside_search_bodies(card: AgentProfileCard) -> list[dict]:
+    interests = _card_interests(card)
+    titles = [_TITLE_FOR[k] for k in KEYWORDS["connect_with"] if k in interests["connect_with"]][:2]
+    keywords = list(dict.fromkeys(
+        _KEYWORD_FOR[k]
+        for field in ("working_on", "love_talking_about")
+        for k in KEYWORDS[field]
+        if k in interests[field] and k in _KEYWORD_FOR
+    ))[:2]
+    bodies = []
+    if titles:
+        bodies.append({**_india_body(), "title": {"include": titles, "exclude": []}})
+    if keywords:
+        bodies.append({**_india_body(), "bio_li": {"include": keywords, "exclude": []}})
+    return bodies
+
+
+def _shape_outside(record: dict) -> dict | None:
+    def field(key: str) -> str:
+        text = str(record.get(key) or "").strip()
+        return "" if text.upper() == "N/A" else text
+
+    url = field("employee_linkedin")
+    name = " ".join(part for part in (field("first_name"), field("last_name")) if part)
+    if not url and not name:
+        return None
+    return {
+        "name": name,
+        "title": field("title"),
+        "company": field("company_name"),
+        "linkedin_url": url,
+        "source": "quickenrich",
+    }
+
+
+def _qe_headers() -> dict:
+    header = (config.QUICKENRICH_AUTH_HEADER or "Authorization").strip()
+    key = config.QUICKENRICH_API_KEY
+    return {
+        header: f"Bearer {key}" if header.lower() == "authorization" else key,
+        "Accept": "application/json",
+    }
+
+
+def _contact_finder(body: dict) -> list[dict]:
+    url = f"{config.QUICKENRICH_BASE_URL.rstrip('/')}/api/employees/contact-finder"
+    try:
+        resp = httpx.post(url, json=body, headers=_qe_headers(), timeout=config.QUICKENRICH_TIMEOUT)
+        if resp.status_code >= 400:
+            logger.warning(
+                "QuickEnrich contact-finder failed: HTTP %s %s",
+                resp.status_code,
+                (resp.text or "")[:300],
+            )
+            return []
+        payload = resp.json()
+    except Exception as exc:
+        logger.warning("QuickEnrich contact-finder failed: %s", exc)
+        return []
+    records = payload.get("data") if isinstance(payload, dict) else payload
+    return [
+        shaped
+        for record in records or []
+        if isinstance(record, dict) and (shaped := _shape_outside(record))
+    ]
+
+
+def search_outside(card: AgentProfileCard, _taken: set[str]) -> list[dict]:
+    if not config.QUICKENRICH_BASE_URL or not config.QUICKENRICH_API_KEY:
+        return []
+    out = []
+    seen = set()
+    for body in outside_search_bodies(card):
+        for person in _contact_finder(body):
+            url = person.get("linkedin_url") or ""
+            key = url or person.get("name")
+            if not key or key in seen:
+                continue
+            seen.add(key)
+            out.append(person)
+            break
+    return out
+
+
 def _public_person(card: AgentProfileCard, handle: str, matched: dict[str, list[str]]) -> dict:
     return {
         "handle": handle,
@@ -108,23 +257,34 @@ def _public_person(card: AgentProfileCard, handle: str, matched: dict[str, list[
     }
 
 
+def _card_linkedin_key(card: AgentProfileCard) -> str:
+    links = card.identity.links or {}
+    return _linkedin_key(links.get("linkedin") or "")
+
+
 def get_suggested_people(handle: str) -> dict | None:
     requester = cards_service.get_card_by_handle(handle)
     if not requester:
         return None
     interests = _card_interests(requester)
     if not any(interests.values()):
-        return {"handle": handle, "people": []}
+        return {"handle": handle, "people": [], "outside": []}
 
     results = []
-    for row in cards_service.list_published_rows(columns="card,handle"):
+    taken = set()
+    if key := _card_linkedin_key(requester):
+        taken.add(key)
+    rows = cards_service.list_published_rows(columns="card,handle")
+    for row in rows:
         candidate_handle = row.get("handle")
         if not candidate_handle or candidate_handle == handle:
             continue
         try:
             candidate = AgentProfileCard.model_validate(row.get("card") or {})
         except Exception:
             continue
+        if key := _card_linkedin_key(candidate):
+            taken.add(key)
         candidate_interests = _card_interests(candidate)
         matched = {
             field: sorted(interests[field] & candidate_interests[field], key=str.casefold)
@@ -136,4 +296,17 @@ def get_suggested_people(handle: str) -> dict | None:
             results.append(_public_person(candidate, candidate_handle, matched))
 
     results.sort(key=lambda person: (-person["match_score"], person["handle"].casefold()))
-    return {"handle": handle, "people": results[:2]}
+    outside = []
+    if config.QUICKENRICH_BASE_URL and config.QUICKENRICH_API_KEY:
+        for person in search_outside(requester, taken):
+            url = person.get("linkedin_url") or ""
+            key = _linkedin_key(url)
+            if key and key in taken:
+                continue
+            if key:
+                taken.add(key)
+            person.setdefault("source", "quickenrich")
+            outside.append(person)
+            if len(outside) == 2:
+                break
+    return {"handle": handle, "people": results[:2], "outside": outside}

services/cards-api/tests/test_suggested_people.py

@@ -1,7 +1,15 @@
+import pytest
+
 from models.card import AgentProfileCard
 from services import suggested_people as people_service
 
 
+@pytest.fixture(autouse=True)
+def _disable_quickenrich(monkeypatch):
+    monkeypatch.setattr(people_service.config, "QUICKENRICH_BASE_URL", "")
+    monkeypatch.setattr(people_service.config, "QUICKENRICH_API_KEY", "")
+
+
 def _card(handle, **fields):
     return AgentProfileCard(
         id=handle,
@@ -124,7 +132,7 @@ def test_no_matches_and_unknown_interest_return_empty(monkeypatch):
         lambda **_kw: (_ for _ in ()).throw(AssertionError("should not scan without recognized interests")),
     )
 
-    assert people_service.get_suggested_people("jane") == {"handle": "jane", "people": []}
+    assert people_service.get_suggested_people("jane") == {"handle": "jane", "people": [], "outside": []}
 
 
 def test_missing_card_returns_none(monkeypatch):
@@ -182,10 +190,106 @@ def test_public_endpoint_returns_people(monkeypatch):
             "card": _card("alex", working_on=["Building with AI"]).model_dump(),
         }],
     )
+    monkeypatch.setattr(people_service.config, "QUICKENRICH_BASE_URL", "")
+    monkeypatch.setattr(people_service.config, "QUICKENRICH_API_KEY", "")
     app = FastAPI()
     app.include_router(cards_api.router, prefix="/cards")
 
     response = TestClient(app).get("/cards/by-handle/jane/suggested-people")
 
     assert response.status_code == 200
     assert response.json()["people"][0]["handle"] == "alex"
+    assert response.json()["outside"] == []
+
+
+def test_outside_empty_when_quickenrich_not_configured(monkeypatch):
+    requester = _card("jane", connect_with=["Founders"])
+    monkeypatch.setattr(people_service.cards_service, "get_card_by_handle", lambda _h: requester)
+    monkeypatch.setattr(people_service.cards_service, "list_published_rows", lambda **_kw: [])
+    monkeypatch.setattr(people_service, "search_outside", lambda _card, _rows: (_ for _ in ()).throw(AssertionError("should not search")))
+    monkeypatch.setattr(people_service.config, "QUICKENRICH_BASE_URL", "")
+    monkeypatch.setattr(people_service.config, "QUICKENRICH_API_KEY", "")
+
+    result = people_service.get_suggested_people("jane")
+
+    assert result["outside"] == []
+
+
+def test_outside_drops_zynd_linkedin_and_caps_at_two(monkeypatch):
+    requester = _card("jane", connect_with=["Founders"])
+    zynd = _card("alex", working_on=["Building with AI"])
+    zynd.identity.links = {"linkedin": "https://www.linkedin.com/in/alex"}
+    monkeypatch.setattr(people_service.cards_service, "get_card_by_handle", lambda _h: requester)
+    monkeypatch.setattr(
+        people_service.cards_service,
+        "list_published_rows",
+        lambda **_kw: [{"handle": "alex", "card": zynd.model_dump()}],
+    )
+    monkeypatch.setattr(people_service.config, "QUICKENRICH_BASE_URL", "https://qe.example")
+    monkeypatch.setattr(people_service.config, "QUICKENRICH_API_KEY", "k")
+    monkeypatch.setattr(
+        people_service,
+        "search_outside",
+        lambda _card, taken: [
+            {"name": "Alex", "title": "Founder", "company": "Acme", "linkedin_url": "https://www.linkedin.com/in/alex"},
+            {"name": "Pat", "title": "Founder", "company": "Beta", "linkedin_url": "https://linkedin.com/in/pat"},
+            {"name": "Sam", "title": "Founder", "company": "Gamma", "linkedin_url": "https://linkedin.com/in/sam"},
+            {"name": "NoUrl", "title": "Founder", "company": "Delta", "linkedin_url": ""},
+        ],
+    )
+
+    result = people_service.get_suggested_people("jane")
+
+    assert [p["linkedin_url"] for p in result["outside"]] == [
+        "https://linkedin.com/in/pat",
+        "https://linkedin.com/in/sam",
+    ]
+
+
+def test_outside_keeps_people_without_linkedin(monkeypatch):
+    requester = _card("jane", connect_with=["Founders"])
+    monkeypatch.setattr(people_service.cards_service, "get_card_by_handle", lambda _h: requester)
+    monkeypatch.setattr(people_service.cards_service, "list_published_rows", lambda **_kw: [])
+    monkeypatch.setattr(people_service.config, "QUICKENRICH_BASE_URL", "https://qe.example")
+    monkeypatch.setattr(people_service.config, "QUICKENRICH_API_KEY", "k")
+    monkeypatch.setattr(
+        people_service,
+        "search_outside",
+        lambda _card, taken: [
+            {"name": "Pat", "title": "Mentor", "company": "Acme", "linkedin_url": ""},
+        ],
+    )
+
+    result = people_service.get_suggested_people("jane")
+
+    assert result["outside"][0]["name"] == "Pat"
+    assert "email" not in result["outside"][0]
+    assert result["outside"][0]["source"] == "quickenrich"
+
+
+def test_outside_search_bodies_titles_then_keywords_india_only():
+    card = _card(
+        "jane",
+        connect_with=["Founders", "Engineers", "Investors"],
+        working_on=["Building with AI", "Open source", "B2B SaaS"],
+        love_talking_about=["Agentic AI"],
+    )
+    bodies = people_service.outside_search_bodies(card)
+    assert len(bodies) == 2
+    assert bodies[0]["title"]["include"] == ["Founder", "Investor"]
+    assert "bio_li" not in bodies[0]
+    assert bodies[1]["bio_li"]["include"] == ["AI", "open source"]
+    assert "title" not in bodies[1]
+    for body in bodies:
+        assert body["country_code"]["include"] == ["IN"]
+        assert body["per_page"] == 4
+        assert "has_email" not in body
+
+
+def test_outside_search_bodies_keywords_only_when_no_titles():
+    card = _card("jane", working_on=["Building with AI"], connect_with=[])
+    bodies = people_service.outside_search_bodies(card)
+    assert len(bodies) == 1
+    assert bodies[0]["bio_li"]["include"] == ["AI"]
+    assert "title" not in bodies[0]
+    assert bodies[0]["country_code"]["include"] == ["IN"]