#!/usr/bin/env python3 """Python 3.9+; credentials stay on your server. No pip packages required.""" import base64 import http.client import json import os import sys import time import urllib.error import urllib.parse import urllib.request class NoRedirects(urllib.request.HTTPRedirectHandler): def redirect_request(self, request, response, code, message, headers, new_url): return None def uploadCourse(filename): api_key = os.environ.get("SCORMBOX_API_KEY") api_secret = os.environ.get("SCORMBOX_API_SECRET") idempotency_key = os.environ.get("SCORMBOX_IDEMPOTENCY_KEY") if not api_key or not api_secret or not idempotency_key: raise RuntimeError( "Set SCORMBOX_API_KEY, SCORMBOX_API_SECRET and a stable " "SCORMBOX_IDEMPOTENCY_KEY for this content version." ) base_url = os.environ.get("SCORMBOX_API_BASE_URL") or "https://scormbox.com/api/v1" service = urllib.parse.urlsplit(base_url) if service.scheme != "https" and not ( service.scheme == "http" and service.hostname == "localhost" ): raise RuntimeError("Use HTTPS for server credentials, or HTTP localhost for development.") if not service.hostname or service.username or service.password or service.query or service.fragment: raise RuntimeError("Use an API base URL without credentials, a query or a fragment.") base_url = base_url.rstrip("/") authorization = "Basic " + base64.b64encode( (api_key + ":" + api_secret).encode("utf-8") ).decode("ascii") opener = urllib.request.build_opener(NoRedirects()) retryable = {409, 429, 500, 502, 503, 504} def request(path, method="GET", body=None, headers=None, deadline=None): binary = isinstance(body, (bytes, bytearray, memoryview)) data = body if binary else None if body is None else json.dumps(body).encode("utf-8") request_headers = { "Authorization": authorization, "Content-Type": "application/octet-stream" if binary else "application/json", } if binary: request_headers["Content-Length"] = str(len(data)) request_headers.update(headers or {}) # Creation has an idempotency key; parts, completion and reads are safe to retry. for retry in range(5): remaining = 120 if deadline is None else deadline - time.monotonic() if remaining <= 0: raise RuntimeError("Preparation timed out; check the course_id before trying again.") try: operation = urllib.request.Request( base_url + path, data=data, headers=request_headers, method=method ) try: response = opener.open(operation, timeout=min(120, remaining)) except urllib.error.HTTPError as error: response = error with response: status = response.status payload = response.read() except (urllib.error.URLError, OSError, http.client.HTTPException) as error: if retry == 4: raise RuntimeError("Request failed after retries: " + str(error)) from error else: if 300 <= status < 400: raise RuntimeError("HTTP " + str(status) + ": redirects are refused.") try: result = json.loads(payload) if not isinstance(result, dict): raise ValueError("Expected a JSON object") except (ValueError, UnicodeDecodeError): if 200 <= status < 300: raise RuntimeError("HTTP " + str(status) + ": invalid JSON response.") result = {} if 200 <= status < 300: return result if status not in retryable or retry == 4: raise RuntimeError("HTTP " + str(status) + ": " + str(result.get("error", "Request failed."))) delay = 2 ** retry # Four retries: 1, 2, 4 and 8 seconds. if deadline is not None: delay = min(delay, max(0, deadline - time.monotonic())) time.sleep(delay) size = os.path.getsize(filename) properties = { "filename": os.path.basename(filename), "size": size, "shareable": os.environ.get("SCORMBOX_SHAREABLE") == "true", } if os.environ.get("SCORMBOX_TITLE"): properties["title"] = os.environ["SCORMBOX_TITLE"] course = request("/courses", "POST", properties, {"Idempotency-Key": idempotency_key}) if course["status"] in {"awaiting_upload", "uploading"}: upload = course["upload"] saved = {part["part_number"] for part in upload["parts"]} upload_path = "/uploads/" + urllib.parse.quote(upload["upload_id"], safe="") with open(filename, "rb", buffering=0) as file: for part_number in range(1, upload["part_count"] + 1): if part_number in saved: continue offset = (part_number - 1) * upload["part_size"] length = min(upload["part_size"], size - offset) chunk = bytearray(length) file.seek(offset) used = 0 while used < length: count = file.readinto(memoryview(chunk)[used:]) if not count: raise RuntimeError("The package changed while being read.") used += count request(upload_path + "/parts/" + str(part_number), "PUT", chunk) del chunk # Release this part before allocating the next one. print("Uploaded part " + str(part_number) + " of " + str(upload["part_count"]), flush=True) request(upload_path + "/complete", "POST", {}) course_id = course["course_id"] deadline = time.monotonic() + 20 * 60 while True: if time.monotonic() >= deadline: raise RuntimeError("Preparation is still running. Check course_id " + course_id + " before trying again.") course = request("/courses/" + urllib.parse.quote(course_id, safe=""), deadline=deadline) if course["status"] == "ready": break if course["status"] == "failed": raise RuntimeError(course.get("error") or "Package preparation failed.") preparation = course.get("preparation") or {} print( "Course " + course["status"] + ": " + str(preparation.get("completed_files", 0)) + " of " + str(preparation.get("total_files", 0)) + " files", flush=True, ) time.sleep(min(2, max(0, deadline - time.monotonic()))) print(json.dumps({ "course_id": course["course_id"], "format": course["format"], "management_url": course["management_url"], "learner_url": course["learner_url"], }, indent=2)) if __name__ == "__main__": try: if len(sys.argv) != 2: raise RuntimeError("Usage: python3 " + os.path.basename(sys.argv[0]) + " /path/to/course.zip") uploadCourse(sys.argv[1]) except (OSError, RuntimeError, KeyError, TypeError, ValueError) as error: print("Error: " + str(error), file=sys.stderr) sys.exit(1)