o
    hjU                     @  s  d Z ddlmZ ddlZddlZddlZddlZddlmZ ddlm	Z	m
Z
mZmZmZ ddlmZ ddlZddlmZ ddlmZmZ dd	lmZmZmZ dd
lmZ ddlmZ eeZ djddZ!dkddZ"dlddZ#dmddZ$dndd Z%dod"d#Z&dpd&d'Z'dqd(d)Z(drd,d-Z)drd.d/Z*dsd2d3Z+dd4dtd7d8Z,dud<d=Z-dvd?d@Z.dAdBdCdwdIdJZ/dxdLdMZ0dydOdPZ1dzdQdRZ2dSdTd{dWdXZ3d|dZd[Z4dSdTd}d]d^Z5d~d_d`Z6dddddadbddedfZ7ddddgddhdiZ8dS )uV   Shared call ingest: normalize, upsert, filters, STT queue — webhook, poll, and sync.    )annotationsN)datetime)AnyDictListOptionalTuple)urlparse)module_active)pick_mcube_customer_phonesanitize_agent_callinfo)call_duration_seconds evaluate_min_duration_for_ingestpurge_unprocessed_if_below_min)!pause_pipeline_if_usage_exhausted)evaluate_usage_allocationpayloadDict[str, Any]keysstrreturnr   c                 G  sB   |D ]}|| v r| | d urt | |  dkr| |   S qd S )N )r   strip)r   r   key r   call_ingest_core.py_first_value   s
   (r   Optional[datetime]c                 C  sv   | d u rd S t | tr| jr| jd dS | S zt| dd}t|}|jr.|jd dW S |W S  ty:   Y d S w )N)tzinfoZz+00:00)
isinstancer   r   replacer   fromisoformat	Exception)valuetextparsedr   r   r   _parse_datetime#   s   

r'   Optional[str]c                 C  s.   t | }|s| d urt|  S d S |dS )N%Y-%m-%d %H:%M:%S)r'   r   r   strftime)r$   r&   r   r   r   _format_db_datetime0   s   
r+   filename
source_bid	List[str]c              	   C  s   t | pd } | sg S | dr| gS z#t|}|s&t|t r&t|d}|s,| gW S |d}|d}W n tyC   | g Y S w t	j
| }t |pNd }d| d| d| d| d| d| d| d	| gS )
Nr   httpr)   z%Yz%mz4https://recordings.mcube.com/mcubefiles112/appmcube//z3https://recordings.mcube.com/mcubefiles112/classic/z	/inbound/)r   r   
startswithr'   r    r   strptimer*   r#   ospathbasename)r,   	starttimer-   dtyearmonthnamebidr   r   r   recording_url_candidates7   s*   


r<   c                 C  s$   t | ||}|r|d S t| pdS )Nr   r   )r<   r   )r,   r6   r-   
candidatesr   r   r   build_recording_urlO   s   r>    Tuple[Dict[str, Any], List[str]]c                C  sB  g }t | ddd}t | ddd}t | ddd	d
}t | dddd}t | ddddd}t | dddd}|s8|d |s?|d |sF|d |sM|d t|pQd }	t|}
|	rp|	dspt|	|
pe||}|rn|d n|	}	tt | dpwd   p~d }tt | d!d"pd}tt | d#d$d%pd}t | d&d't | d(d)t | d*d+t | d,||d-}t|pd t|pd  |
t||	|tt | d.d/pd|t	||t
|d0tt | d1d2pdd3}|d0urt| dkrztdtt||d< W n ttfy   Y nw |d d0u rt|d |d |d4|d< ||fS )5z?Map canonical or common alias fields to raw_calls column names.call_idcallidcallIdcall_status
dialstatus
callStatuscall_start_timecall_starttimer6   
call_startcall_end_timecall_endtimeendtimecall_endrecording_urlfileurlfileUrlfile_urlr,   duration_secondsdurationansweredtimetalktimezcall_id is requiredzcall_status is requiredzcall_start_time is requiredzrecording_url is requiredr   r/   r   	directioninbound
agent_name	agentnameagent_phoneagent_callinfo	emp_phonecustomer_phonecustomer_callinfocalltocall_tocallfrom	call_fromclicktocalldid)r]   r^   r`   rb   rU   rA   
group_name	groupnameNsourceprovider)rA   rC   rG   rJ   rN   rX   rd   rU   rZ   r]   rQ   re   )rG   rJ   rQ   )r   appendr   r   r+   r1   r<   lowerupperr   r   maxintfloat	TypeError
ValueErrorr   )r   r-   errorsr@   rC   rH   rL   	recordingrR   rN   	start_fmtr=   rU   rW   rY   	phone_row
normalizedr   r   r   normalize_webhook_payloadT   sr   







rt   callr;   c                C  s  t | d| d|}t| dpd}| d| d| d| d| d| d	| d| dd
}t| d ||t| dpFd| d| dt| dpVdt|t| dp`dt| t| dpld  psdti | |i | |dS )z4Map a Mcube source DB row to raw_calls column names.r,   r6   rX   r   rA   rK   rQ   rS   pulse)rA   r6   rK   rQ   rS   rv   rG   rJ   rd   rD   ANSWERr[   rU   rV   )rA   rN   rX   rd   rG   rJ   rC   rZ   r]   rU   rQ   _duration_probe_row)r>   getr   r   r   r   rh   r   )ru   r;   r-   rM   rW   duration_rowr   r   r   normalize_poll_call   s0   

