o SiF @s dZddlZddlZddlZddlmZmZddlmZddl m Z ddl m Z iZ Gdd d Zd'd d Zd(ddZd(ddZd(ddZd(ddZd(ddZd(ddZd(ddZddZ eeeeeeeeeeeed Zdejvrejdd D]Zeee<qzd!d"Zd(d#d$Zd%d&ZdS))z!Open URLs by calling subcommands.N)PIPEPopen)urlparse) url2pathname)utilsc@steZdZdZdddgdddZdd Zd d Zd d ZddZddZ ddZ ddZ ddZ ddZ ddZdS)PipeaAWrapper class for subprocess.Pipe. This class looks like a stream from the outside, but it checks subprocess status and handles timeouts with exceptions. This way, clients of the class do not need to know that they are dealing with subprocesses. Args: *args: Passed to `subprocess.Pipe` mode: The mode for opening the pipe. timeout: Timeout for closing/waiting. ignore_errors: Don't raise exceptions on subprocess errors. ignore_status: List of status codes to ignore. **kw: Passed to `subprocess.Pipe` Ng @F)modetimeout ignore_errors ignore_statuscOs||_dg||_||_||f|_|ddkr3t|dti||_|jj|_|jdur2t |dn!|ddkrTt|dti||_|jj |_|jdurTt |dd|_ dS)zCreate an IO Pipe.rrstdoutNz: couldn't openwstdin) r r r argsrrprocrstream ValueErrorrstatus)selfr r r r rkwrD/home/ubuntu/.local/lib/python3.10/site-packages/webdataset/gopen.py__init__+s          z Pipe.__init__cCsd|jdS)z2Return a string representation of the Pipe object.z)rrrrr__str__Esz Pipe.__str__cCs"|j}|dur|dSdS)z'Poll the process and handle any errors.N)rpollwait_for_child)rrrrr check_statusIs  zPipe.check_statusc Csttjdd}|jdur|rdS|j|_|r6td|jdtd|jj d|j dt t j d|j|jvrM|jsOt|j d |jd t dSdS) z>Check the status variable and raise an exception if necessary. GOPEN_VERBOSErNz pipe exit [ :z] filez: exit z (read) )intosenvirongetrrwaitprintgetpidpidrinfosysstderrr r IOError)rverboserrrrOs ,zPipe.wait_for_childcO|jj|i|}||S)zWrap stream.read and checks status. Args: *args: Arguments to pass to stream.read **kw: Keyword arguments to pass to stream.read Returns: The result of stream.read )rreadr rrrresultrrrr4^ z Pipe.readcOr3)zWrap stream.write and checks status. Args: *args: Arguments to pass to stream.write **kw: Keyword arguments to pass to stream.write Returns: The result of stream.write )rwriter r5rrrr8lr7z Pipe.writecOs*|jj|i|}|j|_||S)zWrap stream.readLine and checks status. Args: *args: Arguments to pass to stream.readLine **kw: Keyword arguments to pass to stream.readLine Returns: The result of stream.readLine )rreadLinerrrr r5rrrr9zs z Pipe.readLinecCs2|jjs|j|j|j|_|dSdS)z>Wrap stream.close, wait for the subprocess, and handle errors.N)rclosedcloserr*r rrrrrrr;s   z Pipe.closecCs|S)Context handler.rrrrr __enter__szPipe.__enter__cCs |dS)r<N)r;)retypevalue tracebackrrr__exit__s z Pipe.__exit__cCs$z|WdStyYdSw)zClose the stream upon delete. This is a fallback for when users can't use context managers. We catch all exceptions since __del__ should never raise exceptions during garbage collection. N)r; Exceptionrrrr__del__s  z Pipe.__del__)__name__ __module__ __qualname____doc__rrr rr4r8r9r;r=rArCrrrrrs"  rcCsJt|tsdS|dur||_|dur||_|dur||_|dur#||_dS)aSet options for Pipes. This function can be called on any stream. It will set pipe options only when its argument is a pipe. Args: obj: Any kind of stream timeout: Desired timeout ignore_errors: Desired ignore_errors setting ignore_status: Desired ignore_status setting handler: Desired error handler Returns: True if options were set, False otherwise FNT) isinstancerr r r handler)objr r r rIrrr set_optionss rKrb cCs"|dr tdd|}t||S)zOpen a file. This works for local files; path names only. Args: url: URL to be opened mode: Mode to open it with bufsize: Requested buffer size Returns: An opened file object zfile:z ^file://?) startswithresubopen)urlr bufsizerrr gopen_files  rUcCsV|dsJtjrtd|dd}|ddvr$t||d|dgd St|d ) aUse gopen to open a pipe. This function deliberately uses shell=True with the pipe URL to enable shell command execution directly from URLs. This is an intentional design feature that allows users to construct processing pipelines using shell commands via the pipe: URL scheme. The purpose is to enable flexible data processing directly within data loading pipelines. Note: This feature requires careful use with trusted input sources only, as it will execute arbitrary shell commands specified in the URL. Args: url: A pipe: URL mode: Desired mode bufsize: Desired buffer size Returns: A Pipe object Raises: ValueError: If the mode is unknown zpipe:z8gopen_pipe: unsafe_gopen is False, cannot open pipe URLsNr)r rT)r shellrTr : unknown mode)rOrenforce_securityrr)rSr rTcmdrrr gopen_pipes  r\c Cs~|ddkrdddddddd d d |g }t|||d d gdS|ddkr8dd d ddd dd|g }t|||d dgdSt|d)zOpen a URL with `curl`. Args: url: URL (usually, http:// etc.) mode: File mode bufsize: Buffer size Returns: A Pipe object Raises: ValueError: If the mode is unknown rr curl--connect-timeout30--retry --retry-delay2-f-s-LrWr rTr rz-XPUTz-T-rYrrrSr rTcmd_argsrrr gopen_curls"  rncCsb|ddkrtdd|}ddd|g}t|||dd gd S|dd kr*t|d t|d )zOpen a URL with `curl`. Args: url: URL (usually, http:// etc.) mode: File mode bufsize: Buffer size Returns: A Pipe object Raises: ValueError: If the mode is write or unknown rr z (?i)^htgs://zgs://r]rdrerWrfrgr: cannot writerY)rPrQrrrlrrr gopen_htgss   rpc Csddlm}m}m}|ddkrA||}||j|j|j|jd}|}dddddd d d d d dd||g } t | ||ddgdS|ddkrNt |dt |d)zOpen a URL with `curl`. Args: url: URL (usually, hf:// etc.) mode: File mode bufsize: Buffer size Returns: A Pipe object Raises: ValueError: If the mode is write or unknown r) HfFileSystem get_token hf_hub_urlr )repo_idfilename repo_typerevisionr]r^r_r`rarbrcrdrez-HzAuthorization:Bearer rWrfrgrrorY) huggingface_hubrqrrrs resolve_pathrt path_in_reporvrwrr) rSr rTrqrrrs resolved_pathhttp_urltokenrmrrrgopen_hf;sB   r~cCsd|ddkrdd|g}t|||ddgdS|ddkr+dd d |g}t|||dd gdSt|d ) zOpen a URL with `gsutil`. Args: url: URL (usually, gs:// etc.) mode: File mode bufsize: Buffer size Returns: A Pipe object Raises: ValueError: If the mode is unknown rr gsutilcatrWrfrgrcprirjrYrkrlrrr gopen_gsutilos"    rcCsf|ddkrdd|dg}t|||ddgdS|dd kr,dd d|g}t|||dd gdSt|d ) zOpen a URL with `ais`. Args: url: URL (usually, ais:// etc.) mode: File mode bufsize: Buffer size Returns: A Pipe object Raises: ValueError: If the mode is unknown rr aisr)rirWrfrgrputrjrYrkrlrrr gopen_aiss"    rcOst|d)zRaise a value error. Args: url: URL *args: Other arguments **kw: Other keywords Raises: ValueError: Always raised with the URL and a message z: no gopen handler defined)r)rSrrrrr gopen_errors r) __default__pipehttphttpsrsftpftpsscpgshtgshf USE_AIS_FORr#cCsd}ttjdd}|tjvr|Stjrtdtj|dD]'}|dd\}}t d|||}||krH|rDt d |d ||Sq!|S) aYRewrite the URL based on environment variables. This function checks for URL rewrite rules defined in the GOPEN_REWRITE environment variable and applies them to the given URL. The rewrite rules allow for flexible modification of URLs before they are processed by the gopen system. The GOPEN_REWRITE environment variable should contain one or more rewrite rules separated by semicolons. Each rule consists of two parts separated by an equals sign: a pattern to match at the start of the URL, and a replacement string. Format of GOPEN_REWRITE: GOPEN_REWRITE="pattern1=replacement1;pattern2=replacement2;..." The function applies these rules in order, stopping at the first match. If a match is found, the pattern is replaced with the corresponding replacement at the start of the URL. The GOPEN_VERBOSE environment variable can be set to control logging. If GOPEN_VERBOSE is set to a non-zero value, the function will print information about any URL rewrites that occur. Note: This function performs basic URL rewriting without validation. Validation of the resulting URL for security concerns (such as path traversal) is the responsibility of the caller. Args: url (str): The original URL to potentially rewrite. Returns: str: The rewritten URL if a rewrite rule matches, otherwise the original URL. Example: If GOPEN_REWRITE is set to "http://old.com/=http://new.com/;ftp://=http://" and the input URL is "http://old.com/file.txt", the function will return "http://new.com/file.txt". GOPEN_REWRITEr!rzSrewrite_url: unsafe_gopen is False, cannot rewrite URLs using environment variables;=r^zGOPEN REWRITE z -> ) r&r'r(r)rrZrsplitrPrQr+)rSnamer2r kvnurlrrr rewrite_urls' rcKsttjdd}|rtd|ttjd|dvsJ||dkr7|dkr(tjj S|dkr0tj j St d |t |}t |}|jd kr[tjrKt d ttjd d }t|||dS|jdkrztjrgt d ttjd d }tt|j||dStd}t|j|}||||fi|S)aOpen the URL using various schemes and protocols. This function provides a unified interface for opening resources specified by URLs, supporting multiple schemes and protocols. It uses the `gopen_schemes` dispatch table to handle different URL schemes. Built-in support is provided for the following schemes: - pipe: for opening named pipes - file: for local file system access - http, https: for web resources - sftp, ftps: for secure file transfer - scp: for secure copy protocol When no scheme is specified in the URL, it is treated as a local file path. Environment Variables: - GOPEN_VERBOSE: Set to a non-zero value to enable verbose logging of file operations. Format: GOPEN_VERBOSE=1 - USE_AIS_FOR: Specifies which cloud storage services should use AIS (and its cache) for access. Format: USE_AIS_FOR=aws:gs:s3 - GOPEN_BUFFER: Sets the buffer size for file operations (in bytes). Format: GOPEN_BUFFER=8192 Args: url (str): The source URL or file path to open. mode (str): The mode for opening the resource. Only "rb" (read binary) and "wb" (write binary) are supported. bufsize (int): The buffer size for file operations. Default is 8192 bytes. **kw: Additional keyword arguments to pass to the underlying open function. Returns: file-like object: An opened file-like object for the specified resource. Raises: ValueError: If an unsupported mode is specified. Other exceptions may be raised depending on the specific handler used for the URL scheme. Note: - For stdin/stdout operations, use "-" as the URL. - The function applies URL rewriting based on the GOPEN_REWRITE environment variable before processing. r!rGOPENr$)rLwbrirLrz unknown mode rNz5gopen: unsafe_gopen is False, cannot open local files GOPEN_BUFFER) bufferingr%r)r&r'r(r)r+r.r/r0rbufferrrrrschemerrZrRrpath gopen_schemes)rSr rTrr2prrIrrrgopen s2*  rcKst|dfi|S)zOpen url with gopen and mode "rb". Args: url: Source URL **kw: Other keywords forwarded to gopen Returns: An opened file-like object in read mode rL)r)rSrrrrreaderRs r)NNNN)rLrM) rGr'rPr/ subprocessrr urllib.parserurllib.requestrrNrr.rrKrUr\rnrpr~rrrdictrr(rprotorrrrrrrsP       & "  4 ""   7 F