o i= @sdZddlZddlZddlZddlZddlZddlZddlZddlmZm Z ddl m Z m Z m Z mZmZddlZddlZddlmZddlmZejejeZejdeeeeddlmZddlm Z dd l!m"Z"dd l#m$Z$ej%ej&d e'ej(ed e)gd e*dZ+da,ddZ-eej.e-eej/e-GdddZ0Gddde0Z1Gddde0Z2ddZ3e4dkre3dSdS)a Unified Call Processing Pipeline. This module combines the polling-based sync from call_processor.py and the orchestration logic from orchestrate_pipeline.py into a single, RabbitMQ-powered system. Modes: --mode orchestrator : Scans for new calls, syncs them, and queues tasks in RabbitMQ. --mode worker : Consumes tasks from RabbitMQ and performs Transcription/Analysis. Usage: python3 unified_pipeline.py --mode orchestrator --bid 1713 python3 unified_pipeline.py --mode worker N)datetime timedelta)AnyDictListOptionalSet) DictCursor) load_dotenv)Config)DatabaseHandler)get_stt_provider) AgentRunneru4%(asctime)s [%(levelname)s] %(name)s — %(message)szunified_pipeline.log)levelformathandlersunified_pipelineFcCstddadS)Nz4Shutdown signal received - finishing current task...T)loggerinfo _shutdown)sigframer;/home/aiteam/pcaa-dev/dashboard-backend/unified_pipeline.py_handle_signal;s rc@s(eZdZdZddZddZddZdS) SharedPipelineBasez0Base class for shared DB and RabbitMQ resources.cCs6t|_Gddd}t||j|_d|_d|_dS)Nc@s&eZdZddZdddZddZdS) z2SharedPipelineBase.__init__..ConfigWrappercSs ||_dSN)_cfg)selfcfgrrr__init__Js z;SharedPipelineBase.__init__..ConfigWrapper.__init__NcSst|j||Srgetattrr)rkeydefaultrrrgetLsz6SharedPipelineBase.__init__..ConfigWrapper.getcSs t|j|Srr!)rr#rrr __getattr__Ns z>SharedPipelineBase.__init__..ConfigWrapper.__getattr__r)__name__ __module__ __qualname__r r%r&rrrr ConfigWrapperIs  r*)r configr db rabbitmq_connrabbitmq_channel)rr*rrrr Es  zSharedPipelineBase.__init__c CszGt|jj|jj}tj|jj||jj|jjd}t ||_ |j |_ dD]}|jj |}|j j|ddq(td|jjWdSty\}z td|d}~ww)z!Establish connection to RabbitMQ.)host credentials heartbeatblocked_connection_timeout transcriptionanalysisT)queuedurablezConnected to RabbitMQ at zRabbitMQ connection failed: N)pikaPlainCredentialsr+ RABBITMQ_USER RABBITMQ_PASSConnectionParameters RABBITMQ_HOSTRABBITMQ_HEARTBEATRABBITMQ_TIMEOUTBlockingConnectionr-channelr.RABBITMQ_QUEUE_PREFIX queue_declarerr Exceptionerror)rr0 parametersstage queue_nameerrrconnect_rabbitmqUs&  z#SharedPipelineBase.connect_rabbitmqcCs$|jr|jjs|jdSdSdSr)r- is_closedcloserrrrclose_rabbitmqlsz!SharedPipelineBase.close_rabbitmqN)r'r(r)__doc__r rJrNrrrrrCs  rcseZdZdZfddZddedefddZded ed ed efd d Z dede defddZ dededefddZ defddZ ZS) Orchestratorz0Responsible for syncing calls and queuing tasks.cst|dSr)superr rJrM __class__rrr ss  zOrchestrator.__init__2bidlimitcCs~td|d|j|}|r|dds!td|dn|||||j|d|d|j|d|d||d S) z,Run the orchestrator steps for a single bid.[z] Starting orchestrator runpipeline_enabledTz5] Pipeline disabled or missing config - skipping syncr4rVr5N) rrr,get_pipeline_configr%warning sync_calls queue_tasksrecover_stuck_calls)rrUrVrrrrrun_bidws zOrchestrator.run_bidfilename starttimereturnc Cs|sdS|dr |Sd|vr|d}d|Sz(t|tr&tt|n|}|d|d}}d|d|d|d|WStyc}zt d |d |d ||WYd }~Sd }~ww) z`Reconstruct the full recording URL if it's just a filename (Logic from orchestrate_pipeline.py).http/zhttps://recordings.mcube.com/z%Yz%mz3https://recordings.mcube.com/mcubefiles112/classic/z /inbound/rWz] Error reconstructing URL for : N) startswithlstrip isinstancestrr fromisoformatstrftimerDrrE) rrUr`rapathdtyearmonthrIrrr_get_reconstructed_urls  z#Orchestrator._get_reconstructed_url pipeline_cfgc Csddlm}|||j|}|dkrr|jD}|}|d}|d|d|p.g} | D]!} ||| d| d} | | dkrR|d|d | | d fq1Wd n1s]wYt d |d t | dd Sd S)z6Pull new calls from source DB into {bid}_call_records.r) stage_sync _call_recordsz*SELECT callid, file_url, call_start FROM `z8` WHERE status = 'pending' AND file_url NOT LIKE 'http%'file_url call_startzUPDATE `z%` SET file_url = %s WHERE callid = %scallidNrWz"] URL reconstruction complete for z records) call_processorrsr,get_connectioncursorexecutefetchallrqrrlen) rrUrrrVrsinsertedconnrztable to_repairrowrepairedrrrr\s"      zOrchestrator.sync_callsrGc Cs|jj|}|dkr|jj||d}n|jj||d}|s!dStd|dt|d||D]2}||d|t d}|j j d |t |tjd d d |dkrYd nd}|j||d|q3dS)z/Find calls in the DB and push them to RabbitMQ.r4)batchNrWz ] Queuing z calls for rw)rUrwrG timestamprc) delivery_mode)exchange routing_keybody properties transcribing analyzing)r+rBr,get_calls_to_transcribeget_calls_to_analyzerrr}rnow isoformatr. basic_publishjsondumpsr8BasicPropertiesset_call_status) rrUrGrVrHcallscallpayload new_statusrrrr]s,    zOrchestrator.queue_tasksc Cs|d}ttdd}|jD}|}|d|d|f|j}|d|d|f|j}|s8|rOt d|d|d |d Wd d SWd d S1sZwYd S) zAResets calls stuck in 'transcribing' or 'analyzing' for too long.rt)hoursz UPDATE `z` SET status = 'pending' WHERE status = 'transcribing' AND (transcript IS NULL OR transcript = '') AND updated_at < %s z` SET status = 'transcribed' WHERE status = 'analyzing' AND updated_at < %s rWz ] Recovered z transcription and z analysis tasksN) rrrr,ryrzr{rowcountrr)rrUr stuck_timerrz t_recovered a_recoveredrrrr^s*   "z Orchestrator.recover_stuck_calls)rT)r'r(r)rOr rjintr_rrqrr\r]r^ __classcell__rrrRrrPps  rPcs\eZdZdZfddZddZddZded ed efd d Z ded ed efd dZ Z S)Workerz/Responsible for processing tasks from RabbitMQ.cs"tt|j|_|dSr)rQr rr, agent_runnerrJrMrRrrr s   zWorker.__init__cCsfdD]}|jj|}|jj||jdqtdz|jWdSty2|j YdSw)z Start consuming from all queues.r3)r6on_message_callbackz%Worker started - waiting for tasks...N) r+rBr. basic_consumeprocess_messagerrstart_consumingKeyboardInterruptstop_consuming)rrGrHrrrstarts  z Worker.startc Cs2zit|}|d}|d}|d}td|d|d|d} |dkr-|||} n |d kr7|||} | rQ|j|jd td|d |d |WdS|j|jdd t d|d|d |WdSt y} z#tj d| dd|jr|j|jdd WYd} ~ dSWYd} ~ dSd} ~ ww)z7Dispatches the message to the correct processing logic.rUrwrGrWz ] Processing z for call Fr4r5) delivery_tagz] Successfully processed z for )rrequeuez] Failed to process z#Critical error in worker callback: Texc_infoN) rloadsrrdo_transcription do_analysis basic_ackr basic_nackrErD) rchmethodrrtaskrUrwrGsuccessrIrrrrs.  ""zWorker.process_messagerUrwrbc Cs^|j|}|s dS|dpd}ddlm}||j|}zt||}WntyB}ztd|d|WYd}~dSd}~ww|j ||} | rQ| d s]|jj ||d d d dS|j ||d z| | d |} | r~| j r~|j||| WdStdty} z!td|d|d| |jj ||d t| d WYd} ~ dSd} ~ ww)z+Logic from call_processor.stage_transcribe.F stt_providersarvamr)_decrypt_stt_keyrWz] STT provider init failed: Nru transcribezNo audio URL or record missingrGreasonrTzEmpty transcript returnedz] Transcription failed for rf)r,rZr%rxrr rDrrEget_call_record_detail fail_callrr transcriptsave_call_transcription RuntimeErrorrj) rrUrwrstt_provider_namerapi_keysttrIrresultexcrrrr s:    zWorker.do_transcriptionc CsJ|j||}|std|d|ddS|dpd}|s.td|d|ddS|d p4g}|d p;d|d pAdt|d pHdt|d pPdd}|j||dz|jj |||||d}|ru|j |||WdSWdSt y}z!td|d|d||jj ||dt|dWYd}~dSd}~ww)z(Logic from call_processor.stage_analyze.rWz] Call record z not found for analysisFrrcz] Call z has no transcript for analysisspeaker_segments agent_namecustomer_phonervcall_duration_s)rrrvrr)rUrwrr call_metadataTz] Analysis failed for rfanalyzerN) r,rrrEr%r[rjrrrunsave_call_analysisrDr) rrUrwrrr call_metar5rrrrrBsB  zWorker.do_analysis) r'r(r)rOr rrrjboolrrrrrrRrrs "rc CsHtjdd}|jdddgddd|jd d d |jd td dd|jdtddd|jdddd|}|jdkrt} |jrF|jgn|j }|D]/}t rSn*z |j ||j dWqMt y|}ztjd|d|ddWYd}~qMd}~ww|jrt rnt|jq?|dS|jdkrt}||dSdS)Nz Unified Call Processing Pipeline) descriptionz--mode orchestratorworkerTz:Mode: orchestrator (queue tasks) or worker (process tasks))choicesrequiredhelpz--bidz Orchestrator mode: only this bid)rz--limitrTz Batch limit for sync and queuing)typer$rz --interval<zLoop interval for orchestratorz --continuous store_truezLoop orchestrator indefinitely)actionrrYzOrchestrator error for bid rfr)argparseArgumentParser add_argumentr parse_argsmoderPrUr,get_enabled_pipeline_bidsrr_rVrDrrE continuoustimesleepintervalrNrr)parserargsrbidsrUrIrrrrmainls<  &    r__main__)5rOrrloggingossignalsysrrrtypingrrrrrr8pymysqlpymysql.cursorsr dotenvr rmdirnameabspath__file__ BACKEND_DIRinsertchdirr+r db_handlerr rr rr basicConfigINFO FileHandlerjoin StreamHandler getLoggerrrrSIGTERMSIGINTrrPrrr'rrrrsR        -~~