r{   c                C  sd   |  d|  d|  d|  d|  d|  d|  d|  d|  d	|  d
d
}t|||dS )z>Map legacy sync_calls.py source row to raw_calls column names.rA   r,   r6   rK   rX   rd   rD   r[   rb   rU   )
rA   r,   r6   rK   rX   rd   rD   r[   rb   rU   )r;   r-   )ry   r{   )ru   r;   	poll_liker   r   r   normalize_sync_call   s   r}   urlboolc                 C  sF   t t| pd}|jpd dkrdS |jpd }d|v o"|dS )Nr   zrecordings.mcube.comFz/mcubefiles)z.wavz.mp3z.oggz.mpegz.m4az.webmz.flac)r	   r   hostnamerh   r4   endswith)r~   r&   r4   r   r   r   _is_trusted_mcube_recording_url   s   r   c                 C  s   t | pd } | dsdS z'tjdddddd	| gd
d
d}|jdkr5tdd |jp,d D r5W d
S W n	 t	y?   Y nw z+tjddddd	dddddd| gd
d
d}|jdkrf|jp^d dv riW d
S W dS W dS  t	yu   Y dS w )Nr   )zhttp://zhttps://Fcurlz-Iz-Lz
--max-time20z-sT)capture_outputr%   r   c                 3  s0    | ]   d ot fdddD V  qdS )zHTTP/c                 3  s    | ]}| v V  qd S )Nr   ).0codeliner   r   	<genexpr>   s    z6_http_recording_reachable.<locals>.<genexpr>.<genexpr>)z 200z 206N)r1   any)r   r   r   r   r      s
    
z,_http_recording_reachable.<locals>.<genexpr>z-oz	/dev/nullz-wz%{http_code}z-rz0-1023)200206)
r   r   r1   
subprocessrun
returncoder   stdout
splitlinesr#   )r~   headrangedr   r   r   _http_recording_reachable   sZ   
r   rowOptional[Dict[str, Any]]c                 C  sX   | sdS |  d}t|tr zt|}W n
 ty   Y dS w t|to+t| dS )NFextra_fieldsbulk_csv)ry   r    r   jsonloadsr#   dictr   )r   extrar   r   r   _is_bulk_csv_upload   s   

r   r   rN   rA   c                C  sf   t | rdS |rt|rt| rtd||  dS tdd dv }|r1t| r1td| dS dS )NTz.[%s] Trusting bulk CSV Mcube recording URL: %sORCHESTRATOR_TRUST_MCUBE_URLSr   )1trueyeszA[%s] Trusting Mcube recording URL (ORCHESTRATOR_TRUST_MCUBE_URLS)F)r   r   r   loggerinfor3   getenvrh   )rN   rA   r   trust_mcuber   r   r   recording_url_ready  s   r   rd   cfgTuple[bool, Optional[str]]c                 C  sj   t t|dpdsdS dd |dpg D }|sdS t| p!d  }||vr3d	d
|  dfS dS )Ngroup_filter_enabledr   )TNc                 S  s(   g | ]}t | rt |  qS r   )r   r   rh   )r   gr   r   r   
<listcomp>!  s    
z!group_allowed.<locals>.<listcomp>_allowed_groupnames)Fz5group_filter_enabled but no allowed groups configuredr   Fzgroup 'z' not in allowed list)r   rk   ry   r   r   rh   )rd   r   allowedr:   r   r   r   group_allowed  s   r   rk   c                 C  s*   t dt| dp	d}t| dsdS |S )Nr   min_call_duration_smin_duration_filter_enabled)rj   rk   ry   r
   )r   min_duration_sr   r   r   effective_min_duration_s.  s   
