Spaces:
Sleeping
Sleeping
| 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) | |
| 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') | |
| ) |