"""Python 3.10+; submit once, resume a saved task, download from this site's API.""" import argparse import json import os import re import sys import time import urllib.error import urllib.parse import urllib.request from pathlib import Path class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None # Never forward the API key to another URL. class ApiError(Exception): def __init__(self, status): self.status = status super().__init__(f"HTTP {status}; check the task and usage logs") class VideoClient: def __init__(self, key, base="https://api.manxiaobai.online"): if base.rstrip("/") not in ("https://api.manxiaobai.online", "https://openpix.cc"): raise ValueError("Use one of the documented site origins") if not key or "\n" in key or "\r" in key: raise ValueError("Set MXB_API_KEY in your server environment") self.base, self.key = base.rstrip("/"), key self.opener = urllib.request.build_opener(NoRedirect()) def open(self, path, payload=None, timeout=30): data = None if payload is None else json.dumps(payload).encode("utf-8") headers = {"Authorization": "Bearer " + self.key} if data is not None: headers["Content-Type"] = "application/json" request = urllib.request.Request(self.base + path, data=data, headers=headers) try: return self.opener.open(request, timeout=timeout) except urllib.error.HTTPError as error: error.close() raise ApiError(error.code) from None def json(self, path, payload=None): with self.open(path, payload, timeout=120 if payload is not None else 30) as response: result = json.load(response) if not isinstance(result, dict): raise ValueError("Unexpected API response; check logs before submitting again") return result @staticmethod def task_path(task): if not isinstance(task, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,200}", task): raise ValueError("Invalid task ID") return "/v1/videos/" + urllib.parse.quote(task, safe="") def wait(self, task, seconds): deadline = time.monotonic() + seconds while time.monotonic() < deadline: try: result = self.json(self.task_path(task)) except (urllib.error.URLError, TimeoutError): time.sleep(5) continue except ApiError as error: if error.status not in (429, 500, 502, 503, 504): raise time.sleep(5) continue if result.get("id") != task: raise ValueError("Response task ID mismatch; saved task remains available") status = str(result.get("status", "")).lower() if status in ("completed", "succeeded", "success"): return result if status in ("failed", "failure", "cancelled", "canceled", "expired"): raise ValueError("Task failed; inspect its error and refund in the site logs") if status not in ("queued", "pending", "submitted", "in_progress", "running", "processing"): raise ValueError("Unknown task state; keep the ID and check site logs") time.sleep(5) raise TimeoutError("Waiting stopped; run again with the saved state or --task to resume") def download(self, task, output): temporary = output.with_name(output.name + ".part") try: with self.open(self.task_path(task) + "/content", timeout=180) as response: if response.status != 200: raise ValueError("Unexpected download response") first = response.read(32) if b"ftyp" not in first: raise ValueError("The content response is not an MP4 file") with temporary.open("wb") as stream: stream.write(first) size = len(first) while chunk := response.read(1024 * 1024): size += len(chunk) if size > 512 * 1024 * 1024: raise ValueError("Download exceeds the example's 512 MiB limit") stream.write(chunk) temporary.replace(output) except BaseException: temporary.unlink(missing_ok=True) raise def get_task(client, state, payload): if state.exists(): saved = json.loads(state.read_text(encoding="utf-8")) if saved.get("base") != client.base: raise ValueError("Saved origin differs; use the same --base as the original task") if not saved.get("task_id"): raise ValueError("Submission result unknown: check site logs before creating another task") client.task_path(saved["task_id"]) return saved["task_id"] # Reserve the state file BEFORE sending POST. A crash must not trigger a duplicate. with state.open("x", encoding="utf-8") as stream: json.dump({"base": client.base, "status": "submission_unknown"}, stream) stream.flush() os.fsync(stream.fileno()) result = client.json("/v1/videos", payload) task = result.get("id") client.task_path(task) print("Accepted task:", task, flush=True) temporary = state.with_name(state.name + ".tmp") temporary.write_text(json.dumps({"base": client.base, "task_id": task}), encoding="utf-8") temporary.replace(state) return task def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--base", default="https://api.manxiaobai.online") parser.add_argument("--prompt", default="A blue paper boat floating gently on calm water") parser.add_argument("--model", choices=["dreamina-seedance-2-0-mini-260615", "dreamina-seedance-2-0-fast-260128"], default="dreamina-seedance-2-0-mini-260615") parser.add_argument("--request", type=Path, help="UTF-8 JSON request file") parser.add_argument("--state", type=Path, default=Path("seedance-task.json")) parser.add_argument("--task", help="Existing task ID; never submits a generation") parser.add_argument("--output", type=Path, default=Path("seedance-result.mp4")) parser.add_argument("--wait", type=int, default=1800, help="Polling deadline in seconds") args = parser.parse_args() client = VideoClient(os.environ.get("MXB_API_KEY", ""), args.base) payload = None if not args.task and not args.state.exists(): payload = json.loads(args.request.read_text(encoding="utf-8-sig")) if args.request else { "model": args.model, "prompt": args.prompt, "seconds": "4", "resolution": "480p", "ratio": "16:9", "generate_audio": False, "watermark": False, } if not isinstance(payload, dict): raise ValueError("Request file must contain a JSON object") task = args.task or get_task(client, args.state, payload) client.task_path(task) print("Querying task:", task, flush=True) client.wait(task, args.wait) client.download(task, args.output) print("Saved video:", args.output) if __name__ == "__main__": try: main() except (OSError, ValueError, ApiError) as error: # Do not dump request headers, raw API responses or signed media addresses. print(f"Stopped ({type(error).__name__}). Keep the task state and check site logs; do not auto-resubmit.", file=sys.stderr) sys.exit(1)