Spaces:
Sleeping
Sleeping
File size: 4,159 Bytes
1de8ce6 6d5c3a1 73e6ba6 36a486d 6d5c3a1 1de8ce6 fdff54c 73e6ba6 6d5c3a1 c6d0d33 1de8ce6 0e6aefc 36a486d 7d72bcf c6d0d33 c83ffdd 36a486d c9906d2 1de8ce6 96c5d66 1de8ce6 96c5d66 1de8ce6 96c5d66 1de8ce6 7d72bcf 6d5c3a1 0e6aefc 73e6ba6 6d5c3a1 fdff54c 1de8ce6 fdff54c |
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 |
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Boolean, create_engine, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.sql import func
from datetime import datetime
import requests
from typing import Optional, List, Dict
from .settings import API_BASE_URL
from .database import get_db
import os
# Create declarative base
Base = declarative_base()
class Customer(Base):
__tablename__ = "customers"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(100), nullable=False)
company_name = Column(String(100), nullable=False)
email = Column(String(100), unique=True, nullable=False)
api_key = Column(String(64), unique=True, nullable=False)
webhook_url = Column(String(255), nullable=True, default=None) # URL where call results will be sent
is_active = Column(Boolean, default=True)
created_at = Column(DateTime(timezone=True), default=datetime.utcnow)
updated_at = Column(DateTime(timezone=True), default=datetime.utcnow, onupdate=datetime.utcnow)
def __repr__(self):
return f"<Customer {self.name}>"
def get_call_records(self, start_date: Optional[str] = None, end_date: Optional[str] = None) -> List[Dict]:
"""Get call records via API"""
url = f"{API_BASE_URL}/api/v1/calls"
params = {}
if start_date:
params['start_date'] = start_date
if end_date:
params['end_date'] = end_date
response = requests.get(
url,
headers={"api-key": self.api_key},
params=params
)
response.raise_for_status()
return response.json()
def get_call_details(self, call_id: str) -> Dict:
"""Get specific call details via API"""
url = f"{API_BASE_URL}/api/v1/calls/{call_id}"
response = requests.get(
url,
headers={"api-key": self.api_key}
)
response.raise_for_status()
return response.json()
def search_calls(self, query: Dict) -> List[Dict]:
"""Search calls via API"""
url = f"{API_BASE_URL}/api/v1/calls/search"
response = requests.get(
url,
headers={"api-key": self.api_key},
json=query
)
response.raise_for_status()
return response.json()
def send_webhook(self, data: Dict) -> bool:
"""Send call results to webhook URL if configured"""
if not self.webhook_url:
return False
try:
response = requests.post(
self.webhook_url,
json=data,
headers={"Content-Type": "application/json"},
timeout=5
)
response.raise_for_status()
return True
except Exception as e:
print(f"Webhook delivery failed: {str(e)}")
return False
class CallRecord(Base):
__tablename__ = "call_records"
id = Column(Integer, primary_key=True, index=True)
customer_id = Column(Integer, ForeignKey("customers.id"), nullable=False)
caller_number = Column(String(20), nullable=False)
called_number = Column(String(20), nullable=False)
file_path = Column(String(255), nullable=False)
transcription = Column(String(10000))
summary = Column(String(1000))
sentiment = Column(String(50))
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
@classmethod
def create_from_api(cls, customer_id: int, data: Dict) -> 'CallRecord':
"""Create a call record from API data"""
return cls(
id=data['id'],
customer_id=customer_id,
caller_number=data['caller_number'],
called_number=data['called_number'],
file_path=data.get('file_path'),
transcription=data.get('transcription'),
summary=data.get('summary'),
sentiment=data.get('sentiment'),
keywords=data.get('keywords')
) |