Ë
    tPêjx  ã                   ó¤   — d Z ddlZddlZddlZddlZddlmZmZmZ ddl	m
Z
 erddlmZ neZddlZddlmZ ddlmZ ddlmZmZ  G d	„ d
e«      Zy)z¹
BETA

This is the PubSub logger for GCS PubSub, this sends LiteLLM SpendLogs Payloads to GCS PubSub.

Users can use this instead of sending their SpendLogs to their Postgres database.
é    N)ÚTYPE_CHECKINGÚAnyÚFinal)ÚStandardLoggingPayload)ÚSpendLogsPayload)Úverbose_logger)ÚCustomBatchLogger)Úget_async_httpx_clientÚhttpxSpecialProviderc                   óŽ   ‡ — e Zd Z	 	 	 ddedz  dedz  dedz  fˆ fd„Zdeeef   fd„Zd„ Zd	„ Zd
e	e
z  deeef   dz  fd„Zˆ xZS )ÚGcsPubSubLoggerNÚ
project_idÚtopic_idÚcredentials_pathc                 ó  •— ddl m}  |«        t        t        j                  ¬«      | _        |xs t        j                  d«      | _        |xs t        j                  d«      | _	        |xs t        j                  d«      | _
        | j                  r| j                  st        d«      ‚t        j                  «       | _        t        ‰| �@  d
i |¤d| j                  i¤Ž t        j"                  | j%                  «       «       g | _        y	)a  
        Initialize Google Cloud Pub/Sub publisher

        Args:
            project_id (str): Google Cloud project ID
            topic_id (str): Pub/Sub topic ID
            credentials_path (str, optional): Path to Google Cloud credentials JSON file
        r   ©Ú_premium_user_check)Úllm_providerÚGCS_PUBSUB_PROJECT_IDÚGCS_PUBSUB_TOPIC_IDÚGCS_PATH_SERVICE_ACCOUNTz-Both project_id and topic_id must be providedÚ
flush_lockN© )Úlitellm.proxy.utilsr   r
   r   ÚLoggingCallbackÚasync_httpx_clientÚosÚgetenvr   r   Úpath_service_account_jsonÚ
ValueErrorÚasyncioÚLockr   ÚsuperÚ__init__Úcreate_taskÚperiodic_flushÚ	log_queue)Úselfr   r   r   Úkwargsr   Ú	__class__s         €úl/home/oem/CyberBrain26-unified/.venv/lib/python3.12/site-packages/litellm/integrations/gcs_pubsub/pub_sub.pyr$   zGcsPubSubLogger.__init__    sÂ   ø€ õ 	<áÔä"8ÔFZ×FjÑFjÔ"kˆÔà$ÒJ¬¯	©	Ð2IÓ(JˆŒØ ÒD¤B§I¡IÐ.CÓ$DˆŒØ)9Ò)b¼R¿Y¹YÐGaÓ=bˆÔ&à�Š d§m¢mÜÐLÓMÐMä!Ÿ,™,›.ˆŒÜ‰ÑÑ>˜6Ñ>¨d¯o©oÓ>Ü×Ñ˜D×/Ñ/Ó1Ô2ØJLˆ�ó    Úreturnc              ƒ   óâ   K  — ddl m} |j                  | j                  | j                  d¬«      ƒ d{  –—† \  }}|j                  d|| j                  |ddddd¬«	      \  }}d|› �d	d
œ}|S 7 Œ8­w)z4Construct authorization headers using Vertex AI authr   )Úvertex_chat_completionÚ	vertex_ai)Úcredentialsr   Úcustom_llm_providerNzpub-sub)	ÚmodelÚauth_headerÚvertex_credentialsÚvertex_projectÚvertex_locationÚgemini_api_keyÚstreamr2   Úapi_basezBearer zapplication/json)ÚAuthorizationzContent-Type)Úlitellmr/   Ú_ensure_access_token_asyncr   r   Ú_get_token_and_url)r(   r/   Ú_auth_headerr6   r4   Ú_Úheaderss          r+   Úconstruct_request_headersz)GcsPubSubLogger.construct_request_headersA   s¡   è ø€ å2ð
 )×CÑCØ×6Ñ6Ø—‘Ø +ð Dó 
