"""Shared call ingest: normalize, upsert, filters, STT queue — webhook, poll, and sync.""" from __future__ import annotations import json import logging import os import subprocess from datetime import datetime from typing import Any, Dict, List, Optional, Tuple from urllib.parse import urlparse import pika from feature_entitlements import module_active from mcube_phone_util import pick_mcube_customer_phone, sanitize_agent_callinfo from min_duration_util import ( call_duration_seconds, evaluate_min_duration_for_ingest, purge_unprocessed_if_below_min, ) from pipeline_usage_control import pause_pipeline_if_usage_exhausted from usage_allocation_util import evaluate_usage_allocation logger = logging.getLogger(__name__) def _first_value(payload: Dict[str, Any], *keys: str) -> Any: for key in keys: if key in payload and payload[key] is not None and str(payload[key]).strip() != "": return payload[key] return None def _parse_datetime(value) -> Optional[datetime]: if value is None: return None if isinstance(value, datetime): return value.replace(tzinfo=None) if value.tzinfo else value try: text = str(value).replace("Z", "+00:00") parsed = datetime.fromisoformat(text) return parsed.replace(tzinfo=None) if parsed.tzinfo else parsed except Exception: return None def _format_db_datetime(value) -> Optional[str]: parsed = _parse_datetime(value) if not parsed: return str(value).strip() if value is not None else None return parsed.strftime("%Y-%m-%d %H:%M:%S") def recording_url_candidates(filename: str, starttime, source_bid: str) -> List[str]: filename = str(filename or "").strip() if not filename: return [] if filename.startswith("http"): return [filename] try: dt = _parse_datetime(starttime) if not dt and isinstance(starttime, str): dt = datetime.strptime(starttime, "%Y-%m-%d %H:%M:%S") if not dt: return [filename] year = dt.strftime("%Y") month = dt.strftime("%m") except Exception: return [filename] name = os.path.basename(filename) bid = str(source_bid or "").strip() return [ f"https://recordings.mcube.com/mcubefiles112/appmcube/{year}/{month}/{bid}/{name}", f"https://recordings.mcube.com/mcubefiles112/classic/{year}/{month}/{bid}/inbound/{name}", ] def build_recording_url(filename: str, starttime, source_bid: str) -> str: candidates = recording_url_candidates(filename, starttime, source_bid) return candidates[0] if candidates else str(filename or "") def normalize_webhook_payload( payload: Dict[str, Any], *, source_bid: str ) -> Tuple[Dict[str, Any], List[str]]: """Map canonical or common alias fields to raw_calls column names.""" errors: List[str] = [] call_id = _first_value(payload, "call_id", "callid", "callId") call_status = _first_value(payload, "call_status", "dialstatus", "callStatus") call_start = _first_value(payload, "call_start_time", "call_starttime", "starttime", "call_start") call_end = _first_value(payload, "call_end_time", "call_endtime", "endtime", "call_end") recording = _first_value(payload, "recording_url", "fileurl", "fileUrl", "file_url", "filename") duration = _first_value(payload, "duration_seconds", "duration", "answeredtime", "talktime") if not call_id: errors.append("call_id is required") if not call_status: errors.append("call_status is required") if not call_start: errors.append("call_start_time is required") if not recording: errors.append("recording_url is required") fileurl = str(recording or "").strip() start_fmt = _format_db_datetime(call_start) if fileurl and not fileurl.startswith("http"): candidates = recording_url_candidates(fileurl, start_fmt or call_start, source_bid) fileurl = candidates[0] if candidates else fileurl direction = str(_first_value(payload, "direction") or "inbound").strip().lower() or "inbound" agent_name = str(_first_value(payload, "agent_name", "agentname") or "") agent_phone = str(_first_value(payload, "agent_phone", "agent_callinfo", "emp_phone") or "") phone_row = { "customer_callinfo": _first_value(payload, "customer_phone", "customer_callinfo"), "callto": _first_value(payload, "callto", "call_to"), "callfrom": _first_value(payload, "callfrom", "call_from"), "clicktocalldid": _first_value(payload, "clicktocalldid"), "direction": direction, "callid": call_id, } normalized = { "callid": str(call_id or "").strip(), "call_status": str(call_status or "").strip().upper(), "call_starttime": start_fmt, "call_endtime": _format_db_datetime(call_end), "fileurl": fileurl, "agentname": agent_name, "groupname": str(_first_value(payload, "group_name", "groupname") or ""), "direction": direction, "agent_callinfo": sanitize_agent_callinfo(agent_name, agent_phone), "customer_callinfo": pick_mcube_customer_phone(phone_row), "duration_seconds": None, "source": str(_first_value(payload, "source", "provider") or ""), } if duration is not None and str(duration).strip() != "": try: normalized["duration_seconds"] = max(0, int(float(duration))) except (TypeError, ValueError): pass if normalized["duration_seconds"] is None: normalized["duration_seconds"] = call_duration_seconds( { "call_starttime": normalized["call_starttime"], "call_endtime": normalized["call_endtime"], "duration_seconds": duration, } ) return normalized, errors def normalize_poll_call(call: Dict[str, Any], *, bid: str, source_bid: str) -> Dict[str, Any]: """Map a Mcube source DB row to raw_calls column names.""" recording_url = build_recording_url(call.get("filename"), call.get("starttime"), source_bid) agent_name = str(call.get("agentname") or "") duration_row = { "callid": call.get("callid"), "starttime": call.get("starttime"), "endtime": call.get("endtime"), "duration_seconds": call.get("duration_seconds"), "answeredtime": call.get("answeredtime"), "pulse": call.get("pulse"), "call_starttime": call.get("starttime"), "call_endtime": call.get("endtime"), } return { "callid": str(call["callid"]), "fileurl": recording_url, "agentname": agent_name, "groupname": str(call.get("groupname") or ""), "call_starttime": call.get("starttime"), "call_endtime": call.get("endtime"), "call_status": str(call.get("dialstatus") or "ANSWER"), "agent_callinfo": sanitize_agent_callinfo(agent_name, str(call.get("emp_phone") or "")), "customer_callinfo": pick_mcube_customer_phone(call), "direction": (str(call.get("direction") or "inbound").strip().lower() or "inbound"), "duration_seconds": call_duration_seconds({**call, **duration_row}), "_duration_probe_row": {**call, **duration_row}, } def normalize_sync_call(call: Dict[str, Any], *, bid: str) -> Dict[str, Any]: """Map legacy sync_calls.py source row to raw_calls column names.""" poll_like = { "callid": call.get("callid"), "filename": call.get("filename"), "starttime": call.get("starttime"), "endtime": call.get("endtime"), "agentname": call.get("agentname"), "groupname": call.get("groupname"), "dialstatus": call.get("dialstatus"), "emp_phone": call.get("emp_phone"), "clicktocalldid": call.get("clicktocalldid"), "direction": call.get("direction"), } return normalize_poll_call(poll_like, bid=bid, source_bid=bid) def _is_trusted_mcube_recording_url(url: str) -> bool: parsed = urlparse(str(url or "")) if (parsed.hostname or "").lower() != "recordings.mcube.com": return False path = (parsed.path or "").lower() return "/mcubefiles" in path and path.endswith( (".wav", ".mp3", ".ogg", ".mpeg", ".m4a", ".webm", ".flac") ) def _http_recording_reachable(url: str) -> bool: url = str(url or "").strip() if not url.startswith(("http://", "https://")): return False try: head = subprocess.run( ["curl", "-I", "-L", "--max-time", "20", "-s", url], capture_output=True, text=True, ) if head.returncode == 0 and any( line.startswith("HTTP/") and any(code in line for code in (" 200", " 206")) for line in (head.stdout or "").splitlines() ): return True except Exception: pass try: ranged = subprocess.run( [ "curl", "-L", "--max-time", "20", "-s", "-o", "/dev/null", "-w", "%{http_code}", "-r", "0-1023", url, ], capture_output=True, text=True, ) if ranged.returncode == 0 and (ranged.stdout or "").strip() in ("200", "206"): return True except Exception: pass return False def _is_bulk_csv_upload(row: Optional[Dict[str, Any]]) -> bool: if not row: return False extra = row.get("extra_fields") if isinstance(extra, str): try: extra = json.loads(extra) except Exception: return False return isinstance(extra, dict) and bool(extra.get("bulk_csv")) def recording_url_ready( fileurl: str, callid: str, *, row: Optional[Dict[str, Any]] = None, ) -> bool: if _http_recording_reachable(fileurl): return True if row and _is_bulk_csv_upload(row) and _is_trusted_mcube_recording_url(fileurl): logger.info("[%s] Trusting bulk CSV Mcube recording URL: %s", callid, fileurl) return True trust_mcube = os.getenv("ORCHESTRATOR_TRUST_MCUBE_URLS", "").lower() in ("1", "true", "yes") if trust_mcube and _is_trusted_mcube_recording_url(fileurl): logger.info("[%s] Trusting Mcube recording URL (ORCHESTRATOR_TRUST_MCUBE_URLS)", callid) return True return False def group_allowed(groupname: str, cfg: Dict[str, Any]) -> Tuple[bool, Optional[str]]: if not bool(int(cfg.get("group_filter_enabled") or 0)): return True, None allowed = [ str(g).strip().lower() for g in (cfg.get("_allowed_groupnames") or []) if str(g).strip() ] if not allowed: return False, "group_filter_enabled but no allowed groups configured" name = str(groupname or "").strip().lower() if name not in allowed: return False, f"group '{groupname}' not in allowed list" return True, None def effective_min_duration_s(cfg: Dict[str, Any]) -> int: min_duration_s = max(0, int(cfg.get("min_call_duration_s") or 0)) if not module_active(cfg, "min_duration_filter_enabled"): return 0 return min_duration_s def check_min_duration_for_ingest( row: Dict[str, Any], *, min_duration_s: int, effective_at, recording_url: str, probe_audio: bool = True, log_prefix: str = "Not ingested", ) -> Tuple[bool, Optional[str], Optional[float]]: """Return (should_skip, reason, probed_audio_seconds).""" if min_duration_s <= 0: return False, None, None skip, reason, probed = evaluate_min_duration_for_ingest( row, min_duration_s, effective_at, recording_url, probe_audio=probe_audio, ) if skip: logger.info("[%s] %s (%s)", row.get("callid"), log_prefix, reason) return True, reason, probed return False, None, probed def ensure_call_ingest_exclusions_table(cursor) -> None: cursor.execute( """ CREATE TABLE IF NOT EXISTS call_ingest_exclusions ( id BIGINT AUTO_INCREMENT PRIMARY KEY, bid VARCHAR(50) NOT NULL, callid VARCHAR(64) NOT NULL, reason VARCHAR(255) NULL, excluded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uniq_bid_callid (bid, callid), KEY idx_bid (bid) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """ ) def load_excluded_callids(cursor, bid: str) -> set: """Call IDs blocked from Mcube/orchestrator re-ingest after permanent delete.""" ensure_call_ingest_exclusions_table(cursor) cursor.execute( "SELECT callid FROM call_ingest_exclusions WHERE bid = %s", (str(bid),), ) return { str(row["callid"]) for row in (cursor.fetchall() or []) if row.get("callid") } def is_call_ingest_excluded(cursor, bid: str, callid: str) -> bool: ensure_call_ingest_exclusions_table(cursor) cursor.execute( "SELECT 1 FROM call_ingest_exclusions WHERE bid = %s AND callid = %s LIMIT 1", (str(bid), str(callid)), ) return cursor.fetchone() is not None def exclude_calls_from_ingest( cursor, bid: str, callids: List[str], *, reason: str = "deleted", ) -> int: ensure_call_ingest_exclusions_table(cursor) added = 0 for callid in callids: cid = str(callid or "").strip() if not cid: continue cursor.execute( """ INSERT INTO call_ingest_exclusions (bid, callid, reason) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE reason = VALUES(reason) """, (str(bid), cid, str(reason or "deleted")[:255]), ) if cursor.rowcount >= 1: added += 1 return added def _table_exists_on_cursor(cursor, table_name: str) -> bool: cursor.execute("SHOW TABLES LIKE %s", (table_name,)) return cursor.fetchone() is not None def permanently_delete_calls( cursor, bid: str, callids: List[str], *, reason: str = "deleted", ) -> Dict[str, int]: """Remove call data and block future sync from Mcube source.""" unique_ids = [] seen = set() for callid in callids: cid = str(callid or "").strip() if not cid or cid in seen: continue seen.add(cid) unique_ids.append(cid) if not unique_ids: return {"deleted_raw": 0, "excluded": 0} placeholders = ", ".join(["%s"] * len(unique_ids)) for suffix in ("sarvamresponse", "callanalytics"): table = f"{bid}_{suffix}" if _table_exists_on_cursor(cursor, table): cursor.execute( f"DELETE FROM `{table}` WHERE callid IN ({placeholders})", unique_ids, ) deleted_raw = 0 raw_table = f"{bid}_raw_calls" if _table_exists_on_cursor(cursor, raw_table): cursor.execute( f"DELETE FROM `{raw_table}` WHERE callid IN ({placeholders})", unique_ids, ) deleted_raw = int(cursor.rowcount or 0) excluded = exclude_calls_from_ingest(cursor, bid, unique_ids, reason=reason) return {"deleted_raw": deleted_raw, "excluded": excluded} def upsert_raw_call(cursor, bid: str, row: Dict[str, Any]) -> None: """Insert or update raw_calls without resetting in-flight status.""" cursor.execute( f""" INSERT INTO `{bid}_raw_calls` (bid, callid, fileurl, status, agentname, groupname, call_starttime, call_endtime, call_status, agent_callinfo, customer_callinfo, direction, transcription_requested, transcription_status, selected_for_processing, duration_seconds) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE fileurl = VALUES(fileurl), agentname = VALUES(agentname), groupname = VALUES(groupname), call_starttime = VALUES(call_starttime), call_endtime = VALUES(call_endtime), call_status = VALUES(call_status), agent_callinfo = VALUES(agent_callinfo), customer_callinfo = VALUES(customer_callinfo), direction = VALUES(direction), duration_seconds = VALUES(duration_seconds) """, ( bid, row["callid"], row["fileurl"], 0, row.get("agentname") or "", row.get("groupname") or "", row.get("call_starttime"), row.get("call_endtime"), row.get("call_status") or "ANSWER", row.get("agent_callinfo") or "", row.get("customer_callinfo") or "", row.get("direction") or "inbound", None, None, None, row.get("duration_seconds"), ), ) def queue_for_transcription( cursor, bid: str, call_id: str, recording_url: str, *, db_handler=None, min_duration_s: int = 0, effective_at=None, duration_row: Optional[Dict[str, Any]] = None, log_prefix: str = "STT queue skipped", ) -> bool: """Queue one call for STT. Shared by webhook, orchestrator, and sync.""" call_id = str(call_id).strip() usage = evaluate_usage_allocation(cursor, bid) if usage["blocked"]: if db_handler is not None: pause_pipeline_if_usage_exhausted(db_handler, bid) logger.warning( "[%s] %s: allocation exhausted (%s/%s min)", call_id, log_prefix, usage["used_minutes"], usage["monthly_minute_limit"], ) return False if min_duration_s > 0 and duration_row is not None: skip_min, min_reason, probed_audio = evaluate_min_duration_for_ingest( duration_row, min_duration_s, effective_at, recording_url, probe_audio=True, ) if skip_min: purge_unprocessed_if_below_min( cursor, bid, call_id, duration_row, min_duration_s, effective_at, audio_duration_s=probed_audio, ) logger.info("[%s] %s (%s)", call_id, log_prefix, min_reason) return False if probed_audio is not None: cursor.execute( f""" UPDATE `{bid}_raw_calls` SET duration_seconds = %s WHERE callid = %s """, (max(0, int(round(probed_audio))), call_id), ) resp_table = f"`{bid}_sarvamresponse`" cursor.execute( f""" SELECT id FROM {resp_table} WHERE callid = %s AND transcript IS NOT NULL AND TRIM(transcript) != '' """, (call_id,), ) if cursor.fetchone(): cursor.execute( f""" UPDATE `{bid}_raw_calls` SET status = 2, selected_for_processing = 0 WHERE callid = %s """, (call_id,), ) return True cursor.execute( f"SELECT status, transcription_status FROM `{bid}_raw_calls` WHERE callid = %s LIMIT 1", (call_id,), ) current = cursor.fetchone() or {} current_status = int(current.get("status") if current.get("status") is not None else -99) if current_status in (1, 2, 3): logger.info( "[%s] Skip RabbitMQ publish (raw_calls status=%s already queued or done).", call_id, current_status, ) return True if str(current.get("transcription_status") or "") == "backlog_cleared": logger.info("[%s] Skip RabbitMQ publish (backlog_cleared terminal failure).", call_id) return False cursor.execute( f""" UPDATE `{bid}_raw_calls` SET status = 1, selected_for_processing = 0 WHERE callid = %s AND status IN (0, -2) AND COALESCE(transcription_status, '') != 'backlog_cleared' """, (call_id,), ) if cursor.rowcount < 1: logger.info("[%s] Skip RabbitMQ publish (status changed concurrently).", call_id) return False rabbitmq_host = os.getenv("RABBITMQ_HOST", "localhost") rabbitmq_queue = os.getenv("RABBITMQ_QUEUE", "stt_jobs") rmq_conn = pika.BlockingConnection(pika.ConnectionParameters(host=rabbitmq_host)) try: channel = rmq_conn.channel() channel.queue_declare(queue=rabbitmq_queue, durable=True) channel.basic_publish( exchange="", routing_key=rabbitmq_queue, body=json.dumps({"bid": bid, "call_id": call_id, "recording_url": recording_url}), properties=pika.BasicProperties(delivery_mode=2), ) finally: rmq_conn.close() logger.info("[%s] Successfully queued for transcription.", call_id) return True def maybe_queue_after_ingest( cursor, bid: str, row: Dict[str, Any], *, db_handler=None, min_duration_s: int = 0, effective_at=None, ) -> bool: """Queue STT when recording URL is ready; otherwise leave at status=0 for orchestrator retry.""" call_id = str(row["callid"]) fileurl = str(row.get("fileurl") or "") if not recording_url_ready(fileurl, call_id, row=row): return False duration_row = { "callid": call_id, "call_starttime": row.get("call_starttime"), "call_endtime": row.get("call_endtime"), "duration_seconds": row.get("duration_seconds"), } return queue_for_transcription( cursor, bid, call_id, fileurl, db_handler=db_handler, min_duration_s=min_duration_s, effective_at=effective_at, duration_row=duration_row, )