Create and advance CRM deals from replies and meetings

// handlers/deals.mjs
const API_URL = "https://api.versionseven.ai/v1";
const API_KEY = process.env.VICTORIA_API_KEY;
const REPLY_STAGE = process.env.DEAL_STAGE_REPLY ?? "Qualified";

export async function call(method, path, body) {
  const response = await fetch(`${API_URL}${path}`, {
    method,
    headers: { Authorization: `Bearer ${API_KEY}`, "Content-Type": "application/json" },
    body: body === undefined ? undefined : JSON.stringify(body),
    signal: AbortSignal.timeout(30_000),
  });
  const data = await response.json().catch(() => ({}));
  if (!response.ok) throw new Error(`${method} ${path} answered ${response.status} ${data.error ?? ""}`);
  return data;
}

// The lead the payload describes, by exact email or by name and company; null when ambiguous.
export async function findLead(lead) {
  if (lead.email) {
    const { leads } = await call("GET", `/leads?search=${encodeURIComponent(lead.email)}&limit=100`);
    const exact = leads.filter((candidate) => candidate.email?.toLowerCase() === lead.email.toLowerCase());
    return exact.length === 1 ? exact[0] : null;
  }
  const name = [lead.first_name, lead.last_name].filter(Boolean).join(" ");
  if (!name || !lead.company) return null;
  const { leads } = await call("GET", `/leads?search=${encodeURIComponent(name)}&limit=100`);
  const matches = leads.filter((candidate) => candidate.company?.toLowerCase() === lead.company.toLowerCase());
  return matches.length === 1 ? matches[0] : null;
}

// The default pipeline's stages, by lower-case name.
export async function stages() {
  const { pipelines } = await call("GET", "/crm/pipelines?limit=100");
  const pipeline = pipelines.find((candidate) => candidate.is_default) ?? pipelines[0];
  if (!pipeline) throw new Error("No CRM pipeline; create one in the app first");
  const detail = await call("GET", `/crm/pipelines/${pipeline.id}`);
  return new Map(detail.stages.filter((stage) => stage.is_active).map((stage) => [stage.name.toLowerCase(), stage]));
}

export async function existingDeal(leadId) {
  const { deals } = await call("GET", `/crm/deals?lead_id=${leadId}&limit=10`);
  return deals[0] ?? null;
}

export default async function handle({ event }) {
  if (event.ai_response?.sentiment !== "positive") return;
  const payloadLead = event.lead ?? {};
  const lead = await findLead(payloadLead);
  if (!lead) {
    console.log(`No single lead matches ${payloadLead.email ?? payloadLead.first_name}; no deal created`);
    return;
  }
  if (await existingDeal(lead.id)) return; // one deal per lead

  const stage = (await stages()).get(REPLY_STAGE.toLowerCase());
  if (!stage) throw new Error(`No stage called "${REPLY_STAGE}" in the default pipeline`);
  const name = [lead.first_name, lead.last_name].filter(Boolean).join(" ");
  const deal = await call("POST", "/crm/deals", {
    name: lead.company ? `${lead.company}: ${name}` : name,
    lead_id: lead.id,
    stage_id: stage.id,
    notes: `Positive reply to "${event.campaign}" (${event.channel}):\n${event.prospect_message ?? ""}`,
  });
  console.log(`Deal ${deal.deal.id} opened for ${name} in ${stage.name}`);
}

# handlers/deals.py
import os

import requests

API_URL = "https://api.versionseven.ai/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['VICTORIA_API_KEY']}"}
REPLY_STAGE = os.environ.get("DEAL_STAGE_REPLY", "Qualified")


def call(method, path, body=None):
    response = requests.request(method, f"{API_URL}{path}", headers=HEADERS, json=body, timeout=30)
    try:
        data = response.json()
    except ValueError:
        data = {}
    if not response.ok:
        raise RuntimeError(f"{method} {path} answered {response.status_code} {data.get('error', '')}")
    return data