÷ 
ñ	
ØØð 0×BÑBØØ$Ø#×=Ñ=Ø)Ø ØØØ +Øð Có 

‰ˆ�Qð  ' { mÐ4Ø.ñ
ˆð ˆð-
ús   ‚2A/´A-µ9A/c              ƒ   ó&  K  — ddl m} ddlm}  |«        	 t	        j
                  d|«       |j                  dd«      }t        j                  du r( |||||¬«      }| j                  j                  |«       n| j                  j                  |«       t        | j                  «      | j                  k\  r| j                  «       ƒ d{  –—†  yy7 Œ# t        $ r3}	t	        j                  d	|	t!        j"                  «       «       Y d}	~	yd}	~	ww xY w­w)
a  
        Async Log success events to GCS PubSub Topic

        - Creates a SpendLogsPayload
        - Adds to batch queue
        - Flushes based on CustomBatchLogger settings

        Raises:
            Raises a NON Blocking verbose_logger.exception if an error occurs
        r   )Úget_logging_payloadr   z6PubSub: Logging - Enters logging function for model %sÚstandard_logging_objectNT)r)   Úresponse_objÚ
start_timeÚend_timezPubSub Layer Error - %s
%s)Ú1litellm.proxy.spend_tracking.spend_tracking_utilsrD   r   r   r   ÚdebugÚgetr<   Úgcs_pub_sub_use_v1r'   ÚappendÚlenÚ
batch_sizeÚasync_send_batchÚ	ExceptionÚ	exceptionÚ	tracebackÚ
format_exc)
r(   r)   rF   rG   rH   rD   r   Ústandard_logging_payloadÚspend_logs_payloadÚes
             r+   Úasync_log_success_eventz'GcsPubSubLogger.async_log_success_event`   sñ   è ø€ õ	
õ 	<áÔð	_Ü× Ñ Ð!YÐ[aÔbØ.4¯j©jÐ9RÐTXÓ.YÐ$ô ×)Ñ)¨TÑ1Ù,?Ø!Ø!-Ø)Ø%ô	-Ð"ð —‘×%Ñ%Ð&8Õ9ð —‘×%Ñ%Ð&>Ô?ä�4—>‘>Ó" d§o¡oÒ5Ø×+Ñ+Ó-×-Ñ-ð 6Ø-ùäò 	_Ü×$Ñ$Ð%BÀAÄy×G[ÑG[ÓG]×^Ñ^ûð	_üsA   ‚D—B2C Ã	CÃ
C ÃDÃC Ã	DÃ)D	ÄDÄ	DÄDc              ƒ   óú  K  — 	 | j                   s	 | j                   j                  «        yt        j                  dt	        | j                   «      «       | j                   D ]  }| j                  |«      ƒ d{  –—†  Œ 	 | j                   j                  «        y7 Œ"# t        $ r3}t        j                  d|t        j                  «       «       Y d}~ŒTd}~ww xY w# | j                   j                  «        w xY w­w)z8
        Sends the batch of messages to Pub/Sub
        Nz!PubSub - about to flush %s eventsz"PubSub Error sending batch - %s
%s)
r'   Úclearr   rJ   rN   Úpublish_messagerQ   rR   rS   rT   )r(   ÚmessagerW   s      r+   rP   z GcsPubSubLogger.async_send_batch‰   sÄ   è ø€ ð	#Ø—>’>Øð �N‰N× Ñ Õ"ô × Ñ Ð!DÄcÈ$Ï.É.ÓFYÔZàŸ>™>ò 4�Ø×*Ñ*¨7Ó3×3Ñ3ñ4ð �N‰N× Ñ Õ"ð 4ùäò 	gÜ×$Ñ$Ð%JÈAÌy×OcÑOcÓOe×fÑfûð	gûð �N‰N× Ñ Õ"üs]   ‚C;„B ‘C;¬AB Á8BÁ9B Á?C Â C;ÂB Â	CÂ&)CÃC ÃCÃC ÃC8Ã8C;r\   c              ƒ   ó  K  — 	 | j                  «       ƒ d{  –—† }t        |t        «      r|}nt        j                  |t        ¬«      }ddl}|j                  |j                  d«      «      j                  d«      }dd|igi}d| j                  › d| j                  › d	�}| j                  j                  |||¬
«      ƒ d{  –—† }|j                  dvrAt        j                  dt        |j                   «      «       t#        d|j                   › �«      ‚t        j$                  d|j                   «       |j                  «       S 7 �Œ/7 Œ†# t"        $ r)}	t        j                  dt        |	«      «       Y d}	~	yd}	~	ww xY w­w)zÏ
        Publish message to Google Cloud Pub/Sub using REST API

        Args:
            message: Message to publish (dict or string)

        Returns:
            dict: Published message response
        N)Údefaultr   zutf-8ÚmessagesÚdataz*https://pubsub.googleapis.com/v1/projects/z/topics/z:publish)ÚurlrA   Újson)éÈ   éÊ   zPub/Sub publish error: %szFailed to publish message: zPub/Sub response: %s)rB   Ú
