feat(api): 支持外部客户端创建作品和更新版本
This commit is contained in:
yuzhe
174
tests/api_create_work.py
Normal file
174
tests/api_create_work.py
Normal file
@@ -0,0 +1,174 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user