import os import re import requests import time import threading from datetime import datetime from itertools import cycle from dotenv import load_dotenv from colorama import Fore, init from openai import OpenAI init(autoreset=True) load_dotenv() # === CONFIG / ENV === ZOHO_ACCESS_TOKEN = os.getenv("ZOHO_ACCESS_TOKEN") VAPI_API_KEY = os.getenv("VAPI_API_KEY") VAPI_ASSISTANT_IDS = [a.strip() for a in os.getenv("VAPI_ASSISTANT_IDS", "").split(",") if a.strip()] VAPI_PHONE_NUMBER_ID = os.getenv("VAPI_PHONE_NUMBER_ID") OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") if not (ZOHO_ACCESS_TOKEN and VAPI_API_KEY and VAPI_ASSISTANT_IDS and VAPI_PHONE_NUMBER_ID and OPENAI_API_KEY): raise SystemExit("Missing required env vars") client = OpenAI(api_key=OPENAI_API_KEY) vapi_headers = {"Authorization": f"Bearer {VAPI_API_KEY}", "Content-Type": "application/json"} zoho_headers = {"Authorization": f"Zoho-oauthtoken {ZOHO_ACCESS_TOKEN}", "Content-Type": "application/json"} COLORS = [Fore.MAGENTA, Fore.CYAN, Fore.YELLOW, Fore.GREEN, Fore.BLUE, Fore.WHITE] CALLABLE_STATUSES = {"attempted to contact", "contacted", "", "not contacted", "new"} SKIP_STATUSES = {"junk lead", "contact in future"} ASSISTANT_CYCLE = cycle(VAPI_ASSISTANT_IDS) def is_valid_zoho_id(zoho_id: str) -> bool: return bool(zoho_id) and isinstance(zoho_id, str) def get_zoho_leads(): url = "https://www.zohoapis.in/crm/v2/Leads" params = {"fields": "id,Full_Name,Phone,Email,Company,Lead_Status"} r = requests.get(url, headers=zoho_headers, params=params) r.raise_for_status() return r.json().get("data", []) def format_phone(phone: str) -> str: phone = (phone or "").strip() if not phone: return phone if not phone.startswith("+"): phone = re.sub(r"\D", "", phone) phone = f"+91{phone}" return phone def initiate_vapi_call(lead, assistant_id, max_attempts=3): phone = format_phone(lead.get("Phone") or "") if not phone: print(Fore.YELLOW + f"⚠️ No phone for {lead.get('Full_Name')}, skipping.") return None payload = { "assistantId": assistant_id.strip(), "phoneNumberId": VAPI_PHONE_NUMBER_ID.strip(), "customer": {"name": lead.get("Full_Name"), "number": phone}, "metadata": {"email": lead.get("Email") or "", "company": lead.get("Company") or ""} } for attempt in range(1, max_attempts + 1): try: resp = requests.post("https://api.vapi.ai/call", headers=vapi_headers, json=payload) resp.raise_for_status() call_id = resp.json().get("id") print(Fore.GREEN + f"✅ Call queued for {lead.get('Full_Name')} (call_id={call_id})") return call_id except Exception as e: print(Fore.RED + f"❌ Attempt {attempt}/{max_attempts} failed: {e}") time.sleep(2) return None def fetch_call_status(call_id, max_wait_minutes=15): url = f"https://api.vapi.ai/call/{call_id}" start_time = datetime.now() transcript = "" final_status = "pending" elapsed = 0 while elapsed < max_wait_minutes * 60: try: r = requests.get(url, headers=vapi_headers) r.raise_for_status() data = r.json() current_status = data.get("status", "pending") transcript = data.get("transcript", "") or transcript if current_status in ("completed", "missed", "failed", "ended"): final_status = current_status break time.sleep(5) elapsed += 5 except Exception: time.sleep(5) elapsed += 5 end_time = datetime.now() duration_min = max(0, int((end_time - start_time).seconds // 60)) return transcript, start_time, end_time, duration_min, final_status def summarize_transcript(transcript: str): if not transcript.strip(): return "No conversation recorded.", "neutral" prompt = ( transcript + "\n\nSummarize in 3-4 lines and classify lead as:\n- interested\n- not interested\n- neutral\n- call me later\n\n" "Respond as:\nSummary: ...\nStatus: ..." ) try: resp = client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}], max_tokens=200, ) content = resp.choices[0].message.content.strip() except Exception: return "Failed to summarize.", "neutral" summary = "" status = "neutral" for line in content.splitlines(): if line.lower().startswith("summary:"): summary = line.split(":", 1)[1].strip() if line.lower().startswith("status:"): status = line.split(":", 1)[1].strip().lower() return summary, status def log_call_in_zoho(lead, start_time, end_time, duration, transcript, summary, ai_status, call_status): lead_id = lead.get("id") if not is_valid_zoho_id(lead_id): print(Fore.RED + f"⚠️ Invalid Zoho ID for {lead.get('Full_Name')}, skipping update") return # Determine CRM Lead Status if call_status in ("missed", "failed"): crm_status = "Attempted to Contact" elif ai_status == "interested": crm_status = "Contact In Future" elif ai_status == "not interested": crm_status = "Junk Lead" else: crm_status = "Contacted" # Log call with retries call_payload = { "data": [ { "Subject": f"Call with {lead.get('Full_Name')}", "Call_Type": "Outbound", "Call_Start_Time": start_time.strftime("%Y-%m-%dT%H:%M:%S+05:30"), "Call_Duration": str(duration), "Call_Purpose": "Client Discussion", "Description": f"Transcript:\n{transcript}\n\nSummary:\n{summary}", "Call_Status": crm_status, "What_Id": lead_id, "$se_module": "Leads" } ] } for attempt in range(1, 4): try: r = requests.post("https://www.zohoapis.in/crm/v2/Calls", headers=zoho_headers, json=call_payload) if r.status_code in (200, 201): print(Fore.GREEN + f"📋 Call logged for {lead.get('Full_Name')}") break else: print(Fore.RED + f"Attempt {attempt}: Call log failed {r.status_code} | {r.text}") except Exception as e: print(Fore.RED + f"Attempt {attempt}: Call log exception: {e}") time.sleep(2) # Update Lead Status + add summary with retries upd_payload = { "data": [ { "Lead_Status": crm_status, "Description": f"Transcript:\n{transcript}\n\nSummary:\n{summary}" } ] } update_url = f"https://www.zohoapis.in/crm/v2/Leads/{lead_id}" for attempt in range(1, 4): try: r = requests.put(update_url, headers=zoho_headers, json=upd_payload) if r.status_code in (200, 201): print(Fore.BLUE + f"🔄 Lead '{lead.get('Full_Name')}' updated to '{crm_status}'") break else: print(Fore.RED + f"Attempt {attempt}: Lead update failed {r.status_code} | {r.text}") except Exception as e: print(Fore.RED + f"Attempt {attempt}: Lead update exception: {e}") time.sleep(2) def schedule_callback(lead, minutes, assistant_id): def _job(): call_id = initiate_vapi_call(lead, assistant_id) if call_id: transcript, st, en, dur, call_status = fetch_call_status(call_id) summary, ai_status = summarize_transcript(transcript) log_call_in_zoho(lead, st, en, dur, transcript, summary, ai_status, call_status) t = threading.Timer(minutes * 60, _job) t.daemon = True t.start() def main(): leads = get_zoho_leads() print(Fore.GREEN + f"Retrieved {len(leads)} leads from Zoho\n") for idx, lead in enumerate(leads): print("=====================",lead) color = COLORS[idx % len(COLORS)] name = lead.get("Full_Name") or "Unknown" status = (lead.get("Lead_Status") or "").strip().lower() phone = lead.get("Phone") or "" if status in SKIP_STATUSES: print(color + f"Skipping {name} (status={status})") continue if status not in CALLABLE_STATUSES: print(color + f"Skipping {name} (status not eligible)") continue if not phone.strip(): print(Fore.YELLOW + f"No phone for {name}, skipping.") continue assistant_id = next(ASSISTANT_CYCLE) print(color + f"\nCalling {name} ({phone}) ...") call_id = initiate_vapi_call(lead, assistant_id) if not call_id: continue transcript, start_time, end_time, duration, call_status = fetch_call_status(call_id) summary, ai_status = summarize_transcript(transcript) print(color + f"Transcript:\n{transcript}\n") print(color + f"AI Summary: {summary} | Status: {ai_status} | Call status: {call_status}") if ai_status == "call me later": try: minutes = int(input(f"Enter minutes to re-call {name}: ").strip() or "2") except Exception: minutes = 2 schedule_callback(lead, minutes, assistant_id) log_call_in_zoho(lead, start_time, end_time, duration, transcript, summary, ai_status, call_status) print(Fore.GREEN + "\nAll leads processed. Scheduled callbacks will run in background.") if __name__ == "__main__": main()