isinstanceÚstrrb   ÚdumpsÚbase64Ú	b64encodeÚencodeÚdecoder   r   r   ÚpostÚstatus_coder   ÚerrorÚtextrQ   rJ   )
r(   r\   rA   Úmessage_datarh   Úencoded_messageÚrequest_bodyra   ÚresponserW   s
             r+   r[   zGcsPubSubLogger.publish_message›   sX  è ø€ ð	Ø#'×#AÑ#AÓ#C×CˆGô ˜'¤3Ô'Ø&‘ä#Ÿz™z¨'¼3Ô?�ó à%+×%5Ñ%5°l×6IÑ6IÈ'Ó6RÓ%S×%ZÑ%ZÐ[bÓ%cˆOð $.°¸Ð0IÐ/JÐ"KˆLàEÀdÇoÁoÐEVÐV^Ð_c×_lÑ_lÐ^mÐmuÐvˆCà$(×$;Ñ$;×$@Ñ$@ÀSÐRYÐ`lÐ$@Ó$m×mˆHà×#Ñ#¨:Ñ5Ü×$Ñ$Ð%@Ä#ÀhÇmÁmÓBTÔUÜÐ"=¸h¿m¹m¸_Ð MÓNÐNä× Ñ Ð!7¸¿¹ÔGØ—=‘=“?Ð"ð3 Dùð$ nùô ò 	Ü× Ñ Ð!<¼cÀ!»fÔEÜûð	üsR   ‚F „E —E˜B*E ÃE	ÃBE ÅF ÅE Å	E Å	E=ÅE8Å3F Å8E=Å=F )NNN)Ú__name__Ú
__module__Ú__qualname__rf   r$   ÚdictrB   rX   rP   r   r   r   r[   Ú__classcell__)r*   s   @r+   r   r      s�   ø„ ð "&Ø#Ø'+ñ	Mà˜$‘JðMð ˜‘*ðMð  ™*õ	MðB°°c¸3°h±ó ò>'_òR#ð$(Ð-=Ð@VÑ-Vð (Ð[_Ð`cÐehÐ`hÑ[iÐlpÑ[p÷ (r,   r   )Ú__doc__r!   rb   r   rS   Útypingr   r   r   Úlitellm.types.utilsr   Úlitellm.proxy._typesr   r<   Úlitellm._loggingr   Ú(litellm.integrations.custom_batch_loggerr	   Ú&litellm.llms.custom_httpx.http_handlerr
   r   r   r   r,   r+   ú<module>r€      sJ   ðñó Û Û 	Û ß ,Ñ ,å 6áÞ5àÐã Ý +Ý F÷ôdÐ'õ dr,   