a bgc@s ddlmZddlZddlZddlZddlZddlZddlmZddl 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!eZ2d"d"d#d$d%Z3d"d"d&d'd(Z4d)d"d&d*d+Z5d,d-d.d/Z6d"d-d0d1Z7dCd2d3d4d5d6d7Z8d"d"d8d9d:Z9d"d3d;dd?d@Z;GdAdBdBeZdictloader_detailsbool loading_end source_ownerclassifier_locationanonymize_snippetsNr(r#r#r#r$rBs rBr&)r.returncCsF|r d|vs d|dks |dvr$|St|}|r>|}t|S)zReturn an absolute local path for a local file/directory, for a network related path, return as is. Args: path (str): Relative path to be resolved. Returns: str: Resolved absolute path. z:///r)unknown-r)pathlibPathexistsresolver&)r. full_pathr#r#r$ get_full_paths   rU)loaderrLcCs&tD]\}}||vr|SqdS)zReturn loader type among, file, dir or in-memory. Args: loader (str): Name of the loader, whose type is to be resolved. Returns: str: One of the loader type among, file/dir/in-memory. unsupported)LOADER_TYPE_MAPPINGitems)rV loader_typeloadersr#r#r$get_loader_types  r\rc Csddlm}m}m}m}d}t|ts4td|S|j }zd|vrt||rfd|j d|j }nt||rd|j d|j }nRd |vr|d }|rd |vr|d }|r|d|}nd |vr|d }nd |vr|d }nd |vr$|d }|rt|t rt|dkr|d}nt||r6d}nt||rPd|j}n|jjdkr|dr|d} d| }nZ|dr|dg} ddd| D}n,|dr|dg} ddd| D}WntyYn0tt|S)zReturn an absolute source path of source of loader based on the keys present in Document. Args: loader (BaseLoader): Langchain document loader, derived from Baseloader. r)rrrrrOzGloader is not derived from BaseLoader, source location will be unknown!bucketzgc://rMzs3://sourcechannelr. file_path web_pathsrz notiondb://r folder_idz+https://drive.google.com/drive/u/2/folders/file_idsz, cSsg|]}d|dqS)z https://drive.google.com/file/d/z/viewr#).0Zfile_idr#r#r$ sz(get_loader_full_path.. document_idscSsg|]}d|dqS)z#https://docs.google.com/document/d/z/editr#)rddoc_idr#r#r$res)Z$langchain_community.document_loadersrrrr isinstancerloggererror__dict__r]ZblobkeyrClenZ database_id __class__rgetjoin ExceptionrUr&) rVrrrrlocationZ loader_dictr_rarbrcrfr#r#r$get_loader_full_pathsj                  rszTuple[Framework, Runtime])rLc Cst}td|ddd}t}t|jtjd|dd|j |j t |dd|d dd }d |jvrvd |_ d |_ td|td|||fS)zFetch the current Framework and Runtime details. Returns: Tuple[Framework, Runtime]: Framework and Runtime for the current app instance. Z langchainZlibrary_versionN)r9r:ZPWDr2rNr7Zruntime_version)r-r.r2r3r4r1r5r6DarwinZdesktopzMac OSXz framework zruntime )r r8ror2unamer*noder3environsystemr:get_ipr,r7ridebug)Z runtime_envr?rur7r#r#r$ get_runtimes*     r{cCs@ddl}|}z||}Wnty:|d}Yn0|S)zJFetch local runtime ip address. Returns: str: IP address rN localhost)socket gethostname gethostbynamerq)r}r-Z public_ipr#r#r$ry.s ryzList[Document]intzList[List[Document]])rDmax_batch_sizerLcCs~g}g}d}|D]Z}t|jd}||kr:||gq|||krX||g}d}||||7}q|rz|||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)rm page_contentencodeappend)rDrZbatchesZ current_batchZcurrent_batch_sizedocZdoc_sizer#r#r$generate_size_based_batches>s      r)r`rLcCs@z$ddl}t|j}||j}Wnty:d}Yn0|S)zFetch owner of local file path. Args: file_path (str): Local file path. Returns: str: Name of owner. rNrN)pwdr3statst_uidgetpwuidpw_namerq)r`rZfile_owner_uidZfile_owner_namer#r#r$get_file_owner_from_pathfs    r) source_pathrLcCs|sdSd}tj|r&tj|}n^tj|rd}t|D]>\}}}|D].}tj||}tj|sN|tj|7}qNq@|}|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)r3r.isfilegetsizeisdirwalkrpislink)rsize total_sizedirpath_ filenamesffpr#r#r$get_source_sizeys    r)datarLcCs|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)rrm)rZencoded_contentrr#r#r$calculate_content_sizes rc seZdZUdZded<dZded<ded<ded<d Zd ed <d d fdd ZdddddZd3dddd ddddZ dddddZ d4d ddddZ dd ddd!d dd"d#d$Z e d5dddd'd!d(d)d*d+Ze dddd,d-d.d/Ze d ddd0d1d2ZZS)6PebbloLoaderAPIWrapperzWrapper for Pebblo Loader API.r0api_keyr+r&rJclassifier_url cloud_urlFrGrKr)kwargsc sLt|ddd|d<t|ddt|d<t|ddt|d<tjfi|dS) z%Validate that api key in environment.rZPEBBLO_API_KEYr/rZPEBBLO_CLASSIFIER_URLrZPEBBLO_CLOUD_URLN)r _DEFAULT_CLASSIFIER_URL_DEFAULT_PEBBLO_CLOUD_URLsuper__init__)selfrrnr#r$rszPebbloLoaderAPIWrapper.__init__r;None)apprLc Csd}|jdd}|jdkrD|}|jtjj}|d|||}|jr|jdd}|rzt |j  d}| d|i| dti|jtjj}|d|||}dS) z Send app discovery request to Pebblo server & cloud. Args: app (App): App instance to be discovered. NTZ exclude_unsetr+POST cloud_requestpebblo_server_versionZpebblo_client_version)rErJ _make_headersrrr"value make_requestrjsonloadstextroupdatePLUGIN_VERSIONr) rr pebblo_resppayloadheadersZapp_discover_urlrpebblo_cloud_urlrr#r#r$send_loader_discovers"    z+PebbloLoaderAPIWrapper.send_loader_discoverzList[IndexedDocument]rE) docs_with_idrrFrHrLc Cs6|dd}t|}||||\}}|||||||} i} |jdkr|} |jtjj } zF| d| | | d} | rt | j dgD]}| |d|iqWn0ty}ztd|WYd }~n d }~00|jr|jdkr|| d| | d d || n|jd kr2td 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,rDr'z3An Exception caught in classify_documents: local %sNrKz pebblo-cloudz4API key is missing for sending docs to Pebblo cloud.)rorprepare_docs_for_classificationbuild_classification_payloadrJrrrr!rrrrrrrqriwarningrupdate_doc_datapopsend_docs_to_pebblo_cloud NameError)rrrrFrHrrIrDsource_aggregate_sizerclassified_docsrZ load_doc_urlrZclassified_docer#r#r$classify_documentss@     "     z)PebbloLoaderAPIWrapper.classify_documents)rrLc Csh|jdd}|jtjj}z|d|||}Wn0tyb}ztd|WYd}~n d}~00dS)z Send documents to Pebblo cloud. Args: payload (dict): The payload containing documents to be sent. Trrz3An Exception caught in classify_documents: cloud %sN) rrrr!rrrqrir)rrrrrrr#r#r$rs  z0PebbloLoaderAPIWrapper.send_docs_to_pebblo_cloud)rrLcCs4ddd}|r0|jr&|d|jin td|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.)rrrir)rrrr#r#r$r(s  z$PebbloLoaderAPIWrapper._make_headersz List[dict]r)rrDrFrIrrHrLc Csb|j|j|t|j|d||j|jd }|durHd|d<d|vrH||dd<tfi|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) r9r<rDr@r>rFrHrIrJrKTtruerHrFrr)r9r<rr>rJrKrBrE)rrrDrFrIrrHrr#r#r$r?s$  z3PebbloLoaderAPIWrapper.build_classification_payloadNzOptional[dict]zOptional[Response])methodurlrrtimeoutrLc Cszt|||||d}td||jjtt|jjr6|jjngt|j|jtj krht d|jn>|jtj krt d|j n|jtj krt d|j|WStyt d|Yn0ty}zt d|WYd}~n d}~00dS) 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)rrirzrr&rmbody status_coderINTERNAL_SERVER_ERRORr BAD_REQUESTrOKrrq)rrrrrresponserr#r#r$rks6     "z#PebbloLoaderAPIWrapper.make_requestzTuple[List[dict], int])rrrFrLc Cs:g}d}dd|D}d}|D]}|di}|dg} |ddkr^t|d |d } nt|d |d |} |d t| } |d t| } t|d} t| }||7}|ddpd}|| | ||did| d| rd| 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|] }|qSr#)rE)rdrr#r#r$rezJPebbloLoaderAPIWrapper.prepare_docs_for_classification..FmetadataZauthorized_identitiesrVrr^rrTr<rrr'N last_modified)rrr'rZ file_ownerZsource_path_sizeZsource_full_urlT)rorUrrr&rr)rrrFrDrZ doc_contentZsource_path_updaterZ doc_metadataZdoc_authorized_identitiesZdoc_source_pathZdoc_source_ownerZdoc_source_sizerZpage_content_sizergr#r#r$rs`         z6PebbloLoaderAPIWrapper.prepare_docs_for_classification)rDrrLc CsX|D]N}||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. r' pb_checksumloader_source_pathentitiestopics)rrrrrN)rorr)rDrZdoc_dataZclassified_datar#r#r$rs    z&PebbloLoaderAPIWrapper.update_doc_data)F)F)Nr)rrrr r)rJrKrrrrrr staticmethodrrr __classcell__r#r#rr$rs*    $>,1Fr)r)= __future__rrloggingr3rPr2enumrhttprtypingrrrrr Zlangchain_core.documentsr Zlangchain_core.envr Zlangchain_core.utilsr Zpydanticr requestsrrZrequests.exceptionsrZ)langchain_community.document_loaders.baser getLoggerrrirrrZBATCH_SIZE_BYTESZ file_loaderZ dir_loaderZ in_memoryZ cloud_folderrXr&rr%r*r8r;rBrUr\rsr{ryrrrrrr#r#r#r$s\           H(