gofile

api.py (6428B)

---

 #!/usr/bin/python3
 
 import asyncio
 import collections
 import enum
 import io
 import pathlib
 import random
 from typing import Any, AsyncIterator, Generic, Iterator, Optional, TypeVar
 
 # normally I'd prefer using standard modules but streaming POST data is really important
 import httpx
 import msgspec
 
 T = TypeVar("T")
 
 GofileUpload = collections.namedtuple("GofileUpload", ["file", "result"])
 
 
 class GofileStatus(enum.Enum):
     OK = "ok"
 
     # header is incorrect
     # this may happen if uploadFile was called without using multipart/form-data
     ERROR_HEADERS = "headersError"
 
     # no file was provided
     # POST data may be malformed
     ERROR_NOFILE = "error-noFile"
 
     # no owner token was provided
     # happens if a folderId was provided without the correct owner
     ERROR_OWNER = "error-owner"
 
     # token is invalid
     # either a malformed token was initially provided or the guest token expired
     ERROR_TOKEN = "error-token"
 
     # token is not associated with a premium user
     # as the API notes, guest tokens cannot access the API methods marked as premium
     ERROR_NOT_PREMIUM = "error-notPremium"
 
     # no server is available to process this request
     ERROR_NO_SERVER = "noServer"
 
 
 class GofileServerResult(msgspec.Struct):
     server: Optional[str] = None
 
 
 class GofileZonedServerResult(msgspec.Struct):
     name: Optional[str] = None
     zone: Optional[str] = None
 
     def to_base_server(self):
         return GofileServerResult(self.name)
 
 
 class GofileServerListResult(msgspec.Struct):
     servers: list[GofileZonedServerResult]
 
 
 class GofileUploadResult(msgspec.Struct):
     download_page: str = msgspec.field(name="downloadPage")
     code: str = msgspec.field(name="parentFolderCode")
     parent_folder: str = msgspec.field(name="parentFolder")
     file_id: str = msgspec.field(name="id")
     file_name: str = msgspec.field(name="name")
     md5_hash: str = msgspec.field(name="md5")
     servers: list[str]
 
     # a guestToken field is provided if no access token was given and no folderID was specified
     guest_token: Optional[str] = msgspec.field(default=None, name="guestToken")
 
 
 class GofileServerResponse(msgspec.Struct, Generic[T]):
     status: GofileStatus
     data: T
 
 
 async def _gofile_api_get(*args, type: type[T], **kwargs) -> T:
     # performs a GET request and extracts the 'data' property from the response as a given type
     # if the status is not 'ok', an exception is raised
     async with httpx.AsyncClient() as client:
         r = await client.get(*args, **kwargs)
 
         # suppress the valid-type error since mypy doesn't cannot use runtime-specialized types
         # but msgspec needs to do so
         # https://stackoverflow.com/a/59636248
         result = msgspec.json.decode(r.text, type=GofileServerResponse[type])  # type: ignore[valid-type]
 
         if result.status != GofileStatus.OK:
             raise Exception(result)
         return result.data
 
 
 async def _gofile_api_post(*args, type: type[T], **kwargs) -> T:
     # performs a POST request and extracts the 'data' property from the response as a given type
     # if the status is not 'ok', an exception is raised
     async with httpx.AsyncClient() as client:
         r = await client.post(*args, **kwargs)
 
         # see typing woes at _gofile_api_get
         result = msgspec.json.decode(r.text, type=GofileServerResponse[type])  # type: ignore[valid-type]
 
         if result.status != GofileStatus.OK:
             raise Exception(result)
         return result.data
 
 
 async def get_upload_server() -> GofileServerResult:
     server_list = await _gofile_api_get(
         "https://api.gofile.io/servers", type=GofileServerListResult
     )
 
     if not server_list.servers:
         return GofileServerResult()
     return random.choice(server_list.servers).to_base_server()
 
 
 async def upload_single(
     file: io.FileIO,
     token: Optional[str] = None,
     folder_id: Optional[str] = None,
     server: Optional[str] = None,
 ) -> GofileUpload:
     """
     Uploads a single file.
 
     :param file: An open file handle.
     :param token: Token used for uploading.  If not specified, the returned
                   ``GofileUpload.result.guest_token`` should be used for subsequent uploads to
                   the same folder.
     :param folder_id: Folder to upload to.  If not specified, the returned
                       ``GofileUpload.result.parent_folder`` should be used for subsequent
                       uploads to the same folder.
     :param server: A specific subdomain to upload to.
     """
     # we return a GofileUpload instead of a GofileUploadResult so there's consistency between upload_single / upload_multiple
 
     if not server:
         upload_server_result = await get_upload_server()
         server = upload_server_result.server
 
     # automatically shorten long file names in small terminals (e.g. split panes)
     while True:
         try:
             filepath = pathlib.Path(str(file.name))
             post_data: dict[str, Any] = {
                 "file": (filepath.name, file, "application/octet-stream"),
             }
 
             if token:
                 post_data["token"] = token
             if folder_id:
                 post_data["folderId"] = folder_id
 
             upload_result = await _gofile_api_post(
                 f"https://{server}.gofile.io/contents/uploadfile",
                 files=post_data,
                 type=GofileUploadResult,
             )
             break
         except httpx.HTTPError as e:
             print(e)
             pass
 
     return GofileUpload(file, upload_result)
 
 
 async def upload_multiple(
     files: Iterator[io.FileIO],
     token: Optional[str] = None,
     folder_id: Optional[str] = None,
     server: Optional[str] = None,
 ) -> AsyncIterator[GofileUpload]:
     """
     Uploads multiple files to the same folder, returning an interator of results.
     """
     first_file, *other_files = files
 
     if not server:
         upload_server_result = await get_upload_server()
         server = upload_server_result.server
 
     first_upload = await upload_single(first_file, token, folder_id, server)
 
     if not token:
         token = first_upload.result.guest_token
     if not folder_id:
         folder_id = first_upload.result.parent_folder
 
     yield first_upload
 
     uploads = [upload_single(file, token, folder_id, server) for file in other_files]
     for upload in asyncio.as_completed(uploads):
         yield await upload