
    exi&L                        d dl mZ d dlZd dlZd dlmZ d dlmZ d dlm	Z	 d dl
mZ d dlmc mc mZ d dlmZ d dlmZ d d	lmZ d d
lmZmZmZ  edej4                  ej6                        Zej:                  j=                  ddgddddddddddddddddd       ej?                  dd      de fd       Z!ej?                  d      d        Z"ej?                  d de#fdd!dd"#      de d$e$fd%       Z%ej?                  d&de#fdd!dd"#      de d'e&d(e'fd)       Z(ej?                  d*de#fdd!dd"#      d1de d'e&d+e)fd,       Z*ej?                  d-      d.e&d/e'fd0       Z+y)2    )CeleryN)settings)logger)SessionLocal)BnkAIReqRes) MultiDocumentVerificationRequest)datetime)pusher_client)task_extract_and_chunktask_process_chunks_llmtask_consolidate_outputsbankruptcy_tasks)brokerbackendjsonUTCTi  i     (   i i  2   z100/m)task_serializeraccept_contentresult_serializertimezone
enable_utctask_track_startedtask_time_limittask_soft_time_limitworker_prefetch_multiplierworker_max_tasks_per_childworker_max_memory_per_childtask_acks_latetask_reject_on_worker_lostresult_expiresbroker_pool_limit"broker_connection_retry_on_startupworker_send_task_eventstask_send_sent_eventtask_default_rate_limitprocess_verification_task)bindnamejob_idc                    t               }d}	 |j                  t              j                  t        j                  |k(        j                         }|s+dd| i|j                          t        j                          S |j                  }t        j                  |      }|j                  r#|j                  D cg c]  }t        |       c}ng }|st        d      t        j                   d|        t"        j%                  ||g       d|dd	|j                          t        j                          S c c}w # t        $ rz}|r<t        |      d
d|_        t)        j*                         |_        |j/                          t        |      |dcY d}~|j                          t        j                          S d}~ww xY w# |j                          t        j                          w xY w)aX  
    Orchestrator task that dispatches the chunking pipeline:
    - Dispatches Task 1 (extract_and_chunk_task)
    - Task 1 auto-triggers Task 2 (process_chunks_llm_task)
    - Task 2 auto-triggers Task 3 (consolidate_outputs_task)
    
    Args:
        self: Task instance (bound)
        job_id: Primary key ID from bnk_ai_req_res table
    NerrorJob not found: z No file URLs provided in requestz=[Orchestrator] Dispatching extract_and_chunk_task for job_id=argsTpipeline_dispatched)successr,   statusF)r.   r3   r.   r,   )r   queryr   filteridfirstclosegccollectrequestr   from_raw	file_urlsstr	Exceptionr   infoextract_and_chunk_taskapply_asyncresponser	   utcnow
updated_atcommit)	selfr,   dbjobrequest_dataparsed_requesturlr?   es	            F/var/www/html/BankruptcyAI-uat/bankruptcy-ai/src/utils/celery_tasks.pyr)   r)   6   s    
B
C#hh{#**;>>V+CDJJLvh788 	



7 {{9BB<P GUF^F^)A)AB#SXBdf	>?? 	STZS[\]**0C*D6=RS 	



/ C  3%(V>CL%__.CNIIKQ622 	



3 	



sJ   AD< >;D< 9D7AD< 7D< <	F?AF:F?G :F??G &G(health_check)r+   c                  L    dt        j                         j                         dS )z#Health check task for Celery workerhealthy)r4   	timestamp)r	   rF   	isoformat     rP   rQ   rQ   k   s      hoo.?.I.I.KLLrW   rC   iX     )r+   r*   autoretry_forretry_backoffretry_backoff_maxretry_jittermax_retriesr?   c           
      X   t               }d}	 |j                  t              j                  t        j                  |k(        j                         }|s+dd| i|j                          t        j                          S t        r%t        j                  dddd|j                  d       d	dl}|j                  t        |||
            }|j                  d      st!        |j                  dd            t        r9t        j                  dddd|j                  dd	       d|j                  d       |j"                  }|j                  d      }|j                  di       }	|st!        d      t%        j&                  d|        t(        j+                  |||	g       d||j                  d      |j                  d      dd|j                          t        j                          S # t         $ r}
