#!/usr/bin/env python3 """Two-phase, target-safe Delivery Desk work uploader.""" from __future__ import annotations import argparse import hashlib import json import os import re import sys from pathlib import Path from typing import Any from urllib.error import HTTPError, URLError from urllib.parse import quote from urllib.request import Request, urlopen class UploadError(RuntimeError): pass def api_key() -> str: value = os.getenv("DELIVERY_DESK_API_KEY", "").strip() if not value: raise UploadError("缺少 DELIVERY_DESK_API_KEY；请由操作者在环境变量中配置，不要粘贴到计划文件") return value def base_url(value: str | None = None) -> str: return (value or os.getenv("DELIVERY_DESK_BASE_URL") or "http://127.0.0.1:3010").rstrip("/") def request_json(base: str, path: str, *, method: str = "GET", body: dict[str, Any] | None = None) -> tuple[int, Any]: data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None headers = {"Authorization": f"Bearer {api_key()}", "Accept": "application/json"} if data is not None: headers["Content-Type"] = "application/json" request = Request(f"{base}{path}", data=data, method=method, headers=headers) try: with urlopen(request, timeout=20) as response: raw = response.read().decode("utf-8") return response.status, json.loads(raw) if raw else None except HTTPError as error: raw = error.read().decode("utf-8", errors="replace") try: payload = json.loads(raw) detail = payload.get("error", raw) if isinstance(payload, dict) else raw except json.JSONDecodeError: detail = raw raise UploadError(f"{method} {path} 返回 {error.code}: {detail}") from error except URLError as error: if method == "POST": raise UploadError(f"state_unknown: {method} {path} 的结果未知，禁止自动重试；请先检查服务器状态：{error.reason}") from error raise UploadError(f"无法连接 {base}: {error.reason}") from error def print_json(value: Any) -> None: print(json.dumps(value, ensure_ascii=False, indent=2)) def projects(base: str) -> list[dict[str, Any]]: _, value = request_json(base, "/api/projects") if not isinstance(value, list): raise UploadError("项目发现接口返回格式无效") return value def exact_project(base: str, project_id: int) -> dict[str, Any]: available = projects(base) matches = [item for item in available if int(item.get("id", 0)) == project_id] if len(matches) != 1: choices = [{"group_id": p.get("group_id"), "group_name": p.get("group_name"), "project_id": p.get("id"), "project_name": p.get("name"), "slug": p.get("slug")} for p in available] raise UploadError(f"项目 ID {project_id} 不存在或当前 Key 无权访问。可访问项目：{json.dumps(choices, ensure_ascii=False)}") return matches[0] def project_identity(project: dict[str, Any]) -> dict[str, Any]: return { "group_id": int(project["group_id"]), "group_name": str(project["group_name"]), "project_id": int(project["id"]), "project_name": str(project["name"]), "project_slug": str(project["slug"]), "project_status": str(project["status"]), } def validate_images(values: list[str]) -> list[str]: if not 1 <= len(values) <= 30: raise UploadError("必须提供 1-30 个图片 URL") cleaned = [value.strip() for value in values] if any(not re.match(r"^https?://[^\s]+$", value, re.IGNORECASE) or len(value) > 2048 for value in cleaned): raise UploadError("图片必须是长度不超过 2048 的公开 HTTP/HTTPS URL") return cleaned def confirmation_code(plan: dict[str, Any]) -> str: material = {key: value for key, value in plan.items() if key != "confirmation_code"} canonical = json.dumps(material, ensure_ascii=False, sort_keys=True, separators=(",", ":")) return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:12].upper() def save_plan(plan: dict[str, Any], output: str) -> None: plan["confirmation_code"] = confirmation_code(plan) destination = Path(output) destination.parent.mkdir(parents=True, exist_ok=True) destination.write_text(json.dumps(plan, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print_json({"plan_file": str(destination), "plan": plan}) def get_work(base: str, work_id: int) -> dict[str, Any]: _, value = request_json(base, f"/api/works/{work_id}") if not isinstance(value, dict): raise UploadError("作品详情接口返回格式无效") return value def assert_work_project(work: dict[str, Any], project_id: int) -> None: actual = int((work.get("project") or {}).get("id", 0)) if actual != project_id: raise UploadError(f"作品属于项目 {actual}，不是已确认项目 {project_id}") def content_from_args(args: argparse.Namespace, current: dict[str, Any] | None = None) -> dict[str, Any]: current = current or {} if args.clear_description and args.description is not None: raise UploadError("--description 与 --clear-description 不能同时使用") if args.clear_tags and args.tags is not None: raise UploadError("--tag 与 --clear-tags 不能同时使用") title = args.title if args.title is not None else current.get("title") description = "" if args.clear_description else args.description if args.description is not None else current.get("description", "") tags = [] if args.clear_tags else args.tags if args.tags is not None else current.get("tags", []) if not str(title or "").strip(): raise UploadError("标题不能为空") return { "title": str(title).strip(), "description": str(description or ""), "tags": [str(tag).strip() for tag in tags if str(tag).strip()], "images": validate_images(args.image_urls), } def cmd_projects(args: argparse.Namespace) -> None: base = base_url(args.base_url) print_json([project_identity(item) | {"review_status": item.get("review_status")} for item in projects(base)]) def cmd_works(args: argparse.Namespace) -> None: base = base_url(args.base_url) project = exact_project(base, args.project_id) query = f"?externalId={quote(args.external_id)}" if args.external_id else "" _, value = request_json(base, f"/api/projects/{args.project_id}/works{query}") print_json({"target": project_identity(project), "works": value}) def cmd_inspect(args: argparse.Namespace) -> None: base = base_url(args.base_url) project = exact_project(base, args.project_id) work = get_work(base, args.work_id) assert_work_project(work, args.project_id) print_json({"target": project_identity(project), "work": work}) def cmd_plan_work(args: argparse.Namespace) -> None: base = base_url(args.base_url) project = exact_project(base, args.project_id) if project.get("status") != "active": raise UploadError("目标项目不是 active，不能创建作品") if not re.fullmatch(r"[A-Za-z0-9._:-]{1,128}", args.external_id): raise UploadError("externalId 格式无效") plan = { "schema_version": 1, "operation": "create_work", "base_url": base, "target": project_identity(project), "external_id": args.external_id, "content": content_from_args(args), } save_plan(plan, args.output) def cmd_plan_round(args: argparse.Namespace) -> None: base = base_url(args.base_url) project = exact_project(base, args.project_id) if project.get("status") != "active": raise UploadError("目标项目不是 active，不能创建验收轮次") work = get_work(base, args.work_id) assert_work_project(work, args.project_id) rounds = work.get("rounds") or [] current_round = max((int(item.get("round_number", 0)) for item in rounds), default=0) plan = { "schema_version": 1, "operation": "create_round", "base_url": base, "target": project_identity(project), "work": { "work_id": args.work_id, "external_id": work.get("external_id"), "expected_version_number": int(work.get("version_number", 0)), "current_round": current_round, "next_round": current_round + 1, }, "content": content_from_args(args, work), } save_plan(plan, args.output) def same_content(detail: dict[str, Any], expected: dict[str, Any]) -> bool: images = [str(item.get("url") or "") for item in detail.get("images", [])] try: returned_images = validate_images(images) except UploadError: return False return ( detail.get("title") == expected["title"] and detail.get("description", "") == expected["description"] and detail.get("tags", []) == expected["tags"] and len(returned_images) == len(expected["images"]) ) def cmd_apply(args: argparse.Namespace) -> None: plan = json.loads(Path(args.plan).read_text(encoding="utf-8")) expected_code = confirmation_code(plan) if args.confirm != expected_code or plan.get("confirmation_code") != expected_code: raise UploadError("确认码不匹配；计划可能已改变，禁止执行") if plan.get("schema_version") != 1 or plan.get("operation") not in {"create_work", "create_round"}: raise UploadError("不支持的计划格式") base = base_url(plan.get("base_url")) target = plan["target"] project = exact_project(base, int(target["project_id"])) actual_identity = project_identity(project) if actual_identity != target: raise UploadError(f"项目身份或状态已变化，禁止执行。计划={target}，当前={actual_identity}") content = plan["content"] if plan["operation"] == "create_work": external_id = plan["external_id"] _, existing = request_json(base, f"/api/projects/{target['project_id']}/works?externalId={quote(external_id)}") if existing: print_json({"success": True, "idempotent": True, "message": "externalId 已存在，未发送创建请求", "target": target, "work": existing[0]}) return _, created = request_json(base, f"/api/projects/{target['project_id']}/works", method="POST", body={**content, "externalId": external_id}) work_id = int(created["id"]) verified = get_work(base, work_id) assert_work_project(verified, int(target["project_id"])) if verified.get("external_id") != external_id or not same_content(verified, content): raise UploadError("创建请求返回成功，但回读内容不一致") print_json({"success": True, "idempotent": bool(created.get("idempotent")), "target": target, "work_id": work_id, "external_id": external_id, "round_number": 1, "title": verified["title"], "image_count": len(verified["images"])}) return work_plan = plan["work"] before = get_work(base, int(work_plan["work_id"])) assert_work_project(before, int(target["project_id"])) if int(before.get("version_number", 0)) != int(work_plan["expected_version_number"]): raise UploadError("作品当前版本已变化，旧计划失效；请重新生成计划并确认") _, created = request_json(base, f"/api/works/{work_plan['work_id']}/rounds", method="POST", body=content) verified = get_work(base, int(work_plan["work_id"])) if int(verified.get("version_number", 0)) != int(work_plan["expected_version_number"]) + 1 or not same_content(verified, content): raise UploadError("新轮次请求返回成功，但回读轮次或内容不一致") rounds = verified.get("rounds") or [] round_number = max((int(item.get("round_number", 0)) for item in rounds), default=0) print_json({"success": True, "target": target, "work_id": int(work_plan["work_id"]), "external_id": verified.get("external_id"), "round_number": round_number, "title": verified["title"], "image_count": len(verified["images"]), "server_response_id": created.get("id")}) def add_common(command: argparse.ArgumentParser) -> None: command.add_argument("--base-url", default=None) def add_content(command: argparse.ArgumentParser, *, title_required: bool) -> None: command.add_argument("--title", required=title_required, default=None) command.add_argument("--description", default=None) command.add_argument("--clear-description", action="store_true") command.add_argument("--tag", action="append", dest="tags", default=None) command.add_argument("--clear-tags", action="store_true") command.add_argument("--image-url", action="append", dest="image_urls", required=True) def build_parser() -> argparse.ArgumentParser: root = argparse.ArgumentParser(description="安全发现、规划并上传 Delivery Desk 作品") commands = root.add_subparsers(dest="command", required=True) command = commands.add_parser("projects") add_common(command) command.set_defaults(handler=cmd_projects) command = commands.add_parser("works") add_common(command) command.add_argument("--project-id", type=int, required=True) command.add_argument("--external-id") command.set_defaults(handler=cmd_works) command = commands.add_parser("inspect-work") add_common(command) command.add_argument("--project-id", type=int, required=True) command.add_argument("--work-id", type=int, required=True) command.set_defaults(handler=cmd_inspect) command = commands.add_parser("plan-work") add_common(command) command.add_argument("--project-id", type=int, required=True) command.add_argument("--external-id", required=True) add_content(command, title_required=True) command.add_argument("--output", required=True) command.set_defaults(handler=cmd_plan_work) command = commands.add_parser("plan-round") add_common(command) command.add_argument("--project-id", type=int, required=True) command.add_argument("--work-id", type=int, required=True) add_content(command, title_required=False) command.add_argument("--output", required=True) command.set_defaults(handler=cmd_plan_round) command = commands.add_parser("apply") command.add_argument("--plan", required=True) command.add_argument("--confirm", required=True) command.set_defaults(handler=cmd_apply) return root def main() -> int: args = build_parser().parse_args() args.handler(args) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except (UploadError, KeyError, TypeError, ValueError, json.JSONDecodeError) as error: print(f"失败: {error}", file=sys.stderr) raise SystemExit(1)