o ȳgc@sddlmZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z m Z mZmZmZddlmZddlmZddlmZdd lmZdd lmZmZdd lmZdd lmZe e!Z"d Z#dZ$dZ%dZ&gdZ'gdZ(dgZ)gdZ*e'e(e)e*dZ+Gddde,eZ-GdddeZ.GdddeZ/GdddeZ0GdddeZ1Gd d!d!eZ2dDd%d&Z3dEd(d)Z4dFd+d,Z5dGd.d/Z6dHd0d1Z7 dIdJd7d8Z8dKd:d;Z9dLd=d>Z:dMd@dAZ;GdBdCdCeZ z(get_loader_full_path.. document_idscSrs)z#https://docs.google.com/document/d/z/editr2)rtdoc_idr2r2r3rvrw)$langchain_community.document_loadersr$rr%r isinstancerloggererror__dict__rlblobkeyrRlen database_id __class__r,getjoin Exceptionrdr5) rer$rr%rlocation loader_dictrnrprqrrrxr2r2r3get_loader_full_pathst                  rTuple[Framework, Runtime]c Cst}td|ddd}t}t|jtjd|dd|j |j t |dd|d dd }d |jvr;d |_ d |_ td|td|||fS)zFetch the current Framework and Runtime details. Returns: Tuple[Framework, Runtime]: Framework and Runtime for the current app instance. langchainlibrary_versionN)rHrIPWDrAr]rFruntime_version)r<r=rArBrCr@rDrEDarwindesktopzMac OSXz framework zruntime )r rGrrAunamer9noderBenvironsystemrIget_ipr;rFr|debug) runtime_envrNrrFr2r2r3 get_runtimes*     rcCs@ddl}|}z||}W|Sty|d}Y|Sw)zJFetch local runtime ip address. Returns: str: IP address rN localhost)socket gethostname gethostbynamer)rr< public_ipr2r2r3r.s  rrSList[Document]max_batch_sizeintList[List[Document]]cCs~g}g}d}|D]-}t|jd}||kr||gq|||kr,||g}d}||||7}q|r=|||S)a Generate batches of documents based on page_content size. Args: docs: List of documents to be batched. max_batch_size: Maximum size of each batch in bytes. Defaults to 100*1024(100KB) Returns: List[List[Document]]: List of batches of documents rutf-8)r page_contentencodeappend)rSrbatches current_batchcurrent_batch_sizedocdoc_sizer2r2r3generate_size_based_batches>s       rrocCs@zddl}t|j}||j}W|Styd}Y|Sw)zFetch owner of local file path. Args: file_path (str): Local file path. Returns: str: Name of owner. rNr])pwdrBstatst_uidgetpwuidpw_namer)rorfile_owner_uidfile_owner_namer2r2r3get_file_owner_from_pathfs   r source_pathcCs|sdSd}tj|rtj|}|Stj|rCd}t|D]\}}}|D]}tj||}tj|s?|tj|7}q(q!|}|S)zFetch size of source path. Source can be a directory or a file. Args: source_path (str): Local path of data source. Returns: int: Source size in bytes. r)rBr=isfilegetsizeisdirwalkrislink)rsize total_sizedirpath_ filenamesffpr2r2r3get_source_sizeys"     rdatacCs|d}t|}|S)zCalculate the content size in bytes: - Encode the string to bytes using a specific encoding (e.g., UTF-8) - Get the length of the encoded bytes. Args: data (str): Data string. Returns: int: Size of string in bytes. r)rr)rencoded_contentrr2r2r3calculate_content_sizes rcseZdZUdZded< dZded< ded< ded< d Zd ed < d;fdd ZdddZ d?ddZ d=d@d!d"Z dAd(d)Z e * +dBdCd2d3Ze dDd6d7Ze dEd9d:ZZS)FPebbloLoaderAPIWrapperzWrapper for Pebblo Loader API.r?api_keyr:r5rYclassifier_url cloud_urlFrVrZkwargsrc sLt|ddd|d<t|ddt|d<t|ddt|d<tjd i|dS) z%Validate that api key in environment.rPEBBLO_API_KEYr>rPEBBLO_CLASSIFIER_URLrPEBBLO_CLOUD_URLNr2)r _DEFAULT_CLASSIFIER_URL_DEFAULT_PEBBLO_CLOUD_URLsuper__init__)selfrrr2r3rszPebbloLoaderAPIWrapper.__init__apprJr[Nonec Csd}|jdd}|jdkr"|}|jtjj}|d|||}|jrW|jdd}|r=t |j  d}| d|i| dti|jtjj}|d|||}dSdS) z Send app discovery request to Pebblo server & cloud. Args: app (App): App instance to be discovered. NT exclude_unsetr:POST cloud_requestpebblo_server_versionpebblo_client_version)rTrY _make_headersrr+r1value make_requestrjsonloadstextrupdatePLUGIN_VERSIONr) rr pebblo_resppayloadheadersapp_discover_urlrpebblo_cloud_urlrr2r2r3send_loader_discovers$    z+PebbloLoaderAPIWrapper.send_loader_discover docs_with_idList[IndexedDocument]rUrTrWc Cs4|dd}t|}||||\}}|||||||} i} |jdkrm|} |jtjj } z#| d| | | d} | rSt | j dgD] }| |d|iqGWntyl}z td|WYd }~nd }~ww|jr|jdkr}|| d| | d d || | S|jd krtd td | S) a Send documents to Pebblo server for classification. Then send classified documents to Daxa cloud(If api_key is present). Args: docs_with_id (List[IndexedDocument]): List of documents to be classified. app (App): App instance. loader_details (dict): Loader details. loading_end (bool): Boolean, indicating the halt of data loading by loader. rr>r:ri,rSr6z3An Exception caught in classify_documents: local %sNrZz pebblo-cloudz4API key is missing for sending docs to Pebblo cloud.)rrprepare_docs_for_classificationbuild_classification_payloadrYrrr+r0rrrrrrrr|warningrupdate_doc_datapopsend_docs_to_pebblo_cloud NameError)rrrrUrWrrXrSsource_aggregate_sizerclassified_docsr load_doc_urlrclassified_docer2r2r3classify_documentssH         z)PebbloLoaderAPIWrapper.classify_documentsrc Csh|jdd}|jtjj}z |d|||}WdSty3}z td|WYd}~dSd}~ww)z Send documents to Pebblo cloud. Args: payload (dict): The payload containing documents to be sent. Trrz3An Exception caught in classify_documents: cloud %sN) rrr+r0rrrr|r)rrrrrrr2r2r3rs z0PebbloLoaderAPIWrapper.send_docs_to_pebblo_cloudrcCs6ddd}|r|jr|d|ji|Std|S)z Generate headers for the request. args: cloud_request (bool): flag indicating whether the request is for Pebblo cloud. returns: dict: Headers for the request. zapplication/json)Acceptz Content-Typez x-api-keyz,API key is missing for Pebblo cloud request.)rrr|r)rrrr2r2r3r(s  z$PebbloLoaderAPIWrapper._make_headersrS List[dict]rXrrc Csb|j|j|t|j|d||j|jd }|dur$d|d<d|vr$||dd<td i|jdd}|S) a Build the payload for document classification. Args: app (App): App instance. docs (List[dict]): List of documents to be classified. loader_details (dict): Loader details. source_owner (str): Owner of the source. source_aggregate_size (int): Aggregate size of the source. loading_end (bool): Boolean indicating the halt of data loading by loader. Returns: dict: Payload for document classification. false) rHrKrSrOrMrUrWrXrYrZTtruerWrUrrNr2)rHrKrrMrYrZrQrT)rrrSrUrXrrWrr2r2r3r?s$  z3PebbloLoaderAPIWrapper.build_classification_payloadNmethodurlrOptional[dict]timeoutOptional[Response]c CszYt|||||d}td||jjtt|jjr|jjngt|j|jtj kr6t d|j|WS|jtj krHt d|j |WS|jtj krWt d|j|WStyit d|YdSty}z t d|WYd}~dSd}~ww) a Make a request to the Pebblo API Args: method (str): HTTP method (GET, POST, PUT, DELETE, etc.). url (str): URL for the request. headers (dict): Headers for the request. payload (Optional[dict]): Payload for the request (for POST, PUT, etc.). timeout (int): Timeout for the request in seconds. Returns: Optional[Response]: Response object if the request is successful. )rrrrrz5Request: method %s, url %s, len %s response status %szPebblo Server: Error z$Pebblo received an invalid payload: z-Pebblo returned an unexpected response code: zUnable to reach server %sz'An Exception caught in make_request: %sN)rr|rrr5rbody status_coderINTERNAL_SERVER_ERRORr BAD_REQUESTrOKrr)rrrrrresponserr2r2r3rks@     z#PebbloLoaderAPIWrapper.make_requestrTuple[List[dict], int]c Cs6g}d}dd|D}d}|D]}|di}|dg} |ddkr.t|d |d } n t|d |d |} |d t| } |d t| } t|d} t| }||7}|ddp`d}|| | ||did| d| rwd| ini| durd| ini|ddkr|s|d|d <d}q||fS)a Prepare documents for classification. Args: docs_with_id (List[IndexedDocument]): List of documents to be classified. source_path (str): Source path of the documents. loader_details (dict): Contains loader info. Returns: Tuple[List[dict], int]: Documents and the aggregate size of the source. rcSsg|]}|qSr2)rT)rtrr2r2r3rvszJPebbloLoaderAPIWrapper.prepare_docs_for_classification..Fmetadataauthorized_identitiesrer'rmrrcrKrrr6N last_modified)rrr6r  file_ownersource_path_sizesource_full_urlT)rrdrrr5rr)rrrUrSr doc_contentsource_path_updater doc_metadatadoc_authorized_identitiesdoc_source_pathdoc_source_ownerdoc_source_sizerpage_content_sizeryr2r2r3rs`       z6PebbloLoaderAPIWrapper.prepare_docs_for_classificationrc CsX|D]'}||di}||d|d|di|did|dqdS) z Update the document data with classified information. Args: docs (List[dict]): List of document data to be updated. classified_docs (dict): The dictionary containing classified documents. r6 pb_checksumloader_source_pathentitiestopics)rrrrrN)rrr)rSrdoc_dataclassified_datar2r2r3rs    z&PebbloLoaderAPIWrapper.update_doc_data)rr)rrJr[r)F) rrrrJrUrTrWrVr[rT)rrTr[r)rrVr[rT)rrJrSrrUrTrXr5rrrWrVr[rT)Nr) rr5rr5rrTrrrrr[r)rrrr5rUrTr[r )rSrrrTr[r)r,r-r.r/r8rYrZrrrrrr staticmethodrrr __classcell__r2r2rr3rs6    $ >  ,1 Fr)r=r5r[r5)rer5r[r5)rerr[r5)r[r)r[r5)r)rSrrrr[r)ror5r[r5)rr5r[r)rr5r[r)= __future__rrloggingrBr_rAenumrhttprtypingrrrrr langchain_core.documentsr langchain_core.envr langchain_core.utilsr pydanticr requestsrrrequests.exceptionsr)langchain_community.document_loaders.baser getLoggerr,r|rrrBATCH_SIZE_BYTES file_loader dir_loader in_memory cloud_folderrgr5r+r4r9rGrJrQrdrkrrrrrrrrr2r2r2r3s^               H  (