#!/usr/bin/env python3 """通过 Delivery Desk API 创建作品或提交新的验收轮次。""" from __future__ import annotations import argparse import getpass import json import os import sys from datetime import datetime from http.cookies import SimpleCookie from http.cookiejar import CookieJar from urllib.error import HTTPError, URLError from urllib.request import HTTPCookieProcessor, Request, build_opener class ApiError(RuntimeError): pass def request_json(opener, url: str, *, method: str = "GET", data: bytes | None = None, headers: dict[str, str] | None = None): request = Request(url, data=data, method=method, headers=headers or {}) try: with opener.open(request, timeout=15) as response: body = response.read().decode("utf-8") return response.status, json.loads(body) if body else None, response.headers except HTTPError as error: body = error.read().decode("utf-8", errors="replace") try: detail = json.loads(body).get("error", body) except json.JSONDecodeError: detail = body raise ApiError(f"{method} {url} 返回 {error.code}: {detail}") from error except URLError as error: raise ApiError(f"无法连接 {url}: {error.reason}") from error def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="通过 API 创建测试作品或提交新的验收轮次") parser.add_argument("--base-url", default=os.getenv("DELIVERY_DESK_BASE_URL", "http://127.0.0.1:3010")) parser.add_argument("--project-id", type=int, default=None, help="不传时自动选择唯一可访问的项目") parser.add_argument("--work-id", type=int, default=None, help="传入后为该作品提交新的验收轮次") parser.add_argument("--external-id", default=None, help="调用方作品唯一标识，用于幂等创建和找回作品") parser.add_argument("--api-key", default=os.getenv("DELIVERY_DESK_API_KEY")) parser.add_argument("--username", default=os.getenv("DELIVERY_DESK_USERNAME", "operator")) parser.add_argument("--password", default=os.getenv("DELIVERY_DESK_PASSWORD")) parser.add_argument("--title", default=None) parser.add_argument("--description", default="由 tests/api_create_work.py 通过 API 创建。") parser.add_argument("--tags", default="API 测试") parser.add_argument( "--image-url", action="append", dest="image_urls", default=None, help="公开可读的图片 URL，可重复传入；默认使用一张公共占位图", ) return parser.parse_args() def choose_item(items: list[dict], requested_id: int | None, label: str) -> dict: if requested_id is not None: item = next((candidate for candidate in items if int(candidate["id"]) == requested_id), None) if item is None: raise ApiError(f"无权访问或不存在的{label} ID: {requested_id}") return item if len(items) == 1: return items[0] choices = ", ".join(f'{item["id"]}:{item["name"]}' for item in items) or "无" raise ApiError(f"可访问的{label}不是唯一项，请显式传入对应 ID。当前可选：{choices}") def main() -> int: args = parse_args() base_url = args.base_url.rstrip("/") opener = build_opener(HTTPCookieProcessor(CookieJar())) auth_headers: dict[str, str] = {} if args.api_key: auth_headers["Authorization"] = f"Bearer {args.api_key}" else: password = args.password or getpass.getpass(f"请输入账号 {args.username} 的密码: ") login_data = json.dumps({"username": args.username, "password": password}).encode("utf-8") _, _, login_headers = request_json( opener, f"{base_url}/api/auth/login", method="POST", data=login_data, headers={"Content-Type": "application/json"}, ) cookies = SimpleCookie() cookies.load(login_headers.get("Set-Cookie", "")) session = cookies.get("proofing_session") if session is None: raise ApiError("登录成功，但接口没有返回会话 Cookie") auth_headers["Cookie"] = f"proofing_session={session.value}" _, projects, _ = request_json(opener, f"{base_url}/api/projects", headers=auth_headers) project = choose_item(projects, args.project_id, "项目") project_id = int(project["id"]) title = args.title or f"API 测试作品 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" image_urls = args.image_urls or ["https://placehold.co/1200x800/png?text=Delivery+Desk+API+Test"] upload_headers = {**auth_headers, "Content-Type": "application/json"} payload = { "title": title, "description": args.description, "tags": [args.tags] if args.tags else [], "images": image_urls, } if args.work_id is not None: _, current, _ = request_json(opener, f"{base_url}/api/works/{args.work_id}", headers=auth_headers) if int(current["project"]["id"]) != project_id: raise ApiError(f"作品 {args.work_id} 不属于选定的项目") status, work, _ = request_json( opener, f"{base_url}/api/works/{args.work_id}/rounds", method="POST", data=json.dumps(payload).encode("utf-8"), headers=upload_headers, ) if status != 201 or not work or int(work.get("id", 0)) != args.work_id: raise ApiError("新验收轮次接口没有返回目标作品") action = "round_created" external_id = work.get("external_id") else: external_id = args.external_id or f"api-smoke-{datetime.now().strftime('%Y%m%d-%H%M%S')}" create_payload = {**payload, "externalId": external_id} body = json.dumps(create_payload).encode("utf-8") works_url = f"{base_url}/api/projects/{project_id}/works" status, work, _ = request_json(opener, works_url, method="POST", data=body, headers=upload_headers) if status not in (200, 201) or not work or int(work.get("project_id", 0)) != project_id: raise ApiError("接口未返回属于目标项目的作品") _, repeated, _ = request_json(opener, works_url, method="POST", data=body, headers=upload_headers) if int(repeated.get("id", 0)) != int(work["id"]) or not repeated.get("idempotent"): raise ApiError("相同 externalId 的重复请求未通过幂等校验") action = "work_created" if status == 201 else "existing_work_returned" print( json.dumps( { "success": True, "action": action, "group": project.get("group_name"), "project": project.get("name"), "work_id": work.get("id"), "external_id": external_id, "version_number": work.get("version_number"), "title": work.get("title"), }, ensure_ascii=False, indent=2, ) ) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except (ApiError, ValueError) as error: print(f"测试失败: {error}", file=sys.stderr) raise SystemExit(1)