delivery-desk/.agents/skills/upload-delivery-desk-work/scripts/delivery_desk_upload.py

#!/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):
    def __init__(self, message: str, *, action: str = "ask_operator", next_step: str = "停止操作，将错误和目标信息告知调用者并等待处理") -> None:
        super().__init__(message)
        self.action = action
        self.next_step = next_step

    def payload(self) -> dict[str, Any]:
        return {"success": False, "action": self.action, "error": str(self), "next_step": self.next_step}


def revise(message: str, next_step: str) -> UploadError:
    return UploadError(message, action="revise", next_step=next_step)


def retry(message: str, next_step: str) -> UploadError:
    return UploadError(message, action="retry", next_step=next_step)


def http_failure(method: str, path: str, status: int, detail: str) -> UploadError:
    message = f"{method} {path} 返回 {status}: {detail}"
    if status in {400, 413, 422}:
        return revise(message, "根据错误修改输入或图片 URL，重新生成计划，并取得新的确认码后再执行")
    if status == 401:
        return UploadError(message, next_step="停止操作，请调用者配置或更换有效的 DELIVERY_DESK_API_KEY")
    if status == 403:
        return UploadError(message, next_step="停止操作，向调用者报告已确认的运营组和项目，请管理员修正 Key 权限；不要改选其他项目")
    if status == 404:
        return UploadError(message, next_step="重新执行只读发现命令核对目标；若目标已删除或不可见，询问调用者，不要猜测新目标")
    if status == 409:
        return UploadError(message, next_step="停止操作并报告目标当前状态，请调用者或管理员解除冲突；不要更换目标或自动重试")
    if method == "GET" and status in {429, 502, 503, 504}:
        return retry(message, "等待片刻后原样重试当前只读命令；不要改变目标或参数")
    if method == "POST":
        return UploadError(message, next_step="写入结果可能不确定，禁止自动重试；先只读检查目标状态，再询问调用者")
    return UploadError(message, next_step="停止操作并向调用者报告服务异常；确认服务恢复后再重新执行只读发现")


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 http_failure(method, path, error.code, str(detail)) from error
    except URLError as error:
        if method == "POST":
            raise UploadError(
                f"state_unknown: {method} {path} 的结果未知，禁止自动重试；请先检查服务器状态：{error.reason}",
                next_step="使用 works --external-id 或 inspect-work 只读核对服务器状态；若仍无法确认，询问调用者后再决定",
            ) from error
        raise retry(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 revise("必须提供 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 revise("图片必须是长度不超过 2048 的公开 HTTP/HTTPS URL", "替换为可公开访问的 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 revise("--description 与 --clear-description 不能同时使用", "只保留其中一个参数后重新生成计划")
    if args.clear_tags and args.tags is not None:
        raise revise("--tag 与 --clear-tags 不能同时使用", "只保留其中一种 Tag 操作后重新生成计划")
    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 revise("标题不能为空", "补充非空标题后重新生成计划并取得新的确认码")
    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 revise("externalId 格式无效", "改用符合 [A-Za-z0-9._:-]{1,128} 的稳定 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 revise("确认码不匹配；计划可能已改变，禁止执行", "重新展示当前计划并取得与当前计划一致的新确认码")
    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:
        failure = error if isinstance(error, UploadError) else UploadError(f"输入或响应格式无效: {error}")
        print(json.dumps(failure.payload(), ensure_ascii=False), file=sys.stderr)
        raise SystemExit(1)