import re import os import pytz import logging import tempfile import assemblyai as aai from datetime import datetime from django.conf import settings from utils.logger import log_step from .twilio import get_twilio_client from .json_sanitizer import ModelJSONParser from .recording_service import RecordingService logger = logging.getLogger(__name__) # add check to check if transfer successful then add 3 speaker other wise two only def extract_datetime_str(text): """Extract datetime in YYYY-MM-DD HH:MM:SS format from any messy text.""" match = re.search(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", text) if match: return match.group(0) return None class AssemblyAIService: MODEL = aai.LemurModel.claude3_7_sonnet_20250219 _instance = None def __new__(cls, *args, **kwargs): if cls._instance is None: cls._instance = super(AssemblyAIService, cls).__new__(cls) return cls._instance def __init__(self): if hasattr(self, '_initialized') and self._initialized: return aai.settings.api_key = getattr( settings, 'ASSEMBLY_AI_API_KEY', '889d64c86df04953b93f13008eb8cd43' ) self.call_sid = None self.call_time_utc = None self.transcriber = aai.Transcriber() self.twilio_client = get_twilio_client() self.recording_service = RecordingService(get_twilio_client()) self._initialized = True def get_current_time_pst(self): pst = pytz.timezone('US/Pacific') return datetime.now(pst) def get_call_date_time_pst(self): if not self.call_time_utc: return None pst = pytz.timezone('US/Pacific') call_time_pst = self.call_time_utc.astimezone(pst) return call_time_pst def download_recording(self): recording = self.recording_service.get_latest_call_recording(self.call_sid) self.call_time_utc = recording.start_time response = self.recording_service.download(recording.sid) return response.content def transcribe_audio(self, use_speaker_labels=False): raw_cfg = aai.TranscriptionConfig( speaker_labels=False, punctuate=True, format_text=True, disfluencies=False, ) labeled_cfg = aai.TranscriptionConfig( speaker_labels=True, punctuate=True, format_text=True, speakers_expected=3, ) temp_file_path = None speaker_utterances = None try: log_step("DOWNLOAD_RECORDING", self.call_sid) audio_bytes = self.download_recording() if audio_bytes: with tempfile.NamedTemporaryFile(suffix='.mp3', delete=False) as f: f.write(audio_bytes) temp_file_path = f.name log_step("START_TRANSCRIBE", self.call_sid, bytes=len(audio_bytes)) raw_transcript = self.transcriber.transcribe(audio_bytes, config=raw_cfg) if use_speaker_labels: labeled = self.transcriber.transcribe(audio_bytes, config=labeled_cfg) speaker_utterances = labeled.utterances finally: if temp_file_path and os.path.exists(temp_file_path): os.unlink(temp_file_path) return raw_transcript, speaker_utterances def extract_datetime_from_text(self, text): match = re.search(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", text) if not match: logger.warning(f"No valid datetime found in response: {text}") return None return match.group(0) def analyze_transcript(self, transcript): call_time_pst = self.get_call_date_time_pst() # 2025-12-01 11:11:24-08:00 if call_time_pst is None: call_time_pst = self.get_current_time_pst().strftime('%Y-%m-%d %H:%M:%S') prompt = f""" You are an AI assistant analyzing a phone call transcript. Your job is to extract booking intent with extremely high accuracy. TRANSCRIPT: \"\"\"{transcript.text}\"\"\" CURRENT PST TIME: {call_time_pst} RULES: 1. booking_intent MUST be 1 or 0 ONLY. 2. booking_datetime MUST be in exact format "YYYY-MM-DD HH:MM:SS" or null. 3. If customer expresses ANY intent to schedule, set booking_intent = 1. 4. Consider phrases: - "yes let's book" - "I want an appointment" - "schedule me" - "come tomorrow", "next Monday", "after 2 hours" - ANY mention of date/time relating to booking. 5. You MUST convert relative times (tomorrow, next Sunday, in 3 hours) using PST timezone. 6. If date but no time → use "00:00:00". 7. If unclear → booking_intent = 0 and booking_datetime = null. OUTPUT STRICTLY IN JSON: {{ "booking_intent": <1 or 0>, "booking_datetime": "", "summary": "" }} """ result = transcript.lemur.task(prompt, final_model=self.MODEL).response parsed = ModelJSONParser.parse(result) if not parsed: logger.warning("LLM returned invalid JSON") return 0, None, "", result booking_intent = parsed.get("booking_intent", 0) booking_datetime = parsed.get("booking_datetime", None) summary = parsed.get("summary", "") logger.info(f"booking_datetime directly from LLM: {booking_datetime}") # 2025-12-02 17:00:00 PST Time return booking_intent, booking_datetime, summary, result def convert_booking_datetime_to_utc(self, booking_datetime): if not booking_datetime or booking_datetime.lower() in ['no', 'none', 'n/a', '']: return None try: dt = datetime.strptime(booking_datetime, '%Y-%m-%d %H:%M:%S') pst = pytz.timezone('US/Pacific') dt_pst = pst.localize(dt) dt_utc = dt_pst.astimezone(pytz.UTC) return dt_utc.strftime('%Y-%m-%d %H:%M:%S') except ValueError as e: logger.warning(f"Invalid datetime format received: {booking_datetime}. Error: {e}") return None except Exception as e: logger.error(f"Error converting datetime: {e}") return None def process_transcription(self, call_sid, use_speaker=False): self.call_sid = call_sid try: raw_transcript, speaker_utterances = self.transcribe_audio( use_speaker_labels=use_speaker ) log_step("TRANSCRIBED", call_sid) booking_intent, dt_raw, summary, llm_raw = self.analyze_transcript(raw_transcript) log_step("AI_ANALYSIS", call_sid, booking_intent=booking_intent) booking_datetime = self.convert_booking_datetime_to_utc(dt_raw) log_step("DATETIME_CONVERTED", call_sid, utc=booking_datetime) # "utc": "2025-12-03 01:00:00" logger.info(f"summary: {summary}") logger.info(f"llm_raw: {llm_raw}") return { 'transcript': raw_transcript.text, 'booking_intent': booking_intent, 'booking_datetime': booking_datetime, "speaker_labels": speaker_utterances, 'results': { "summary": summary, "llm_raw": llm_raw } } except Exception as e: log_step("TRANSCRIPTION_FAILED", call_sid, error=repr(e)) return { 'transcript': '', 'booking_intent': 0, 'booking_datetime': None, "speaker_labels": None, 'results': '' } def get_speech_to_text(call_sid: str, use_speaker=False): if call_sid is None: logger.info(f"No Twilio call ID for call {call.id}") return _,_,_,_,_ service = AssemblyAIService() result = service.process_transcription(call_sid, use_speaker=use_speaker) return ( result['transcript'], result['booking_intent'], result['booking_datetime'], result["speaker_labels"], result['results'] )