|j-                          |r=t/        |
      ddd|_        t3        j4                         |_        |j9                          t/        |
      |dcY d}
~
|j                          t        j                          S d}
~
ww xY w# |j                          t        j                          w xY w)a#  
    Task 1: Extract text from documents using LlamaParse and store chunks with overlap
    Auto-triggers Task 2 (process_chunks_llm_task) upon completion
    
    Args:
        job_id: Primary key ID from bnk_ai_req_res table
        file_urls: List of document URLs (can be 1 or many)
    Nr.   r/   notification-channelnotification-eventFzExtracting text from documentsr4   message
request_idr   )rJ   ai_req_res_idr?   r3   Unknown errorzExtraction complete: total_chunksz chunks created	file_typedataz file_type is required in requestz@[Task 1] Triggering Task 2 (process_chunks_llm_task) for job_id=r0   Tdocuments_processedprocess_chunks_llm_task)r3   r,   ri   rf   	next_taskextract_and_chunkr.   r3   stager5   )r   r6   r   r7   r8   r9   r:   r;   r<   r
   triggerrc   asynciorunr   getrA   r=   r   rB   rj   rD   rollbackr@   rE   r	   rF   rG   rH   )rI   r,   r?   rJ   rK   rp   resultrL   rg   rh   rO   s              rP   rC   rC   t   sJ   $ 
B
CAhh{#**;>>V+CDJJLvh78x 	



w !!"8:N;!nnQ  	3 
  zz)$FJJw@AA !!"8:N26::na3P2QQ`a!nnQ  {{ $$[1	+>?? 	VW]V^_`++&)T1J+K #)::.C#D"JJ~62
( 	



  	3
 %(VH[\CL%__.CNIIKQ622 	



	3 	



s8   AG, >E	G, ,	J 5AI;J J ;J  J &J)rj   rg   rh   c           
      J   t               }d}	 |j                  t              j                  t        j                  |k(        j                         }|s+dd| i|j                          t        j                          S t        r%t        j                  dddd|j                  d       t        j                  d	|        d
dl}|j                  t!        |||||j                              }|j#                  d      st%        |j#                  dd            t        rJ|j#                  d      s9t        j                  dddd|j#                  dd
       d|j                  d       t        j                  d|        t&        j)                  |||j#                  dd      g       d||j#                  d      dd|j                          t        j                          S # t$        $ r}|j+                          |r=t-        |      ddd|_        t1        j2                         |_        |j7                          t-        |      |dcY d}~|j                          t        j                          S d}~ww xY w# |j                          t        j                          w xY w)a6  
    Task 2: Loop over chunks, send each to OpenAI, store responses
    Auto-triggers Task 3 (consolidate_outputs_task) upon completion
    
    Args:
        job_id: Primary key ID from bnk_ai_req_res table
        file_type: Document type for prompt selection
        data: Data payload for verification
    Nr.   r/   r_   r`   FzProcessing documents with AIra   z[Task 2] Using file_type=r   )rJ   rd   rg   rh   rc   r3   re   jailbreak_detectedzAI analysis complete: chunks_processedz chunks processedzA[Task 2] Triggering Task 3 (consolidate_outputs_task) for job_id=r0   Tconsolidate_outputs_task)r3   r,   rw   rk   process_chunks_llmrm   r5   )r   r6   r   r7   r8   r9   r:   r;   r<   r
   ro   rc   r   rB   rp   rq   r   rr   rA   rx   rD   rs   r@   rE   r	   rF   rG   rH   )	rI   r,   rg   rh   rJ   rK   rp   rt   rO   s	            rP   rj   rj      s:   & 
B
C<hh{#**;>>V+CDJJLvh78n 	



m !!"8:N9!nnQ  	/	{;<4 ~~
  zz)$FJJw@AA ,@!A!!"8:N3FJJ?QST4U3VVgh!nnQ  	WX^W_`a ,,69fjjQeglFm2n,o  &