r   TzNot ingested)probe_audio
log_prefixr   rM   r   r   +Tuple[bool, Optional[str], Optional[float]]c          	      C  sR   |dkrdS t | ||||d\}}}|r$td| d|| d||fS dd|fS )	z3Return (should_skip, reason, probed_audio_seconds).r   )FNNr   [%s] %s (%s)rA   TFN)r   r   r   ry   )	r   r   effective_atrM   r   r   skipreasonprobedr   r   r   check_min_duration_for_ingest5  s   


r   Nonec                 C  s   |  d d S )Na  
        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
        )execute)cursorr   r   r   #ensure_call_ingest_exclusions_tableN  s   r   setc                 C  s0   t |  | dt|f dd |  pg D S )zJCall IDs blocked from Mcube/orchestrator re-ingest after permanent delete.z8SELECT callid FROM call_ingest_exclusions WHERE bid = %sc                 S  s"   h | ]}| d rt|d  qS )rA   )ry   r   )r   r   r   r   r   	<setcomp>e  s    
z(load_excluded_callids.<locals>.<setcomp>)r   r   r   fetchall)r   r;   r   r   r   load_excluded_callids^  s   
r   c                 C  s,   t |  | dt|t|f |  d uS )NzKSELECT 1 FROM call_ingest_exclusions WHERE bid = %s AND callid = %s LIMIT 1)r   r   r   fetchone)r   r;   rA   r   r   r   is_call_ingest_excludedl  s   r   deletedr   callidsr   c             	   C  sh   t |  d}|D ])}t|pd }|sq| dt||t|p dd d f | jdkr1|d7 }q|S )Nr   r   z
            INSERT INTO call_ingest_exclusions (bid, callid, reason)
            VALUES (%s, %s, %s)
            ON DUPLICATE KEY UPDATE reason = VALUES(reason)
            r         )r   r   r   r   rowcount)r   r;   r   r   addedrA   cidr   r   r   exclude_calls_from_ingestu  s   
r   
table_namec                 C  s   |  d|f |  d uS )NzSHOW TABLES LIKE %s)r   r   )r   r   r   r   r   _table_exists_on_cursor  s   r   Dict[str, int]c                C  s   g }t  }|D ]}t|pd }|r||v rq|| || q|s*dddS ddgt| }dD ]}	| d|	 }
t| |
rQ| d|
 d	| d
| q6d}| d}t| |rr| d| d	| d
| t	| j
ppd}t| |||d}||dS )z9Remove call data and block future sync from Mcube source.r   r   )deleted_rawexcludedz, z%s)sarvamresponsecallanalytics_zDELETE FROM `z` WHERE callid IN ()
_raw_callsr   )r   r   r   addrg   joinlenr   r   rk   r   r   )r   r;   r   r   
unique_idsseenrA   r   placeholderssuffixtabler   	raw_tabler   r   r   r   permanently_delete_calls  s:   





