"""Author RAG — ConvertKit / Mailchimp Subscriber Sync. When a visitor shows buy intent or clicks a buy link, their email (if captured) is optionally synced to the author's email list. Authors configure their ConvertKit or Mailchimp API key from the admin panel. Sync happens asynchronously via Celery. Routes: GET /api/admin/{slug}/integrations/email → current config (keys masked) PUT /api/admin/{slug}/integrations/email → save API key + list ID POST /api/admin/{slug}/integrations/email/test → send test ping to provider """ import hashlib import structlog from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field from sqlalchemy.ext.asyncio import AsyncSession from app.dependencies import get_admin_author_access, get_db logger = structlog.get_logger(__name__) router = APIRouter() PROVIDERS = ("convertkit", "mailchimp") class EmailIntegrationConfig(BaseModel): provider: str = Field(..., pattern="^(convertkit|mailchimp)$") api_key: str = Field(..., min_length=8, max_length=255) list_id: str = Field(..., max_length=100, description="ConvertKit Form ID or Mailchimp Audience ID") sync_on_buy_intent: bool = True sync_on_link_click: bool = False # In-memory store (Phase 3: encrypted DB column) _configs: dict[str, dict] = {} def _mask(key: str) -> str: return f"{key[:4]}...{key[-4:]}" if key and len(key) > 8 else "***" @router.get("/{slug}/integrations/email") async def get_email_integration(slug: str, access=Depends(get_admin_author_access)): cfg = _configs.get(slug) if not cfg: return {"configured": False} return {"configured": True, "provider": cfg["provider"], "api_key": _mask(cfg["api_key"]), "list_id": cfg["list_id"], "sync_on_buy_intent": cfg["sync_on_buy_intent"], "sync_on_link_click": cfg["sync_on_link_click"]} @router.put("/{slug}/integrations/email") async def save_email_integration(slug: str, body: EmailIntegrationConfig, access=Depends(get_admin_author_access)): _configs[slug] = body.model_dump() logger.info("Email integration saved", author_id=slug, provider=body.provider) return {"message": f"{body.provider.title()} integration saved."} @router.post("/{slug}/integrations/email/test") async def test_email_integration(slug: str, access=Depends(get_admin_author_access)): cfg = _configs.get(slug) if not cfg: raise HTTPException(400, "No email integration configured.") provider = cfg["provider"] try: import httpx async with httpx.AsyncClient(timeout=8.0) as client: if provider == "convertkit": r = await client.get( f"https://api.convertkit.com/v3/forms/{cfg['list_id']}/subscriptions", params={"api_secret": cfg["api_key"]}, ) ok = r.status_code in (200, 201) else: # mailchimp # Mailchimp API server determined from key suffix (e.g. us1) dc = cfg["api_key"].split("-")[-1] if "-" in cfg["api_key"] else "us1" r = await client.get( f"https://{dc}.api.mailchimp.com/3.0/lists/{cfg['list_id']}", auth=("anystring", cfg["api_key"]), ) ok = r.status_code == 200 return {"ok": ok, "provider": provider, "status_code": r.status_code} except Exception as e: raise HTTPException(502, f"Connection to {provider} failed: {e}") async def sync_subscriber(author_id: str, email: str, first_name: str = "") -> None: """Called by the chat pipeline when a visitor reveals their email.""" cfg = _configs.get(author_id) if not cfg: return try: import httpx async with httpx.AsyncClient(timeout=8.0) as client: if cfg["provider"] == "convertkit": await client.post( f"https://api.convertkit.com/v3/forms/{cfg['list_id']}/subscribe", json={"api_key": cfg["api_key"], "email": email, "first_name": first_name, "tags": ["chatbot-visitor"]}, ) else: dc = cfg["api_key"].split("-")[-1] if "-" in cfg["api_key"] else "us1" subscriber_hash = hashlib.md5(email.lower().encode()).hexdigest() await client.put( f"https://{dc}.api.mailchimp.com/3.0/lists/{cfg['list_id']}/members/{subscriber_hash}", auth=("anystring", cfg["api_key"]), json={"email_address": email, "status": "subscribed", "merge_fields": {"FNAME": first_name}, "tags": ["chatbot-visitor"]}, ) logger.info("Subscriber synced", author_id=author_id, provider=cfg["provider"]) except Exception as e: logger.warning("Subscriber sync failed", error=str(e))