"""糯米 API · Python 3 标准库媒体示例；导入不会创建任务或扣费。 在服务端保存 API Key。提交前保存业务编号、用户已确认的报价， 成功提交后保存 task_id；程序重启后用这些编号恢复查询。 """ from __future__ import annotations import json import math import os import re import time from urllib.error import HTTPError, URLError from urllib.parse import quote, urlencode, urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener class NuomiApiError(Exception): def __init__(self, message, **fields): super().__init__(message) self.status = None self.code = "request_error" self.task_id = None self.client_request_id = None self.uncertain = False self.requires_confirmation = False self.new_request_id_required = False self.__dict__.update(fields) class _NoRedirect(HTTPRedirectHandler): def redirect_request(self, request, response, code, message, headers, new_url): # Never forward a customer's Authorization header to another address. return None def _task_id(task): return (task.get("id") or task.get("task_id")) if isinstance(task, dict) else None def _result_url(task): metadata = task.get("metadata") metadata = metadata if isinstance(metadata, dict) else {} for value in (task.get("video_url"), task.get("url"), task.get("result_url"), metadata.get("url")): if isinstance(value, str) and value.lower().startswith(("https://", "http://")): return value return None class NuomiMediaClient: def __init__(self, base_url=None, api_key=None, request_timeout=15, *, opener=None): base_url = base_url or os.environ.get("NUOMI_BASE_URL", "") api_key = api_key or os.environ.get("NUOMI_API_KEY", "") try: parsed = urlsplit(base_url) valid = parsed.scheme in {"http", "https"} and parsed.hostname and parsed.path in {"", "/"} and not ( parsed.username or parsed.password or parsed.query or parsed.fragment or any(char in base_url for char in "?#\\") or any(char.isspace() or ord(char) < 32 for char in base_url)) parsed.port except (TypeError, ValueError): valid = False if not valid: raise ValueError("Base URL 只填糯米 API 的 HTTP(S) 根域名与端口，不含 /v1、账号、查询参数或片段。") if not isinstance(api_key, str) or not api_key.strip(): raise ValueError("请设置服务端 NUOMI_API_KEY。") if (type(request_timeout) not in {int, float} or not math.isfinite(request_timeout) or request_timeout <= 0): raise ValueError("请求超时必须大于零。") self.base_url = base_url.rstrip("/") self.api_key = api_key.strip() self.request_timeout = request_timeout self._open = opener or build_opener(_NoRedirect).open def request(self, path, *, method="GET", body=None, task_id=None, client_request_id=None): submission = method == "POST" headers = {"Authorization": "Bearer " + self.api_key, "Accept": "application/json"} encoded = None if body is not None: headers["Content-Type"] = "application/json" encoded = json.dumps(body, ensure_ascii=False).encode("utf-8") request = Request(self.base_url + path, data=encoded, headers=headers, method=method) try: with self._open(request, timeout=self.request_timeout) as response: status, content = response.status, response.read() except HTTPError as error: status, content = error.code, error.read() error.close() except (URLError, TimeoutError, OSError): raise NuomiApiError( "提交结果不确定；保留原业务编号并先查询原任务。" if submission else "查询暂时失败；保留任务编号继续核对。", code="request_uncertain", uncertain=submission, task_id=task_id, client_request_id=client_request_id) from None try: data = json.loads(content) except (ValueError, UnicodeError): data = None if not 200 <= status < 300: detail = (data.get("error") or data.get("detail")) if isinstance(data, dict) else None fields = detail if isinstance(detail, dict) else {} code = fields.get("code") or "http_error" message = detail if isinstance(detail, str) else fields.get("message") or "HTTP " + str(status) raise NuomiApiError(message, status=status, code=code, task_id=task_id, client_request_id=client_request_id, uncertain=submission and (status >= 500 or status == 408), requires_confirmation=code == "model_price_changed", price_revision=fields.get("price_revision"), model=fields.get("model")) if not isinstance(data, dict): raise NuomiApiError("接口返回内容暂不可读；保留原编号核查。", status=status, code="invalid_response", uncertain=submission, task_id=task_id, client_request_id=client_request_id) return data def models(self): return self.request("/v1/models") def balance(self): return self.request("/v1/balance") def lookup(self, client_request_id): identifier = str(client_request_id or "").strip() if not identifier: raise ValueError("请输入原业务编号。") try: return self.request("/v1/tasks/by-client-request-id?" + urlencode({"client_request_id": identifier}), client_request_id=identifier) except NuomiApiError as error: if error.status == 404: return None raise def submit(self, payload): if not isinstance(payload, dict): raise ValueError("payload 必须是字典。") identifier = payload.get("client_request_id") identifier = identifier.strip() if isinstance(identifier, str) else "" if not identifier or len(identifier) > 191 or any(ord(char) < 32 for char in identifier): raise ValueError("请提供稳定的 client_request_id，最多 191 字符。") revision = payload.get("price_revision") if not isinstance(revision, str) or not re.fullmatch(r"[0-9a-f]{64}", revision): raise ValueError("请先从 /v1/models 读取并确认模型报价，再传入对应的 price_revision。") try: task = self.request("/v1/videos", method="POST", body={**payload, "client_request_id": identifier}, client_request_id=identifier) if not _task_id(task): raise NuomiApiError("提交响应缺少任务编号，请按原业务编号查询。", code="invalid_response", uncertain=True, client_request_id=identifier) return task except NuomiApiError as error: if not error.uncertain: raise # Only GET recovery; never silently issue a second paid POST. recovered = self.lookup(identifier) if _task_id(recovered): return recovered raise def get_task(self, task_id): if not isinstance(task_id, str) or not task_id: raise ValueError("请输入已保存的任务编号。") # Deferred installations may start this existing task on its first GET. return self.request("/v1/videos/" + quote(task_id, safe=""), task_id=task_id) def wait_for_task(self, task_id, *, timeout_seconds=600, interval_seconds=3, client_request_id=None, on_progress=None): if (type(timeout_seconds) not in {int, float} or not math.isfinite(timeout_seconds) or timeout_seconds <= 0 or type(interval_seconds) not in {int, float} or not math.isfinite(interval_seconds) or interval_seconds < 1): raise ValueError("请设置正数等待时限，轮询间隔至少 1 秒。") deadline, last_error = time.monotonic() + timeout_seconds, None while time.monotonic() < deadline: task = None try: task = self.get_task(task_id) except NuomiApiError as error: error.client_request_id = client_request_id if error.status in {401, 403}: raise # Fix credentials; do not create a new order. last_error = error if task is not None: if on_progress is not None: on_progress(task) status, url = str(task.get("status", "")).lower(), _result_url(task) if status in {"completed", "succeeded"} and url: return {"task": task, "url": url} if status in {"failed", "error", "cancelled", "canceled"}: detail = task.get("error") detail = detail if isinstance(detail, dict) else {} code = detail.get("code") or "media_task_failed" raise NuomiApiError(detail.get("message") or "任务已明确失败；请先查看错误，再决定是否重新生成。", code=code, task_id=task_id, client_request_id=task.get("client_request_id") or client_request_id, requires_confirmation=code == "model_price_changed", new_request_id_required=True) # Query errors, unknown states, or completed without a URL stay pending. remaining = deadline - time.monotonic() if remaining > 0: time.sleep(min(interval_seconds, remaining)) raise NuomiApiError("等待已到时限；保存编号稍后继续查询，不要新建重复任务。", code="poll_timeout", task_id=task_id, client_request_id=client_request_id, uncertain=True, last_query_status=last_error.status if last_error else None, last_query_code=last_error.code if last_error else None)