def find_lead(lead: dict):
    """The lead the payload describes, by exact email or by name and company; None when ambiguous."""
    if lead.get("email"):
        leads = call("GET", f"/leads?search={requests.utils.quote(lead['email'])}&limit=100")["leads"]
        exact = [c for c in leads if (c.get("email") or "").lower() == lead["email"].lower()]
        return exact[0] if len(exact) == 1 else None
    name = " ".join(part for part in (lead.get("first_name"), lead.get("last_name")) if part)
    if not name or not lead.get("company"):
        return None
    leads = call("GET", f"/leads?search={requests.utils.quote(name)}&limit=100")["leads"]
    matches = [c for c in leads if (c.get("company") or "").lower() == lead["company"].lower()]
    return matches[0] if len(matches) == 1 else None


def stages():
    """The default pipeline's stages, by lower-case name."""
    pipelines = call("GET", "/crm/pipelines?limit=100")["pipelines"]
    pipeline = next((p for p in pipelines if p.get("is_default")), pipelines[0] if pipelines else None)
    if not pipeline:
        raise RuntimeError("No CRM pipeline; create one in the app first")
    detail = call("GET", f"/crm/pipelines/{pipeline['id']}")
    return {stage["name"].lower(): stage for stage in detail["stages"] if stage.get("is_active")}


def existing_deal(lead_id):
    deals = call("GET", f"/crm/deals?lead_id={lead_id}&limit=10")["deals"]
    return deals[0] if deals else None


def handle(delivery: dict) -> None:
    event = delivery["event"]
    if (event.get("ai_response") or {}).get("sentiment") != "positive":
        return
    payload_lead = event.get("lead") or {}
    lead = find_lead(payload_lead)
    if not lead:
        print(f"No single lead matches {payload_lead.get('email') or payload_lead.get('first_name')}; no deal created", flush=True)
        return
    if existing_deal(lead["id"]):
        return  # one deal per lead

    stage = stages().get(REPLY_STAGE.lower())
    if not stage:
        raise RuntimeError(f'No stage called "{REPLY_STAGE}" in the default pipeline')
    name = " ".join(part for part in (lead.get("first_name"), lead.get("last_name")) if part)
    deal = call(
        "POST",
        "/crm/deals",
        {
            "name": f"{lead['company']}: {name}" if lead.get("company") else name,
            "lead_id": lead["id"],
            "stage_id": stage["id"],
            "notes": f'Positive reply to "{event.get("campaign")}" ({event.get("channel")}):\n{event.get("prospect_message") or ""}',
        },
    )
    print(f"Deal {deal['deal']['id']} opened for {name} in {stage['name']}", flush=True)

// bookings.mjs
import { readFile, writeFile } from "node:fs/promises";
import { call, existingDeal, stages } from "./handlers/deals.mjs";

const STATE_FILE = process.env.STATE_FILE ?? "bookings-state.json";
const MEETING_STAGE = process.env.DEAL_STAGE_MEETING;

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const state = await readFile(STATE_FILE, "utf8").then(JSON.parse).catch(() => ({}));
const runStarted = new Date().toISOString();
const allStages = await stages();
const meetingStage =
  (MEETING_STAGE && allStages.get(MEETING_STAGE.toLowerCase())) ||
  allStages.get("meeting booked") ||
  allStages.get("booked") ||
  allStages.get("closed won");
if (!meetingStage) throw new Error("No meeting stage found; set DEAL_STAGE_MEETING");

