Ë
    ÌAHj¥  ã                   óÒ   — U d dl Z d dlZd dlZd dlmZmZ d dlmZ ddlm	Z	 ddl
mZmZ  ej                  d«      ZdZi Zeeef   ed	<   d
„ Zdd„ZdZd„ Zi Zeeef   ed<   d„ Zd„ Zy)é    N)Ú	timedeltaÚtimezone)ÚAnyé   )ÚDuplicateJobError)ÚblueÚgreenz
rq.scriptsa†  
    -- KEYS[1] = job key (rq:job:{job_id})
    -- KEYS[2] = queue key (rq:queue:{queue_name})
    -- ARGV[1] = job_id
    -- ARGV[2] = push direction ("L", "R", or "N" for no push)
    -- ARGV[3] = TTL in seconds (-1 for no TTL)
    -- ARGV[4+] = field1, value1, field2, value2, ... for HSET

    -- Check if job hash already exists
    if redis.call("EXISTS", KEYS[1]) == 1 then
        return 0  -- Duplicate, reject
    end

    -- Save job hash
    if #ARGV > 3 then
        redis.call("HSET", KEYS[1], unpack(ARGV, 4))
    end

    -- Set TTL if specified
    local ttl = tonumber(ARGV[3])
    if ttl and ttl > 0 then
        redis.call("EXPIRE", KEYS[1], ttl)
    end

    -- Push job ID to queue (skip if "N")
    if ARGV[2] == "L" then
        redis.call("LPUSH", KEYS[2], ARGV[1])
    elseif ARGV[2] == "R" then
        redis.call("RPUSH", KEYS[2], ARGV[1])
    end

    return 1  -- Success
Ú_registered_scriptsc                 ó\   — | t         vr| j                  t        «      t         | <   t         |    S )z;Get or create the registered Lua script for unique enqueue.)r
   Úregister_scriptÚUNIQUE_ENQUEUE_SCRIPT©Ú
connections    úB/root/tools/cai/cai_env/lib/python3.12/site-packages/rq/scripts.pyÚget_unique_enqueue_scriptr   3   s-   € àÔ,Ñ,Ø*4×*DÑ*DÔEZÓ*[Ô˜JÑ'Ü˜zÑ*Ð*ó    c                 ój  — t        | «      }t        |«      }|j                  |j                  nd}|sd}n|rd}nd} ||j                  |g|j                  ||g|z   ¬«      }	|	dk(  rt        d|j                  › d«      ‚t        j                  d	t        |j                  «      t        |«      «       y
)aU  Atomically check uniqueness, save job, and optionally push to queue using Lua script.

    Args:
        connection: Redis connection
        queue_key (str): The Redis key for the queue
        job (Job): The job to enqueue
        enqueue (bool): Whether to push job ID to the queue. Defaults to True.
            Set to False for sync jobs that don't need to be queued.
        at_front (bool): Whether to push to front of queue

    Returns:
        bool: True if job was enqueued, False if duplicate exists

    Raises:
        DuplicateJobError: If a job with the same ID already exists
    éÿÿÿÿÚNÚLÚR©ÚkeysÚargsr   úJob with ID 'ú' already existsz Uniquely enqueued job %s into %sT)
r   Ú_build_hset_argsÚttlÚkeyÚidr   ÚloggerÚdebugr   r	   )
r   Ú	queue_keyÚjobÚenqueueÚat_frontÚscriptÚ	hset_argsr   Úpush_directionÚresults
             r   Úsave_unique_jobr+   :   s³   € ô" ' zÓ2€Fä  Ó%€Ið —W‘WÐ(ˆ#'Š'¨b€Cñ Ø‰Ù	Ø‰àˆñ Øg‰gyÐ!Øf‰fn cÐ*¨YÑ6ô€Fð
 ‚{Ü -°·±¨xÐ7GÐ HÓIÐIä
‡LLÐ3´T¸#¿&¹&³\Ä5ÈÓCSÔTØr   a¾  
    -- KEYS[1] = job key (rq:job:{job_id})
    -- KEYS[2] = scheduled registry key (rq:scheduled:{queue_name})
    -- KEYS[3] = queues key (rq:queues)
    -- ARGV[1] = job_id
    -- ARGV[2] = TTL in seconds (-1 for no TTL)
    -- ARGV[3] = scheduled timestamp (UTC)
    -- ARGV[4] = queue key (rq:queue:{queue_name})
    -- ARGV[5+] = field1, value1, field2, value2, ... for HSET

    -- Check if job hash already exists
    if redis.call("EXISTS", KEYS[1]) == 1 then
        return 0  -- Duplicate, reject
    end

    -- Save job hash
    if #ARGV > 4 then
        redis.call("HSET", KEYS[1], unpack(ARGV, 5))
    end

    -- Set TTL if specified
    local ttl = tonumber(ARGV[2])
    if ttl and ttl > 0 then
        redis.call("EXPIRE", KEYS[1], ttl)
    end

    -- Add to scheduled registry sorted set
    redis.call("ZADD", KEYS[2], tonumber(ARGV[3]), ARGV[1])

    -- Register queue
    redis.call("SADD", KEYS[3], ARGV[4])

    return 1  -- Success
c                 ó  — | j                  «       }g }|j                  «       D ]g  \  }}|€Œ	|j                  |«       t        |t        «      r|j                  |«       Œ<|j                  t        |t
        «      st        |«      n|«       Œi |S )zRBuild flat list of field/value pairs from job.to_dict() for use in Lua HSET calls.)Úto_dictÚitemsÚappendÚ
