di/UdZddlZddlZddlZddlmZmZddlmZmZm Z m Z m Z ddl m Z ddlmZej eZGddee Zej*gej,gej.ej*gej0ej*gej2ej,ej*giZeeeefed <ej*d ej,d ej.d ej0d ej2d iZeeefed <ej*gej*gej*gej*gej,gej,gej,gej,gej,gej,gej,gej,gd Zeeeefed<ej*ej.ej0ej2gej,ej2gej.gej0gej2giZeeeefed<ej*dej,dej.dej0dej2diZ eeefed<defdZ!dedeee fdede"fdZ#deede efdZ$de edeefd Z%d!eded"ed#ed$ede"f d%Z& d7d!eded"ed#ed$ed&e"de efd'Z' d8d!edeed"ed(eed$e edeee ff d)Z(d!eded"ed#ed$ed*ed+e ed,eede efd-Z)d!ed.ed"ed$ed,eede"f d/Z*d9d!ed"e ed$e edefd0Z+d8d!ed"e ede"fd1Z,d!ed"ed$efd2Z-d!ed$edeee ffd3Z. d8ded4ed5e efd6Z/y):a Task Orchestrator for Smart Inventory ===================================== This module provides utilities to automatically trigger Celery tasks based on affected database tables. It handles: - Task dependency ordering - HOLD status for tasks waiting on dependencies - Polling for HOLD tasks to start when dependencies complete - Tracking via CeleryTaskTracker This is a table-based orchestrator - it determines which tasks to run based on which database tables were modified. It can be used from: - CSV upload processing (data_import/controller.py) - Webhook event processing (inventory/webhookcontroller.py) - Any other data modification flow Usage: from src.smart_inventory.utils.task_orchestrator import trigger_tasks_for_affected_tables # After any data modification trigger_tasks_for_affected_tables( db=db, affected_tables=["sales_orders", "sales_order_lines", "inventory_movements"], company_id=1, unique_dates=["2026-01-01", "2026-01-02"] ) N)datetimedate)ListDictSetOptionalAny)Enum)Sessionc$eZdZdZdZdZdZdZdZy)TaskNamez*Celery task names for inventory processingcompute_daily_salescompute_inventory_snapshotcompute_service_level_dailycompute_slow_movers_90d#compute_inventory_planning_snapshotN) __name__ __module__ __qualname____doc__ DAILY_SALESINVENTORY_SNAPSHOT SERVICE_LEVEL SLOW_MOVERSINVENTORY_PLANNINGJ/var/www/html/hubwallet-dev/src/smart_inventory/utils/task_orchestrator.pyr r -s 4'K51M+K>rr TASK_DEPENDENCIES TASK_PRIORITY) sales_orderssales_order_linessales_return_orderssales_return_order_linesinventory_movementsinventory_movementinventory_batchesinventory_batchpurchase_order_receivepurchase_order_receive_linespurchase_order_returnpurchase_order_return_linesTABLE_TO_TASKSDOWNSTREAM_TASKSz>src.smart_inventory.tasks.daily_sales_task.compute_daily_saleszBsrc.smart_inventory.tasks.snapshot_task.compute_inventory_snapshotzHsrc.smart_inventory.tasks.service_level_task.compute_service_level_dailyzBsrc.smart_inventory.tasks.slow_movers_task.compute_slow_movers_90dzUsrc.smart_inventory.tasks.inventory_planning_task.compute_inventory_planning_snapshotTASK_NAME_TO_CELERY_NAME task_namecx |tjk(rddlm}|S|tjk(rddlm}|S|tjk(rddlm }|S|tjk(rddl m }|S|tjk(rddlm}|S y #t $r%}t"j%d|d|Yd }~y d }~wwxYw) z9Get the actual Celery task function by name (for Windows)r)r)r)r)r)rzCould not import task : N)r r*src.smart_inventory.tasks.daily_sales_taskrr1src.smart_inventory.tasks.inventory_snapshot_taskrr,src.smart_inventory.tasks.service_level_taskrr*src.smart_inventory.tasks.slow_movers_taskrr1src.smart_inventory.tasks.inventory_planning_taskr ImportErrorloggerwarning)r3rrrrres r_get_celery_taskr?s ,, , V& & (55 5 d- - (00 0 `. . (.. . Z* * (55 5 m6 66  / {"QC@As-B B B B .B B9B44B9kwargstask_idreturnc6 tjdk(r"t|}|r|j||yyddlm}t j|}|r|j|||yy#t$r%}tjd|d|Yd }~yd }~wwxYw) a Dispatch a Celery task with platform-specific handling. Windows (solo pool): Uses direct task import and apply_async Linux (prefork pool): Uses send_task with task name string Args: task_name: The TaskName enum kwargs: Task keyword arguments task_id: Task ID to use Returns: True if task was dispatched successfully win32)r@rATr celery_appFzFailed to dispatch task r5N) sysplatformr? apply_asyncsrc.utils.celery_workerrFr2get send_task Exceptionr<error)r3r@rA celery_taskrF celery_namer>s r_dispatch_celery_taskrQs <<7 "*95K''vw'G ;266yAK$$[$Q  / {"QC@As3A*1A** B3BBaffected_tablesct}|Dcgc]"}|jjdd$}}|D]*}|tvs t|D]}|j |,t |}|rU|j }tj|gD])}||vs|j ||j|+|rU|Scc}w)a Determine which tasks need to run based on affected tables. Also includes downstream dependent tasks. Args: affected_tables: List of table names that were modified Returns: Set of TaskName enums for tasks that should run -_) setlowerreplacer0addlistpopr1rKappend)rRrequired_taskstnormalized_tablestabletasktasks_to_checkdownstream_tasks rget_required_tasksrds%(EN?NN**34NN#) N "&u- )""4( )) .)N !!#/33D"= 7On4""?3%%o6 7  #Os'C tasksct|dS)z Sort tasks by priority/dependency order. Args: tasks: Set of tasks to sort Returns: List of tasks sorted by execution priority c.tj|dS)Nc)r#rK)r^s rz$get_tasks_in_order..s}'8'8B'?r)key)sorted)res rget_tasks_in_orderrls %? @@rdb company_id target_datebatch_idctddlm}m}ddlm}t j |g}|sy|jj} |tjk(xr|| k(} |D]} | r| tjk(rtjjd|} |j|j|j|k(|j j#| j%} | D]M}|j&|j(k7st*j-d|j d|jyt*j-d |j| jd |d |}|j|j|j|k(|j |k(j/}|s(t*j-d | jd |d |j&|j(k7st*j-d | jd|jyy)a Check if all dependencies for a task have completed successfully. Only checks dependencies that were actually triggered in the current batch. If a dependency task doesn't exist in this batch, it's considered "not required" and skipped - this handles cases like purchase_receive where DAILY_SALES is never triggered but INVENTORY_PLANNING still needs to run. Special case: INVENTORY_PLANNING for today's date needs to wait for ALL DAILY_SALES tasks in the batch (not just today's), because the planning calculation uses historical sales data from the past 90 days. Args: db: Database session task_name: The task to check dependencies for company_id: Company ID target_date: Target date string (YYYY-MM-DD) batch_id: Batch ID to group related tasks Returns: True if all dependencies are met (SUCCESS status) or not required rCeleryTaskTrackerCeleryTaskStatus)rT_%_z Waiting for z to complete for Fz#All daily_sales tasks complete for rUz Dependency z not in batch z , skippingz not yet complete for ))src.smart_inventory.apps.inventory.modelsrsrtrrrrKtoday isoformatr rrvaluequeryfilterrnr3likeallstatusSUCCESSr<debugfirst)rmr3rnrorprsrt date_type dependencies today_stris_planning_for_todaydep_taskpatternall_daily_sales ds_tracker dep_task_name dep_trackers rcheck_dependencies_metrs:^*$((B7L  !++-IX000 !y  !$ X1E1E%E!--334CzBG hh'89@@!,, :!++009ce  . ! $$(8(@(@@LL< 0D0D/EEVW`WfWfVg!hi  ! LL>y>OP Q $>>*!K=(D hh0188  ( (J 6  ' '= 8  %'  LL;x~~&6nXJjY Z    !1!9!9 9 LL;x~~&66LY__L]^ _I$L rcheck_dependenciescJddlm}m}|jd|d|}|j |j |j |k(|j|k(j} | r1tjd|d| j| jSttj} d} |rt!|||||} | r |j"n |j$} || ||| | rt'j(nd} |j+| |j-| r|t.j0k(rd |i}ny|t.j2k(rd |i}na|t.j4k(rd |i}nI|t.j6k(rd |i}n1|t.j8k(rd |i}ntj;d |y t=||| r&tjd |jd|n$tj;d|jdy | Stj|jd| S#t>$r`}tjAd|jd||jB| _ t|| _"|j-Yd}~yd}~wwxYw)a} Trigger a single Celery task with tracking. Args: db: Database session task_name: Task to trigger company_id: Company ID target_date: Target date (YYYY-MM-DD) batch_id: Batch ID for grouping check_dependencies: Whether to check and set HOLD status Returns: Task ID if triggered, None if failed rrrrUTask exists: TNrAr3rnr~ started_atrotarget_date_strsnapshot_date_strUnknown task type: z Triggered  for Celery task  not availableFailed to trigger task r5 set to HOLD)#rvrsrtryrzr{r3rnrr<rr~rAstruuiduuid4rPENDINGHOLDrnowrYcommitr rrrrrr=rQrMrNFAILURE error_message)rmr3rnrorprrsrttracker_task_nameexistingrAdependencies_metinitial_statustracker task_kwargsr>s rtrigger_single_taskrEs,^%??+1[M8*Exx)*11##'88$$ 2 eg  u./y8IJK$**,G1"i[Zbc2B%--GWG\G\N #%58<<>4 GFF7OIIK ,, ,(+6K (55 5(+6K (00 0,k:K (.. .. nMNO N  ( 56 N  LL29??2C2aSI J-55GN$'FG ! IIK  s;AH99 J"AJJ" unique_datesc ddlm}|s ttjdd}t |}|st jd||ddgdStjtjtjtjtjg}|Dcgc] }||vs| } }|} tjj!} | g} t j#d| Dcgc]}|j$c}t j#d| d t'| d t j#d | d g} d}d}t)| D]\}}|tjk(r| }n| }|D]}t+|||||||dkDr| |d z nd| }|s&ddlm}|j/|j1|j2|k(j5}|r|j6j$nd}| j9|j$||||d zd||j:j$k(r|d z }|d z }|dkDr t=||||||| dScc}wcc}w)a% Main function to trigger all required tasks based on affected tables. This is the primary entry point for the task orchestrator. Call this after CSV processing to automatically trigger all necessary background tasks. EXECUTION ORDER (wave-based): 1. DAILY_SALES for ALL unique dates in CSV (computes historical sales data) 2. INVENTORY_SNAPSHOT for TODAY only (current inventory state) 3. SERVICE_LEVEL for TODAY only (current service level) 4. SLOW_MOVERS for TODAY only (current slow mover analysis) 5. INVENTORY_PLANNING for TODAY only (current planning snapshot) Each wave waits for the previous wave to complete before starting. This ensures proper data availability for dependent calculations. Args: db: Database session affected_tables: List of table names that were modified by CSV upload company_id: Company ID the data belongs to unique_dates: List of unique dates (YYYY-MM-DD) in the uploaded data batch_id: Optional batch ID for grouping (auto-generated if not provided) Returns: Dict with triggered task information: { "batch_id": str, "tasks_triggered": int, "tasks_on_hold": int, "task_details": [...] } r)rtNzNo tasks for tables: )rptasks_triggered tasks_on_hold task_detailszWave-based execution: zDAILY_SALES dates:  (z dates)zSnapshot/Planning dates: z (today only)r )rmr3rnrorp wave_indexprevious_wave_task all_dates)rsunknown)r3rAror~wave)rvrtrrrrdr<rr rrrrrrrwrxinforylen enumeratetrigger_single_task_waversrzr{rArr~r\r_ensure_hold_poller_running)rmrRrnrrprtr]EXECUTION_ORDERr^ ordered_tasksrr today_onlyrrrrr3dates_to_processrorArsrr~s r!trigger_tasks_for_affected_tablesrsNK tzz|$Ra((8N  ,_,=>?      #### O!0G113FQGMGI &&(IJ KK(=)Ia!'')I(JKL KK%i[3y>2B'JK KK+J<}EFLOM "+=!9)) I ,, ,(  * + )K.#%'!%DNQRN=a#@X\# GW((#45<<%--8%'29--i##!*&#.$&N %-22888!Q&M#q(OA )))Xq#B H=*&$  EH*Js& I"0I"1I' rrrc hddlm}m} |jd|d|} |j |j |j | k(|j|k(j} | r1tjd| d| j| jSttj} d} |dkDr|rt!|||||} | r | j"n | j$}|| | ||| rt'j(nd}|j+||j-| r|t.j0k(rd |i}ny|t.j2k(rd |i}na|t.j4k(rd |i}nI|t.j6k(rd |i}n1|t.j8k(rd |i}ntj;d |y t=||| r)tjd |d|jd|n$tj;d|jdy | Stjd |d|jd|d| S#t>$r`}tjAd|jd|| jB|_ t||_"|j-Yd}~yd}~wwxYw)a Trigger a single task with wave-based dependency checking. Wave 0 tasks run immediately. Wave N tasks wait for ALL tasks in wave N-1 to complete. Args: db: Database session task_name: Task to trigger company_id: Company ID target_date: Target date (YYYY-MM-DD) batch_id: Batch ID for grouping wave_index: Which wave this task belongs to (0-based) previous_wave_task: The task type from the previous wave (to wait for) all_dates: All dates in this batch (to check all previous wave tasks) Returns: Task ID if triggered, None if failed rrrrUrrTNrrorrrzWave z : Triggered rrrrr5r)#rvrsrtryrzr{r3rnrr<rr~rArrrcheck_previous_wave_completerrrrrYrr rrrrrr=rQrMrNrr)rmr3rnrorprrrrsrtrrrArrrrr>s rrr1s:^%??+1[M8*Exx)*11##'88$$ 2 eg  u./y8IJK$**,GA~,7 "J)  2B%--GWG\G\N #%58<<>4 GFF7OIIK ,, ,(+6K (55 5(+6K (00 0,k:K (.. .. nMNb N  uZL9??*;5 \Z[ N  LL29??2C2aSI J-55GN$'FG ! IIK  sAI J1AJ,,J1 previous_taskcddlm}m}|D]}|jd|d|}|j |j |j |k(|j|k(j} | stjd|y| j|jk7stjd|d| jjdytjd |jd t|d y ) az Check if ALL tasks of the previous wave (for all dates) have completed. Args: db: Database session previous_task: The task type from the previous wave company_id: Company ID batch_id: Batch ID all_dates: All dates in this batch Returns: True if all previous wave tasks are complete (SUCCESS status) rrrrUzPrevious wave task not found: Fz!Previous wave task not complete: r)zPrevious wave (z) complete for all z datesT)rvrsrtryrzr{r3rnrr<rr~rr) rmrrnrprrsrtror3rs rrrs(^  $**+1[M8*E ((,-44  ' '9 4  ( (J 6  %'   LL9)E F >>-55 5 LLJZZ`ab rcBddlm}m}tjtj tj tjtjg}|j|j|j|jk(}|r|j|j|k(}|j}|syd}|D]j} | jj!dd} t#| dkr/| d} | d} | d} |r| |k7rFd}d}t%|D]\}}|j&| k(s|}|}n|r|dkrz|dk(rd }n||dz }|j&d | }|j|j|j| jk(|jj)|j}d }|sd }n$|D]}|j|j*k7sd }n|s)|tjk(rd | i}nb|tj k(rd | i}nJ|tj k(rd | i}n2|tjk(rd| i}n|tjk(rd| i}n |j,| _ t/j0| _|j5t7||| j8r,t:j=d| jd|d|dz }nJt:j?d| j|j@| _ d| _!|j5m|S#tD$ra}t:j?d| jd||j@| _ tG|| _!|j5Yd}~d}~wwxYw)a\ Check HOLD tasks and start them if previous wave is complete. Wave-based approach: tasks wait for ALL tasks of the previous wave to complete. Args: db: Database session company_id: Optional filter by company batch_id: Optional filter by batch Returns: Number of tasks started rrrrUr!r"r NTruFrorrzStarted HOLD task: z (wave rzFailed to dispatch HOLD task zFailed to dispatch taskzFailed to start HOLD task r5)$rvrsrtr rrrrrrzr{r~rrnr}r3rsplitrrryr|rrrrrrrQrAr<rrNrrrMr)rmrnrprsrtrrz hold_tasks tasks_startedrparts task_valuero task_batch_idr3ritnrrrprevious_wave_tasks prev_trackerrr>s rprocess_hold_tasksrs^ #### O HH& ' . .  $4$9$99 E .99ZGHJ MU!!((a0 u:> 1X Ah a  1   / EArxx:%     JN  ?# ,JN;M&,,-S@G"$((+<"="D"D!,,0B0BB!++009#ce $ &#' $7L#**.>.F.FF+0( H000,k: h999,k: h4440+> h2222K@ h9992K@  !1!9!9%-\\^" )KQLL#6w7H7H6IQ[P\\]!^_!Q&MLL#@ARAR@S!TU%5%=%=GN,EG)IIK_Un    9':K:K9LBqcRS!1!9!9(+A%  s(CL44 N=ANNcddlm}m}|j|j |j |j k(}|r|j |j|k(}|jduS)z+Check if there are any tasks in HOLD statusrrrN) rvrsrtrzr{r~rrnr)rmrnrsrtrzs rhas_hold_tasksrJsd] HH& ' . .  $4$9$99 E .99ZGH ;;= $$rc$ tjdk(rddlm}|j ||nddlm}|jd||dtjd|y #t$r"}tjd |Yd }~y d }~wwxYw) zu Ensure the background poller for HOLD tasks is running. Uses a Celery task that polls every 20 seconds. rDr)poll_hold_tasks)rnrprEz:src.smart_inventory.tasks.hold_task_poller.poll_hold_tasks)r@zStarted hold poller for batch z"Failed to start hold task poller: N) rGrH*src.smart_inventory.tasks.hold_task_pollerrdelayrJrFrLr<rrMr=)rmrnrprrFr>s rrrXs A <<7 " R  ! !Z( ! K ;  L&0hG !   4XJ?@ A;A3?@@AsA!A$$ B-B  Bc ddlm}m}|j|j |j j d|j}dddddt|d}g}|D]}|j|jjddz||jj<|j|j|j |jj|jr|jjnd|j r|j jnd|j"d|d|d z|d k(}||||d S) z Get the status of all tasks in a batch. Args: db: Database session batch_id: Batch ID to check Returns: Dict with batch status information rrrz%_)pendingstartedsuccessfailureholdtotalr N)rAr3r~r completed_atrrrr)rp status_counts all_completere)rvrsrtrzr{r3r|r}rrKr~ryr\rArrxrr) rmrprsrtrer task_listrars rget_batch_statusrrsR^ HH& ' . .##((2hZ9  ce U MI  +8+<+$r"} t.jAd| Yd} ~ 7d} ~ wwxYw#|jCwxYw#t>$r"}t.jEd|Yd}~yd}~wwxYw)a Update CeleryTaskTracker status for a task. Call this from within Celery tasks to update their tracking status. When status is 'success', automatically triggers processing of HOLD tasks that may have been waiting for this task to complete. Args: task_id: The Celery task ID status: New status (started, success, failure) error_message: Optional error message for failure status r)get_db_sessionrr)rrrrretryr)rrNiz Updated task z status to rrUr!r"z Auto-started z HOLD tasks after z completedzFailed to process hold tasks: zNo tracker found for task_id z&Failed to update task tracker status: )# src.utils.dbrrvrsrtrzr{rArSTARTEDrrrRETRYrKrWr~rrrrrrr<rrnr3rrrrrMr=closerN)rAr~rrrsrtrmr status_map new_statusrrprhold_errr>s rupdate_task_tracker_statusrs%"4C/a  - hh0188!))W4eg  077/77/77/77-33  (^^FLLN; %/GN||~2-5\\^*+AA/7||~,$0=et0D-IIKLL= VH!MN||~2w7I7IDKDUDU 1 1 8 8a @[]/25zQ58DX&8W=O=OQY&ZG&{ & mG9DVW^V__i,j k!>wiHI HHJ )X"NN-KH:+VWWX HHJ C =aSABBCsYI FH7'8H H78I H4H/*H7/H44H77I  I I7I22I7)T)N)NN)0rloggingrGrrrtypingrrrrr enumr sqlalchemy.ormr getLoggerrr<rr rrrrrr__annotations__r#intr0r1r2r?boolrQrdrlrrrrrrrrrrrrrrsv: #11"   8 $?sD? "  X112 8//0 ("="=x?S?S!T 54$x.01 !  A !  & tHcM"))*"../$001!)!5!5 6%778#667#556 334 (::;%-%@%@$A'99:$,$?$?#@--S$x.()8 81183G3GIdIde ("="=!> B "  4$xh/0 Z !e f ^ !x 1$x}-0 X tCH~ PS X\ NS c(mD Ac(m AX ATTTT T  T  Tz $ \\\\ \  \  \c]\H# JJ#YJJs) J sm J  #s(^ JZeeee e  e  e!*eCyec]eP(((( ( Cy (  (^7 PXY\P]ilD %w %HSM %T %AGAAA4--C-DcN-n$(EC EC ECC=ECr