�
���g� � � � d dl Z d dlZd dlmZ d dlZd dlZd dlmZmZm Z d dl
m Z n# e $ r d dl
m Z Y nw xY wdZdZ G d� de� � ZdS ) � N)�Thread)�APIError�DatetimeSerializer�
batch_post)�Emptyi i P c �h � e Zd ZdZ ej d� � Z dd �Zd
� Zd � Z d � Z
d
� Z d� Z dS )�Consumerz.Consumes the messages from the client's queue.�posthog�d N� �?F�
� c �� � t j | � � d| _ || _ || _ || _ || _ || _ || _ || _ d| _
|| _ | | _ |
| _
dS )zCreate a consumer thread.TN)r �__init__�daemon�flush_at�flush_interval�api_key�host�on_error�queue�gzip�running�retries�timeout�historical_migration) �selfr r r r r r r r r r s �`/home/asafur/pinokio/api/open-webui.git/app/env/lib/python3.11/site-packages/posthog/consumer.pyr zConsumer.__init__ sr � � ��� � � ��� � ��
� ,��� �� ��� � ��
���
��� �
�� � �� � �� �$8��!�!�!� c � � | j � d� � | j r| � � � | j �| j � d� � dS )zRuns the consumer.zconsumer is running...zconsumer exited.N)�log�debugr �upload�r s r �runz Consumer.run= sW � � ����/�0�0�0��l� � �K�K�M�M�M� �l� �
����)�*�*�*�*�*r c � � d| _ dS )zPause the consumer.FN)r r$ s r �pausezConsumer.pauseE s
� � �� � � r c � � d}| � � � }t |� � dk rdS | � |� � d}nQ# t $ rD}| j � d|� � d}| j r| � ||� � Y d}~nd}~ww xY w|D ]}| j � � � � |S # |D ]}| j � � � � |c cY S xY w)z:Upload the next batch of items, return whether successful.Fr Tzerror uploading: %sN) �next�len�request� Exceptionr! �errorr r � task_done)r �success�batch�e�items r r# zConsumer.uploadI s
� ���� � � � �� �u�:�:��?�?��5� � �L�L�� � � ��G�G��� (� (� (� �H�N�N�0�!� 4� 4� 4��G��}�
(��
�
�a��'�'�'���������� (���� �
'�
'���
�$�$�&�&�&�&��N�� �
'�
'���
�$�$�&�&�&�&��N�N�N�N�N�N�N�Ns/ �A �B6 �
B�:B� B6 �B�B6 �6#Cc � � | j }g }t j � � }d}t |� � | j k �rt j � � |z
}|| j k rn� |� d| j |z
�� � }t t
j |t �� � � � � � � }|t k r)| j � dt |� � � � ��|� |� � ||z
}|t k r | j � d|� � n*n# t" $ r Y n w xY wt |� � | j k ��|S )z)Return the next batch of items to upload.r T)�blockr )�clsz)Item exceeds 900kib limit, dropping. (%s)z hit batch size limit (size: %d))r � monotonicr* r r �get�json�dumpsr �encode� MAX_MSG_SIZEr! r- �str�append�BATCH_SIZE_LIMITr" r )r r �items�
start_time�
total_size�elapsedr2 � item_sizes r r) z
Consumer.next^ sW � ��
���� �(�*�*�
��
��%�j�j�4�=�(�(� �)�+�+�j�8�G��$�-�-�-��
� �y�y�t�T�5H�7�5R�y�S�S�� ��
�4�5G� H� H� H� O� O� Q� Q� R� R� � �|�+�+��H�N�N�#N�PS�TX�PY�PY�Z�Z�Z� �� � �T�"�"�"��i�'�
� �!1�1�1��H�N�N�#D�j�Q�Q�Q�� 2�� �
�
�
���
���� �%�j�j�4�=�(�(�$ � s �B D* �(A D* �*
D7�6D7c � � �� d� }t j t j t � j dz |�� � �� fd�� � } |� � dS )z=Attempt to upload the batch and retry before raising an errorc � � t | t � � r-| j dk rdS d| j cxk odk nc o
| j dk S dS )NzN/AFi� i� i� )�
isinstancer �status)�excs r �fatal_exceptionz)Consumer.request.<locals>.fatal_exception} s_ � ��#�x�(�(�
� �:��&�&� �5��s�z�/�/�/�/�C�/�/�/�/�F�S�Z�3�5F�F� �ur � )� max_tries�giveupc �d �� t �j �j �j �j � �j �� � d S )N)r r r0 r )r r r r r r )r0 r s ��r � send_requestz&Consumer.request.<locals>.send_request� sA �� � �� �� ��Y� � ��%)�%>�
�
�
�
�
�
r N)�backoff� on_exception�expor, r )r r0 rI rN s `` r r+ zConsumer.requestz so ��� �
�
�
�
� �g�l�I�� �PQ�AQ�Zi� j� j� j� � � � � �
k� j� � � �����r )r NNr Fr
r F)
�__name__�
__module__� __qualname__�__doc__�logging� getLoggerr! r r% r' r# r) r+ � r r r r s� � � � � � �8�8�
�'�
�I�
&�
&�C� �
���
��� "� 9� 9� 9� 9�B+� +� +� � � �� � �*� � �8� � � � r r )r8 rV � threadingr rO r6 �posthog.requestr r r r r � ImportError�Queuer; r> r rX r r �<module>r] s� �� � � � � ���� � � � � � � ���� � � � � D� D� D� D� D� D� D� D� D� D� ���������� � � ��������� ���� � � #� �}� }� }� }� }�v� }� }� }� }� }s �) � 7�7