+= >3	
& 	



  	3
 %(VH\]CL%__.CNIIKQ622 	



	3 	



s8   AG% >EG% %	I9.AI4
I9I< 4I99I< <&J"rx   rv   c           	      	   t               }d}	 |j                  t              j                  t        j                  |k(        j                         }|s+dd| i|j                          t        j                          S t        r'|s%t        j                  dddd|j                  d       d	dl}|j                  t        |||
            }|j                  d      st!        |j                  dd            |j                  d      }||_        t%        j&                         |_        |j+                          |j,                  }	|	r|	j                  d      nd}
t/        j0                  d|        t/        j0                  d|
        |
r|
|_        |j+                          	 d	dl}|}t7        dd        t7        d|
        t7        d d       |j9                  |
|d      }t7        d|j:                          t7        dt=        |j>                                |j:                  dk(  r	 |jA                         |_!        |j:                  tE        |jF                        d|_$        |j+                          t        rH|sFt        j                  dddd|j                  t%        j&                         jK                         d       t/        j0                  d|        nVd|j:                   |j>                  dd  d!|_!        |j+                          t/        jL                  d"|j:                          nLt        rFt        j                  dddd|j                  t%        j&                         jK                         d       d|d|
dud$|j                          t        j                          S #  d|j>                  i|_!        Y xY w# t         $ rD}t/        jL                  d#|        dtO        |      i|_!        |j+                          Y d}~d}~ww xY w# t         $ r}|jQ                          t/        jL                  d%|        |r=tO        |      dd&d'|_        t%        j&                         |_        |j+                          tO        |      |d(cY d}~|j                          t        j                          S d}~ww xY w# |j                          t        j                          w xY w))aE  
    Task 3: Consolidate all chunk LLM outputs into final response
    Sends callback to frontend when complete
    
    Args:
        job_id: Primary key ID from bnk_ai_req_res table
        file_type: Document type for consolidation logic
        jailbreak_detected: If True, skip notifications (only do consolidation)
    Nr.   r/   r_   r`   FzConsolidating resultsra   r   )rJ   rd   rg   r3   re   consolidated_responsecallback_urlz+[Task 3] Consolidation complete for job_id=z[Task 3] callback_url: 
<============================================================z[Task 3 Callback] Sending to:    r   timeoutz#[Task 3 Callback] Response status: z([Task 3 Callback] Response body length:    raw_text)status_codeheadersTz#Verification completed successfully)r4   rb   rc   rT   z([Task 3] Callback successful for job_id=zCallback returned   )r.   rE   z%[Task 3] Callback failed with status z[Task 3] Callback exception: )r3   r,   consolidatedcallback_sentz[Task 3] Error: consolidate_outputsrm   r5   ))r   r6   r   r7   r8   r9   r:   r;   r<   r
   ro   rc   rp   rq   r   rr   rA   rE   r	   rF   rG   rH   r=   r   rB   r|   requestsprintpostr   lentextr   callback_resdictr   callback_res_headerrU   r.   r@   rs   )rI   r,   rg   rv   rJ   rK   rp   rt   r{   rL   r|   r   payloadrE   callback_errorrO   s                   rP   rx   rx   (  s<   & 
B
Cqhh{#**;>>V+CDJJLvh78X 	



W !3!!"8:N2!nnQ  	5 
  zz)$FJJw@AA !'

+B C,!*
		 {{;G|''7TA&JK-l^<=+CIIK(/6(m$6|nEFm$#==GR=P;H<P<P;QRS@X]]AS@TUV''3.G+3==?( ?G>R>R_cdldtdt_u.vC+IIK %-?%--.DFZ&*'L*-..)1):)D)D)F	]  KK"J6( ST3EhFZFZE[1\jrjwjwx|y|j}'~C$IIKLL#HI]I]H^!_` %%&<>R"D"%..!)!2!<!<!>	U   )5	
