t-deckard/deck/http.py

# 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 url_escape(s):
  """Prepare parameter value data for x-www-form-urlencoded format"""
  return ''.join(c if c.isalpha() or c.isdigit() else '%%%02x' % ord(c) for c in s)

def form_encode(params):
  """Prepare parameter object for x-www-form-urlencoded format"""
  out = []
  for param, val in params.items():
    out.append(url_escape(param) + '=' + url_escape(val))
  return '&'.join(out)

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))