diff --git a/README.md b/README.md index b208fea..72e7e1e 100644 --- a/README.md +++ b/README.md @@ -158,7 +158,9 @@ Exports the following functions: - `http_set_ua(ua_str='Mozilla/5.0')`: set a default HTTP user agent string - `http_fetch(url, useragent=None, hdrs={})`: fetch text content from a URL into a string (returns `(status_code, content)` tuple) +- `auth_header(username, password)`: get an dictionary fragment with the corresponding HTTP Basic Auth header according to the username and password - `wget(url, filename='', useragent=None, hdrs={})`: download a file into a local path (returns `(status_code, filename)` tuple) +- `wput(url, file_path, field_name="file", method="POST", format="x-www-form-urlencoded", headers=None)`: upload a file from a local path in the `x-www-form-urlencoded` or JSON format (returns `(status_code, content)` tuple) ### `deck.llm` diff --git a/deck/http.py b/deck/http.py index e4cf0ce..47cd583 100644 --- a/deck/http.py +++ b/deck/http.py @@ -11,6 +11,8 @@ try: except: import requests +import json, binascii + DEFAULT_UA = 'Mozilla/5.0' def http_set_ua(ua_str='Mozilla/5.0'): """Set a default user agent""" @@ -28,6 +30,12 @@ def http_fetch(url, useragent=None, hdrs={}): except: return (404, '') +def auth_header(auth_username=None, auth_password=None): + """Get the value for HTTP Basic Auth header""" + auth_str = f"{auth_username}:{auth_password}" + auth_b64 = binascii.b2a_base64(auth_str.encode("utf-8")).decode("utf-8").strip() + return {"Authorization": f"Basic {auth_b64}"} + def wget(url, filename='', useragent=None, hdrs={}): """Download a file""" if filename == '': @@ -45,5 +53,44 @@ def wget(url, filename='', useragent=None, hdrs={}): f.write(r.content) else: filename = '' return (r.status_code, filename) - + +def wput(url, file_path, field_name="file", method="POST", format="x-www-form-urlencoded", headers=None): + """Upload a file""" + if headers is None: + headers = {} + method = method.upper() + try: + with open(file_path, "rb") as f: + file_data = f.read() + response = None + # Handle x-www-form-urlencoded (Key-Value pair) + if format.lower() == "x-www-form-urlencoded": + payload = {field_name: file_data} + response = requests.request(method, url, data=payload, headers=headers) + # Handle JSON format + elif format.lower() == "json": + if "Content-Type" not in headers: + headers["Content-Type"] = "application/json" + try: + content = file_data.decode("utf-8") + is_base64 = False + except UnicodeError: + # Encode binary data to Base64 and decode to string for JSON serialization + content = binascii.b2a_base64(file_data).decode("utf-8").strip() + is_base64 = True + payload_dict = { + field_name: content, + "encoding": "base64" if is_base64 else "text" + } + payload = json.dumps(payload_dict) + response = requests.request(method, url, data=payload, headers=headers) + else: + raise ValueError("Unsupported format. Use 'x-www-form-urlencoded' or 'json'.") + return (response.status_code, response.text) + except OSError as e: + print(f"File error: {e}") + return (0, str(e)) + except Exception as e: + print(f"Upload error: {e}") + return (0, str(e))