( 	



sG,6+F((  <^<LMN$+S-@#A 		*  
3
's+, %(VH]^CL%__.CNIIKQ622 	




3 	



s}   AP  >D&P  %BO 2N5 C4O ;AP  5O
O 	P:PP  PP   	S)A4SSS SS &S5#process_verification_task_withoutdbrc   rL   c                    	 t        dd        t        d|         t        dt        |              t        dt        |t              r|j	                         nd        t        dt        |t              rd|v nd	        t        d
t        |t              r|j                  d      nd	        t        d d       t        j                  |      }ddl}|j                  t        j                  d|d            }|j                  d      }t        d|        |r	 ddl}ddl}|j                         }t        dd        t        d|        t        d       t        |j                  |dt                     t        d d       |j!                  ||d      }	t        d|	j"                          t        d|	j$                  dd         |	j"                  dk(  r	 |	j                         }
t        d|
        d| dS #  Y 	xY w# t&        $ r!}t        dt        |              Y d}~1d}~ww xY w# t&        $ r~}t        |t              r|j                  d      nd}|rA	 ddl}dt        |      t        |      d}|j!                  ||d       n# t&        $ r Y nw xY wt        |      | dcY d}~S d}~ww xY w)z
    Celery task to process document verification WITHOUT database
    
    Args:
        request_id: Unique request ID (correlation ID)
        request_data: Full request payload dict (includes callback_url)
    r}   r~   z"[NO-DB TASK] Received request_id: z [NO-DB TASK] request_data type: z [NO-DB TASK] request_data keys: z
Not a dictz+[NO-DB TASK] callback_url in request_data: r|   zN/Az![NO-DB TASK] callback_url value: r   Nz.[NO-DB TASK] After processing - callback_url: z[Callback NO-DB] Sending to: z[Callback NO-DB] Payload:   )indentdefaultr   r   z"[Callback NO-DB] Response status: z [Callback NO-DB] Response body: r   r   z)[Callback NO-DB] Webhook responded with: z[Callback NO-DB] Failed: T)r3   rc   F)r3   rb   r.   )r.   rc   )r   type
isinstancer   keysrr   r   r>   rp   rq   prompt_controller'verify_multi_documents_celery_withoutdbr   r   dumpsr@   r   r   r   rA   )rc   rL   rM   rp   rt   r|   r   r   r   rE   webhook_responser   rO   error_payloads                 rP   r   r     s   B;6(m2:,?@0l1C0DEF0
S_aeHf1B1B1Dlx0yz{;^hiuw{^|Nl<Z  CH  <I  J  	K1V`amosVt,2B2B>2Rz  2A  B  	Cm :BB<P 	.VVW[]kmqrs $''7>|nMNI !++- 6(m$5l^DE13djjCj@Am$#==GR=P:8;O;O:PQR8t9L8MNO ''3.+3==?( IJZI[\]  z:: I1#n2E1FGHHI
  ;;ElTX;Y|''7_c	$"1v V!
 lK  Qz::;s   D H? #CH (H H? HH 	H<H72H? 7H<<H? ?	K%K.0JK	J+(K*J++K;KK)F),celeryr   osr;   src.core.configr   src.core.loggingr   src.database.sessionr   src.modules.modelsr   src.modules.prompt.controllermodulesprompt
controllerr   #src.modules.prompt.schemas.requestsr   r	   src.utils.pusher_configr
   src.utils.file_chunkr   r   r   CELERY_BROKER_URLCELERY_RESULT_BACKEND
celery_appconfupdatetaskintr)   rQ   rA   listrC   r@   r   rj   boolrx   r   rV   rW   rP   <module>r      s    	 	 $ # - * 9 9 P  1  %%**
   8  ! &#'+ #)  8 d!<=1C 1 >1h n%M &M 	!	,  M M MMf 	"	,  I# I# IT II^ 	#	,  ~3 ~3 ~TX ~~F ;<J;C J;t J; =J;rW   