isinstanceÚbytesÚstr)r$   Újob_datar(   r   Úvalues        r   r   r   Œ   s   € à{‰{‹}€HØ€IØ—n‘nÓ&ò V‰
ˆˆUØÑØ×Ñ˜SÔ!Ü˜%¤Ô'Ø× Ñ  Õ'à× Ñ ´:¸eÄSÔ3I¤ U¤ÈuÕUðVð Ðr   Ú_registered_schedule_scriptsc                 ó\   — | t         vr| j                  t        «      t         | <   t         |    S )z<Get or create the registered Lua script for unique schedule.)r5   r   ÚUNIQUE_SCHEDULE_SCRIPTr   s    r   Úget_unique_schedule_scriptr8      s-   € àÔ5Ñ5Ø3=×3MÑ3MÔNdÓ3eÔ$ ZÑ0Ü'¨
Ñ3Ð3r   c                 ój  — t        | «      }t        |«      }|j                  |j                  nd}|j                  sYt	        t        t        j                  dk(  rt        j                  nt        j                   ¬«      «      }|j                  |¬«      }t        j                  |j                  «       «      }	d}
 ||j                  ||
g|j                  ||	|g|z   ¬«      }|dk(  rt        d|j                  › d«      ‚t         j#                  d	t%        |j                  «      t'        |«      «       y
)a8  Atomically check uniqueness, save job, and add to scheduled registry using Lua script.

    Args:
        connection: Redis connection
        queue_key (str): The Redis key for the queue (e.g. rq:queue:default)
        registry_key (str): The Redis key for the scheduled registry (e.g. rq:scheduled:default)
        job (Job): The job to schedule
        scheduled_datetime (datetime): The scheduled execution time

    Returns:
        bool: True if job was scheduled successfully

    Raises:
        DuplicateJobError: If a job with the same ID already exists
    r   r   )Úseconds)Útzinfoz	rq:queuesr   r   r   zUniquely scheduled job %s in %sT)r8   r   r   r;   r   r   ÚtimeÚdaylightÚaltzoneÚreplaceÚcalendarÚtimegmÚutctimetupler   r    r   r!   r"   r   r	   )r   r#   Úregistry_keyr$   Úscheduled_datetimer'   r(   r   ÚtzÚ	timestampÚ
queues_keyr*   s               r   Úschedule_unique_jobrH   ¤   s  € ô  (¨
Ó3€Fä  Ó%€Ið —W‘WÐ(ˆ#'Š'¨b€Cð ×$Ò$Ü”i¼4¿=¹=ÈAÒ;M¬$¯-ª-ÔSW×S_ÑS_Ð(`ÔaÓbˆØ/×7Ñ7¸rÐ7ÓBÐÜ—‘Ð 2× ?Ñ ?Ó AÓB€Ià€JáØg‰g| ZÐ0Øf‰fc˜9 iÐ0°9Ñ<ô€Fð
 ‚{Ü -°·±¨xÐ7GÐ HÓIÐIä
‡LLÐ2´D¸¿¹³LÄ%ÈÓBUÔVØr   )TF)r@   Úloggingr<   Údatetimer   r   Útypingr   Ú
exceptionsr   Úlogutilsr   r	   Ú	getLoggerr!   r   r
   ÚdictÚ__annotations__r   r+   r7   r   r5   r8   rH   © r   r   ú<module>rR      s‚   ðÜ Û Û ß (Ý å )ß !à	ˆ×	Ñ	˜<Ó	(€ð Ð ðF ')Ð T˜#˜s˜(‘^Ó (ò+ó*ð\!Ð òHð 02Ð ˜d 3¨ 8™nÓ 1ò4ó(r   