o Bpj@sdZddlmZddlZddlmZmZmZmZm Z ddl m Z ddl m Z eeZe eegefZdd d Zdd dddZdddddZddddddZdS)zKAuto pause/resume per-BID orchestration when usage allocation is exhausted.) annotationsN)AnyCallableDictListOptional)evaluate_usage_allocation)maybe_send_usage_validity_alertbidstrreturnboolc CsLz ddlm}||WSty%}ztd||WYd}~dSd}~ww)Nr) stop_loopz)[%s] Failed to stop orchestrator loop: %sF)orchestrator_supervisorr Exceptionloggerwarning)r rexcrA/home/aiteam/pcaa-dev/dashboard-backend/pipeline_usage_control.py_default_stop_loops  r) stop_loop_fnr StopLoopFnc Cs t|}|s dS|||pi}t|dpds dS|;}|'}t||}|dsC WdWddSt |||Wdn1sSwYWdn1sbwYt d||d|d| |dd d |pt }||d S) z Disable pipeline_enabled and stop the orchestrator loop when usage is blocked. Returns True if the pipeline was paused by this call. Fpipeline_enabledrblockedNzDUsage limit exhausted for BID %s (%s/%s min); auto-stopping pipeline used_minutesmonthly_minute_limitrusage_limit_pausedT)r strip%ensure_business_pipeline_config_tableget_pipeline_configintgetget_connectioncursorrr rrsave_pipeline_configr) db_handlerr rcfgconnr&usagestop_fnrrr!pause_pipeline_if_usage_exhaustedsB       r-) start_loop_fnr.c Cs"t|}|s dS|||pi}t|dpds dSt|dp'dr3||ddidS|5}|!}t ||}|drV WdWddSWdn1s`wYWdn1sowYt d||d|d ||d dd |r||d S) z Re-enable pipeline when it was auto-paused for usage and headroom is available again. Returns True if the pipeline was resumed by this call. FrrrrNzFUsage headroom restored for BID %s (%s/%s min); auto-resuming pipelinerrrrT) r r r!r"r#r$r'r%r&rrinfo)r(r r.r)r*r&r+rrr"resume_pipeline_if_usage_availableHsH      r0)rr.Dict[str, Any]c Cs|g}g}|}|}|d|pg}Wdn1s&wY|p.t}|D]} t| dp:d} | sAq1t | dpHd} t | dpQd} |'}|}t || } t || | Wdn1srwYWdn1swY| dr| r| | dd d || | | td | | d | d q1| r| ds| | d dd |r|| | | td| | d | d q1||dS)zMSupervisor hook: pause exhausted BIDs; resume auto-paused BIDs with headroom.zt SELECT bid, pipeline_enabled, usage_limit_paused FROM business_pipeline_config Nr rrrrrrz3Supervisor paused BID %s pipeline (usage %s/%s min)rrz4Supervisor resumed BID %s pipeline (usage %s/%s min))pausedresumed)r!r%r&executefetchallrr r$r r#rr r'appendrrr/)r(rr.r3r4r*r&rowsr,rowr enabled usage_pausedr+rrr sync_usage_limit_pipeline_statesvsj        r<)r r r r )r r rrr r )r r r.rr r )rrr.rr r1)__doc__ __future__rloggingtypingrrrrrusage_allocation_utilrusage_validity_alert_servicer getLogger__name__rr r rrr-r0r<rrrrs      21