Spaces:
Sleeping
Sleeping
File size: 6,833 Bytes
9c6a172 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | """
CRM / Ticketing Integration
============================
Supports Zendesk out of the box; stub adapters for Freshdesk, HubSpot,
Salesforce, and a generic REST endpoint.
Environment variables (set in HF Space secrets or .env):
CRM_PROVIDER = zendesk | freshdesk | hubspot | generic
ZENDESK_SUBDOMAIN = yourcompany
ZENDESK_EMAIL = agent@yourcompany.com
ZENDESK_API_TOKEN = xxxxx
CRM_GENERIC_URL = https://your-crm.example.com/api/tickets
CRM_GENERIC_TOKEN = bearer_token
"""
import os
import uuid
import logging
from datetime import datetime
from typing import Optional
logger = logging.getLogger(__name__)
class CRMClient:
def __init__(self):
self.provider = os.getenv("CRM_PROVIDER", "demo")
self._demo = self.provider == "demo"
# ── Public API ────────────────────────────────────────────────────────────
def create_ticket(self, subject: str, description: str,
requester_name: str = "Voice User",
requester_phone: str = "",
priority: str = "normal",
tags: list[str] = None) -> dict:
"""Create a support ticket and return {ticket_id, url, status}."""
tags = tags or ["hausa-voice-agent", "auto-created"]
payload = {
"subject": subject,
"description": description,
"requester_name": requester_name,
"requester_phone": requester_phone,
"priority": priority,
"tags": tags,
"created_at": datetime.utcnow().isoformat(),
"channel": "voice",
}
if self._demo:
return self._demo_ticket(payload)
if self.provider == "zendesk":
return self._zendesk_create(payload)
return self._generic_create(payload)
def get_ticket(self, ticket_id: str) -> Optional[dict]:
if self._demo:
return {"ticket_id": ticket_id, "status": "open",
"created_at": datetime.utcnow().isoformat()}
if self.provider == "zendesk":
return self._zendesk_get(ticket_id)
return None
def update_ticket(self, ticket_id: str, comment: str,
status: str = "pending") -> dict:
if self._demo:
logger.info(f"[DEMO] Update ticket {ticket_id}: {comment}")
return {"ticket_id": ticket_id, "status": status}
if self.provider == "zendesk":
return self._zendesk_update(ticket_id, comment, status)
return {}
def assign_to_human(self, ticket_id: str,
agent_group: str = "frontline") -> dict:
"""Transfer ticket to human agent queue."""
comment = f"[Voice AI] Escalated to human agent. Group: {agent_group}"
return self.update_ticket(ticket_id, comment, status="open")
# ── Demo stub ─────────────────────────────────────────────────────────────
@staticmethod
def _demo_ticket(payload: dict) -> dict:
ticket_id = "TKT-" + str(uuid.uuid4())[:8].upper()
logger.info(f"[DEMO] CRM ticket created: {ticket_id} | {payload['subject']}")
return {
"ticket_id": ticket_id,
"url": f"https://demo.zendesk.com/tickets/{ticket_id}",
"status": "created",
"provider": "demo",
}
# ── Zendesk ───────────────────────────────────────────────────────────────
def _zendesk_create(self, payload: dict) -> dict:
import requests
subdomain = os.environ["ZENDESK_SUBDOMAIN"]
email = os.environ["ZENDESK_EMAIL"]
token = os.environ["ZENDESK_API_TOKEN"]
url = f"https://{subdomain}.zendesk.com/api/v2/tickets.json"
body = {
"ticket": {
"subject": payload["subject"],
"comment": {"body": payload["description"]},
"requester": {"name": payload["requester_name"]},
"priority": payload["priority"],
"tags": payload["tags"],
}
}
r = requests.post(url, json=body,
auth=(f"{email}/token", token), timeout=10)
r.raise_for_status()
data = r.json()["ticket"]
return {
"ticket_id": str(data["id"]),
"url": data["url"],
"status": "created",
"provider": "zendesk",
}
def _zendesk_get(self, ticket_id: str) -> dict:
import requests
subdomain = os.environ["ZENDESK_SUBDOMAIN"]
email = os.environ["ZENDESK_EMAIL"]
token = os.environ["ZENDESK_API_TOKEN"]
url = f"https://{subdomain}.zendesk.com/api/v2/tickets/{ticket_id}.json"
r = requests.get(url, auth=(f"{email}/token", token), timeout=10)
r.raise_for_status()
data = r.json()["ticket"]
return {"ticket_id": ticket_id, "status": data["status"],
"subject": data["subject"]}
def _zendesk_update(self, ticket_id: str, comment: str, status: str) -> dict:
import requests
subdomain = os.environ["ZENDESK_SUBDOMAIN"]
email = os.environ["ZENDESK_EMAIL"]
token = os.environ["ZENDESK_API_TOKEN"]
url = f"https://{subdomain}.zendesk.com/api/v2/tickets/{ticket_id}.json"
body = {"ticket": {"status": status,
"comment": {"body": comment, "public": False}}}
r = requests.put(url, json=body,
auth=(f"{email}/token", token), timeout=10)
r.raise_for_status()
return {"ticket_id": ticket_id, "status": status}
# ── Generic REST ──────────────────────────────────────────────────────────
def _generic_create(self, payload: dict) -> dict:
import requests
url = os.environ["CRM_GENERIC_URL"]
token = os.environ.get("CRM_GENERIC_TOKEN", "")
headers = {"Authorization": f"Bearer {token}",
"Content-Type": "application/json"}
r = requests.post(url, json=payload, headers=headers, timeout=10)
r.raise_for_status()
data = r.json()
return {"ticket_id": str(data.get("id", "?")), "status": "created",
"provider": "generic"}
|