jRddlZddlZddlZddlZddlZddlmZm Z ddl m Z m Z ddl mZddlmZddlZddlmZddlmZddlmZdd lmZmZmZmZmZmZdd lm Z m!Z!e Gd d eZ"e Gd deeZ#e Gdde#Z$e Gdde#Z%e Gdde#Z&e Gdde#eZ'e Gdde#Z(e Gdde#Z)e Gdde#Z*dS)N)ABCabstractmethod) dataclassfield) DictProxy)Path)DataClassJsonMixin)SourceConnectionNetworkError) RetryHandler)BaseDestinationConnectorBaseSourceConnectorPartitionConfigProcessorConfig ReadConfigRetryStrategyConfig)ingest_log_streaming_initloggercZeZdZdZdZedefdZejdefdZdS)PipelineContextz9 Data that gets shared across each pipeline node cd|_dSN_ingest_docs_mapselfs q/lsinfo/ai/hellotax_ai/base_platform/venv/lib/python3.11/site-packages/unstructured/ingest/pipeline/interfaces.py __post_init__zPipelineContext.__post_init__!s7;returnc<|jtd|jS)Nz!ingest_docs_map never initialized)r ValueErrorrs ringest_docs_mapzPipelineContext.ingest_docs_map$s#  (@AA A$$rvaluec||_dSrr)rr#s rr"zPipelineContext.ingest_docs_map*s %rN) __name__ __module__ __qualname____doc__rpropertyrr"setterrrrrs<<<%%%%X% &Y&&&&&&rrceZdZUdZeed<d dejejej dej fdZ de fdZ e dejej fdZd Zdejefd ZdS) PipelineNodezL Class that encapsulates logic to run during a single pipeline step pipeline_contextNiterablerc0|r|ng}|r2tjdjjdt |ds7|r|_nۉ_nj j dkr0|rfd|D_n_ntj j j tj jr tjn tjf5}|j|_dddn #1swxYwYt'jt(jrdjD_jS)NzCalling z with z docsc:g|]}|Sr+)run).0itrs r z)PipelineNode.__call__..Fs#???txx||???r) processes initializerinitargscg|]}||Srr+)r4rs rr6z)PipelineNode.__call__..RsCCCQ]1]]]r)rinfo __class__r%len initializesupported_multiprocessingr3resultr. num_processesmpPoolrverboseloggingDEBUGINFOmap isinstancetIterable)rr/pools` r__call__zPipelineNode.__call__7s'/88R   KR4>2RRc(mmRRR    --// ; )"hhx00 "hhjj  " 0A 5 5 )????h??? "hhjj /=5+/+@+HZ'--gl\ ;"hhtx::  ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; dk1: . . DCCdkCCCDK{s$!EEEcdS)NTr+rs rr@z&PipelineNode.supported_multiprocessingUtrcdSrr+rargskwargss rr3zPipelineNode.runX rc|x}r.tjd||ddt |jjr tjn tj dS)Nz Creating T)parentsexist_ok) get_pathrr<mkdirrr.rErFrGrH)rpaths rr?zPipelineNode.initialize\so==?? "4 4 K*D** + + + JJtdJ 3 3 3!43H3P"b'--V]VbcccccrcdSrr+rs rrYzPipelineNode.get_pathbrPrr)r%r&r'r(r__annotations__rKOptionalrLAnyrNboolr@rr3r?rrYr+rrr-r-/s&%%%AJqu,=!>!%<4 aj&7   ^ ddd !*T*rr-cleZdZUdZeed<fdZedej e fdZ de fdZ xZS)DocFactoryNodez> Encapsulated logic to generate a list of ingest docs source_doc_connectorctjd|jt |jdS)Nz?Running doc factory to generate ingest docs. Source connector: )rr<rcto_jsonsuperr?rr=s rr?zDocFactoryNode.initializensn  G!%!:!B!B!D!D G G     !,,.....rrcdSrr+rRs rr3zDocFactoryNode.runvrUrcdSNFr+rs rr@z(DocFactoryNode.supported_multiprocessingzur)r%r&r'r(r r]r?rrKrLdictr3r`r@ __classcell__r=s@rrbrbfs.---///// aj&6   ^ 4rrbceZdZUdZeed<dZeje ed<e deje fdZ fdZ ededejefd ZxZS) SourceNodez Encapsulated logic to pull from a data source via base ingest docs Output of logic expected to be the json outputs of the data itself read_configNretry_strategy_configrc |jx}rHttjt|j|jttjtjSdS)N)max_time max_triesrstart_log_levelbackoff_log_level) rrr backoffexpor max_retry_time max_retriesrlevel)rrrs rretry_strategyzSourceNode.retry_strategysR$($> >  ,.=/; & "(, trcptjdtdS)Nz@Running source node to download data associated with ingest docsrr<rfr?rgs rr?zSourceNode.initializes0 VWWW ringest_doc_jsoncdSrr+)rrs rr3zSourceNode.runrUr)r%r&r'r(rr]rrrKr^rr)r r}r?rstrr3rmrns@rrprp~s =A1:&9:AAA   < 8   X  3 1:c?   ^     rrpceZdZUdZeed<eeZeed<fdZ de fdZ e de de je fd Zdefd ZxZS) PartitionNodez` Encapsulates logic to run partition on the json files as the output of the source node partition_config)default_factorypartition_kwargsctjd|jdt j|jdtdS)NzCRunning partition node to extract content from json files. Config: z, partition kwargs: ]) rr<rrejsondumpsrrfr?rgs rr?zPartitionNode.initializes{  F,4466 F F!%D,A!B!B F F F   rrc|j}|j|d<tjt j|dddS)NrT) sort_keys ) rto_dictrhashlibsha256rrencode hexdigest)r hash_dicts r create_hashzPartitionNode.create_hashsg)1133 (,(= $%~djdCCCJJLLMMWWYYZ][]Z]^^r json_pathcdSrr+rrs rr3zPartitionNode.runrUrc^t|jjdz S)N partitioned)rr.work_dirresolvers rrYzPartitionNode.get_paths'T*344}DMMOOOr)r%r&r'r(rr]rrlrr?rrrrKr^r3rrYrmrns@rrrs&%%%"U4888d888_S____  S QZ_   ^ P$PPPPPPPPrrcHeZdZdZededejefdZdS) ReformatNodezr Encapsulated any logic to reformat the output List[Element] content from partition before writing it elements_jsonrcdSrr+)rrs rr3zReformatNode.runrUrN) r%r&r'r(rrrKr^r3r+rrrrsR   C   ^   rrcleZdZUdZeed<edeje fdZ fdZ de fdZ xZS) WriteNodezV Encapsulated logic to write the final result to a downstream data connection dest_doc_connector json_pathscdSrr+)rrs rr3z WriteNode.runrUrctjd|jdt |jdS)Nz=Running write node to upload content. Destination connector: r)rr<rrerfr?rgs rr?zWriteNode.initializest  K&*&=&E&E&G&G K K K     **,,,,,rrcdSrjr+rs rr@z#WriteNode.supported_multiprocessingrkr)r%r&r'r(r r]rrKListrr3r?r`r@rmrns@rrrs1000 afSk   ^ -----4rrc>eZdZdZfdZedefdZxZS)CopyNodezh Encapsulated logic to copy the final result of the pipeline to the designated output location. cptjdtdS)Nzrs5 ########((((((((......//////BBBBBB;;;;;;IHHHHHHH &&&&&o&& && 33333%s33 3l \ .          @ PPPPPLPP P:      <       .       |             l        r