const summary = { bookings: 0, moved: 0, created: 0, unmatched: 0 };
const { campaigns } = await call("GET", "/campaigns?limit=500");
for (const campaign of campaigns) {
  const since = state[campaign.id] ?? "1970-01-01T00:00:00Z";
  const { bookings } = await call("GET", `/campaigns/${campaign.id}/bookings?since=${encodeURIComponent(since)}`);
  for (const booking of bookings) {
    if (booking.canceled_at || booking.status === "canceled") continue;
    summary.bookings++;
    if (!booking.lead) {
      summary.unmatched++;
      console.log(`${campaign.name}: booking ${booking.id} isn't matched to a lead (${booking.invitee_email ?? "no invitee email"})`);
      continue;
    }
    const deal = await existingDeal(booking.lead.id);
    const name = [booking.lead.first_name, booking.lead.last_name].filter(Boolean).join(" ");
    if (deal) {
      if (deal.stage_id !== meetingStage.id) {
        await call("PATCH", `/crm/deals/${deal.id}`, { stage_id: meetingStage.id });
        summary.moved++;
        console.log(`${campaign.name}: ${name} → ${meetingStage.name}`);
      }
    } else {
      await call("POST", "/crm/deals", {
        name: booking.lead.company ? `${booking.lead.company}: ${name}` : name,
        lead_id: booking.lead.id,
        stage_id: meetingStage.id,
        notes: `Meeting booked from "${campaign.name}" (${booking.kind}, ${booking.source})${booking.start_time ? ` for ${booking.start_time}` : ""}`,
      });
      summary.created++;
      console.log(`${campaign.name}: ${name} → new deal in ${meetingStage.name}`);
    }
    await sleep(650);
  }
  state[campaign.id] = runStarted; // next run reads only bookings made after this one started
}

await writeFile(STATE_FILE, JSON.stringify(state, null, 2));
console.log(JSON.stringify(summary));

# bookings.py
import json
import os
import time
from datetime import datetime, timezone
from urllib.parse import quote

from handlers.deals import call, existing_deal, stages

STATE_FILE = os.environ.get("STATE_FILE", "bookings-state.json")
MEETING_STAGE = os.environ.get("DEAL_STAGE_MEETING")


def main():
    try:
        with open(STATE_FILE, encoding="utf-8") as file:
            state = json.load(file)
    except (FileNotFoundError, ValueError):
        state = {}
    run_started = datetime.now(timezone.utc).isoformat()
    all_stages = stages()
    meeting_stage = (
        (MEETING_STAGE and all_stages.get(MEETING_STAGE.lower()))
        or all_stages.get("meeting booked")
        or all_stages.get("booked")
        or all_stages.get("closed won")
    )
    if not meeting_stage:
        raise SystemExit("No meeting stage found; set DEAL_STAGE_MEETING")

    summary = {"bookings": 0, "moved": 0, "created": 0, "unmatched": 0}
    for campaign in call("GET", "/campaigns?limit=500")["campaigns"]:
        since = state.get(campaign["id"], "1970-01-01T00:00:00Z")
        bookings = call("GET", f"/campaigns/{campaign['id']}/bookings?since={quote(since)}")["bookings"]
        for booking in bookings:
            if booking.get("canceled_at") or booking.get("status") == "canceled":
                continue
            summary["bookings"] += 1
            lead = booking.get("lead")
            if not lead:
                summary["unmatched"] += 1
                print(f"{campaign['name']}: booking {booking['id']} isn't matched to a lead ({booking.get('invitee_email') or 'no invitee email'})")
                continue
            name = " ".join(part for part in (lead.get("first_name"), lead.get("last_name")) if part)
            deal = existing_deal(lead["id"])
            if deal:
                if deal.get("stage_id") != meeting_stage["id"]:
                    call("PATCH", f"/crm/deals/{deal['id']}", {"stage_id": meeting_stage["id"]})
                    summary["moved"] += 1
                    print(f"{campaign['name']}: {name} → {meeting_stage['name']}")
            else:
                when = f" for {booking['start_time']}" if booking.get("start_time") else ""
                call(
                    "POST",
                    "/crm/deals",
                    {
                        "name": f"{lead['company']}: {name}" if lead.get("company") else name,
                        "lead_id": lead["id"],
                        "stage_id": meeting_stage["id"],
                        "notes": f'Meeting booked from "{campaign["name"]}" ({booking["kind"]}, {booking["source"]}){when}',
                    },
                )
                summary["created"] += 1
                print(f"{campaign['name']}: {name} → new deal in {meeting_stage['name']}")
            time.sleep(0.65)
        state[campaign["id"]] = run_started  # next run reads only bookings made after this one started

    with open(STATE_FILE, "w", encoding="utf-8") as file:
        json.dump(state, file, indent=2)
    print(json.dumps(summary))


if __name__ == "__main__":
    main()