102 lines
4.0 KiB
Python
102 lines
4.0 KiB
Python
#!/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)
|