feat(review): 重构项目级单方案验收协作

tests/api_get_work_annotations.py

@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+"""获取指定作品验收轮次的全部批注与反馈。"""
+
+from __future__ import annotations
+
+import argparse
+import getpass
+import json
+import os
+import sys
+from http.cookiejar import CookieJar
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlencode
+from urllib.request import HTTPCookieProcessor, Request, build_opener
+
+
+class ApiError(RuntimeError):
+    """Delivery Desk API 请求失败。"""
+
+
+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
+    except HTTPError as error:
+        body = error.read().decode("utf-8", errors="replace")
+        try:
+            parsed = json.loads(body)
+            detail = parsed.get("error", body) if isinstance(parsed, dict) else 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="获取作品指定验收轮次的全部批注")
+    parser.add_argument("--base-url", default=os.getenv("DELIVERY_DESK_BASE_URL", "http://127.0.0.1:3010"))
+    parser.add_argument("--project-name", default="光影内容计划", help="用于校验作品所属项目")
+    parser.add_argument("--work-id", type=int, default=13)
+    parser.add_argument("--round", type=int, default=2, dest="round_number")
+    parser.add_argument("--include-history", action="store_true", 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"))
+    return parser.parse_args()
+
+
+def authenticate(opener, base_url: str, args: argparse.Namespace) -> dict[str, str]:
+    if args.api_key:
+        return {"Authorization": f"Bearer {args.api_key}"}
+
+    password = args.password or getpass.getpass(f"请输入账号 {args.username} 的密码: ")
+    payload = json.dumps({"username": args.username, "password": password}).encode("utf-8")
+    status, _ = request_json(
+        opener,
+        f"{base_url}/api/auth/login",
+        method="POST",
+        data=payload,
+        headers={"Content-Type": "application/json"},
+    )
+    if status != 200:
+        raise ApiError("登录接口未返回成功状态")
+    return {}
+
+
+def main() -> int:
+    args = parse_args()
+    if args.work_id < 1 or args.round_number < 1:
+        raise ApiError("作品 ID 和轮次必须是正整数")
+
+    base_url = args.base_url.rstrip("/")
+    opener = build_opener(HTTPCookieProcessor(CookieJar()))
+    auth_headers = authenticate(opener, base_url, args)
+
+    query = urlencode({"round": args.round_number, "include_history": str(args.include_history).lower()})
+    _, context = request_json(
+        opener,
+        f"{base_url}/api/works/{args.work_id}/optimization-context?{query}",
+        headers=auth_headers,
+    )
+    if not isinstance(context, dict):
+        raise ApiError("内容优化接口返回格式无效")
+
+    project = context.get("project")
+    actual_project_name = project.get("name") if isinstance(project, dict) else None
+    if actual_project_name != args.project_name:
+        raise ApiError(f"Work {args.work_id:03d} 属于项目“{actual_project_name}”，不是“{args.project_name}”")
+    print(json.dumps(context, 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)