98 lines
3.1 KiB
Python
98 lines
3.1 KiB
Python
# High-level HTTP protocol interaction
|
|
# Created by Luxferre in 2025, released into public domain
|
|
|
|
# init requests library in a platform-agnostic way
|
|
requests = None
|
|
try:
|
|
from deck import net
|
|
status, stext, myip = net.wifi_connect_wait()
|
|
if status:
|
|
requests = net.init_requests()
|
|
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"""
|
|
global DEFAULT_UA
|
|
DEFAULT_UA = ua_str
|
|
|
|
def http_fetch(url, useragent=None, hdrs={}):
|
|
"""Fetch text content from URL"""
|
|
if useragent is None:
|
|
useragent = DEFAULT_UA
|
|
hdrs['User-Agent'] = useragent
|
|
try:
|
|
r = requests.get(url, headers=hdrs)
|
|
return (r.status_code, r.text)
|
|
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 == '':
|
|
filename = url.split('/')[-1]
|
|
if useragent is None:
|
|
useragent = DEFAULT_UA
|
|
hdrs['User-Agent'] = useragent
|
|
r = requests.get(url, stream=True, headers=hdrs)
|
|
if r.status_code >= 200 and r.status_code < 400:
|
|
with open(filename, 'wb') as f:
|
|
try:
|
|
for chunk in r.iter_content(chunk_size=8192):
|
|
f.write(chunk)
|
|
except AttributeError:
|
|
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={}, useragent=None):
|
|
"""Upload a file"""
|
|
if useragent is None:
|
|
useragent = DEFAULT_UA
|
|
headers['User-Agent'] = useragent
|
|
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))
|
|
|