U hb @s(ddlmZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z m Z mZmZmZddlmZddlmZddlmZdd lmZdd lmZmZdd lmZdd lmZe e!Z"d Z#dZ$dZ%dZ&ddddddddddddg Z'dddd d!gZ(d"gZ)d#d$d%gZ*e'e(e)e*d&Z+Gd'd(d(e,eZ-Gd)d*d*eZ.Gd+d,d,eZ/Gd-d.d.eZ0Gd/d0d0eZ1Gd1d2d2eZ2d3d3d4d5d6Z3d3d3d7d8d9Z4d:d3d7d;d<Z5d=d>d?d@Z6d3d>dAdBZ7dTdCdDdEdFdGdHZ8d3d3dIdJdKZ9d3dDdLdMdNZ:d3dDdOdPdQZ;GdRdSdSeZdictloader_detailsbool loading_end source_ownerclassifier_locationNr(r#r#r#r$rBs rBr&)r.returncCsF|r d|ks d|dks |dkr$|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   rT)loaderrKcCs&tD]\}}||kr|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)rU loader_typeloadersr#r#r$get_loader_types  r[rc Csddlm}m}m}m}d}t|ts4td|S|j }zd|krt||rfd|j d|j }nt||rd|j d|j }nRd |kr|d }|rd |kr|d }|r|d|}nd |kr|d }nd |kr|d }nd |kr$|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}Wntk rYnXtt|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)rrrrrNzGloader is not derived from BaseLoader, source location will be unknown!bucketzgc://rLzs3://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#)rcdoc_idr#r#r$rds)Z$langchain_community.document_loadersrrrr isinstancerloggererror__dict__r\ZblobkeyrClenZ database_id __class__rgetjoin ExceptionrTr&) rUrrrrlocationZ loader_dictr^r`rarbrer#r#r$get_loader_full_pathsj                  rrzTuple[Framework, Runtime])rKc Cst}td|ddd}t}t|jtjd|dd|j |j t |dd|d dd }d |jkrvd |_ 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:ZPWDr2rMr7Zruntime_version)r-r.r2r3r4r1r5r6DarwinZdesktopzMac OSXz framework zruntime )r r8rnr2unamer*noder3environsystemr:get_ipr,r7rhdebug)Z runtime_envr?rtr7r#r#r$ get_runtime s,     rzcCsBddl}|}z||}Wntk r<|d}YnX|S)zJFetch local runtime ip address. Returns: str: IP address rN localhost)socket gethostname gethostbynamerp)r|r-Z public_ipr#r#r$rx,srxzList[Document]intzList[List[Document]])rDmax_batch_sizerKcCs~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)rl page_contentencodeappend)rDrZbatchesZ current_batchZcurrent_batch_sizedocZdoc_sizer#r#r$generate_size_based_batches<s      r)r_rKcCsBz$ddl}t|j}||j}Wntk r<d}YnX|S)zFetch owner of local file path. Args: file_path (str): Local file path. Returns: str: Name of owner. rNrM)pwdr3statst_uidgetpwuidpw_namerp)r_rZfile_owner_uidZfile_owner_namer#r#r$get_file_owner_from_pathds   r) source_pathrKcCs|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.isfilegetsizeisdirwalkroislink)rsizeZ total_sizedirpath_ filenamesffpr#r#r$get_source_sizews    r)datarKcCs|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)rrl)rZencoded_contentrr#r#r$calculate_content_sizes rc seZdZUdZded<dZded<ded<ded<d d fd d Zd ddddZd2dd ddddddZdddddZ d3dddddZ d dddd ddd!d"d#Z e d4dddd&d d'd(d)d*Z e dddd+d,d-d.Ze dddd/d0d1ZZS)5PebbloLoaderAPIWrapperzWrapper for Pebblo Loader API.r0api_keyr+r&rJclassifier_url cloud_urlr)kwargsc sHt|ddd|d<t|ddt|d<t|ddt|d<tjf|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__)selfrrmr#r$rs&zPebbloLoaderAPIWrapper.__init__r;None)apprKc 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_requestrjsonloadstextrnupdatePLUGIN_VERSIONr) rr pebblo_resppayloadheadersZapp_discover_urlrpebblo_cloud_urlrr#r#r$send_loader_discovers"    z+PebbloLoaderAPIWrapper.send_loader_discoverFzList[IndexedDocument]rErG) docs_with_idrrFrHrKc Cs(|dd}t|}||||\}}|||||||} i} |jdkr|} |jtjj } zF| d| | | d} | rt | j dgD]}| |d|iqWn.tk r}ztd|W5d }~XYnX|jr|jdkr|| d| || n|jd kr$td 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 %sNz pebblo-cloudz4API key is missing for sending docs to Pebblo cloud.)rnrprepare_docs_for_classificationbuild_classification_payloadrJrrrr!rrrrrrrprhwarningrupdate_doc_datasend_docs_to_pebblo_cloud NameError)rrrrFrHrrIrDsource_aggregate_sizerclassified_docsrZ load_doc_urlrZclassified_docer#r#r$classify_documentssT       z)PebbloLoaderAPIWrapper.classify_documents)rrKc Csf|jdd}|jtjj}z|d|||}Wn.tk r`}ztd|W5d}~XYnXdS)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!rrrprhr)rrrrrrr#r#r$rs  z0PebbloLoaderAPIWrapper.send_docs_to_pebblo_cloud)rrKcCs4ddd}|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.)rrrhr)rrrr#r#r$r"s  z$PebbloLoaderAPIWrapper._make_headersz List[dict]r)rrDrFrIrrHrKc CsZ|j|j|t|j|d||jd }|dkrDd|d<d|krD||dd<tf|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>rFrHrIrJTtruerHrFrr)r9r<rr>rJrBrE)rrrDrFrIrrHrr#r#r$r9s"  z3PebbloLoaderAPIWrapper.build_classification_payloadNzOptional[dict]zOptional[Response])methodurlrrtimeoutrKc Cszt|||||d}td||jjtt|jjr6|jjngt|j|jtj krht d|jn>|jtj krt d|j n|jtj krt d|j|WStk rt d|Yn.tk r}zt d|W5d}~XYnXdS) 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)rrhryrr&rlbody status_coderINTERNAL_SERVER_ERRORr BAD_REQUESTrOKrrp)rrrrrresponserr#r#r$rds:    z#PebbloLoaderAPIWrapper.make_requestzTuple[List[dict], int])rrrFrKc Cs8g}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| dk rd| 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)rcrr#r#r$rdszJPebbloLoaderAPIWrapper.prepare_docs_for_classification..FmetadataZauthorized_identitiesrUrr]rrSr<rrr'N last_modified)rrr'rZ file_ownerZsource_path_sizeZsource_full_urlT)rnrTrrr&rr)rrrFrDrZ doc_contentZsource_path_updaterZ doc_metadataZdoc_authorized_identitiesZdoc_source_pathZdoc_source_ownerZdoc_source_sizerZpage_content_sizerfr#r#r$rs`        z6PebbloLoaderAPIWrapper.prepare_docs_for_classification)rDrrKc 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)rnrpop)rDrZdoc_dataZclassified_datar#r#r$rs    z&PebbloLoaderAPIWrapper.update_doc_data)F)F)Nr)rrrr r)rJrrrrrr staticmethodrrr __classcell__r#r#rr$rs(   $<+1Fr)r)= __future__rrloggingr3rOr2enumrhttprtypingrrrrr Zlangchain_core.documentsr Zlangchain_core.envr Zlangchain_core.pydantic_v1r Zlangchain_core.utilsr requestsrrZrequests.exceptionsrZ)langchain_community.document_loaders.baser getLoggerrrhrrrZBATCH_SIZE_BYTESZ file_loaderZ dir_loaderZ in_memoryZ cloud_folderrWr&rr%r*r8r;rBrTr[rrrzrxrrrrrr#r#r#r$s           H(