commit 3ec9bf27bd6deb9d8ff3934af900892add1d1d56
Author: archiveanon <>
Date: Thu, 21 Sep 2023 02:33:40 +0000
Initial commit
Diffstat:
2 files changed, 161 insertions(+), 0 deletions(-)
diff --git a/src/gofile/__init__.py b/src/gofile/__init__.py
diff --git a/src/gofile/api.py b/src/gofile/api.py
@@ -0,0 +1,161 @@
+#!/usr/bin/python3
+
+import argparse
+import collections
+import enum
+import pathlib
+from typing import Generic, Iterator, Optional, TypeVar
+
+import msgspec
+
+# normally I'd prefer using standard modules but streaming POST data is really important
+import requests
+from requests_toolbelt.multipart.encoder import MultipartEncoder
+
+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"
+
+
+class GofileServerResult(msgspec.Struct):
+ server: str
+
+
+class GofileUploadResult(msgspec.Struct):
+ download_page: str = msgspec.field(name="downloadPage")
+ code: str = msgspec.field(name="code")
+ parent_folder: str = msgspec.field(name="parentFolder")
+ file_id: str = msgspec.field(name="fileId")
+ file_name: str = msgspec.field(name="fileName")
+ md5_hash: str = msgspec.field(name="md5")
+
+ # 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
+
+
+def _gofile_api_get(*args, 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
+ r = requests.get(*args, **kwargs)
+ result = msgspec.json.decode(r.text, type=GofileServerResponse[type])
+
+ if result.status != GofileStatus.OK:
+ raise Exception(result)
+ return result.data
+
+
+def _gofile_api_post(*args, 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
+ r = requests.post(*args, **kwargs)
+ result = msgspec.json.decode(r.text, type=GofileServerResponse[type])
+
+ if result.status != GofileStatus.OK:
+ raise Exception(result)
+ return result.data
+
+
+def get_upload_server() -> GofileServerResult:
+ return _gofile_api_get("https://api.gofile.io/getServer", type=GofileServerResult)
+
+
+def upload_single(
+ file: pathlib.Path,
+ token: Optional[str] = None,
+ folder_id: Optional[str] = None,
+ server: Optional[str] = None,
+) -> GofileUpload:
+ # we return a GofileUpload instead of a GofileUploadResult so there's consistency between upload_single / upload_multiple
+ if not server:
+ upload_server_result = get_upload_server()
+ server = upload_server_result.server
+
+ post_data = {
+ "file": (file.name, file.open("rb"), "application/octet-stream"),
+ }
+
+ if token:
+ post_data["token"] = token
+ if folder_id:
+ post_data["folderId"] = folder_id
+
+ multipart_data = MultipartEncoder(fields=post_data)
+ upload_result = _gofile_api_post(
+ f"https://{server}.gofile.io/uploadFile",
+ data=multipart_data,
+ headers={"Content-Type": multipart_data.content_type},
+ type=GofileUploadResult,
+ )
+
+ return GofileUpload(file, upload_result)
+
+
+def upload_multiple(
+ files: Iterator[pathlib.Path], token: Optional[str] = None, folder_id: Optional[str] = None
+) -> list[GofileUpload]:
+ # returns a generator of GofileUpload instances in the same order files were given
+ first_file, *other_files = files
+
+ upload_server_result = get_upload_server()
+ server = upload_server_result.server
+
+ first_upload = 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
+
+ for file in other_files:
+ upload = upload_single(file, token, folder_id, server)
+ yield upload
+
+
+def main():
+ # you may save and reuse a guest token
+ # there's no clear indicator on whether or not it expires
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("path", type=pathlib.Path, help="Path to single file or folder")
+
+ args = parser.parse_args()
+
+ upload = None
+
+ if not args.path.exists():
+ print(f"Path '{args.path}' does not exist")
+ return
+ elif args.path.is_dir():
+ results = upload_multiple(args.path.glob("*"))
+ upload, *other_uploads = results
+ elif args.path.is_file():
+ upload = upload_single(args.path)
+
+ print(upload.result)
+
+
+if __name__ == "__main__":
+ main()
