175 lines
7.7 KiB
Python
175 lines
7.7 KiB
Python
|
#!/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("--collection-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"])
|
|||
|
|
_, collections, _ = request_json(
|
|||
|
|
opener,
|
|||
|
|
f"{base_url}/api/projects/{project_id}/collections",
|
|||
|
|
headers=auth_headers,
|
|||
|
|
)
|
|||
|
|
collection = choose_item(collections, args.collection_id, "作品交付集")
|
|||
|
|
collection_id = int(collection["id"])
|
|||
|
|
if int(collection["project_id"]) != project_id:
|
|||
|
|
raise ApiError(f"作品交付集 {collection_id} 不属于项目 {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/notes/{args.work_id}", headers=auth_headers)
|
|||
|
|
if int(current["project"]["id"]) != project_id or int(current["collection"]["id"]) != collection_id:
|
|||
|
|
raise ApiError(f"作品 {args.work_id} 不属于选定的项目和作品交付集")
|
|||
|
|
status, work, _ = request_json(
|
|||
|
|
opener,
|
|||
|
|
f"{base_url}/api/notes/{args.work_id}/versions",
|
|||
|
|
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 = "version_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, "collectionId": collection_id, "externalId": external_id}
|
|||
|
|
body = json.dumps(create_payload).encode("utf-8")
|
|||
|
|
status, work, _ = request_json(opener, f"{base_url}/api/notes", method="POST", data=body, headers=upload_headers)
|
|||
|
|
if status not in (200, 201) or not work or int(work.get("collection_id", 0)) != collection_id:
|
|||
|
|
raise ApiError("接口未返回属于目标作品交付集的作品")
|
|||
|
|
_, repeated, _ = request_json(opener, f"{base_url}/api/notes", 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"),
|
|||
|
|
"collection": collection.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)
|