qiDddlZddlZddlZddlZddlZddlZddlmZddlmZddl m Z ddl Z ddl m Z mZddlmZmZej$dZGd d eZGd d e Zy) N)suppress)datetime)quote)AbstractBufferedFileAbstractFileSystem)infer_storage_optionstokenizewebhdfscNeZdZdZeej ZdZ dfd Z e dZ dZ d dZ d!dZedZed Zed Zd Zd Zd Zd"dZdZdZdZd#dZdZdZdZd$dZdZ dZ!d"dZ"dZ#d"dZ$dZ%dZ&dZ'xZ(S)%WebHDFSa Interface to HDFS over HTTP using the WebHDFS API. Supports also HttpFS gateways. Four auth mechanisms are supported: insecure: no auth is done, and the user is assumed to be whoever they say they are (parameter ``user``), or a predefined value such as "dr.who" if not given spnego: when kerberos authentication is enabled, auth is negotiated by requests_kerberos https://github.com/requests/requests-kerberos . This establishes a session based on existing kinit login and/or specified principal/password; parameters are passed with ``kerb_kwargs`` token: uses an existing Hadoop delegation token from another secured service. Indeed, this client can also generate such tokens when not insecure. Note that tokens expire, but can be renewed (by a previously specified user) and may allow for proxying. basic-auth: used when both parameter ``user`` and parameter ``password`` are provided. )r webHDFSc  |jryt|di| | rdndd|d|d|_||_|xsi|_i|_| xsi|_||| td||j d<||_ ||_ | |td |||j d <|||j d <|r | td | |_ | |_ |jd t|||_y)a Parameters ---------- host: str Name-node address port: int Port for webHDFS kerberos: bool Whether to authenticate with kerberos for this connection token: str or None If given, use this token on every call to authenticate. A user and user-proxy may be encoded in the token and should not be also given user: str or None If given, assert the user name to connect with password: str or None If given, assert the password to use for basic auth. If password is provided, user must be provided also proxy_to: str or None If given, the user has the authority to proxy, and this value is the user in who's name actions are taken kerb_kwargs: dict Any extra arguments for HTTPKerberosAuth, see ``_ data_proxy: dict, callable or None If given, map data-node addresses. This can be necessary if the HDFS cluster is behind a proxy, running on Docker or otherwise has a mismatch between the host-names given by the name-node and the address by which to refer to them from the client. If a dict, maps host names ``host->data_proxy[host]``; if a callable, full URLs are passed, and function must conform to ``url->data_proxy(url)``. use_https: bool Whether to connect to the Name-node using HTTPS instead of HTTP session_cert: str or Tuple[str, str] or None Path to a certificate file, or tuple of (cert, key) files to use for the requests.Session session_verify: str, bool or None Path to a certificate file to use for verifying the requests.Session. kwargs Nhttpshttpz://:z /webhdfs/v1z_If passing a delegation token, must not set user or proxy_to, as these are encoded in the token delegationzQIf passing a password, the user must also beset in order to set up the basic-authz user.namedoaszJIf using Kerberos auth, do not specify the user, this is handled by kinit.webhdfs_)_cachedsuper__init__urlkerb kerb_kwargsparsproxy ValueErroruserpassword session_certsession_verify_connectr _fsid)selfhostportkerberostokenr r!proxy_tor data_proxy use_httpsr"r#kwargs __class__s P/opt/pipecat/venv/lib/python3.12/site-packages/fsspec/implementations/webhdfs.pyrzWebHDFS.__init__.s9r <<  "6"!*g7s4&${S &," %2  8#7  ',DIIl #    | < )- +&   (DIIf  (2  ), t 456 c|jSN)r%r&s r0fsidz WebHDFS.fsids zzr1ctj|_|jr|j|j_|j |j_|jr'ddlm }|di|j|j_ |j@|j3ddlm}||j|j|j_ yyy)Nr)HTTPKerberosAuth) HTTPBasicAuthr)requestsSessionsessionr"certr#verifyrrequests_kerberosr7rauthr r! requests.authr8)r&r7r8s r0r$zWebHDFS._connects'')    $ 1 1DLL "11  99 : 0 D43C3C DDLL  99 T]]%> 3 -dii GDLL &? r1c ||j|nd}|j|jt|dz}|j }|j |j |j|d<tjd|||jj|j||||} | jdvrY | j} | dd } | dd } | d vr t| | d vr t| | d vr t!| t#| | j'| S#tt$f$rY#wxYw)Nz/=)safeopzsending %s with %s)methodrparamsdataallow_redirects)iiiiiRemoteExceptionmessage exception)IllegalArgumentExceptionUnsupportedOperationException)SecurityExceptionAccessControlException)FileNotFoundException)_strip_protocol _apply_proxyrrcopyupdaterupperloggerdebugr;request status_codejsonrPermissionErrorFileNotFoundError RuntimeErrorKeyErrorraise_for_status) r&rDrEpathrGredirectr.rargsouterrmsgexps r0_callz WebHDFS._callsS-1-=t##D)25D+A AB{{} DIIXXZT  )37ll""<<>$ #  ??7 7 ,hhj+,Y7+,[9WW$S/)KK)#..55+C00&s++  )  s D;;E  E c \|xs |j}t|||||j|||S)a^ Parameters ---------- path: str File location mode: str 'rb', 'wb', etc. block_size: int Client buffer size for read-ahead or write buffer autocommit: bool If False, writes to temporary file that only gets put in final location upon commit replication: int Number of copies of file on the cluster, write mode only permissions: str or int posix permissions, write mode only kwargs Returns ------- WebHDFile instance )mode block_sizetempdir autocommit replication permissions) blocksize WebHDFilerk)r&r`rirjrlrmrnr.s r0_openz WebHDFS._opens<B 14>>   !LL!##  r1cB|dj|d<|d|d<|S)Ntypelengthsize)lower)infos r0 _process_infozWebHDFS._process_infos*F|))+V H~V  r1ct|dS)Nr`)r )clsr`s r0rQzWebHDFS._strip_protocols$T*622r1ct|}|jdd|jddd|vr|jd|d<|S)Nr`protocolusernamer )r pop)urlpathrcs r0_get_kwargs_from_urlszWebHDFS._get_kwargs_from_urlssG#G,   D!  ''*-CK r1cz|jd|}|jd}||d<|j|S)N GETFILESTATUSr` FileStatusname)rgrZrx)r&r`rcrws r0rwz WebHDFS.info s>jjtj4xxz,'V !!$''r1c|j|}|jdd}|tj|dz St d)z=Return the created timestamp of a file as a datetime.datetimemodificationTimeNz5Could not retrieve creation time (modification time).rwgetr fromtimestampr]r&r`rwmtimes r0createdzWebHDFS.createdsIyy+T2  ))%$,7 7RSSr1c|j|}|jdd}|tj|dz St d)z>Return the modified timestamp of a file as a datetime.datetimerNrz%Could not retrieve modification time.rrs r0modifiedzWebHDFS.modifiedsGyy+T2  ))%$,7 7BCCr1c |jd|}|jdd}|D]0}|j||jddz|dz|d<2|rt |d St d |DS) N LISTSTATUSr FileStatusesr/ pathSuffixrc |dS)Nrr)is r0zWebHDFS.ls..(s qyr1)keyc3&K|] }|d yw)rNr).0rws r0 zWebHDFS.ls..*s94$v,9s)rgrZrxrstripsorted)r&r`detailr.rcinfosrws r0lsz WebHDFS.ls!sjjDj1 >*<8 GD   t $;;s+c1D4FFDL G %%89 99599 9r1cN|jd|}|jdS)z8Total numbers of files, directories and bytes under pathGETCONTENTSUMMARYrContentSummaryrgrZ)r&r`rcs r0content_summaryzWebHDFS.content_summary,s'jj,4j8xxz*++r1cD|jd|d}d|jvr\|j|jd}|jj |}|j |j dS|j |j dS)z/Checksum info of file, giving method and resultGETFILECHECKSUMF)r`raLocation FileChecksum)rgheadersrRr;rr_rZ)r&r`rclocationout2s r0ukeyz WebHDFS.ukey1sjj*jF  $((Z)@AH<<##H-D  ! ! #99;~. .  "88:n- -r1cJ|jd}|jdS)zGet user's home directoryGETHOMEDIRECTORYPathr)r&rcs r0home_directoryzWebHDFS.home_directory=s"jj+,xxz&!!r1c|r|jd|}n|jd}|jd}| td|dS)zRetrieve token which can give the same authority to other uses Parameters ---------- renewer: str or None User who may use this token; if None, will be current user GETDELEGATIONTOKEN)renewerTokenz1No token available for this user/security context urlString)rgrZr)r&rrcts r0get_delegation_tokenzWebHDFS.get_delegation_tokenBsT **17*CC**12C HHJw  9PQ Q~r1cP|jdd|}|jdS)z/Make token live longer. Returns new expiry timeRENEWDELEGATIONTOKENputrEr*longr)r&r*rcs r0renew_delegation_tokenzWebHDFS.renew_delegation_tokenSs(jj/UjKxxz&!!r1c,|jdd|y)z Stop the token from being usefulCANCELDELEGATIONTOKENrrNrg)r&r*s r0cancel_delegation_tokenzWebHDFS.cancel_delegation_tokenXs *5 Fr1c.|jdd||y)aSet the permission at path Parameters ---------- path: str location to set (file or directory) mod: str or int posix epresentation or permission, give as oct string, e.g, '777' or 0o777 SETPERMISSIONr)rEr` permissionNr)r&r`mods r0chmodz WebHDFS.chmod\s ?5t Lr1cNi}|||d<|||d<|jdd|d|y)zChange owning user and/or groupNownergrouprrEr`)SETOWNERr)r&r`rrr.s r0chownz WebHDFS.chownis=  #F7O  #F7O Ae$A&Ar1c.|jd|d|y)a9 Set file replication factor Parameters ---------- path: str File location (not for directories) replication: int Number of copies of file on the cluster. Should be smaller than number of data nodes; normally 3 on most systems. SETREPLICATIONr)r`rErmNr)r&r`rms r0set_replicationzWebHDFS.set_replicationrs #$u+ Vr1c ,|jdd|y)NMKDIRSrrrr&r`r.s r0mkdirz WebHDFS.mkdirs 8E 5r1cf|dur|j|r t||j|y)NF)existsFileExistsErrorr)r&r`exist_oks r0makedirszWebHDFS.makedirss, u T!2!$' ' 4r1c .|jdd||y)NRENAMEr)rEr` destinationr)r&path1path2r.s r0mvz WebHDFS.mvs 8E5 Ir1c B|jdd||rdydy)NDELETEdeletetruefalse)rEr` recursiver)r&r`rr.s r0rmz WebHDFS.rms0  )f  07  r1c &|j|yr3)rrs r0rm_filezWebHDFS.rm_files   r1c |j|5}dj|j|dtjdg} |j|d5}t j ||ddd|j|| dddy#1swY%xYw#t$r7tt5|j|ddd#1swYxYwwxYw#1swYyxYw)Nrz.tmp.wb) openjoin_parentsecrets token_hexshutil copyfileobjr BaseExceptionrr\r)r&lpathrpathr.lstream tmp_fnamerstreams r0cp_filezWebHDFS.cp_files YYu  $,,u"5w?P?PQS?T>U7V!WXI YYy$/97&&w89 5)   99! /0'GGI&''   sM9C( B%B5B%B" B%%C%=C C%C! C%%C((C1c|jr(t|jr|j|}|S|jr5|jjD]\}}|j||d}|S)N)rcallableitemsreplace)r&rkvs r0rRzWebHDFS._apply_proxysk ::(4::.zz(+H  ZZ ((* 51#++Aq!4 5r1) iFNNNNNNFNT)rNNT)rbNTNNFr3)NN))__name__ __module__ __qualname____doc__strtempfile gettempdirrkr|rpropertyr5r$rgrq staticmethodrx classmethodrQrrwrrrrrrrrrrrrrrrrrrrR __classcell__r/s@r0r r s-*%(%%'(G#H  c7JH$H+ Z 33( TD :, ." "" G MB W6 J  r1r cBeZdZdZfdZddZdZdZdZdZ xZ S) rpz"A file living in HDFS over webHDFSc t|||fi||j}|jdd|j dd|jdd|j dd|j dd|_|j d}|j dddurR|j |_tj j|ttj|_yy)NrnrmirkrlF) rrrSrr~rnr`targetosrruuiduuid4)r&fsr`r.rkr/s r0rzWebHDFile.__init__s T,V, ::mT * 2 JJ}d + ::mT * 2 JJ}d +!::mS9**Y' ::lE *e 3))DK Wc$**,.?@DI 4r1c|jjj|j|jj ddi}|j y)zWrite one part of a multi-block file upload Parameters ========== final: bool This is the last block, so should complete file, if self.autocommit is True. content-typeapplication/octet-stream)rGrT)rr;postrbuffergetvaluer_)r&finalrcs r0 _upload_chunkzWebHDFile._upload_chunksRggoo"" MM%%'#%?@#  r1cZ|jj}d|jvrd\}}n d\}}d|d<|jj|||j fddi|}|jj |jd}d |jvr|jjj|d d i }|j|jjd d|j fddi|}|jj |jd|_ yy)zCreate remote file/uploada)APPENDPOST)CREATEPUTr overwriteraFrwrr)rr!r"N) r.rSrirrgr`rRrr;rr_r)r&r.rDrErcrrs r0_initiate_uploadzWebHDFile._initiate_uploads !!# $)) )JB(JB"(F; dggmmB LELVL77'' J(?@ $)) 77??&&>3M"N'D  ! ! # 477==6499WuWPVWD GG00j1IJDM r1ct|d}t|j|}||k\s||jk\ry|jj d|j |||z d}|j d|jvrY|jd}|jjj|jj|}|jS|jS)Nrr1OPENF)r`offsetrtrar) maxminrurrgr`r_rr;rrRcontent)r&startendrcrrs r0 _fetch_rangezWebHDFile._fetch_rangesE1 $))S! C<5DII-ggmm 5uu    ${{:.H77??&&tww';';H'EFD<< ;; r1cd|jj|j|jyr3)rrr`rr4s r0commitzWebHDFile.commits  499dkk*r1cN|jj|jyr3)rrr`r4s r0discardzWebHDFile.discards  499r1r) rrrrrrr'r0r2r4rrs@r0rprps&, A"K( +r1rp)loggingrrrr r contextlibrr urllib.parserr9specrrutilsr r getLoggerrVr rprr1r0r;s\   ;3   9 %V Vr I$Ir1