o yq i@sddlmZddlmZddlmZddlZddlmZddlm Z m Z m Z Gd d d e eZ ed d Gd ddZed d GdddZ d'dejdee deefddZdee e fdeefddZe dedee e fdefddZe ddddde de de d ee dee e fdee d!eee e fdefd"d#Ze d'd$e dee e fd!eee e fdefd%d&ZdS)() dataclass)Enum)OptionalN) constants) get_sessionhf_raise_for_statusvalidate_hf_hub_argsc@seZdZdZdZdS) XetTokenTypereadwriteN)__name__ __module__ __qualname__READWRITErra/home/app/PaddleOCR-VL/.venv_paddleocr/lib/python3.10/site-packages/huggingface_hub/utils/_xet.pyr sr T)frozenc@seZdZUeed<eed<dS) XetFileData file_hash refresh_routeN)rrrstr__annotations__rrrrrs  rc@s&eZdZUeed<eed<eed<dS)XetConnectionInfo access_tokenexpiration_unix_epochendpointN)rrrrrintrrrrrs  rresponserreturncCs|durdSz|jtj}tj|jvr|jtjd}n|jtj}Wn ty-YdSw|dur4|ntj}|tj rJ| tj d| d}t ||dS)a^ Parse XET file metadata from an HTTP response. This function extracts XET file metadata from the HTTP headers or HTTP links of a given response object. If the required metadata is not found, it returns `None`. Args: response (`httpx.Response`): The HTTP response object containing headers dict and links dict to extract the XET metadata from. Returns: `Optional[XetFileData]`: An instance of `XetFileData` containing the file hash and refresh route if the metadata is found. Returns `None` if the required metadata is missing. Nurl/)rr) headersrZHUGGINGFACE_HEADER_X_XET_HASHZ$HUGGINGFACE_HEADER_LINK_XET_AUTH_KEYlinksZ&HUGGINGFACE_HEADER_X_XET_REFRESH_ROUTEKeyErrorENDPOINT startswithZHUGGINGFACE_CO_URL_HOMEreplacerstripr)r rrrrrr!parse_xet_file_data_from_responses$     r+r$c CsPz|tj}|tj}t|tj}Wn tttfy YdSwt|||dS)a Parse XET connection info from the HTTP headers or return None if not found. Args: headers (`dict`): HTTP headers to extract the XET metadata from. Returns: `XetConnectionInfo` or `None`: The information needed to connect to the XET storage service. Returns `None` if the headers do not contain the XET connection info. N)rrr) rZ!HUGGINGFACE_HEADER_X_XET_ENDPOINTZ%HUGGINGFACE_HEADER_X_XET_ACCESS_TOKENrZ#HUGGINGFACE_HEADER_X_XET_EXPIRATIONr& ValueError TypeErrorr)r$rrrrrr&parse_xet_connection_info_from_headersBs  r. file_datacCs|jdur tdt|j|S)a Utilizes the information in the parsed metadata to request the Hub xet connection information. This includes the access token, expiration, and XET service URL. Args: file_data: (`XetFileData`): The file data needed to refresh the xet connection information. headers (`dict[str, str]`): Headers to use for the request, including authorization headers and user agent. Returns: `XetConnectionInfo`: The connection information needed to make the request to the xet storage service. Raises: [`~utils.HfHubHTTPError`] If the Hub API returned an error. [`ValueError`](https://docs.python.org/3/library/exceptions.html#ValueError) If the Hub API response is improperly formatted. Nz>The provided xet metadata does not contain a refresh endpoint.)rr,#_fetch_xet_connection_info_with_url)r/r$rrrrefresh_xet_connection_info[s  r1)revisionrparams token_typerepo_id repo_typer2r3c Cs@|dur|ntj}|d|d|d|jd| }t|||S)a Uses the repo info to request a xet access token from Hub. Args: token_type (`XetTokenType`): Type of the token to request: `"read"` or `"write"`. repo_id (`str`): A namespace (user or an organization) and a repo name separated by a `/`. repo_type (`str`): Type of the repo to upload to: `"model"`, `"dataset"` or `"space"`. revision (`str`, `optional`): The revision of the repo to get the token for. headers (`dict[str, str]`): Headers to use for the request, including authorization headers and user agent. endpoint (`str`, `optional`): The endpoint to use for the request. Defaults to the Hub endpoint. params (`dict[str, str]`, `optional`): Additional parameters to pass with the request. Returns: `XetConnectionInfo`: The connection information needed to make the request to the xet storage service. Raises: [`~utils.HfHubHTTPError`] If the Hub API returned an error. [`ValueError`](https://docs.python.org/3/library/exceptions.html#ValueError) If the Hub API response is improperly formatted. Nz/api/zs/z/xet-z-token/)rr'valuer0)r4r5r6r2r$rr3r"rrr(fetch_xet_connection_info_from_repo_infows%" r8r"cCs8tj|||d}t|t|j}|durtd|S)aS Requests the xet connection info from the supplied URL. This includes the access token, expiration time, and endpoint to use for the xet storage service. Args: url: (`str`): The access token endpoint URL. headers (`dict[str, str]`): Headers to use for the request, including authorization headers and user agent. params (`dict[str, str]`, `optional`): Additional parameters to pass with the request. Returns: `XetConnectionInfo`: The connection information needed to make the request to the xet storage service. Raises: [`~utils.HfHubHTTPError`] If the Hub API returned an error. [`ValueError`](https://docs.python.org/3/library/exceptions.html#ValueError) If the Hub API response is improperly formatted. )r$r"r3Nz6Xet headers have not been correctly set by the server.)rgetr r.r$r,)r"r$r3respmetadatarrrr0s  r0)N) dataclassesrenumrtypingrZhttpxrrr r rr rrResponser+dictr.r1r8r0rrrrsx    %    )