r   c                 C  s   |  d| d||d |d d|dpd|dpd|d	|d
|dp)d|dp/d|dp5d|dp;dddd|df dS )z>Insert or update raw_calls without resetting in-flight status.z
        INSERT INTO `aL  _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)
        rA   rN   r   rX   r   rd   rG   rJ   rC   rw   rZ   r]   rU   rV   NrQ   )r   ry   )r   r;   r   r   r   r   upsert_raw_call  s,   r   zSTT queue skipped)
db_handlerr   r   rz   r   r@   rz   c             	   C  sX  t | }t| |}	|	d r'|durt|| td|||	d |	d  dS |dkri|durit||||dd	\}
}}|
rRt| ||||||d
 td||| dS |duri| 	d| dt
dtt||f d| d}| 	d| d|f |  r| 	d| d|f dS | 	d| d|f |  pi }t|ddur|dnd}|dv rtd|| dS t |dpddkrtd| dS | 	d| d|f | jd k rtd!| dS td"d#}td$d%}ttj|d&}z%| }|j|dd' |jd|t|||d(tjd)d*d+ W |  n|  w td,| dS )-zBQueue one call for STT. Shared by webhook, orchestrator, and sync.blockedNz)[%s] %s: allocation exhausted (%s/%s min)used_minutesmonthly_minute_limitFr   Tr   )audio_duration_sr   z
                UPDATE `zh_raw_calls`
                SET duration_seconds = %s
                WHERE callid = %s
                `z_sarvamresponse`z
        SELECT id FROM zm
        WHERE callid = %s
          AND transcript IS NOT NULL
          AND TRIM(transcript) != ''
        z
            UPDATE `zn_raw_calls`
            SET status = 2, selected_for_processing = 0
            WHERE callid = %s
            z*SELECT status, transcription_status FROM `z%_raw_calls` WHERE callid = %s LIMIT 1statusi)r         zH[%s] Skip RabbitMQ publish (raw_calls status=%s already queued or done).transcription_statusr   backlog_clearedz>[%s] Skip RabbitMQ publish (backlog_cleared terminal failure).z
        UPDATE `z_raw_calls`
        SET status = 1, selected_for_processing = 0
        WHERE callid = %s
          AND status IN (0, -2)
          AND COALESCE(transcription_status, '') != 'backlog_cleared'
        r   z9[%s] Skip RabbitMQ publish (status changed concurrently).RABBITMQ_HOST	localhostRABBITMQ_QUEUEstt_jobs)host)queuedurable)r;   r@   rM   r   )delivery_mode)exchangerouting_keybody
propertiesz+[%s] Successfully queued for transcription.)r   r   r   r   r   warningr   r   r   r   rj   rk   roundr   ry   r   r3   r   pikaBlockingConnectionConnectionParameterschannelqueue_declarebasic_publishr   dumpsBasicPropertiesclose)r   r;   r@   rM   r   r   r   rz   r   usageskip_min
min_reasonprobed_audio
resp_tablecurrentcurrent_statusrabbitmq_hostrabbitmq_queuermq_connr  r   r   r   queue_for_transcription  s   

			
 


r  )r   r   r   c          	   
   C  sh   t |d }t |dpd}t|||dsdS ||d|d|dd	}t| |||||||d
S )zZQueue STT when recording URL is ready; otherwise leave at status=0 for orchestrator retry.rA   rN   r   r   FrG   rJ   rQ   )rA   rG   rJ   rQ   )r   r   r   rz   )r   ry   r   r  )	r   r;   r   r   r   r   r@   rN   rz   r   r   r   maybe_queue_after_ingesth  s&   
r  )r   r   r   r   r   r   )r   r   )r   r(   )r,   r   r-   r   r   r.   )r,   r   r-   r   r   r   )r   r   r-   r   r   r?   )ru   r   r;   r   r-   r   r   r   )ru   r   r;   r   r   r   )r~   r   r   r   )r   r   r   r   )rN   r   rA   r   r   r   r   r   )rd   r   r   r   r   r   )r   r   r   rk   )r   r   r   rk   rM   r   r   r   r   r   r   r   )r   r   )r;   r   r   r   )r;   r   rA   r   r   r   )r;   r   r   r.   r   r   r   rk   )r   r   r   r   )r;   r   r   r.   r   r   r   r   )r;   r   r   r   r   r   )r;   r   r@   r   rM   r   r   rk   rz   r   r   r   r   r   )r;   r   r   r   r   rk   r   r   )9__doc__
__future__r   r   loggingr3   r   r   typingr   r   r   r   r   urllib.parser	   r   feature_entitlementsr
   mcube_phone_utilr   r   min_duration_utilr   r   r   pipeline_usage_controlr   usage_allocation_utilr   	getLogger__name__r   r   r'   r+   r<   r>   rt   r{   r}   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r  r  r   r   r   r   <module>   sl    







H




+







)1 