added http auth and upload functions

This commit is contained in:
Luxferre
2025-12-25 15:48:27 +02:00
parent 2e589df792
commit 1d3b5f99f5
2 changed files with 50 additions and 1 deletions
+2
View File
@@ -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`
+47
View File
@@ -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 == '':
@@ -46,4 +54,43 @@ def wget(url, filename='', useragent=None, hdrs={}):
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))