# Parallel-Vibe Plan Template

`auto-draw-plot` 对 `parallel-vibe` 的标准用法是：每一轮都必须先生成合法的 `parallel-vibe` plan，再决定是否真的把“下一轮 prompt 草案”交给独立线程。这样既保留统一协议，又避免把整个出图闭环都塞进 thread workspace。

示例结构：

```json
{
  "threads": [
    {
      "thread_id": "001",
      "title": "Round 1 prompt generation",
      "runner": {
        "type": "shell",
        "profile": "deep",
        "cmd_template": "python3 /abs/path/auto-draw-plot/scripts/parallel_round_worker.py --run-dir /abs/path/.draw-plot/run-xxxx --request-file /abs/path/.draw-plot/run-xxxx/requests/user-need.md --round 1 --result-file RESULT.md"
      },
      "prompt": "读取当前 run 的用户需求和历史 round 反馈，在隔离 workspace 中生成下一轮 prompt 草案，并写出 RESULT.md。"
    }
  ],
  "synthesize": false
}
```

说明：

- 本文件不是可选示例，而是 `auto-draw-plot` 每一轮都必须写出的标准协议。
- `runner.type` 必须是 `parallel-vibe` 当前支持的合法类型；对本 skill，推荐 `shell`。
- thread 的职责只应包括：读取需求、起草 prompt、写出 `RESULT.md`。
- PNG 生成、视觉评估、最佳轮次选择仍由 `auto-draw-plot/scripts/run_draw_plot.py` 主流程负责。
- 实际 plan 由 `scripts/build_parallel_plan.py` 生成，脚本会对路径做 shell quote；示例中的绝对路径必须位于当前 project/workspace 边界内，不应指向父目录或外部敏感位置。

# Prompt Guidelines

1. **理解需求与模式**：
   - 把用户的 `user_need` 拆成三部分：意图（是什么）、视觉元素（有哪些节点/文本/布局）、硬约束（颜色/比例/输出格式）。
   - 先确定 `mode`：默认 `general`；技术路线图/flowchart 用 `roadmap`；原理图/机制图/架构图用 `schematic`。
   - 在 prompt 中用明确的 bullet（如 `- elements`、`- layout`）列出每个部分，避免模糊说法。

2. **Prompt 模板结构**：
   ```
   你是一个严谨的绘图 prompt 规划助手，负责把输入需求转化为适合当前图片 provider 的可执行 prompt。
   场景：{用途 / 受众}
   元素：
   - {元素 1}
   - {元素 2}
   风格：{色调 / 质感 / 字体 / 参考图}
   排除项：{千万别做的事情}
   输出产物：PNG，要求 {比例 / provider 原生尺寸选择参考 / 透明背景 / 其他}
   ```
   把 `visual_constraints` 填入 `风格` 与 `输出` 中，把 `reference_images` 作为“样式参考图”说明。不要把 `4K`、`3840px` 等字样写成最终像素承诺；默认最终 PNG 保留图片 provider 返回的原生尺寸。

3. **Mode Preset 要点**：
   - `general`：忠实复述用户需求，主体清晰，画面稳定，避免无关元素。
   - `roadmap`：白底、A4/打印可读、3-5 个阶段、阶段标题条、圆角节点、主链粗箭头、风险/备选细线或虚线；禁止图内总标题/caption。
   - `schematic`：白底、16:10 友好、分组大框、圆角节点、机制链/模块关系清楚、主链粗箭头、辅助/验证细箭头或虚线；严格保留用户术语。
   - `roadmap` 与 `schematic` 都必须强约束文字：水平、清晰、框内留白、短句自然换行；中文标签使用正常字宽的现代黑体/思源黑体/Noto Sans CJK 风格，无衬线常规到半粗体、深灰或黑色；禁止窄体、长体、压缩体、condensed/narrow/compressed font、横向压缩、瘦长拉伸字体；避免细体/浅灰/过轻笔画；禁止乱码/扭曲/艺术字/手写/透视。

4. **多轮优化提示语**：
   - 第一轮不要带过多假设，直接复述需求。
   - 之后的轮次把第 `n` 轮 `output.png` 作为第 `n+1` 轮第一参考图，并在 prompt 末尾附上 `feedback` 段，比如：
     ```
     Reference image:
     - 第一张参考图是上一轮 output.png，请在此基础上局部微调，不要从零重画。
     Feedback:
     - 上轮评估：文字与背景对比不足
     - 修改方向：增强对比、加边框
     ```
   - 指定 `round` 和 `max_rounds` 以便 meta 记录。

5. **视觉评估入参**：
   - `evaluation` 段至少要覆盖 `score`（0-10）、`passed`（true/false）、`must_fix`、`prompt_patch`。
   - 若本轮未通过，下一轮 prompt 必须把 `must_fix` / `prompt_patch` 显式合并进去。

6. **安全与隐私**：
   - 不要在 prompt 中暴露 API key、绝对路径等敏感信息。
   - 引用参考图时只描述风格，不附带具体文件路径（由 workflow 在 workspace 内提供）。

7. **记录**：
   - 把每轮 prompt 写入 `.draw-plot/run-<timestamp>/rounds/round-XX/prompt.txt`，便于复现与审计。
   - 脚本默认不调用远端文本规划接口；本地模板应稳定组合“用户需求 + 模式护栏 + 上轮修复项”。

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import shlex
from pathlib import Path
from typing import Any, Dict

from common import expand_path, load_config, skill_root, slugify, write_json


def build_parallel_plan(
    *,
    run_dir: Path,
    request_file: Path,
    round_index: int,
    output_plan: Path,
) -> Dict[str, Any]:
    cfg = load_config()
    pv_cfg = cfg.get("parallel_vibe", {}) or {}
    runner_script = skill_root() / str(pv_cfg.get("worker_script", "scripts/parallel_round_worker.py"))
    round_name = f"{cfg.get('generation', {}).get('round_dir_prefix', 'round-')}{round_index:02d}"
    thread_id = f"{round_index:03d}"
    plan = {
        "threads": [
            {
                "thread_id": thread_id,
                "title": f"auto-draw-plot {round_name}",
                "runner": {
                    "type": str(pv_cfg.get("default_runner", "shell")),
                    "profile": str(pv_cfg.get("default_profile", "deep")),
                    "cmd_template": (
                        f"python3 {shlex.quote(str(runner_script))} "
                        f"--run-dir {shlex.quote(str(run_dir))} "
                        f"--request-file {shlex.quote(str(request_file))} "
                        f"--round {round_index} "
                        f"--result-file {shlex.quote(str(pv_cfg.get('result_filename', 'RESULT.md')))}"
                    ),
                },
                "prompt": (
                    "在当前 workspace 内完成本轮 prompt 优化草案：读取用户需求与已有 round 记录，"
                    "输出 `prompt.txt`、`evaluation-request.md` 与 `RESULT.md`。"
                ),
            }
        ],
        "synthesize": False,
        "project_hint": slugify(request_file.stem, max_len=32),
    }
    write_json(output_plan, plan)
    return plan


def main() -> None:
    parser = argparse.ArgumentParser(description="为 auto-draw-plot 生成 parallel-vibe plan.json。")
    parser.add_argument("--run-dir", required=True)
    parser.add_argument("--request-file", required=True)
    parser.add_argument("--round", required=True, type=int)
    parser.add_argument("--output-plan", default="")
    args = parser.parse_args()

    run_dir = expand_path(args.run_dir, base=Path.cwd())
    request_file = expand_path(args.request_file, base=Path.cwd())
    cfg = load_config()
    default_plan_name = str((cfg.get("parallel_vibe", {}) or {}).get("plan_filename", "parallel-plan.json"))
    output_plan = expand_path(args.output_plan, base=Path.cwd()) if args.output_plan else run_dir / "parallel-vibe" / default_plan_name
    build_parallel_plan(run_dir=run_dir, request_file=request_file, round_index=int(args.round), output_plan=output_plan)
    print(output_plan)


if __name__ == "__main__":
    main()

from __future__ import annotations

import datetime as dt
import hashlib
import json
import os
import re
import shutil
import sys
from pathlib import Path
from typing import Any, Dict, Iterable, Optional


def info(message: str) -> None:
    print(f"[auto-draw-plot] {message}")


def warn(message: str) -> None:
    print(f"[auto-draw-plot][warn] {message}", file=sys.stderr)


def fatal(message: str, code: int = 2) -> "None":
    print(f"[auto-draw-plot][error] {message}", file=sys.stderr)
    raise SystemExit(code)


def skill_root() -> Path:
    return Path(__file__).resolve().parents[1]


def load_yaml(path: Path) -> Dict[str, Any]:
    try:
        import yaml  # type: ignore
    except Exception as exc:
        fatal(f"缺少 PyYAML，无法读取配置：{exc}")
    try:
        data = yaml.safe_load(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        fatal(f"配置文件不存在：{path}")
    except Exception as exc:
        fatal(f"读取 YAML 失败：{path} ({exc})")
    return data or {}


def load_config() -> Dict[str, Any]:
    return load_yaml(skill_root() / "config.yaml")


def ensure_dir(path: Path) -> Path:
    path.mkdir(parents=True, exist_ok=True)
    return path


def expand_path(value: str | Path, *, base: Optional[Path] = None) -> Path:
    p = Path(os.path.expanduser(str(value)))
    if p.is_absolute():
        return p
    if base is None:
        base = Path.cwd()
    return (base / p).resolve()


def write_text(path: Path, text: str) -> None:
    ensure_dir(path.parent)
    path.write_text(text, encoding="utf-8")


def read_text(path: Path) -> str:
    return path.read_text(encoding="utf-8")


def write_json(path: Path, payload: Any) -> None:
    ensure_dir(path.parent)
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def read_json(path: Path) -> Any:
    return json.loads(path.read_text(encoding="utf-8"))


def now_tag(fmt: str) -> str:
    return dt.datetime.now().strftime(fmt)


def sha256_file(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as fh:
        for chunk in iter(lambda: fh.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def copy_file(src: Path, dst: Path) -> None:
    ensure_dir(dst.parent)
    shutil.copy2(src, dst)


def slugify(text: str, *, max_len: int = 48) -> str:
    normalized = re.sub(r"[^0-9A-Za-z\u4e00-\u9fff]+", "-", text.strip())
    normalized = normalized.strip("-").lower()
    if not normalized:
        normalized = "draw-plot"
    return normalized[:max_len].rstrip("-") or "draw-plot"


def relative_display(path: Path, *, base: Optional[Path] = None) -> str:
    try:
        if base is None:
            base = Path.cwd()
        return str(path.resolve().relative_to(base.resolve()))
    except Exception:
        return str(path)


def extract_json_from_text(text: str) -> Optional[Dict[str, Any]]:
    raw = (text or "").strip()
    if not raw:
        return None

    fenced = re.findall(r"```(?:json)?\s*(\{.*?\})\s*```", raw, flags=re.DOTALL)
    candidates = fenced + [raw]
    for candidate in candidates:
        candidate = candidate.strip()
        if not candidate:
            continue
        candidate = _repair_json_candidate(candidate)
        try:
            data = json.loads(candidate)
        except Exception:
            data = _try_balanced_json(candidate)
        if isinstance(data, dict):
            return data
    return None


def _try_balanced_json(text: str) -> Optional[Dict[str, Any]]:
    start = text.find("{")
    if start < 0:
        return None
    depth = 0
    for idx in range(start, len(text)):
        ch = text[idx]
        if ch == "{":
            depth += 1
        elif ch == "}":
            depth -= 1
            if depth == 0:
                snippet = text[start : idx + 1]
                snippet = _repair_json_candidate(snippet)
                try:
                    data = json.loads(snippet)
                except Exception:
                    return None
                if isinstance(data, dict):
                    return data
                return None
    return None


def join_lines(lines: Iterable[str]) -> str:
    return "\n".join([line for line in lines if line is not None])


def _repair_json_candidate(text: str) -> str:
    repaired = text
    repaired = re.sub(r'"\s+([A-Za-z_][A-Za-z0-9_]*)"\s*:', r'"\1":', repaired)
    repaired = re.sub(r'"\n+([A-Za-z_][A-Za-z0-9_]*)"\s*:', r'"\1":', repaired)
    repaired = repaired.replace("\r\n", "\n")
    return repaired

from __future__ import annotations

import os
import re
import ntpath
from pathlib import Path
from typing import Dict, Optional


DEFAULT_REMOTE_ENV = Path.home() / ".bensz-skills" / "config" / "remote.env"
_LINE_RE = re.compile(r"^\s*([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(.*?)\s*$")


def resolve_config_path(value: str, *, environ: Optional[Dict[str, str]] = None, platform_name: Optional[str] = None) -> Path:
    """Resolve skill config paths consistently across PowerShell, Git Bash and POSIX shells."""
    raw = str(value or "").strip()
    env = dict(os.environ if environ is None else environ)
    platform = str(platform_name or os.name).lower()
    if raw == "~" or raw.startswith("~/") or raw.startswith("~\\"):
        if platform.startswith("win") or platform == "nt":
            home = env.get("USERPROFILE") or (
                f"{env.get('HOMEDRIVE', '')}{env.get('HOMEPATH', '')}".strip()
            )
            if home:
                suffix = raw[2:] if len(raw) > 1 else ""
                return Path(ntpath.join(home, suffix.replace("/", "\\")))
        home = env.get("HOME")
        if home:
            suffix = raw[2:] if len(raw) > 1 else ""
            return Path(home) / suffix
    return Path(raw).expanduser()


def find_remote_env(preferred: Optional[Path] = None) -> Optional[Path]:
    if preferred is not None and preferred.exists() and preferred.is_file():
        return preferred
    if DEFAULT_REMOTE_ENV.exists() and DEFAULT_REMOTE_ENV.is_file():
        return DEFAULT_REMOTE_ENV
    return None


def _strip_inline_comment(value: str) -> str:
    val = value.strip()
    if not val:
        return ""
    if (val.startswith('"') and val.endswith('"')) or (val.startswith("'") and val.endswith("'")):
        return val[1:-1]
    if "#" in val:
        return val.split("#", 1)[0].rstrip()
    return val


def parse_dotenv(text: str) -> Dict[str, str]:
    out: Dict[str, str] = {}
    for raw_line in (text or "").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        if line.startswith("export "):
            raw_line = line[len("export ") :]
        m = _LINE_RE.match(raw_line)
        if not m:
            continue
        out[m.group(1)] = _strip_inline_comment(m.group(2))
    return out


def load_dotenv(path: Path) -> Dict[str, str]:
    try:
        raw = path.read_text(encoding="utf-8")
    except UnicodeDecodeError:
        raw = path.read_text(encoding="utf-8", errors="replace")
    return parse_dotenv(raw)


def merged_env(dotenv_path: Optional[Path]) -> Dict[str, str]:
    out: Dict[str, str] = {}
    if dotenv_path is not None and dotenv_path.exists():
        out.update(load_dotenv(dotenv_path))
    out.update({k: v for k, v in os.environ.items() if isinstance(v, str)})
    return out


def mask_secret(secret: str, *, keep: int = 4) -> str:
    text = str(secret or "")
    if not text:
        return ""
    if len(text) <= keep:
        return "*" * len(text)
    return "*" * max(8, len(text) - keep) + text[-keep:]

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import re
from pathlib import Path
from typing import Any, Dict, Optional

from common import expand_path, load_config, read_text, sha256_file, write_json
from modes import DrawMode, mode_evaluation_lines, mode_prompt_lines


def evaluate_image(
    *,
    image_path: Path,
    request_text: str,
    image_prompt: str,
    remote_env: Optional[Path],
    output_json: Optional[Path],
    debug_dir: Optional[Path],
    mode: Optional[DrawMode] = None,
) -> Dict[str, Any]:
    normalized = heuristic_evaluation(image_path)
    normalized["review_backend"] = "heuristic"
    normalized["review_context"] = build_review_prompt(request_text=request_text, image_prompt=image_prompt, mode=mode)
    if output_json is not None:
        write_json(output_json, normalized)
    return normalized


def build_review_prompt(*, request_text: str, image_prompt: str, mode: Optional[DrawMode]) -> str:
    return "\n".join(
        [
            "你是一位严格但建设性的视觉评审。",
            "请根据用户需求、当前图片 prompt 与提供的图片本身，判断这张图是否足够满足要求。",
            "输出必须是原始 JSON，不要加 Markdown 代码块，不要解释。",
            "JSON schema:",
            "{",
            '  "score": 0.0,',
            '  "passed": false,',
            '  "summary": "一句话总结",',
            '  "strengths": ["..."],',
            '  "issues": ["..."],',
            '  "must_fix": ["..."],',
            '  "prompt_patch": ["下一轮 prompt 应补充的指令"],',
            '  "confidence": 0.0',
            "}",
            "",
            "评分标准：",
            "- 需求覆盖度",
            "- 主体是否清晰",
            "- 画面结构是否稳定",
            "- 若含文字，文字是否尽量清晰可读",
            "- 是否有明显伪影、水印、乱码或离题",
            *(["", "模式检查：", *mode_prompt_lines(mode)] if mode is not None else []),
            *(mode_evaluation_lines(mode) if mode is not None else []),
            "",
            "用户需求：",
            request_text.strip(),
            "",
            "当前图片 prompt：",
            image_prompt.strip(),
        ]
    )


def normalize_evaluation(
    *,
    payload: Dict[str, Any],
    raw_text: str,
    image_path: Path,
    cfg: Dict[str, Any],
) -> Dict[str, Any]:
    score = _coerce_float(payload.get("score"), default=0.0)
    passed = bool(payload.get("passed", False))
    strengths = _normalize_text_list(payload.get("strengths"))
    issues = _normalize_text_list(payload.get("issues"))
    must_fix = _normalize_text_list(payload.get("must_fix"))
    prompt_patch = _normalize_text_list(payload.get("prompt_patch"))
    confidence = _coerce_float(payload.get("confidence"), default=0.0)
    if confidence > 1.0 and confidence <= 10.0:
        confidence = confidence / 10.0
    summary = str(payload.get("summary") or "").strip()

    if not summary:
        summary = "模型未返回结构化总结，需人工复核。"
    if score >= float(cfg.get("accept_score", 8.5)):
        passed = True
    return {
        "score": round(score, 2),
        "passed": passed,
        "summary": summary,
        "strengths": strengths,
        "issues": issues,
        "must_fix": must_fix,
        "prompt_patch": prompt_patch,
        "confidence": round(confidence, 2),
        "image_sha256": sha256_file(image_path),
        "raw_text": raw_text,
    }


def heuristic_evaluation(image_path: Path) -> Dict[str, Any]:
    cfg = load_config()
    eval_cfg = cfg.get("evaluation", {}) or {}
    long_edge = 0
    try:
        from PIL import Image  # type: ignore

        with Image.open(image_path) as img:
            long_edge = max(img.size)
    except Exception:
        long_edge = 0

    size_bytes = image_path.stat().st_size if image_path.exists() else 0
    # Heuristic fallback is intentionally capped below accept_score; it proves
    # the file is non-empty/resolution-sane but cannot certify semantic quality.
    score = 4.5
    strengths = []
    issues = []
    if size_bytes >= int(eval_cfg.get("heuristic_min_file_size_bytes", 8192)):
        score += 1.0
        strengths.append("PNG 文件体积正常，说明图像并非空白占位。")
    else:
        issues.append("PNG 文件过小，可能生成失败或接近空白。")
    if long_edge >= int(eval_cfg.get("heuristic_min_long_edge_px", 1800)):
        score += 1.0
        strengths.append("PNG 像素尺寸达到基本可检查级别；这不代表模型原生细节或文字清晰度已达标。")
    else:
        issues.append("PNG 像素尺寸偏低，建议重新生成或检查 provider 原生尺寸。")
    return {
        "score": round(score, 2),
        "passed": False,
        "summary": "已退化为启发式评估，请人工确认视觉质量。",
        "strengths": strengths,
        "issues": issues or ["缺少 AI 视觉评估结果。"],
        "must_fix": issues,
        "prompt_patch": ["重新强调主体清晰、结构稳定、文字可读。"],
        "confidence": 0.2,
        "image_sha256": sha256_file(image_path) if image_path.exists() else "",
        "raw_text": "",
        "fallback_mode": "heuristic",
    }


def salvage_evaluation_payload(text: str) -> Dict[str, Any]:
    raw = text or ""
    payload: Dict[str, Any] = {}

    score = _search_number(raw, r'"score"\s*:\s*([0-9]+(?:\.[0-9]+)?)')
    if score is not None:
        payload["score"] = score

    passed = re.search(r'"passed"\s*:\s*(true|false)', raw, flags=re.IGNORECASE)
    if passed:
        payload["passed"] = passed.group(1).lower() == "true"

    summary = _search_string_block(raw, "summary")
    if summary:
        payload["summary"] = summary

    for key in ("strengths", "issues", "must_fix", "prompt_patch"):
        arr = _search_string_array(raw, key)
        if arr:
            payload[key] = arr

    confidence = _search_number(raw, r'confidence"?\s*:\s*([0-9]+(?:\.[0-9]+)?)')
    if confidence is not None:
        payload["confidence"] = confidence
    return payload


def _coerce_float(value: Any, *, default: float) -> float:
    try:
        return float(value)
    except Exception:
        return default


def _normalize_text_list(value: Any) -> list[str]:
    if isinstance(value, list):
        return [str(item).strip() for item in value if str(item).strip()]
    if isinstance(value, str) and value.strip():
        return [value.strip()]
    return []


def _search_number(text: str, pattern: str) -> Optional[float]:
    m = re.search(pattern, text, flags=re.IGNORECASE | re.DOTALL)
    if not m:
        return None
    try:
        return float(m.group(1))
    except Exception:
        return None


def _search_string_block(text: str, key: str) -> str:
    m = re.search(
        rf'"{re.escape(key)}"\s*:\s*"(?P<body>.*?)"\s*,\s*"',
        text,
        flags=re.DOTALL,
    )
    if not m:
        m = re.search(
            rf'"{re.escape(key)}"\s*:\s*"(?P<body>.*?)"\s*[,\n]\s*[A-Za-z"]',
            text,
            flags=re.DOTALL,
        )
    if not m:
        return ""
    return _clean_multiline_fragment(m.group("body"))


def _search_string_array(text: str, key: str) -> list[str]:
    m = re.search(
        rf'"{re.escape(key)}"\s*:\s*\[(?P<body>.*?)\]',
        text,
        flags=re.DOTALL,
    )
    if not m:
        return []
    items = re.findall(r'"(.*?)"', m.group("body"), flags=re.DOTALL)
    return [_clean_multiline_fragment(item) for item in items if _clean_multiline_fragment(item)]


def _clean_multiline_fragment(text: str) -> str:
    cleaned = text.replace("\\n", "\n")
    cleaned = re.sub(r"\s+", " ", cleaned)
    return cleaned.strip(" ,\n\t")


def main() -> None:
    parser = argparse.ArgumentParser(description="对 auto-draw-plot 生成结果做 AI 视觉评估。")
    parser.add_argument("--image", required=True, help="待评估 PNG 路径")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--request-file", help="用户需求文件")
    group.add_argument("--request-text", help="用户需求文本")
    prompt_group = parser.add_mutually_exclusive_group(required=True)
    prompt_group.add_argument("--prompt-file", help="生成该图片的 prompt 文件")
    prompt_group.add_argument("--prompt-text", help="生成该图片的 prompt 文本")
    parser.add_argument("--output-json", default="", help="输出 JSON 路径")
    parser.add_argument("--api-env", default="", help="remote.env 路径")
    parser.add_argument("--debug-dir", default="", help="调试目录")
    args = parser.parse_args()

    request_text = read_text(Path(args.request_file)) if args.request_file else str(args.request_text or "")
    image_prompt = read_text(Path(args.prompt_file)) if args.prompt_file else str(args.prompt_text or "")

    image_path = expand_path(args.image, base=Path.cwd())
    output_json = expand_path(args.output_json, base=Path.cwd()) if args.output_json else None
    debug_dir = expand_path(args.debug_dir, base=Path.cwd()) if args.debug_dir else None
    remote_env = expand_path(args.api_env, base=Path.cwd()) if args.api_env else None

    try:
        evaluation = evaluate_image(
            image_path=image_path,
            request_text=request_text,
            image_prompt=image_prompt,
            remote_env=remote_env,
            output_json=output_json,
            debug_dir=debug_dir,
        )
    except Exception:
        evaluation = heuristic_evaluation(image_path)
        if output_json is not None:
            write_json(output_json, evaluation)
    print(evaluation["summary"])


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations

import argparse
from pathlib import Path
from typing import Any, Dict, List, Optional

from common import expand_path, load_config, warn, write_json
from image_provider_client import (
    ImageProviderConfig,
    generate_image_png,
    is_provider_fallback_allowed,
    provider_error_debug_payload,
    resolve_image_provider,
)
from nano_banana_client import load_gemini_config


def _reference_preservation_prompt(prompt: str, reference_images: Optional[List[Path]]) -> str:
    if not reference_images:
        return prompt
    return "\n".join(
        [
            prompt.rstrip(),
            "",
            "Reference edit contract:",
            "- preserve_subject=true except for the explicitly requested edit",
            "- preserve_composition=true except for the explicitly requested edit",
            "- preserve_background=true except for the explicitly requested edit",
            "- change only what the request explicitly asks to change",
        ]
    )


def generate_image(
    *,
    prompt: str,
    output_png: Path,
    remote_env: Optional[Path],
    canvas_w: int,
    canvas_h: int,
    debug_dir: Optional[Path],
    reference_images: Optional[List[Path]],
    provider_cfg: Optional[ImageProviderConfig] = None,
    provider_name: Optional[str] = None,
    allow_provider_fallback: Optional[bool] = None,
    require_reference_images: bool = False,
    postprocess_resize: Optional[bool] = None,
    postprocess_w: Optional[int] = None,
    postprocess_h: Optional[int] = None,
    quality: Optional[str] = None,
    provider_size: Optional[str] = None,
    output_format: Optional[str] = None,
    output_compression: Optional[int] = None,
) -> Dict[str, Any]:
    cfg = load_config()
    api_cfg = cfg.get("api", {}) or {}
    gen_cfg = cfg.get("generation", {}) or {}
    if allow_provider_fallback is None:
        allow_provider_fallback = bool(api_cfg.get("allow_provider_fallback", False))
    if postprocess_resize is None:
        postprocess_resize = bool(gen_cfg.get("postprocess_resize_default", False))
    if require_reference_images and not reference_images:
        raise ValueError("当前生成轮次要求参考图，但 reference_images 为空。")
    provider_cfg = provider_cfg or resolve_image_provider(
        remote_env_path=remote_env,
        provider_name=provider_name,
        run_healthcheck=False,
    )
    effective_prompt = _reference_preservation_prompt(prompt, reference_images)
    try:
        result = generate_image_png(
            provider_cfg=provider_cfg,
            prompt=effective_prompt,
            output_png=output_png,
            canvas_w=canvas_w,
            canvas_h=canvas_h,
            reference_images=reference_images,
            debug_dir=debug_dir,
            timeout_s=int(api_cfg.get("request_timeout_s", 180)),
            retries=int(api_cfg.get("retry_attempts", 5)),
            postprocess_resize=bool(postprocess_resize),
            postprocess_w=postprocess_w,
            postprocess_h=postprocess_h,
            quality=quality,
            provider_size=provider_size,
            output_format=output_format,
            output_compression=output_compression,
        )
    except Exception as exc:
        if provider_cfg.provider != "gpt-image-2":
            raise
        gpt_error = exc
        model_label = str(provider_cfg.model or provider_cfg.provider)
        if debug_dir is not None:
            write_json(
                debug_dir / "gpt-image-2-error.json",
                {
                    "provider": provider_cfg.provider,
                    "model": provider_cfg.model,
                    "base_url": provider_cfg.base_url,
                    "error": provider_error_debug_payload(gpt_error),
                    "reference_image_count": len(reference_images or []),
                },
            )
        if not allow_provider_fallback:
            raise RuntimeError(
                f"{model_label} 生成失败，未切换到其他图片模型。"
                "只有用户明确要求允许模型回退时，才会改用 Nano Banana/Gemini。"
                f"错误：{gpt_error}"
            ) from gpt_error
        if not is_provider_fallback_allowed(gpt_error):
            raise RuntimeError(
                f"{model_label} 请求已被服务端计费、权限或客户端策略拒绝，未跨 provider 回退。"
                "请根据结构化错误码修复订阅、余额、权限或计费服务状态后重试。"
                f"错误：{gpt_error}"
            ) from gpt_error
        warn(f"{model_label} 生成失败，用户已允许回退，改用 Nano Banana/Gemini：{exc}")
        try:
            gemini_cfg = load_gemini_config(remote_env_path=remote_env)
        except Exception as gemini_exc:
            raise RuntimeError(f"{model_label} 生成失败，且 Nano Banana/Gemini 回退不可用：gpt={gpt_error}; gemini={gemini_exc}") from gemini_exc
        result = generate_image_png(
            provider_cfg=ImageProviderConfig(
                provider="nano_banana",
                base_url=gemini_cfg.base_url,
                api_key=gemini_cfg.api_key,
                model=gemini_cfg.model,
                env_path=gemini_cfg.env_path,
                source="fallback",
            ),
            prompt=effective_prompt,
            output_png=output_png,
            canvas_w=canvas_w,
            canvas_h=canvas_h,
            reference_images=reference_images,
            debug_dir=debug_dir,
            timeout_s=int(api_cfg.get("request_timeout_s", 180)),
            retries=int(api_cfg.get("retry_attempts", 5)),
            postprocess_resize=bool(postprocess_resize),
            postprocess_w=postprocess_w,
            postprocess_h=postprocess_h,
            quality=quality,
            provider_size=provider_size,
            output_format=output_format,
            output_compression=output_compression,
        )
    if debug_dir is not None:
        write_json(debug_dir / "image-generation.json", result)
    return result


def main() -> None:
    parser = argparse.ArgumentParser(description="调用图片 provider 生成图片；OpenAI 图片模型默认请求低画质 JPEG。")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--prompt-file", help="prompt 文件路径")
    group.add_argument("--prompt-text", help="直接传入 prompt 文本")
    parser.add_argument("--output-image", "--output-png", dest="output_image", required=True, help="输出图片路径；--output-png 为兼容别名")
    parser.add_argument("--api-env", default="", help="remote.env 路径，默认 ~/.bensz-skills/config/remote.env")
    parser.add_argument("--canvas-width", type=int, default=1600, help="期望布局宽度/宽高比参考，不承诺最终图片像素")
    parser.add_argument("--canvas-height", type=int, default=900, help="期望布局高度/宽高比参考，不承诺最终图片像素")
    parser.add_argument("--postprocess-resize", action="store_true", default=None, help="显式启用后处理尺寸对齐；默认保留 provider 原生输出")
    parser.add_argument("--postprocess-width", type=int, default=0, help="后处理目标宽度；需配合 --postprocess-resize")
    parser.add_argument("--postprocess-height", type=int, default=0, help="后处理目标高度；需配合 --postprocess-resize")
    parser.add_argument("--quality", default="", help="OpenAI 图片模型 quality：low/medium/high/auto")
    parser.add_argument("--provider-size", default="", help="OpenAI 图片模型原生尺寸枚举，默认 1024x1024")
    parser.add_argument("--output-format", default="", help="OpenAI 图片模型输出格式：jpeg/png/webp")
    parser.add_argument("--output-compression", type=int, default=-1, help="输出压缩 0-100；默认使用配置值")
    parser.add_argument("--debug-dir", default="", help="调试目录")
    parser.add_argument("--reference-image", action="append", default=[], help="可重复传入参考图路径")
    parser.add_argument("--provider", default="auto", help="图片 provider/model：auto（默认）/ gpt-image-2.5-flare / gpt-image-2.5-sunburst / gpt-image-2 / nano_banana")
    parser.add_argument(
        "--allow-provider-fallback",
        action="store_true",
        help="provider 故障时允许从 OpenAI 图片模型切到 Nano Banana/Gemini；计费、权限与客户端策略错误仍不回退",
    )
    parser.add_argument("--require-reference-images", action="store_true", help="传入参考图时必须使用可消费参考图的 provider")
    args = parser.parse_args()

    prompt = (
        Path(args.prompt_file).read_text(encoding="utf-8")
        if args.prompt_file
        else str(args.prompt_text or "").strip()
    )
    if not prompt.strip():
        raise SystemExit("prompt 不能为空。")

    result = generate_image(
        prompt=prompt,
        output_png=expand_path(args.output_image, base=Path.cwd()),
        remote_env=expand_path(args.api_env, base=Path.cwd()) if args.api_env else None,
        canvas_w=int(args.canvas_width),
        canvas_h=int(args.canvas_height),
        debug_dir=expand_path(args.debug_dir, base=Path.cwd()) if args.debug_dir else None,
        reference_images=[expand_path(item, base=Path.cwd()) for item in (args.reference_image or [])],
        provider_name=str(args.provider or "auto"),
        allow_provider_fallback=bool(args.allow_provider_fallback),
        require_reference_images=bool(args.require_reference_images),
        postprocess_resize=args.postprocess_resize,
        postprocess_w=int(args.postprocess_width) or None,
        postprocess_h=int(args.postprocess_height) or None,
        quality=args.quality or None,
        provider_size=args.provider_size or None,
        output_format=args.output_format or None,
        output_compression=args.output_compression if args.output_compression >= 0 else None,
    )
    print(result.get("output_file") or result["output_png"])


if __name__ == "__main__":
    main()

from __future__ import annotations

import base64
import hashlib
import io
import json
import mimetypes
import os
import re
import struct
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

from common import ensure_dir, load_config, warn, write_json
from env_utils import find_remote_env, mask_secret, merged_env, resolve_config_path
from nano_banana_client import (
    generate_png as generate_nano_banana_png,
    load_gemini_config,
    nano_banana_health_check,
)


@dataclass(frozen=True)
class ImageProviderConfig:
    provider: str
    base_url: str
    api_key: str
    model: str
    env_path: Optional[Path]
    source: str
    preflight_status: str = "not_checked"
    preflight_scope: str = "configuration_only"
    generation_eligibility: str = "unknown_until_image_submit"
    config_path: Optional[Path] = None
    auth_path: Optional[Path] = None
    api_key_fingerprint: str = ""
    config_conflicts: Tuple[str, ...] = ()

    def with_preflight(self, *, status: str, scope: str) -> "ImageProviderConfig":
        return replace(
            self,
            preflight_status=str(status),
            preflight_scope=str(scope),
            generation_eligibility="unknown_until_image_submit",
        )


class ProviderUnavailable(RuntimeError):
    pass


class ProviderHTTPError(RuntimeError):
    def __init__(self, code: int, reason: str, detail: str, *, headers: Optional[Dict[str, str]] = None):
        self.code = int(code)
        self.reason = str(reason)
        self.detail = str(detail)
        self.headers = dict(headers or {})
        error = _parse_provider_error_detail(self.detail)
        self.error_type = error["type"]
        self.error_code = error["code"]
        self.retryable = error["retryable"]
        self.safe_message = error["message"]
        self.category = _classify_provider_error(http_status=self.code, error_code=self.error_code)
        machine_code = f" code={self.error_code}" if self.error_code else ""
        super().__init__(f"HTTP {self.code} {self.reason}{machine_code}: {self.safe_message}")


class ProviderProtocolError(RuntimeError):
    def __init__(
        self,
        *,
        error_code: str,
        http_status: int,
        http_reason: str,
        request_url: str,
        final_url: str,
        headers: Dict[str, str],
        body: bytes,
        request_client_request_id: str = "",
    ) -> None:
        self.error_type = "provider_protocol_error"
        self.error_code = str(error_code or "PROVIDER_NON_JSON_RESPONSE")
        self.http_status = int(http_status or 0)
        self.http_reason = re.sub(r"[\r\n]+", " ", str(http_reason or "").strip())[:200]
        self.request_location = _safe_url_origin_path(request_url)
        self.response_location = _safe_url_origin_path(final_url)
        self.was_redirected = str(request_url) != str(final_url)
        self.final_origin_changed = _safe_url_origin(request_url) != _safe_url_origin(final_url)
        self.content_type = _safe_response_header(headers, "Content-Type")
        self.content_length = _safe_response_header(headers, "Content-Length")
        self.request_id = _safe_correlation_id(_safe_response_header(headers, "X-Request-ID"))
        self.client_request_id = _safe_correlation_id(
            _safe_response_header(headers, "X-Client-Request-ID")
        ) or _safe_correlation_id(request_client_request_id)
        self.body_bytes = len(body)
        self.body_sha256 = hashlib.sha256(body).hexdigest()
        self.body_prefix_kind = _body_prefix_kind(body)
        self.retryable = False
        self.category = "provider_protocol_error"
        self.safe_message = (
            "image provider returned an empty success response"
            if self.error_code == "PROVIDER_EMPTY_RESPONSE"
            else "image provider returned a non-JSON success response"
        )
        super().__init__(
            f"{self.error_code}: HTTP {self.http_status} {self.http_reason}; "
            f"location={self.response_location}; body_bytes={self.body_bytes}"
        )


class ProviderJobError(RuntimeError):
    def __init__(
        self,
        *,
        job_id: str,
        status: str,
        error_type: str,
        error_code: str,
        message: str,
    ) -> None:
        self.job_id = str(job_id or "unknown")
        self.status = str(status or "failed")
        self.error_type = str(error_type or "provider_error")
        self.error_code = str(error_code or "IMAGE_GENERATION_FAILED")
        self.safe_message = _redact_sensitive_error_text(message or "image generation job failed")
        self.category = _classify_provider_error(http_status=None, error_code=self.error_code)
        super().__init__(
            "OpenAI 图片异步任务失败："
            f"job_id={self.job_id}; status={self.status}; code={self.error_code}; message={self.safe_message}"
        )


def health_check_image_provider(
    *,
    remote_env_path: Optional[Path] = None,
    provider_name: Optional[str] = None,
    timeout_s: int = 30,
) -> ImageProviderConfig:
    return resolve_image_provider(
        remote_env_path=remote_env_path,
        provider_name=provider_name,
        timeout_s=timeout_s,
        run_healthcheck=True,
    )


def resolve_image_provider(
    *,
    remote_env_path: Optional[Path] = None,
    provider_name: Optional[str] = None,
    timeout_s: int = 30,
    run_healthcheck: bool = False,
) -> ImageProviderConfig:
    cfg = load_config()
    api_cfg = cfg.get("api", {}) if isinstance(cfg.get("api"), dict) else {}
    requested_provider = _normalize_provider(provider_name or "")
    priority = (
        [str(provider_name or "").strip()]
        if requested_provider and requested_provider != "auto"
        else [str(item).strip() for item in (api_cfg.get("provider_priority") or ["gpt-image-2.5-flare", "nano_banana"])]
    )
    errors: List[str] = []
    for provider in priority:
        normalized = _normalize_provider(provider)
        try:
            if normalized == "gpt-image-2":
                image_cfg = load_gpt_image_2_config(
                    remote_env_path=remote_env_path,
                    model_override=_openai_model_from_provider_name(provider),
                )
                if run_healthcheck:
                    preflight_status = _openai_health_check(image_cfg, timeout_s=timeout_s)
                    image_cfg = image_cfg.with_preflight(
                        status=preflight_status,
                        scope=(
                            "connectivity_and_authentication_only"
                            if preflight_status == "connectivity/authentication_ok"
                            else "connectivity_only"
                        ),
                    )
                return image_cfg
            if normalized == "nano_banana":
                gemini_cfg = load_gemini_config(remote_env_path=remote_env_path)
                if run_healthcheck:
                    nano_banana_health_check(remote_env_path=remote_env_path, timeout_s=timeout_s)
                image_cfg = ImageProviderConfig(
                    provider="nano_banana",
                    base_url=gemini_cfg.base_url,
                    api_key=gemini_cfg.api_key,
                    model=gemini_cfg.model,
                    env_path=gemini_cfg.env_path,
                    source="remote_env",
                )
                if run_healthcheck:
                    image_cfg = image_cfg.with_preflight(
                        status="provider_probe_ok",
                        scope="provider_specific_probe_only",
                    )
                return image_cfg
        except Exception as exc:
            errors.append(f"{normalized}: {exc}")
            if requested_provider and requested_provider != "auto":
                raise ProviderUnavailable(
                    f"用户指定的图片 provider `{normalized}` 不可用，未切换到其他模型。原因：{exc}"
                ) from exc
            warn(f"图片 provider `{normalized}` 不可用，尝试下一个 provider。原因：{exc}")
    raise ProviderUnavailable("未找到可用图片生成 provider：" + " | ".join(errors))


def load_gpt_image_2_config(
    *,
    remote_env_path: Optional[Path] = None,
    model_override: Optional[str] = None,
) -> ImageProviderConfig:
    cfg = load_config()
    api_cfg = cfg.get("api", {}) if isinstance(cfg.get("api"), dict) else {}
    gpt_cfg = api_cfg.get("gpt_image_2", {}) if isinstance(api_cfg.get("gpt_image_2"), dict) else {}
    env_path = find_remote_env(remote_env_path)
    env = merged_env(env_path)

    codex_config_path, codex_auth_path = _codex_config_paths(api_cfg=api_cfg)
    codex = _load_codex_provider_config(
        config_path=codex_config_path,
        auth_path=codex_auth_path,
        provider_names=_codex_provider_names(api_cfg=api_cfg, data=None),
        auth_key_names=gpt_cfg.get("codex_auth_key_names") or gpt_cfg.get("env_api_key_keys") or ["OPENAI_API_KEY", "OPENAI_API"],
        provider_auth_key_names=gpt_cfg.get("codex_provider_auth_key_names") or ["experimental_bearer_token", "api_key", "api_token", "token"],
    )
    base_url, api_key, model, source = _resolve_gpt_image_2_inputs(codex=codex, env=env, gpt_cfg=gpt_cfg)
    conflicts = _config_conflicts(codex=codex, env=env, gpt_cfg=gpt_cfg)
    if conflicts:
        raise ProviderUnavailable(
            "检测到图片配置来源冲突（{}），请清理 Windows 环境变量或更新 Codex 配置后重试。".format(
                ", ".join(conflicts)
            )
        )
    model = str(model_override or model or _default_openai_image_model(gpt_cfg)).strip()
    if model_override:
        source = "+".join(dict.fromkeys([*source.split("+"), "provider_arg"])) if source else "provider_arg"
    allowed_models = _allowed_openai_image_models(gpt_cfg)
    if model not in allowed_models:
        raise ProviderUnavailable(
            "OpenAI 图片模型必须是 {} 之一，当前为 {!r}".format(", ".join(allowed_models), model)
        )
    if not base_url:
        raise ProviderUnavailable("缺少 OPENAI_BASE_URL / OPENAI_API_BASE，且未从 Codex 配置找到 BenszAPI base_url")
    if not api_key:
        raise ProviderUnavailable("缺少 OPENAI_API_KEY / OPENAI_API，且未从 Codex auth.json 找到可复用密钥")

    base_url = str(base_url).strip().rstrip("/")
    allowed_domains = [str(item).strip().lower() for item in (gpt_cfg.get("allowed_base_domains") or ["benszresearch.com"])]
    _validate_benszresearch_base_url(base_url, allowed_domains=allowed_domains)
    if urllib.parse.urlparse(base_url).path.rstrip("/") == "":
        base_url = f"{base_url}/v1"
    return ImageProviderConfig(
        provider=str(gpt_cfg.get("provider") or "gpt-image-2"),
        base_url=base_url,
        api_key=str(api_key).strip(),
        model=str(model).strip(),
        env_path=env_path,
        source=source,
        config_path=codex_config_path,
        auth_path=codex_auth_path,
        api_key_fingerprint=_secret_fingerprint(str(api_key).strip()),
        config_conflicts=tuple(conflicts),
    )


def generate_image_png(
    *,
    provider_cfg: ImageProviderConfig,
    prompt: str,
    output_png: Path,
    canvas_w: int,
    canvas_h: int,
    reference_images: Optional[List[Path]] = None,
    debug_dir: Optional[Path] = None,
    timeout_s: int = 180,
    retries: int = 5,
    postprocess_resize: bool = False,
    postprocess_w: Optional[int] = None,
    postprocess_h: Optional[int] = None,
    quality: Optional[str] = None,
    provider_size: Optional[str] = None,
    output_format: Optional[str] = None,
    output_compression: Optional[int] = None,
) -> Dict[str, Any]:
    _validate_postprocess_args(
        postprocess_resize=postprocess_resize,
        postprocess_w=postprocess_w,
        postprocess_h=postprocess_h,
    )
    if _is_openai_image_provider(provider_cfg):
        result = _generate_openai_png(
            cfg=provider_cfg,
            prompt=prompt,
            output_png=output_png,
            canvas_w=canvas_w,
            canvas_h=canvas_h,
            reference_images=reference_images,
            debug_dir=debug_dir,
            timeout_s=timeout_s,
            retries=retries,
            postprocess_resize=postprocess_resize,
            postprocess_w=postprocess_w,
            postprocess_h=postprocess_h,
            quality=quality,
            provider_size=provider_size,
            output_format=output_format,
            output_compression=output_compression,
        )
    else:
        gemini_cfg = load_gemini_config(remote_env_path=provider_cfg.env_path)
        result = generate_nano_banana_png(
            cfg=gemini_cfg,
            prompt=prompt,
            output_png=output_png,
            canvas_w=canvas_w,
            canvas_h=canvas_h,
            reference_images=reference_images,
            debug_dir=debug_dir,
            timeout_s=timeout_s,
            retries=retries,
            postprocess_resize=postprocess_resize,
            postprocess_w=postprocess_w,
            postprocess_h=postprocess_h,
        )
    result.update(
        {
            "provider": provider_cfg.provider,
            "model": provider_cfg.model,
            "base_url": provider_cfg.base_url,
            "env_path": str(provider_cfg.env_path) if provider_cfg.env_path else None,
            "provider_source": provider_cfg.source,
        }
    )
    if debug_dir is not None:
        write_json(debug_dir / "image-provider.json", _debug_provider(provider_cfg))
    return result


def _generate_openai_png(
    *,
    cfg: ImageProviderConfig,
    prompt: str,
    output_png: Path,
    canvas_w: int,
    canvas_h: int,
    reference_images: Optional[List[Path]],
    debug_dir: Optional[Path],
    timeout_s: int,
    retries: int,
    postprocess_resize: bool,
    postprocess_w: Optional[int],
    postprocess_h: Optional[int],
    quality: Optional[str] = None,
    provider_size: Optional[str] = None,
    output_format: Optional[str] = None,
    output_compression: Optional[int] = None,
) -> Dict[str, Any]:
    refs = [Path(p) for p in (reference_images or [])]
    model_label = _openai_model_label(cfg)
    if refs:
        return _generate_openai_edit_png(
            cfg=cfg,
            prompt=prompt,
            output_png=output_png,
            canvas_w=canvas_w,
            canvas_h=canvas_h,
            reference_images=refs,
            debug_dir=debug_dir,
            timeout_s=timeout_s,
            retries=retries,
            postprocess_resize=postprocess_resize,
            postprocess_w=postprocess_w,
            postprocess_h=postprocess_h,
            quality=quality,
            provider_size=provider_size,
            output_format=output_format,
            output_compression=output_compression,
        )
    requested_size = _openai_requested_size(canvas_w, canvas_h, provider_size)
    quality, output_format, output_compression = _openai_generation_options(
        quality, output_format, output_compression
    )
    submit_retries = min(max(1, retries), _openai_submit_retry_attempts())
    payload: Dict[str, Any] = {
        "model": cfg.model,
        "prompt": prompt,
        "size": requested_size,
        "n": 1,
        "quality": quality,
        "output_format": output_format,
    }
    if output_compression is not None:
        payload["output_compression"] = output_compression
    endpoint_path = "/images/jobs/generations" if _use_openai_async_job_endpoint() else "/images/generations"
    submit_mode = _openai_submit_mode(endpoint_path)
    if debug_dir is not None:
        ensure_dir(debug_dir)
        write_json(
            debug_dir / "request.json",
            _openai_request_debug_payload(
                cfg=cfg,
                endpoint=endpoint_path,
                submit_mode=submit_mode,
                payload=payload,
            ),
        )

    try:
        response = _post_json_with_retries(
            url=f"{cfg.base_url}{endpoint_path}",
            payload=payload,
            headers={"Authorization": f"Bearer {cfg.api_key}"},
            timeout_s=timeout_s,
            retries=submit_retries,
            retry_message=f"{model_label} 暂时不可用",
        )
    except ProviderHTTPError as exc:
        if not _can_fallback_openai_async_job_to_sync(exc):
            raise
        if debug_dir is not None:
            write_json(
                debug_dir / "async-job-unsupported.json",
                {
                    "endpoint": endpoint_path,
                    "submit_mode": submit_mode,
                    "error": provider_error_debug_payload(exc),
                    "fallback_endpoint": "/images/generations",
                },
            )
            write_json(
                debug_dir / "request-sync-fallback.json",
                _openai_request_debug_payload(
                    cfg=cfg,
                    endpoint="/images/generations",
                    submit_mode="sync_unsupported_fallback",
                    payload=payload,
                ),
            )
        endpoint_path = "/images/generations"
        submit_mode = "sync_unsupported_fallback"
        response = _post_json_with_retries(
            url=f"{cfg.base_url}{endpoint_path}",
            payload=payload,
            headers={"Authorization": f"Bearer {cfg.api_key}"},
            timeout_s=timeout_s,
            retries=submit_retries,
            retry_message=f"{model_label} 兼容同步接口暂时不可用",
        )
    response = _resolve_openai_image_response(
        initial_response=response,
        cfg=cfg,
        endpoint=endpoint_path,
        debug_dir=debug_dir,
        timeout_s=timeout_s,
    )
    if debug_dir is not None:
        write_json(debug_dir / "response.json", _sanitize_image_response(response))

    best = _best_image(_extract_openai_images(response))
    if best is None:
        excerpt = json.dumps(response, ensure_ascii=False)[:800]
        raise RuntimeError(f"未从 {model_label} 响应中提取到图片。response_excerpt={excerpt}")
    mime, raw = best
    mime = _write_provider_image(output_png, mime, raw)
    size_meta = _build_output_size_meta(
        output_png,
        postprocess_resize=postprocess_resize,
        canvas_w=canvas_w,
        canvas_h=canvas_h,
        postprocess_w=postprocess_w,
        postprocess_h=postprocess_h,
    )
    return {
        "mime_type": mime,
        "output_file": str(output_png),
        "output_png": str(output_png),
        "requested_provider_size": _openai_size_meta(requested_size),
        **size_meta,
        "response_path": str(debug_dir / "response.json") if debug_dir is not None else None,
        "submit_mode": submit_mode,
        "endpoint": endpoint_path,
    }


def _generate_openai_edit_png(
    *,
    cfg: ImageProviderConfig,
    prompt: str,
    output_png: Path,
    canvas_w: int,
    canvas_h: int,
    reference_images: List[Path],
    debug_dir: Optional[Path],
    timeout_s: int,
    retries: int,
    postprocess_resize: bool,
    postprocess_w: Optional[int],
    postprocess_h: Optional[int],
    quality: Optional[str] = None,
    provider_size: Optional[str] = None,
    output_format: Optional[str] = None,
    output_compression: Optional[int] = None,
) -> Dict[str, Any]:
    model_label = _openai_model_label(cfg)
    refs = _existing_reference_images(reference_images)
    requested_size = _openai_requested_size(canvas_w, canvas_h, provider_size)
    quality, output_format, output_compression = _openai_generation_options(
        quality, output_format, output_compression
    )
    submit_retries = min(max(1, retries), _openai_submit_retry_attempts())
    fields: Dict[str, str] = {
        "model": cfg.model,
        "prompt": prompt,
        "size": requested_size,
        "n": "1",
        "quality": quality,
        "output_format": output_format,
    }
    if output_compression is not None:
        fields["output_compression"] = str(output_compression)
    files = _openai_edit_file_parts(refs)
    endpoint_path = "/images/jobs/edits" if _use_openai_async_job_endpoint() else "/images/edits"
    submit_mode = _openai_submit_mode(endpoint_path)
    if debug_dir is not None:
        ensure_dir(debug_dir)
        write_json(
            debug_dir / "request.json",
            {
                "provider": cfg.provider,
                "base_url": cfg.base_url,
                "model": cfg.model,
                "endpoint": endpoint_path,
                "submit_mode": submit_mode,
                "headers": _redacted_openai_headers(cfg),
                "payload": fields,
                "files": [
                    {
                        "field": field,
                        "filename": filename,
                        "mime_type": mime_type,
                        "size_bytes": len(raw),
                        "sha256": hashlib.sha256(raw).hexdigest(),
                    }
                    for field, filename, mime_type, raw in files
                ],
            },
        )

    try:
        response = _post_multipart_with_retries(
            url=f"{cfg.base_url}{endpoint_path}",
            fields=fields,
            files=files,
            headers={"Authorization": f"Bearer {cfg.api_key}"},
            timeout_s=timeout_s,
            retries=submit_retries,
            retry_message=f"{model_label} 编辑暂时不可用",
        )
    except ProviderHTTPError as exc:
        if not _can_fallback_openai_async_job_to_sync(exc):
            raise
        if debug_dir is not None:
            write_json(
                debug_dir / "async-job-unsupported.json",
                {
                    "endpoint": endpoint_path,
                    "submit_mode": submit_mode,
                    "error": provider_error_debug_payload(exc),
                    "fallback_endpoint": "/images/edits",
                },
            )
            write_json(
                debug_dir / "request-sync-fallback.json",
                {
                    "provider": cfg.provider,
                    "base_url": cfg.base_url,
                    "model": cfg.model,
                    "endpoint": "/images/edits",
                    "submit_mode": "sync_unsupported_fallback",
                    "headers": _redacted_openai_headers(cfg),
                    "payload": fields,
                    "files": [
                        {
                            "field": field,
                            "filename": filename,
                            "mime_type": mime_type,
                            "size_bytes": len(raw),
                            "sha256": hashlib.sha256(raw).hexdigest(),
                        }
                        for field, filename, mime_type, raw in files
                    ],
                },
            )
        endpoint_path = "/images/edits"
        submit_mode = "sync_unsupported_fallback"
        response = _post_multipart_with_retries(
            url=f"{cfg.base_url}{endpoint_path}",
            fields=fields,
            files=files,
            headers={"Authorization": f"Bearer {cfg.api_key}"},
            timeout_s=timeout_s,
            retries=submit_retries,
            retry_message=f"{model_label} 兼容编辑接口暂时不可用",
        )
    response = _resolve_openai_image_response(
        initial_response=response,
        cfg=cfg,
        endpoint=endpoint_path,
        debug_dir=debug_dir,
        timeout_s=timeout_s,
    )
    if debug_dir is not None:
        write_json(debug_dir / "response.json", _sanitize_image_response(response))

    best = _best_image(_extract_openai_images(response))
    if best is None:
        excerpt = json.dumps(response, ensure_ascii=False)[:800]
        raise RuntimeError(f"未从 {model_label} 编辑响应中提取到图片。response_excerpt={excerpt}")
    mime, raw = best
    mime = _write_provider_image(output_png, mime, raw)
    size_meta = _build_output_size_meta(
        output_png,
        postprocess_resize=postprocess_resize,
        canvas_w=canvas_w,
        canvas_h=canvas_h,
        postprocess_w=postprocess_w,
        postprocess_h=postprocess_h,
    )
    return {
        "mime_type": mime,
        "output_file": str(output_png),
        "output_png": str(output_png),
        "requested_provider_size": _openai_size_meta(requested_size),
        **size_meta,
        "response_path": str(debug_dir / "response.json") if debug_dir is not None else None,
        "reference_image_count": len(refs),
        "reference_images": [
            {
                "path": str(ref),
                "sha256": hashlib.sha256(ref.read_bytes()).hexdigest(),
                "source": "reference_image",
            }
            for ref in refs
        ],
        "operation": "image-edit",
        "submit_mode": submit_mode,
        "endpoint": endpoint_path,
    }


def _openai_health_check(cfg: ImageProviderConfig, *, timeout_s: int) -> str:
    parsed = urllib.parse.urlparse(cfg.base_url)
    probe_url = urllib.parse.urlunparse((parsed.scheme, parsed.netloc, "/v1/models", "", "", ""))
    if cfg.base_url.endswith("/v1"):
        probe_url = f"{cfg.base_url}/models"
    try:
        _post_json(probe_url, None, method="GET", headers={"Authorization": f"Bearer {cfg.api_key}"}, timeout_s=timeout_s)
    except ProviderHTTPError as exc:
        if exc.code in {404, 405}:
            return "connectivity_ok_models_probe_unsupported"
        raise
    return "connectivity/authentication_ok"


_OPENAI_TRANSIENT_HTTP_CODES = {429, 500, 502, 503, 504}
_OPENAI_ASYNC_UNSUPPORTED_HTTP_CODES = {404, 405, 501}
_CLIENT_POLICY_ERROR_CODES = {
    "BILLING_PRICING_NOT_CONFIGURED",
    "SUBSCRIPTION_REQUIRED",
    "OVERAGE_LIMIT_EXCEEDED",
    "INSUFFICIENT_BALANCE",
    "INSUFFICIENT_QUOTA",
    "PERMISSION_DENIED",
    "ACCESS_DENIED",
    "BILLING_NOT_ALLOWED",
}
_PROVIDER_FALLBACK_BLOCKED_ERROR_CODES = _CLIENT_POLICY_ERROR_CODES | {
    "BILLING_SERVICE_ERROR",
    "PROVIDER_EMPTY_RESPONSE",
    "PROVIDER_NON_JSON_RESPONSE",
}


def _parse_provider_error_detail(detail: str) -> Dict[str, Any]:
    raw = str(detail or "").strip()
    payload: Any = None
    if raw:
        try:
            payload = json.loads(raw)
        except Exception:
            payload = None
    error = payload.get("error") if isinstance(payload, dict) else None
    if not isinstance(error, dict):
        error = payload if isinstance(payload, dict) else {}
    error_type = _safe_scalar(error.get("type"))
    error_code = _safe_scalar(error.get("code"))
    message = _safe_scalar(error.get("message") or error.get("detail"))
    if not message:
        message = raw if payload is None else "image provider request failed"
    return {
        "type": error_type,
        "code": error_code,
        "message": _redact_sensitive_error_text(message),
        "retryable": error.get("retryable") if isinstance(error.get("retryable"), bool) else None,
    }


def _safe_scalar(value: Any) -> str:
    if isinstance(value, (str, int, float)):
        return str(value).strip()[:800]
    return ""


def _redact_sensitive_error_text(value: str) -> str:
    text = str(value or "").strip()[:800]
    text = re.sub(r"(?i)Bearer\s+[^\s,;]+", "Bearer [redacted]", text)
    text = re.sub(r"\bsk-[A-Za-z0-9_-]{8,}\b", "[redacted:api_key]", text)
    return text


def _classify_provider_error(*, http_status: Optional[int], error_code: str) -> str:
    normalized_code = str(error_code or "").strip().upper()
    if normalized_code in _CLIENT_POLICY_ERROR_CODES:
        return "client_policy_error"
    if normalized_code == "BILLING_SERVICE_ERROR":
        return "transient_platform_error"
    if http_status in _OPENAI_TRANSIENT_HTTP_CODES:
        return "transient_platform_error"
    if http_status is not None and 400 <= int(http_status) < 500:
        return "client_policy_error"
    return "provider_error"


def _should_retry_provider_http_error(exc: ProviderHTTPError) -> bool:
    if exc.retryable is False:
        return False
    if exc.retryable is True:
        return exc.code in _OPENAI_TRANSIENT_HTTP_CODES
    return exc.code in _OPENAI_TRANSIENT_HTTP_CODES and exc.category == "transient_platform_error"


def is_provider_fallback_allowed(exc: BaseException) -> bool:
    if isinstance(exc, ProviderProtocolError):
        return False
    error_code = str(getattr(exc, "error_code", "") or "").strip().upper()
    category = str(getattr(exc, "category", "") or "").strip()
    cfg = load_config()
    api_cfg = cfg.get("api", {}) if isinstance(cfg.get("api"), dict) else {}
    configured = api_cfg.get("provider_fallback_blocked_error_codes")
    blocked_codes = (
        {str(item).strip().upper() for item in configured if str(item).strip()}
        if isinstance(configured, list)
        else _PROVIDER_FALLBACK_BLOCKED_ERROR_CODES
    )
    if error_code in blocked_codes:
        return False
    if category == "client_policy_error":
        return False
    return True


def provider_error_debug_payload(exc: BaseException) -> Dict[str, Any]:
    payload: Dict[str, Any] = {
        "category": str(getattr(exc, "category", "provider_error") or "provider_error"),
        "exception_type": type(exc).__name__,
    }
    if isinstance(exc, ProviderHTTPError):
        payload.update(
            {
                "http_status": exc.code,
                "http_reason": exc.reason,
                "error": {
                    "type": exc.error_type,
                    "code": exc.error_code,
                    "message": exc.safe_message,
                },
            }
        )
        return payload
    if isinstance(exc, ProviderProtocolError):
        payload.update(
            {
                "error": {
                    "type": exc.error_type,
                    "code": exc.error_code,
                    "message": exc.safe_message,
                },
                "response": {
                    "http_status": exc.http_status,
                    "http_reason": exc.http_reason,
                    "request_location": exc.request_location,
                    "response_location": exc.response_location,
                    "content_type": exc.content_type,
                    "content_length": exc.content_length,
                    "body_bytes": exc.body_bytes,
                    "body_sha256": exc.body_sha256,
                    "body_prefix_kind": exc.body_prefix_kind,
                    "was_redirected": exc.was_redirected,
                    "final_origin_changed": exc.final_origin_changed,
                    "request_id": exc.request_id,
                    "client_request_id": exc.client_request_id,
                },
            }
        )
        return payload
    if isinstance(exc, ProviderJobError):
        payload.update(
            {
                "job_id": exc.job_id,
                "status": exc.status,
                "error": {
                    "type": exc.error_type,
                    "code": exc.error_code,
                    "message": exc.safe_message,
                },
            }
        )
        return payload
    payload["message"] = _redact_sensitive_error_text(str(exc))
    return payload


def _post_json_with_retries(
    *,
    url: str,
    payload: Optional[Dict[str, Any]],
    headers: Dict[str, str],
    timeout_s: int,
    retries: int,
    retry_message: str,
) -> Dict[str, Any]:
    last_error: Optional[ProviderHTTPError] = None
    for attempt in range(1, max(1, retries) + 1):
        try:
            return _post_json(
                url,
                payload,
                headers=headers,
                timeout_s=timeout_s,
            )
        except ProviderHTTPError as exc:
            last_error = exc
            if not _should_retry_provider_http_error(exc) or attempt >= retries:
                break
            wait_s = _retry_after(exc.headers) or min(30.0, 2.0**attempt)
            warn(f"{retry_message}（HTTP {exc.code}），{wait_s:.1f}s 后重试（{attempt}/{retries}）。")
            time.sleep(max(0.5, wait_s))
    if last_error is not None:
        raise last_error
    raise RuntimeError("图片 provider 请求失败，但没有捕获到明确错误。")


def _post_multipart_with_retries(
    *,
    url: str,
    fields: Dict[str, str],
    files: List[Tuple[str, str, str, bytes]],
    headers: Dict[str, str],
    timeout_s: int,
    retries: int,
    retry_message: str,
) -> Dict[str, Any]:
    last_error: Optional[ProviderHTTPError] = None
    for attempt in range(1, max(1, retries) + 1):
        try:
            return _post_multipart(
                url,
                fields=fields,
                files=files,
                headers=headers,
                timeout_s=timeout_s,
            )
        except ProviderHTTPError as exc:
            last_error = exc
            if not _should_retry_provider_http_error(exc) or attempt >= retries:
                break
            wait_s = _retry_after(exc.headers) or min(30.0, 2.0**attempt)
            warn(f"{retry_message}（HTTP {exc.code}），{wait_s:.1f}s 后重试（{attempt}/{retries}）。")
            time.sleep(max(0.5, wait_s))
    if last_error is not None:
        raise last_error
    raise RuntimeError("图片 provider multipart 请求失败，但没有捕获到明确错误。")


def _use_openai_async_job_endpoint() -> bool:
    async_cfg = _async_image_job_config()
    if not bool(async_cfg.get("enabled", True)):
        return False
    submit_mode = str(async_cfg.get("submit_mode") or "sub2api_job_endpoint").strip().lower()
    return submit_mode in {"sub2api_job_endpoint", "job_endpoint", "jobs"}


def _openai_submit_mode(endpoint_path: str) -> str:
    if str(endpoint_path).startswith("/images/jobs/"):
        return str(_async_image_job_config().get("submit_mode") or "sub2api_job_endpoint")
    return "sync"


def _can_fallback_openai_async_job_to_sync(exc: ProviderHTTPError) -> bool:
    if not _use_openai_async_job_endpoint():
        return False
    async_cfg = _async_image_job_config()
    if not bool(async_cfg.get("fallback_to_sync_on_unsupported", True)):
        return False
    return int(exc.code) in _OPENAI_ASYNC_UNSUPPORTED_HTTP_CODES


def _openai_request_debug_payload(
    *,
    cfg: ImageProviderConfig,
    endpoint: str,
    submit_mode: str,
    payload: Dict[str, Any],
) -> Dict[str, Any]:
    return {
        "provider": cfg.provider,
        "base_url": cfg.base_url,
        "model": cfg.model,
        "endpoint": endpoint,
        "submit_mode": submit_mode,
        "headers": _redacted_openai_headers(cfg),
        "payload": payload,
    }


def _redacted_openai_headers(cfg: ImageProviderConfig) -> Dict[str, str]:
    return {"Authorization": f"Bearer {mask_secret(cfg.api_key)}"}


def _resolve_openai_image_response(
    *,
    initial_response: Dict[str, Any],
    cfg: ImageProviderConfig,
    endpoint: str,
    debug_dir: Optional[Path],
    timeout_s: int,
) -> Dict[str, Any]:
    if _best_image(_extract_openai_images(initial_response)) is not None:
        return initial_response

    job = _extract_openai_async_job(initial_response)
    if not job:
        return initial_response

    async_cfg = _async_image_job_config()
    if not bool(async_cfg.get("enabled", True)):
        return initial_response

    if debug_dir is not None:
        ensure_dir(debug_dir)
        write_json(debug_dir / "async-job-initial.json", _sanitize_image_response(initial_response))

    status_urls = _openai_async_status_urls(
        cfg=cfg,
        job_id=str(job.get("job_id") or ""),
        status_url=str(job.get("status_url") or ""),
        templates=async_cfg.get("status_endpoint_templates"),
    )
    if not status_urls:
        raise RuntimeError(
            f"{_openai_model_label(cfg)} 返回了异步图片任务，但响应中没有可轮询的 status_url 或 job_id。"
            f"endpoint={endpoint}; status={job.get('status') or 'unknown'}"
        )

    max_wait_s = max(1.0, float(async_cfg.get("max_wait_s", timeout_s) or timeout_s))
    poll_interval_s = max(0.5, float(async_cfg.get("poll_interval_s", 5) or 5))
    poll_timeout_s = max(1, int(async_cfg.get("poll_timeout_s", min(timeout_s, 60)) or min(timeout_s, 60)))
    started = time.monotonic()
    current = initial_response
    poll_log: List[Dict[str, Any]] = []
    attempt = 0
    while time.monotonic() - started <= max_wait_s:
        images = _extract_openai_images(current)
        if _best_image(images) is not None:
            if debug_dir is not None:
                write_json(debug_dir / "async-job-polls.json", poll_log)
            return current

        job = _extract_openai_async_job(current) or job
        status = str(job.get("status") or "").strip().lower()
        if status in _OPENAI_ASYNC_FAILURE_STATUSES:
            raise ProviderJobError(
                job_id=str(job.get("job_id") or "unknown"),
                status=status,
                error_type=str(job.get("error_type") or "provider_error"),
                error_code=str(job.get("error_code") or "IMAGE_GENERATION_FAILED"),
                message=str(job.get("error_message") or job.get("detail") or "image generation job failed"),
            )
        if status in _OPENAI_ASYNC_SUCCESS_STATUSES:
            result = _resolve_openai_async_job_result(
                current=current,
                cfg=cfg,
                job=job,
                debug_dir=debug_dir,
                timeout_s=poll_timeout_s,
                result_templates=async_cfg.get("result_endpoint_templates"),
            )
            if _best_image(_extract_openai_images(result)) is not None:
                if debug_dir is not None:
                    write_json(debug_dir / "async-job-polls.json", poll_log)
                return result
            if attempt > 0:
                raise RuntimeError(
                    f"{_openai_model_label(cfg)} 异步图片任务已完成，但响应中没有可提取图片："
                    f"job_id={job.get('job_id') or 'unknown'}"
                )

        if attempt > 0:
            time.sleep(poll_interval_s)
        attempt += 1
        current = _poll_openai_async_job(
            status_urls=status_urls,
            api_key=cfg.api_key,
            timeout_s=poll_timeout_s,
        )
        polled_job = _extract_openai_async_job(current) or {}
        poll_log.append(
            {
                "attempt": attempt,
                "elapsed_s": round(time.monotonic() - started, 3),
                "status": polled_job.get("status") or status or "unknown",
                "job_id": polled_job.get("job_id") or job.get("job_id"),
                "response": _sanitize_image_response(current),
            }
        )
        if debug_dir is not None:
            write_json(debug_dir / "async-job-polls.json", poll_log)

    raise TimeoutError(
        f"{_openai_model_label(cfg)} 异步图片任务轮询超时："
        f"job_id={job.get('job_id') or 'unknown'}; max_wait_s={max_wait_s:g}; endpoint={endpoint}"
    )


def _resolve_openai_async_job_result(
    *,
    current: Dict[str, Any],
    cfg: ImageProviderConfig,
    job: Dict[str, Any],
    debug_dir: Optional[Path],
    timeout_s: int,
    result_templates: Any,
) -> Dict[str, Any]:
    response_obj = _first_dict_value(current, ("response", "result", "output"))
    if response_obj and _best_image(_extract_openai_images(response_obj)) is not None:
        if debug_dir is not None:
            write_json(debug_dir / "async-job-result.json", _sanitize_image_response(response_obj))
        return response_obj

    result_urls = _openai_async_result_urls(
        cfg=cfg,
        job_id=str(job.get("job_id") or ""),
        result_url=str(job.get("result_url") or ""),
        templates=result_templates,
    )
    for result_url in result_urls:
        result = _post_json(
            result_url,
            None,
            method="GET",
            headers={"Authorization": f"Bearer {cfg.api_key}"},
            timeout_s=timeout_s,
        )
        if debug_dir is not None:
            write_json(
                debug_dir / "async-job-result.json",
                {
                    "result_url": result_url,
                    "response": _sanitize_image_response(result),
                },
            )
        return result

    return current


_OPENAI_ASYNC_PENDING_STATUSES = {
    "created",
    "queued",
    "pending",
    "submitted",
    "starting",
    "running",
    "processing",
    "in_progress",
    "in-progress",
}
_OPENAI_ASYNC_SUCCESS_STATUSES = {"completed", "complete", "succeeded", "success", "done", "finished"}
_OPENAI_ASYNC_FAILURE_STATUSES = {"failed", "failure", "error", "cancelled", "canceled", "expired", "rejected"}


def _async_image_job_config() -> Dict[str, Any]:
    cfg = load_config()
    api_cfg = cfg.get("api", {}) if isinstance(cfg.get("api"), dict) else {}
    async_cfg = api_cfg.get("async_image_job", {}) if isinstance(api_cfg.get("async_image_job"), dict) else {}
    return dict(async_cfg)


def _extract_openai_async_job(resp: Dict[str, Any]) -> Dict[str, Any]:
    for item in _iter_response_dicts(resp):
        status = _first_string(item, ("status", "state", "phase"))
        status_normalized = status.strip().lower()
        job_id = _first_string(
            item,
            (
                "job_id",
                "jobId",
                "task_id",
                "taskId",
                "generation_id",
                "generationId",
                "request_id",
                "requestId",
                "id",
            ),
        )
        status_url = _first_string(
            item,
            ("status_url", "statusUrl", "poll_url", "pollUrl", "polling_url", "pollingUrl"),
        )
        result_url = _first_string(
            item,
            ("result_url", "resultUrl", "response_url", "responseUrl", "output_url", "outputUrl"),
        )
        object_type = _first_string(item, ("object", "type", "kind")).strip().lower()
        explicit_job_id = _first_string(
            item,
            ("job_id", "jobId", "task_id", "taskId", "generation_id", "generationId"),
        )
        looks_async = bool(status_url) or bool(result_url) or bool(explicit_job_id) or object_type in {
            "image.job",
            "image_job",
            "job",
            "task",
            "generation.job",
        } or status_normalized in (
            _OPENAI_ASYNC_PENDING_STATUSES | _OPENAI_ASYNC_SUCCESS_STATUSES | _OPENAI_ASYNC_FAILURE_STATUSES
        )
        if not looks_async:
            continue
        error_fields = _structured_error_fields(item)
        return {
            "job_id": job_id,
            "status": status_normalized or status,
            "status_url": status_url,
            "result_url": result_url,
            "detail": _detail_string(item),
            "error_type": error_fields["type"],
            "error_code": error_fields["code"],
            "error_message": error_fields["message"],
        }
    return {}


def _iter_response_dicts(value: Any) -> List[Dict[str, Any]]:
    out: List[Dict[str, Any]] = []

    def visit(node: Any, depth: int) -> None:
        if depth > 5:
            return
        if isinstance(node, dict):
            out.append(node)
            for child in node.values():
                if isinstance(child, (dict, list)):
                    visit(child, depth + 1)
        elif isinstance(node, list):
            for child in node:
                if isinstance(child, (dict, list)):
                    visit(child, depth + 1)

    visit(value, 0)
    return out


def _first_string(data: Dict[str, Any], keys: Tuple[str, ...]) -> str:
    for key in keys:
        value = data.get(key)
        if isinstance(value, str) and value.strip():
            return value.strip()
        if isinstance(value, (int, float)) and str(value).strip():
            return str(value).strip()
    return ""


def _first_dict_value(data: Dict[str, Any], keys: Tuple[str, ...]) -> Dict[str, Any]:
    for key in keys:
        value = data.get(key)
        if isinstance(value, dict):
            return value
    return {}


def _detail_string(data: Dict[str, Any]) -> str:
    value = _first_string(data, ("message", "detail", "failure_reason", "failureReason"))
    if value:
        return value
    error = data.get("error")
    if isinstance(error, str):
        return error.strip()
    if isinstance(error, dict):
        return _first_string(error, ("message", "detail", "code", "type")) or json.dumps(error, ensure_ascii=False)[:800]
    return ""


def _structured_error_fields(data: Dict[str, Any]) -> Dict[str, str]:
    error = data.get("error")
    source = error if isinstance(error, dict) else data
    message = _safe_scalar(source.get("message") or source.get("detail"))
    if not message and isinstance(error, str):
        message = error.strip()
    return {
        "type": _safe_scalar(source.get("type")),
        "code": _safe_scalar(source.get("code")),
        "message": _redact_sensitive_error_text(message),
    }


def _openai_async_status_urls(
    *,
    cfg: ImageProviderConfig,
    job_id: str,
    status_url: str,
    templates: Any,
) -> List[str]:
    urls: List[str] = []
    normalized_status_url = _normalize_openai_status_url(cfg.base_url, status_url)
    if normalized_status_url:
        urls.append(normalized_status_url)
    if job_id:
        if not isinstance(templates, list) or not templates:
            templates = ["{base_url}/images/jobs/{job_id}", "{base_url}/images/generations/{job_id}"]
        quoted_job_id = urllib.parse.quote(job_id, safe="")
        for template in templates:
            try:
                rendered = str(template).format(base_url=cfg.base_url.rstrip("/"), job_id=quoted_job_id)
            except Exception:
                continue
            normalized = _normalize_openai_status_url(cfg.base_url, rendered)
            if normalized:
                urls.append(normalized)
    return list(dict.fromkeys(urls))


def _openai_async_result_urls(
    *,
    cfg: ImageProviderConfig,
    job_id: str,
    result_url: str,
    templates: Any,
) -> List[str]:
    urls: List[str] = []
    normalized_result_url = _normalize_openai_status_url(cfg.base_url, result_url)
    if normalized_result_url:
        urls.append(normalized_result_url)
    if job_id:
        if not isinstance(templates, list) or not templates:
            templates = ["{base_url}/images/jobs/{job_id}/result"]
        quoted_job_id = urllib.parse.quote(job_id, safe="")
        for template in templates:
            try:
                rendered = str(template).format(base_url=cfg.base_url.rstrip("/"), job_id=quoted_job_id)
            except Exception:
                continue
            normalized = _normalize_openai_status_url(cfg.base_url, rendered)
            if normalized:
                urls.append(normalized)
    return list(dict.fromkeys(urls))


def _normalize_openai_status_url(base_url: str, status_url: str) -> str:
    raw = str(status_url or "").strip()
    if not raw:
        return ""
    parsed_base = urllib.parse.urlparse(base_url)
    parsed = urllib.parse.urlparse(raw)
    if not parsed.scheme:
        if not raw.startswith("/"):
            raw = "/" + raw
        raw = urllib.parse.urlunparse((parsed_base.scheme, parsed_base.netloc, raw, "", "", ""))
        parsed = urllib.parse.urlparse(raw)
    if parsed.scheme != "https" or parsed.netloc.lower() != parsed_base.netloc.lower():
        return ""
    return urllib.parse.urlunparse((parsed.scheme, parsed.netloc, parsed.path, "", parsed.query, ""))


def _poll_openai_async_job(*, status_urls: List[str], api_key: str, timeout_s: int) -> Dict[str, Any]:
    errors: List[str] = []
    for url in status_urls:
        try:
            return _post_json(
                url,
                None,
                method="GET",
                headers={"Authorization": f"Bearer {api_key}"},
                timeout_s=timeout_s,
            )
        except ProviderHTTPError as exc:
            errors.append(f"{url}: HTTP {exc.code} {exc.reason}")
            if exc.code not in {404, 405}:
                raise
    raise RuntimeError("异步图片任务状态接口不可用：" + " | ".join(errors))


def _post_json(
    url: str,
    payload: Optional[Dict[str, Any]],
    *,
    method: str = "POST",
    headers: Optional[Dict[str, str]] = None,
    timeout_s: int = 60,
) -> Dict[str, Any]:
    body = None if payload is None else json.dumps(payload, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(url, data=body, method=method.upper())
    if body is not None:
        req.add_header("Content-Type", "application/json")
    req.add_header("Accept", "application/json")
    req.add_header("User-Agent", "OpenAI/Python 1.0.0")
    for key, value in (headers or {}).items():
        req.add_header(key, value)
    client_request_id = _ensure_client_request_id(req)
    try:
        with urllib.request.urlopen(req, timeout=timeout_s) as resp:
            raw = resp.read()
            response_status = _response_status(resp)
            response_reason = str(getattr(resp, "reason", "") or "")
            response_url = str(resp.geturl() or url)
            response_headers = {str(k): str(v) for k, v in getattr(resp, "headers", {}).items()}
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        hdrs = {str(k): str(v) for k, v in getattr(exc, "headers", {}).items()}
        raise ProviderHTTPError(int(exc.code), str(exc.reason), detail[:1600], headers=hdrs) from exc
    except Exception as exc:
        raise RuntimeError(f"请求图片 provider 失败：{exc}") from exc
    data = _decode_provider_json_response(
        raw,
        http_status=response_status,
        http_reason=response_reason,
        request_url=url,
        final_url=response_url,
        headers=response_headers,
        request_client_request_id=client_request_id,
    )
    if not isinstance(data, dict):
        raise RuntimeError("图片 provider 响应不是 JSON object。")
    return data


def _post_multipart(
    url: str,
    *,
    fields: Dict[str, str],
    files: List[Tuple[str, str, str, bytes]],
    headers: Optional[Dict[str, str]] = None,
    timeout_s: int = 60,
) -> Dict[str, Any]:
    boundary = "----auto-draw-plot-" + base64.urlsafe_b64encode(os.urandom(18)).decode("ascii").rstrip("=")
    body = _encode_multipart(fields=fields, files=files, boundary=boundary)
    req = urllib.request.Request(url, data=body, method="POST")
    req.add_header("Content-Type", f"multipart/form-data; boundary={boundary}")
    req.add_header("Accept", "application/json")
    req.add_header("User-Agent", "OpenAI/Python 1.0.0")
    for key, value in (headers or {}).items():
        req.add_header(key, value)
    client_request_id = _ensure_client_request_id(req)
    try:
        with urllib.request.urlopen(req, timeout=timeout_s) as resp:
            raw = resp.read()
            response_status = _response_status(resp)
            response_reason = str(getattr(resp, "reason", "") or "")
            response_url = str(resp.geturl() or url)
            response_headers = {str(k): str(v) for k, v in getattr(resp, "headers", {}).items()}
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        hdrs = {str(k): str(v) for k, v in getattr(exc, "headers", {}).items()}
        raise ProviderHTTPError(int(exc.code), str(exc.reason), detail[:1600], headers=hdrs) from exc
    except Exception as exc:
        raise RuntimeError(f"请求图片 provider 失败：{exc}") from exc
    data = _decode_provider_json_response(
        raw,
        http_status=response_status,
        http_reason=response_reason,
        request_url=url,
        final_url=response_url,
        headers=response_headers,
        request_client_request_id=client_request_id,
    )
    if not isinstance(data, dict):
        raise RuntimeError("图片 provider 响应不是 JSON object。")
    return data


def _decode_provider_json_response(
    raw: bytes,
    *,
    http_status: int,
    http_reason: str,
    request_url: str,
    final_url: str,
    headers: Dict[str, str],
    request_client_request_id: str = "",
) -> Any:
    if not raw.strip():
        raise ProviderProtocolError(
            error_code="PROVIDER_EMPTY_RESPONSE",
            http_status=http_status,
            http_reason=http_reason,
            request_url=request_url,
            final_url=final_url,
            headers=headers,
            body=raw,
            request_client_request_id=request_client_request_id,
        )
    try:
        return json.loads(raw.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ProviderProtocolError(
            error_code="PROVIDER_NON_JSON_RESPONSE",
            http_status=http_status,
            http_reason=http_reason,
            request_url=request_url,
            final_url=final_url,
            headers=headers,
            body=raw,
            request_client_request_id=request_client_request_id,
        ) from exc


def _response_status(response: Any) -> int:
    status = getattr(response, "status", None)
    if status is None and hasattr(response, "getcode"):
        status = response.getcode()
    try:
        return int(status or 0)
    except (TypeError, ValueError):
        return 0


def _safe_response_header(headers: Dict[str, str], name: str) -> str:
    expected = name.lower()
    for key, value in headers.items():
        if str(key).lower() == expected:
            raw = re.sub(r"[\r\n]+", " ", str(value or "").strip())[:200]
            if expected == "content-length" and not raw.isdigit():
                return ""
            return raw
    return ""


def _safe_correlation_id(value: str) -> str:
    value = str(value or "")
    if not value or len(value) > 128 or not re.fullmatch(r"[A-Za-z0-9._:-]+", value):
        return ""
    return value


def _ensure_client_request_id(req: urllib.request.Request) -> str:
    existing = _safe_correlation_id(req.get_header("X-client-request-id") or "")
    client_request_id = existing or f"auto-draw-plot:{uuid.uuid4()}"
    req.add_header("X-Client-Request-ID", client_request_id)
    return client_request_id


def _safe_url_origin_path(value: str) -> str:
    try:
        parsed = urllib.parse.urlparse(str(value or ""))
        hostname = str(parsed.hostname or "").lower()
        if not parsed.scheme or not hostname:
            return str(parsed.path or "/")
        port = parsed.port
        netloc = hostname if port is None else f"{hostname}:{port}"
        return urllib.parse.urlunparse((parsed.scheme.lower(), netloc, parsed.path or "/", "", "", ""))
    except (TypeError, ValueError):
        return ""


def _safe_url_origin(value: str) -> str:
    safe_location = _safe_url_origin_path(value)
    parsed = urllib.parse.urlparse(safe_location)
    if parsed.scheme and parsed.netloc:
        return f"{parsed.scheme}://{parsed.netloc}"
    return ""


def _body_prefix_kind(body: bytes) -> str:
    stripped = body.lstrip()
    if not stripped:
        return "empty"
    first = stripped[:1]
    if first == b"<":
        return "markup"
    if first in {b"{", b"["}:
        return "json_like"
    if all(byte in {9, 10, 13} or 32 <= byte <= 126 for byte in stripped[:32]):
        return "text"
    return "binary"


def _encode_multipart(*, fields: Dict[str, str], files: List[Tuple[str, str, str, bytes]], boundary: str) -> bytes:
    chunks: List[bytes] = []
    for name, value in fields.items():
        chunks.extend(
            [
                f"--{boundary}\r\n".encode("utf-8"),
                f'Content-Disposition: form-data; name="{_escape_multipart_name(name)}"\r\n\r\n'.encode("utf-8"),
                str(value).encode("utf-8"),
                b"\r\n",
            ]
        )
    for field, filename, mime_type, raw in files:
        chunks.extend(
            [
                f"--{boundary}\r\n".encode("utf-8"),
                (
                    'Content-Disposition: form-data; '
                    f'name="{_escape_multipart_name(field)}"; filename="{_escape_multipart_name(filename)}"\r\n'
                ).encode("utf-8"),
                f"Content-Type: {mime_type}\r\n\r\n".encode("utf-8"),
                raw,
                b"\r\n",
            ]
        )
    chunks.append(f"--{boundary}--\r\n".encode("utf-8"))
    return b"".join(chunks)


def _escape_multipart_name(value: str) -> str:
    return str(value).replace("\\", "\\\\").replace('"', '\\"').replace("\r", "").replace("\n", "")


def _existing_reference_images(reference_images: List[Path]) -> List[Path]:
    max_bytes = _max_reference_image_bytes()
    refs: List[Path] = []
    for path in reference_images:
        ref = Path(path)
        if not ref.exists():
            raise FileNotFoundError(f"参考图不存在：{ref}")
        if not ref.is_file():
            raise FileNotFoundError(f"参考图不是文件：{ref}")
        size_bytes = ref.stat().st_size
        if size_bytes > max_bytes:
            raise ValueError(f"参考图过大：{ref} ({size_bytes} bytes > {max_bytes} bytes)")
        refs.append(ref)
    if not refs:
        raise ValueError("OpenAI 图片编辑模式需要至少 1 张参考图。")
    return refs


def _openai_edit_file_parts(reference_images: List[Path]) -> List[Tuple[str, str, str, bytes]]:
    parts: List[Tuple[str, str, str, bytes]] = []
    multi = len(reference_images) > 1
    for ref in reference_images:
        raw = ref.read_bytes()
        mime_type = _infer_reference_image_mime(ref, raw)
        field_name = "image[]" if multi else "image"
        parts.append((field_name, ref.name, mime_type, raw))
    return parts


def _infer_reference_image_mime(path: Path, raw: bytes) -> str:
    if raw.startswith(b"\x89PNG\r\n\x1a\n"):
        return "image/png"
    if raw.startswith(b"\xff\xd8\xff"):
        return "image/jpeg"
    if len(raw) >= 12 and raw[0:4] == b"RIFF" and raw[8:12] == b"WEBP":
        return "image/webp"
    guessed = mimetypes.guess_type(str(path))[0] or "unknown"
    raise RuntimeError(f"参考图内容不是受支持的图片格式：{path} (guessed={guessed})")


def _extract_openai_images(resp: Dict[str, Any]) -> List[Tuple[str, bytes]]:
    out: List[Tuple[str, bytes]] = []
    for item in _iter_response_dicts(resp):
        mime = str(item.get("mime_type") or item.get("mimeType") or "image/png").lower()
        for key in ("b64_json", "b64Json", "image_b64", "imageB64", "base64_image", "base64Image"):
            raw = _decode_b64_image(item.get(key))
            if raw:
                out.append((mime, raw))
        image_url = item.get("url") or item.get("image_url") or item.get("imageUrl")
        if isinstance(image_url, str) and image_url.startswith("data:image/"):
            raw = _decode_data_url_image(image_url)
            if raw:
                out.append(("image/png", raw))
    return out


def _decode_b64_image(value: Any) -> Optional[bytes]:
    if not isinstance(value, str) or not value.strip():
        return None
    try:
        raw = base64.b64decode(value.strip(), validate=True)
    except Exception:
        return None
    if not raw:
        return None
    return raw


def _decode_data_url_image(value: str) -> Optional[bytes]:
    try:
        header, encoded = value.split(",", 1)
    except ValueError:
        return None
    if ";base64" not in header.lower():
        return None
    return _decode_b64_image(

#!/usr/bin/env python3
from __future__ import annotations

import argparse
from pathlib import Path
from typing import Any, Dict, Optional

from common import ensure_dir, expand_path, fatal, load_config, now_tag, write_json, write_text


def _allocate_run_dir(hidden_root: Path, base_run_id: str) -> tuple[str, Path]:
    candidate = hidden_root / base_run_id
    if not candidate.exists():
        return base_run_id, candidate
    for idx in range(2, 100):
        run_id = f"{base_run_id}-{idx:02d}"
        candidate = hidden_root / run_id
        if not candidate.exists():
            return run_id, candidate
    fatal(f"无法在 {hidden_root} 下分配唯一工作目录: {base_run_id}")


def _task_root(project_root: Path, workspace_cfg: Dict[str, Any], run_id: str) -> Path:
    """Return the task-level root for a default workspace allocation."""
    task_base = expand_path(str(workspace_cfg.get("task_root_dir", ".bensz-api")), base=project_root)
    task_prefix = str(workspace_cfg.get("task_prefix", "task")).strip("-") or "task"
    task_label = str(workspace_cfg.get("task_label", "auto-draw-plot")).strip("-") or "auto-draw-plot"
    return task_base / f"{task_prefix}-{run_id}-{task_label}"


def _write_task_readme(task_root: Path, skill_name: str, contract: str) -> None:
    readme = task_root / "README.md"
    if readme.exists():
        return
    write_text(
        readme,
        "# BenszAPI 任务工作区\n\n"
        f"- 工作区契约：`{contract}`\n"
        f"- 本轮 skill：`{skill_name}`\n"
        "- `shared/` 仅用于多个 skill 共享的中间材料；单 skill 任务不必创建。\n"
        f"- `{skill_name}/input|output|log/` 分别保存输入引用、临时产物和日志。\n",
    )


def init_workspace(
    *,
    project_root: Path,
    workspace_base: Optional[str],
    output_png: Optional[str],
    run_id: Optional[str],
    allow_outside_project: bool = False,
) -> Dict[str, Any]:
    cfg = load_config()
    workspace_cfg = cfg.get("workspace", {}) or {}
    reports_cfg = cfg.get("reports", {}) or {}
    generation_cfg = cfg.get("generation", {}) or {}

    prefix = str(workspace_cfg.get("run_prefix", ""))
    timestamp_fmt = str(workspace_cfg.get("timestamp_format", "%Y%m%d%H%M%S%f"))
    if run_id:
        normalized_run_id = run_id if run_id.startswith(prefix) else f"{prefix}{run_id}"
        task_root = _task_root(project_root, workspace_cfg, normalized_run_id)
    else:
        normalized_run_id = f"{prefix}{now_tag(timestamp_fmt)}"
        task_root = _task_root(project_root, workspace_cfg, normalized_run_id)
        normalized_run_id, task_root = _allocate_run_dir(task_root.parent, task_root.name)
        task_root = task_root
        # The generated task directory is the collision-safe run identifier source.
        suffix = f"-{str(workspace_cfg.get('task_label', 'auto-draw-plot')).strip('-')}"
        normalized_run_id = task_root.name[len(str(workspace_cfg.get("task_prefix", "task")).strip("-") or "task") + 1 :]
        if normalized_run_id.endswith(suffix):
            normalized_run_id = normalized_run_id[: -len(suffix)]

    skill_name = str(workspace_cfg.get("task_label", "auto-draw-plot"))
    contract = str(workspace_cfg.get("workspace_contract", "bensz-api-task-v1"))
    if workspace_base:
        # Explicit destinations remain supported for callers that manage their own task root.
        run_dir = expand_path(workspace_base, base=project_root)
        hidden_root = run_dir.parent
    else:
        hidden_root = task_root.parent
        run_dir = task_root / skill_name
        _write_task_readme(task_root, skill_name, contract)

    run_dir = ensure_dir(run_dir)
    subdirs = {name: ensure_dir(run_dir / str(name)) for name in (workspace_cfg.get("subdirs") or [])}
    latest_run_pointer = hidden_root / str(workspace_cfg.get("latest_run_pointer", "latest-run.txt"))
    write_text(latest_run_pointer, normalized_run_id + "\n")

    if output_png:
        public_output_png = expand_path(output_png, base=project_root)
    else:
        public_output_png = project_root / str(generation_cfg.get("default_output_name", "draw-plot.jpg"))
    allow_external = bool(allow_outside_project or workspace_cfg.get("allow_outside_project", False))
    if not allow_external:
        if not _is_within(hidden_root, project_root):
            fatal("workspace_base 必须位于 project_root 内；如确需写到外部路径，请显式允许 outside project。")
        if not _is_within(public_output_png.resolve().parent, project_root):
            fatal("output_png 必须位于 project_root 内；如确需写到外部路径，请显式允许 outside project。")

    manifest = {
        "project_root": str(project_root.resolve()),
        "workspace_root": str(hidden_root.resolve()),
        "task_root": str((run_dir.parent if workspace_base else task_root).resolve()),
        "workspace_contract": contract,
        "run_id": normalized_run_id,
        "run_dir": str(run_dir.resolve()),
        "workspace_inside_project_root": _is_within(hidden_root, project_root),
        "public_output_image": str(public_output_png.resolve()),
        "public_output_png": str(public_output_png.resolve()),
        "public_output_dir": str(public_output_png.resolve().parent),
        "requests_dir": str(subdirs.get("requests", run_dir / "requests")),
        "rounds_dir": str(subdirs.get("rounds", run_dir / "rounds")),
        "meta_dir": str(subdirs.get("meta", run_dir / "meta")),
        "exports_dir": str(subdirs.get("exports", run_dir / "exports")),
        "parallel_vibe_dir": str(subdirs.get("parallel-vibe", run_dir / "parallel-vibe")),
        "request_file": str((subdirs.get("requests", run_dir / "requests") / "user-need.md").resolve()),
        "analysis_json": str((run_dir / str(reports_cfg.get("analysis_json", "meta/analysis.json"))).resolve()),
        "result_json": str((run_dir / str(reports_cfg.get("result_json", "meta/result.json"))).resolve()),
    }
    write_json(run_dir / str(reports_cfg.get("run_manifest", "run-manifest.json")), manifest)
    return manifest


def _is_within(target: Path, base: Path) -> bool:
    try:
        target.resolve().relative_to(base.resolve())
        return True
    except Exception:
        return False


def main() -> None:
    parser = argparse.ArgumentParser(description="初始化 auto-draw-plot 隐藏工作区。")
    parser.add_argument("--project-root", default=".", help="项目根目录，默认当前目录")
    parser.add_argument("--workspace-base", default="", help="自定义隐藏工作区根目录")
    parser.add_argument("--output-image", "--output-png", dest="output_image", default="", help="最终公开输出图片路径；--output-png 为兼容别名")
    parser.add_argument("--run-id", default="", help="指定 run id（可选）")
    parser.add_argument("--allow-outside-project", action="store_true", help="允许 workspace/output 写到 project_root 外部")
    args = parser.parse_args()

    manifest = init_workspace(
        project_root=expand_path(args.project_root, base=Path.cwd()),
        workspace_base=args.workspace_base or None,
        output_png=args.output_image or None,
        run_id=args.run_id or None,
        allow_outside_project=bool(args.allow_outside_project),
    )
    write_json(Path(manifest["run_dir"]) / "init-output.json", manifest)
    print(manifest["run_dir"])


if __name__ == "__main__":
    main()

from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Dict, List


@dataclass(frozen=True)
class DrawMode:
    name: str
    label: str
    canvas_width: int
    canvas_height: int
    accept_score: float
    planner_role: str
    purpose: str
    prompt_sections: List[str]
    guardrails: List[str]
    evaluation_criteria: List[str]


def resolve_mode(config: Dict[str, Any], requested: str | None = None) -> DrawMode:
    modes_cfg = config.get("modes", {}) if isinstance(config.get("modes"), dict) else {}
    default_name = str(modes_cfg.get("default", "general") or "general").strip().lower()
    explicit = requested is not None and str(requested).strip() != ""
    mode_name = _canonical_mode_name(modes_cfg, requested or default_name)
    presets = modes_cfg.get("presets") if isinstance(modes_cfg.get("presets"), dict) else {}
    preset = presets.get(mode_name)
    if not isinstance(preset, dict) and explicit:
        valid = sorted(str(key) for key in presets.keys()) or [default_name]
        raise ValueError(f"不支持的绘图模式：{requested!r}。可用模式：{', '.join(valid)}")
    if not isinstance(preset, dict):
        preset = presets.get(default_name) if isinstance(presets.get(default_name), dict) else {}
        mode_name = default_name if isinstance(presets.get(default_name), dict) else "general"

    gen_cfg = config.get("generation", {}) if isinstance(config.get("generation"), dict) else {}
    eval_cfg = config.get("evaluation", {}) if isinstance(config.get("evaluation"), dict) else {}
    return DrawMode(
        name=mode_name,
        label=str(preset.get("label") or mode_name),
        canvas_width=int(preset.get("canvas_width") or gen_cfg.get("default_canvas_width", 1600) or 1600),
        canvas_height=int(preset.get("canvas_height") or gen_cfg.get("default_canvas_height", 900) or 900),
        accept_score=float(preset.get("accept_score") or eval_cfg.get("accept_score", 8.5) or 8.5),
        planner_role=str(preset.get("planner_role") or "你是一位高质量图片生成 prompt 设计师。"),
        purpose=str(preset.get("purpose") or "生成一张高质量 PNG 图片。"),
        prompt_sections=_as_text_list(preset.get("prompt_sections")),
        guardrails=_as_text_list(preset.get("guardrails")),
        evaluation_criteria=_as_text_list(preset.get("evaluation_criteria")),
    )


def mode_prompt_lines(mode: DrawMode) -> List[str]:
    lines = [
        f"绘图模式：{mode.name}（{mode.label}）",
        f"模式目标：{mode.purpose}",
    ]
    if mode.prompt_sections:
        lines.append("模式构图要求：")
        lines.extend([f"- {item}" for item in mode.prompt_sections])
    if mode.guardrails:
        lines.append("模式硬性约束：")
        lines.extend([f"- {item}" for item in mode.guardrails])
    return lines


def mode_evaluation_lines(mode: DrawMode) -> List[str]:
    if not mode.evaluation_criteria:
        return []
    return ["模式专属评估项：", *[f"- {item}" for item in mode.evaluation_criteria]]


def _canonical_mode_name(modes_cfg: Dict[str, Any], requested: str) -> str:
    raw = str(requested or "").strip().lower()
    if not raw:
        return "general"
    presets = modes_cfg.get("presets") if isinstance(modes_cfg.get("presets"), dict) else {}
    if raw in presets:
        return raw
    aliases = modes_cfg.get("aliases") if isinstance(modes_cfg.get("aliases"), dict) else {}
    for name, values in aliases.items():
        candidates = [str(name).strip().lower()]
        if isinstance(values, list):
            candidates.extend(str(item).strip().lower() for item in values)
        if raw in candidates:
            return str(name).strip().lower()
    return raw


def _as_text_list(value: Any) -> List[str]:
    if isinstance(value, list):
        return [str(item).strip() for item in value if str(item).strip()]
    if isinstance(value, str) and value.strip():
        return [value.strip()]
    return []

#!/usr/bin/env python3
from __future__ import annotations

import argparse

from common import expand_path
from image_provider_client import health_check_image_provider


def main() -> None:
    parser = argparse.ArgumentParser(description="检查 auto-draw-plot 图片 provider 配置（兼容旧 Nano Banana 检查入口）。")
    parser.add_argument("--api-env", default="", help="remote.env 路径，默认 ~/.bensz-skills/config/remote.env")
    parser.add_argument("--provider", default="auto", help="图片 provider/model：auto（默认）/ gpt-image-2.5-flare / gpt-image-2.5-sunburst / gpt-image-2 / nano_banana")
    args = parser.parse_args()

    cfg = health_check_image_provider(
        remote_env_path=expand_path(args.api_env) if args.api_env else None,
        provider_name=str(args.provider or "auto"),
    )
    print(
        f"OK {cfg.preflight_status} provider={cfg.provider} model={cfg.model} base_url={cfg.base_url} "
        f"scope={cfg.preflight_scope} generation_eligible={cfg.generation_eligibility}"
    )


if __name__ == "__main__":
    main()

from __future__ import annotations

import base64
import json
import math
import re
import struct
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

from common import ensure_dir, warn
from env_utils import find_remote_env, mask_secret, merged_env
from common import load_config


@dataclass(frozen=True)
class GeminiConfig:
    base_url: str
    api_key: str
    model: str
    env_path: Optional[Path]


class GeminiHTTPError(RuntimeError):
    def __init__(self, code: int, reason: str, detail: str):
        super().__init__(f"HTTP {code} {reason}: {detail}")
        self.code = int(code)
        self.reason = str(reason)
        self.detail = str(detail)


def load_gemini_config(*, remote_env_path: Optional[Path] = None) -> GeminiConfig:
    env_path = find_remote_env(remote_env_path)
    env = merged_env(env_path)

    base_url = str(env.get("GEMINI_BASE_URL", "") or "").strip().rstrip("/")
    api_key = str(
        env.get("GEMINI_API", "")
        or env.get("GEMINI_API_KEY", "")
        or env.get("GOOGLE_API_KEY", "")
        or ""
    ).strip()
    model = str(env.get("GEMINI_MODEL", "") or "").strip()

    missing: List[str] = []
    if not env_path:
        missing.append("remote.env")
    if not base_url:
        missing.append("GEMINI_BASE_URL")
    if not api_key:
        missing.append("GEMINI_API")
    if not model:
        missing.append("GEMINI_MODEL")
    if missing:
        raise RuntimeError(
            "未检测到可用的 Nano Banana / Gemini 配置。缺少：{}。请先检查 `~/.bensz-skills/config/remote.env`。".format(
                "、".join(missing)
            )
        )

    return GeminiConfig(base_url=base_url, api_key=api_key, model=model, env_path=env_path)


def _post_json(
    url: str,
    payload: Dict[str, Any],
    *,
    headers: Optional[Dict[str, str]] = None,
    timeout_s: int = 60,
) -> Dict[str, Any]:
    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(url, data=body, method="POST")
    req.add_header("Content-Type", "application/json")
    req.add_header("Accept", "application/json")
    if headers:
        for key, value in headers.items():
            req.add_header(key, value)
    try:
        with urllib.request.urlopen(req, timeout=timeout_s) as resp:
            raw = resp.read()
    except urllib.error.HTTPError as exc:
        detail_raw = exc.read()
        detail = detail_raw.decode("utf-8", errors="replace")
        raise GeminiHTTPError(int(exc.code), str(exc.reason), detail[:1600]) from exc
    except Exception as exc:
        raise RuntimeError(f"请求 Gemini 失败：{exc}") from exc

    try:
        data = json.loads(raw.decode("utf-8"))
    except Exception as exc:
        raise RuntimeError(f"Gemini 响应 JSON 解析失败：{exc}") from exc
    if not isinstance(data, dict):
        raise RuntimeError("Gemini 响应不是 JSON object。")
    return data


def generate_content(cfg: GeminiConfig, payload: Dict[str, Any], *, timeout_s: int = 120) -> Dict[str, Any]:
    endpoint = f"{cfg.base_url}/models/{cfg.model}:generateContent"
    try:
        return _post_json(endpoint, payload, headers={"x-goog-api-key": cfg.api_key}, timeout_s=timeout_s)
    except GeminiHTTPError as exc:
        if exc.code not in {401, 403}:
            raise
        warn(f"Gemini header 认证失败（HTTP {exc.code}），回退到 query key。")
        return _post_json(f"{endpoint}?key={cfg.api_key}", payload, headers=None, timeout_s=timeout_s)


def nano_banana_health_check(*, remote_env_path: Optional[Path] = None, timeout_s: int = 30) -> GeminiConfig:
    cfg = load_gemini_config(remote_env_path=remote_env_path)
    text_payload = {
        "contents": [{"role": "user", "parts": [{"text": "ping"}]}],
        "generationConfig": {"temperature": 0.0, "maxOutputTokens": 16},
    }
    try:
        resp = generate_content(cfg, text_payload, timeout_s=timeout_s)
        if not isinstance(resp.get("candidates"), list) or not resp.get("candidates"):
            raise RuntimeError("Gemini 文本连通性检查失败：响应中缺少 candidates。")
        return cfg
    except GeminiHTTPError as exc:
        if exc.code != 400 or "not supported by this model" not in exc.detail.lower():
            raise
        # Some Nano Banana style models only support IMAGE outputs. In that case
        # we fall back to a tiny image-generation probe.
        image_payload = {
            "contents": [{"role": "user", "parts": [{"text": "A tiny blue square on white background"}]}],
            "generationConfig": {
                "temperature": 0.0,
                "imageConfig": {"aspectRatio": "1:1", "imageSize": "1K"},
            },
        }
        resp = generate_content(cfg, image_payload, timeout_s=timeout_s)
        if _best_image(_extract_inline_images(resp)) is None:
            raise RuntimeError("Gemini 图片连通性检查失败：未返回 inline image。")
        return cfg


def _sanitize_payload(payload: Dict[str, Any]) -> Dict[str, Any]:
    cloned = json.loads(json.dumps(payload, ensure_ascii=False))
    contents = cloned.get("contents")
    if not isinstance(contents, list):
        return cloned
    for content in contents:
        if not isinstance(content, dict):
            continue
        parts = content.get("parts")
        if not isinstance(parts, list):
            continue
        for part in parts:
            if not isinstance(part, dict):
                continue
            inline = part.get("inlineData") or part.get("inline_data")
            if isinstance(inline, dict) and isinstance(inline.get("data"), str):
                inline["data"] = "<omitted>"
    return cloned


def _write_debug_request(debug_dir: Optional[Path], cfg: GeminiConfig, payload: Dict[str, Any]) -> None:
    if debug_dir is None:
        return
    ensure_dir(debug_dir)
    body = {
        "base_url": cfg.base_url,
        "model": cfg.model,
        "api_key": mask_secret(cfg.api_key),
        "payload": _sanitize_payload(payload),
    }
    (debug_dir / "request.json").write_text(json.dumps(body, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def _write_debug_response(debug_dir: Optional[Path], payload: Dict[str, Any]) -> None:
    if debug_dir is None:
        return
    ensure_dir(debug_dir)
    (debug_dir / "response.json").write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def _extract_text(resp: Dict[str, Any]) -> str:
    out: List[str] = []
    candidates = resp.get("candidates")
    if not isinstance(candidates, list):
        return ""
    for cand in candidates:
        if not isinstance(cand, dict):
            continue
        content = cand.get("content")
        if not isinstance(content, dict):
            continue
        parts = content.get("parts")
        if not isinstance(parts, list):
            continue
        for part in parts:
            if not isinstance(part, dict):
                continue
            if part.get("thought") is True:
                continue
            text = part.get("text")
            if isinstance(text, str) and text.strip():
                out.append(text)
    return "\n".join(out).strip()


def _extract_inline_images(resp: Dict[str, Any]) -> List[Tuple[str, bytes]]:
    out: List[Tuple[str, bytes]] = []
    candidates = resp.get("candidates")
    if not isinstance(candidates, list):
        return out
    for cand in candidates:
        if not isinstance(cand, dict):
            continue
        content = cand.get("content")
        if not isinstance(content, dict):
            continue
        parts = content.get("parts")
        if not isinstance(parts, list):
            continue
        for part in parts:
            if not isinstance(part, dict):
                continue
            inline = part.get("inlineData") or part.get("inline_data")
            if not isinstance(inline, dict):
                continue
            data_b64 = inline.get("data")
            if not isinstance(data_b64, str) or not data_b64.strip():
                continue
            try:
                raw = base64.b64decode(data_b64)
            except Exception:
                continue
            mime = str(inline.get("mimeType") or inline.get("mime_type") or "application/octet-stream").lower()
            out.append((mime, raw))
    return out


def _best_image(images: List[Tuple[str, bytes]]) -> Optional[Tuple[str, bytes]]:
    if not images:
        return None
    scored: List[Tuple[int, int, Tuple[str, bytes]]] = []
    for mime, raw in images:
        scored.append((1 if "png" in mime else 0, len(raw), (mime, raw)))
    scored.sort(key=lambda item: (item[0], item[1]), reverse=True)
    return scored[0][2]


def _infer_image_mime(path: Path, raw: bytes) -> str:
    if raw.startswith(b"\x89PNG\r\n\x1a\n"):
        return "image/png"
    if raw.startswith(b"\xff\xd8\xff"):
        return "image/jpeg"
    if len(raw) >= 12 and raw[0:4] == b"RIFF" and raw[8:12] == b"WEBP":
        return "image/webp"
    raise RuntimeError(f"不支持的图片格式：{path}")


def part_from_image_path(path: Path) -> Dict[str, Any]:
    size_bytes = path.stat().st_size
    max_bytes = _max_reference_image_bytes()
    if size_bytes > max_bytes:
        raise ValueError(f"参考图过大：{path} ({size_bytes} bytes > {max_bytes} bytes)")
    raw = path.read_bytes()
    mime = _infer_image_mime(path, raw)
    return {
        "inlineData": {
            "mimeType": mime,
            "data": base64.b64encode(raw).decode("ascii"),
        }
    }


def generate_text(
    *,
    cfg: GeminiConfig,
    parts: List[Dict[str, Any]],
    debug_dir: Optional[Path] = None,
    timeout_s: int = 120,
    temperature: float = 0.1,
    max_output_tokens: int = 1200,
) -> Tuple[str, Dict[str, Any]]:
    payload: Dict[str, Any] = {
        "contents": [{"role": "user", "parts": parts}],
        "generationConfig": {
            "temperature": float(temperature),
            "maxOutputTokens": int(max_output_tokens),
        },
    }
    _write_debug_request(debug_dir, cfg, payload)
    resp = generate_content(cfg, payload, timeout_s=timeout_s)
    _write_debug_response(debug_dir, resp)
    return _extract_text(resp), resp


def _choose_aspect_ratio(w: int, h: int) -> str:
    ratios = {
        "16:9": 16 / 9,
        "9:16": 9 / 16,
        "4:3": 4 / 3,
        "3:4": 3 / 4,
        "3:2": 3 / 2,
        "2:3": 2 / 3,
        "1:1": 1.0,
        "5:4": 5 / 4,
        "4:5": 4 / 5,
        "21:9": 21 / 9,
        "1:4": 1 / 4,
        "4:1": 4 / 1,
    }
    target = (w / max(1, h)) if h else 1.0
    return min(ratios.items(), key=lambda item: abs(item[1] - target))[0]


def _target_4k_dims(canvas_w: int, canvas_h: int) -> Tuple[int, int]:
    long_edge = 3840
    w = max(1, int(canvas_w))
    h = max(1, int(canvas_h))
    if w >= h:
        return long_edge, max(1, int(round(long_edge * (h / w))))
    return max(1, int(round(long_edge * (w / h)))), long_edge


def _maybe_resize_to_canvas(path: Path, *, target_w: int, target_h: int) -> None:
    try:
        from PIL import Image  # type: ignore
    except Exception:
        warn("缺少 Pillow，跳过 PNG 尺寸对齐。")
        return
    try:
        with Image.open(path) as img:
            w, h = img.size
            if w == target_w and h == target_h:
                return
            scale = min(target_w / max(1, w), target_h / max(1, h))
            new_w = max(1, int(math.floor(w * scale)))
            new_h = max(1, int(math.floor(h * scale)))
            resized = img.convert("RGBA").resize((new_w, new_h), resample=Image.LANCZOS)
            canvas = Image.new("RGBA", (target_w, target_h), (255, 255, 255, 255))
            ox = (target_w - new_w) // 2
            oy = (target_h - new_h) // 2
            canvas.paste(resized, (ox, oy), resized)
            canvas.convert("RGB").save(path, format="PNG", optimize=True)
    except Exception as exc:
        warn(f"PNG 尺寸对齐失败（已忽略）：{exc}")


def _validate_postprocess_args(
    *,
    postprocess_resize: bool,
    postprocess_w: Optional[int],
    postprocess_h: Optional[int],
) -> None:
    has_w = postprocess_w is not None
    has_h = postprocess_h is not None
    if has_w != has_h:
        raise ValueError("--postprocess-width 与 --postprocess-height 必须同时提供。")
    if (has_w or has_h) and not postprocess_resize:
        raise ValueError("只有启用 --postprocess-resize 时才能指定后处理目标尺寸。")
    if postprocess_resize and not (has_w and has_h):
        raise ValueError("--postprocess-resize 需要同时指定 --postprocess-width 与 --postprocess-height。")
    if has_w and int(postprocess_w or 0) <= 0:
        raise ValueError("--postprocess-width 必须为正整数。")
    if has_h and int(postprocess_h or 0) <= 0:
        raise ValueError("--postprocess-height 必须为正整数。")


def _max_reference_image_bytes() -> int:
    cfg = load_config()
    gen_cfg = cfg.get("generation", {}) if isinstance(cfg.get("generation"), dict) else {}
    return max(1, int(gen_cfg.get("max_reference_image_bytes", 20 * 1024 * 1024)))


def _image_size(path: Path) -> Optional[Dict[str, int]]:
    try:
        raw = path.read_bytes()[:32]
        if raw.startswith(b"\x89PNG\r\n\x1a\n") and len(raw) >= 24:
            width, height = struct.unpack(">II", raw[16:24])
            return {"width": int(width), "height": int(height)}
    except Exception:
        pass
    try:
        from PIL import Image  # type: ignore

        with Image.open(path) as img:
            return {"width": int(img.size[0]), "height": int(img.size[1])}
    except Exception:
        return None


def _parse_retry_after(detail: str) -> Optional[float]:
    m = re.search(r"retry\\s+in\\s+([0-9]+(?:\\.[0-9]+)?)s", detail, flags=re.IGNORECASE)
    if not m:
        return None
    try:
        return float(m.group(1))
    except Exception:
        return None


def generate_png(
    *,
    cfg: GeminiConfig,
    prompt: str,
    output_png: Path,
    canvas_w: int,
    canvas_h: int,
    reference_images: Optional[List[Path]] = None,
    debug_dir: Optional[Path] = None,
    timeout_s: int = 180,
    retries: int = 5,
    postprocess_resize: bool = False,
    postprocess_w: Optional[int] = None,
    postprocess_h: Optional[int] = None,
) -> Dict[str, Any]:
    _validate_postprocess_args(
        postprocess_resize=postprocess_resize,
        postprocess_w=postprocess_w,
        postprocess_h=postprocess_h,
    )
    refs = [Path(p) for p in (reference_images or [])]
    parts = [part_from_image_path(ref) for ref in refs]
    parts.append({"text": prompt})
    payload: Dict[str, Any] = {
        "contents": [{"role": "user", "parts": parts}],
        "generationConfig": {
            "temperature": 0.2,
            "imageConfig": {
                "aspectRatio": _choose_aspect_ratio(canvas_w, canvas_h),
                "imageSize": "4K",
            },
        },
    }
    _write_debug_request(debug_dir, cfg, payload)

    last_error: Optional[Exception] = None
    response: Dict[str, Any] = {}
    for attempt in range(1, max(1, retries) + 1):
        try:
            response = generate_content(cfg, payload, timeout_s=timeout_s)
            last_error = None
            break
        except GeminiHTTPError as exc:
            last_error = exc
            if exc.code not in {429, 503} or attempt >= retries:
                break
            wait_s = _parse_retry_after(exc.detail) or min(30.0, 2.0**attempt)
            warn(f"Gemini 限流/资源忙，{wait_s:.1f}s 后重试（{attempt}/{retries}）。")
            time.sleep(max(0.5, wait_s))
        except Exception as exc:
            last_error = exc
            break

    if last_error is not None:
        raise last_error

    _write_debug_response(debug_dir, response)
    best = _best_image(_extract_inline_images(response))
    if best is None:
        excerpt = json.dumps(response, ensure_ascii=False)[:800]
        raise RuntimeError(f"未从 Gemini 响应中提取到图片。response_excerpt={excerpt}")
    mime, raw = best
    output_png.parent.mkdir(parents=True, exist_ok=True)
    output_png.write_bytes(raw)
    native_size = _image_size(output_png)
    target_w, target_h = (
        (max(1, int(postprocess_w)), max(1, int(postprocess_h)))
        if postprocess_w and postprocess_h
        else _target_4k_dims(canvas_w, canvas_h)
    )
    postprocess_applied = False
    if postprocess_resize:
        before = _image_size(output_png)
        _maybe_resize_to_canvas(output_png, target_w=target_w, target_h=target_h)
        after = _image_size(output_png)
        postprocess_applied = bool(before and after and before != after)
    return {
        "mime_type": mime,
        "output_png": str(output_png),
        "requested_provider_size": {
            "aspect_ratio": _choose_aspect_ratio(canvas_w, canvas_h),
            "image_size": "4K",
        },
        "native_size": native_size,
        "output_size": _image_size(output_png),
        "postprocess_resize_applied": postprocess_applied,
        "postprocess_target_size": {"width": target_w, "height": target_h} if postprocess_resize else None,
        "response_path": str(debug_dir / "response.json") if debug_dir is not None else None,
    }

#!/usr/bin/env python3
from __future__ import annotations

import argparse
from pathlib import Path

from common import ensure_dir, read_text, write_text


def main() -> None:
    parser = argparse.ArgumentParser(description="parallel-vibe round worker for auto-draw-plot")
    parser.add_argument("--run-dir", required=True)
    parser.add_argument("--request-file", required=True)
    parser.add_argument("--round", required=True, type=int)
    parser.add_argument("--result-file", default="RESULT.md")
    parser.add_argument("--wrapped-prompt", default="")
    args = parser.parse_args()

    request_text = read_text(Path(args.request_file))
    round_dir = ensure_dir(Path.cwd() / f"round-{int(args.round):02d}")
    prompt_path = round_dir / "prompt.txt"
    eval_req = round_dir / "evaluation-request.md"
    prompt_body = "\n".join(
        [
            "请基于以下用户需求，为当前图片生成 provider 生成一段更具体的图片 prompt：",
            "",
            request_text.strip(),
            "",
            "请确保主体清晰、结构稳定、文字短且可读，不要引入无关元素。",
        ]
    )
    write_text(prompt_path, prompt_body + "\n")
    write_text(
        eval_req,
        "本文件由 parallel-vibe worker 生成，用于提示下一阶段做视觉评估。\n",
    )
    write_text(
        Path.cwd() / args.result_file,
        "\n".join(
            [
                "# auto-draw-plot parallel round",
                "",
                f"- round: `{int(args.round):02d}`",
                f"- prompt_file: `{prompt_path.name}`",
                f"- evaluation_request: `{eval_req.name}`",
                "",
                "本 worker 只负责在隔离 workspace 内生成本轮 prompt 草案，真正的图片生成与评估由主流程脚本执行。",
            ]
        )
        + "\n",
    )


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
from __future__ import annotations

import argparse
from pathlib import Path
from typing import Any, Dict, List, Optional

from common import (
    copy_file,
    ensure_dir,
    expand_path,
    join_lines,
    load_config,
    read_text,
    sha256_file,
    write_json,
    write_text,
)
from evaluate_image import evaluate_image, heuristic_evaluation
from generate_image import generate_image
from image_provider_client import ImageProviderConfig, health_check_image_provider, write_image_with_format_contract
from init_workspace import init_workspace
from modes import DrawMode, mode_prompt_lines, resolve_mode
from build_parallel_plan import build_parallel_plan as create_parallel_plan


def build_round_prompt(
    *,
    request_text: str,
    round_index: int,
    max_rounds: int,
    prior_rounds: List[Dict[str, Any]],
    remote_env: Optional[Path],
    prompt_dir: Path,
    mode: DrawMode,
) -> Dict[str, Any]:
    cfg = load_config()
    gen_cfg = cfg.get("generation", {}) or {}
    guardrails = gen_cfg.get("prompt_guardrails", []) or []

    feedback_lines: List[str] = []
    if prior_rounds:
        prev_round = prior_rounds[-1]
        evaluation = prev_round.get("evaluation", {}) or {}
        feedback_lines.extend(
            [
                f"- 上一轮得分：{evaluation.get('score', 0)}",
                f"- 上一轮总结：{evaluation.get('summary', '')}",
            ]
        )
        for item in (evaluation.get("must_fix") or [])[:5]:
            feedback_lines.append(f"- 必须修复：{item}")
        for item in (evaluation.get("prompt_patch") or [])[:6]:
            feedback_lines.append(f"- Prompt 增量：{item}")

    planner_lines = [
        mode.planner_role,
        "请把用户需求转成可直接发送给图片生成模型的高质量 prompt。",
        "请输出原始 JSON，不要加 Markdown。",
        "JSON schema:",
        "{",
        '  "image_prompt": "可直接发送给图片模型的完整 prompt",',
        '  "negative_prompt": ["尽量避免的内容"],',
        '  "focus_points": ["本轮重点"],',
        '  "reasoning": "本轮为什么这样写 prompt"',
        "}",
        "",
        f"当前轮次：{round_index}/{max_rounds}",
        "",
        "用户需求：",
        request_text.strip(),
        "",
        *mode_prompt_lines(mode),
        "",
        "硬性护栏：",
        *[f"- {line}" for line in guardrails],
    ]
    if prior_rounds:
        planner_lines.extend(
            [
                "",
                "本轮生成方式：上一轮图片会作为第一张参考图提供给图片模型；请基于它做 image-to-image 微调。",
                "请尽量保留上一轮已经正确的主体、布局、配色和文字，只针对反馈问题做局部优化；除非反馈明确要求重构，不要从零重画。",
            ]
        )
    if feedback_lines:
        planner_lines.extend(["", "上一轮反馈：", *feedback_lines])

    negative_prompt = ["watermark", "logo", "distorted text", "random extra objects"]
    if mode.name in {"roadmap", "schematic"}:
        negative_prompt.extend(
            [
                "condensed font",
                "narrow font",
                "compressed font",
                "horizontally squeezed text",
                "tall skinny Chinese characters",
            ]
        )

    payload: Dict[str, Any] = {
        "image_prompt": fallback_round_prompt(request_text=request_text, feedback_lines=feedback_lines, mode=mode),
        "negative_prompt": negative_prompt,
        "focus_points": ["忠实满足用户要求", "主体清晰", "结构稳定"],
        "reasoning": "使用本地模板拼装 prompt；脚本默认不调用 Gemini 文本规划，宿主 AI 可在运行前自行优化用户需求。",
    }
    normalized = normalize_prompt_payload(
        payload=payload,
        request_text=request_text,
        feedback_lines=feedback_lines,
        guardrails=[str(item) for item in guardrails] + mode.guardrails,
    )
    normalized["raw_text"] = ""
    normalized["planner_backend"] = "local_template"
    normalized["planner_context"] = join_lines(planner_lines)
    write_json(prompt_dir / "prompt-plan.json", normalized)
    write_text(prompt_dir / "prompt.txt", normalized["full_prompt"] + "\n")
    return normalized


def fallback_round_prompt(*, request_text: str, feedback_lines: List[str], mode: DrawMode) -> str:
    lines = [
        "Create one polished, publication-ready image that directly satisfies the user requirement below.",
        f"Mode: {mode.name} - {mode.label}.",
        f"Purpose: {mode.purpose}",
        "",
        "User requirement:",
        request_text.strip(),
        "",
        "Render guidance:",
        *[f"- {item}" for item in mode.prompt_sections],
        *[f"- {item}" for item in mode.guardrails],
        "- Keep the main subject prominent and unambiguous.",
        "- Use a clean composition with readable labels only when necessary.",
        "- Prefer a white or light background unless the user explicitly asked otherwise.",
        "- Make colors intentional and high-contrast.",
    ]
    if mode.name in {"roadmap", "schematic"}:
        lines.extend(
            [
                "- Use normal-width Chinese label typography, like modern Heiti / Source Han Sans / Noto Sans CJK, with natural character proportions.",
                "- Prefer wrapping labels into short lines over squeezing or narrowing glyphs to fit boxes.",
                "- Do not use condensed, narrow, compressed, horizontally squeezed, or tall skinny text for Chinese labels.",
            ]
        )
    if feedback_lines:
        lines.extend(
            [
                "",
                "The previous round image is attached as the primary reference. Edit and improve that image instead of starting from scratch.",
                "Keep correct existing composition details stable; only change what the feedback requires.",
                "Fix these issues from the previous round:",
                *feedback_lines,
            ]
        )
    return join_lines(lines).strip()


def normalize_prompt_payload(
    *,
    payload: Dict[str, Any],
    request_text: str,
    feedback_lines: List[str],
    guardrails: List[str],
) -> Dict[str, Any]:
    image_prompt = str(payload.get("image_prompt") or "").strip()
    negative_prompt = _normalize_text_list(payload.get("negative_prompt"))
    focus_points = _normalize_text_list(payload.get("focus_points"))
    reasoning = str(payload.get("reasoning") or "").strip()
    if not image_prompt:
        sections = [
            "请生成一张高质量、结构清晰、主体明确的图片。",
            "用户需求如下：",
            request_text.strip(),
        ]
        if feedback_lines:
            sections.extend(["", "请同时修复上一轮问题：", *feedback_lines])
        image_prompt = join_lines(sections)
        if not reasoning:
            reasoning = "模型未返回结构化 prompt，已回退到基于原始需求的护栏模板。"

    final_lines = [image_prompt.strip(), "", "必须遵守：", *[f"- {line}" for line in guardrails]]
    if negative_prompt:
        final_lines.extend(["", "尽量避免：", *[f"- {item}" for item in negative_prompt]])
    return {
        "image_prompt": image_prompt.strip(),
        "negative_prompt": negative_prompt,
        "focus_points": focus_points,
        "reasoning": reasoning,
        "full_prompt": join_lines(final_lines).strip(),
    }


def build_round_reference_images(
    *,
    user_reference_images: Optional[List[Path]],
    prior_rounds: List[Dict[str, Any]],
    max_reference_images: int,
) -> Dict[str, Any]:
    refs: List[Path] = []
    seen: set[str] = set()

    def add_ref(path: Path) -> None:
        if len(refs) >= max(1, max_reference_images):
            return
        key = _path_key(path)
        if key in seen:
            return
        refs.append(path)
        seen.add(key)

    prev_image = previous_round_image(prior_rounds)
    if prev_image is not None:
        add_ref(prev_image)
    for item in user_reference_images or []:
        add_ref(Path(item))

    prev_round = prior_rounds[-1] if prior_rounds else None
    if prev_image is not None and user_reference_images:
        source = "mixed"
    elif prev_image is not None:
        source = "previous_round"
    elif user_reference_images:
        source = "user_reference"
    else:
        source = "none"
    return {
        "mode": "image-to-image" if refs else "text-to-image",
        "source": source,
        "source_round": prev_round.get("round") if prev_round else None,
        "source_image": str(prev_image) if prev_image is not None else None,
        "reference_images": [str(item) for item in refs],
        "user_reference_images": [str(item) for item in (user_reference_images or [])],
        "max_reference_images": max_reference_images,
    }


def previous_round_image(prior_rounds: List[Dict[str, Any]]) -> Optional[Path]:
    if not prior_rounds:
        return None
    image_meta = (prior_rounds[-1].get("image", {}) or {})
    output_png = str(image_meta.get("output_file") or image_meta.get("output_png") or "").strip()
    if not output_png:
        return None
    path = Path(output_png)
    if not path.exists():
        return None
    return path


def _path_key(path: Path) -> str:
    try:
        return str(path.resolve())
    except Exception:
        return str(path)


def run_draw_plot(
    *,
    request_text: str,
    project_root: Path,
    output_png: Optional[str],
    workspace_base: Optional[str],
    max_rounds: Optional[int],
    canvas_w: Optional[int],
    canvas_h: Optional[int],
    remote_env: Optional[Path],
    reference_images: Optional[List[Path]],
    mode_name: Optional[str] = None,
    provider_name: Optional[str] = None,
    allow_provider_fallback: bool = False,
    allow_outside_project: bool = False,
    postprocess_resize: Optional[bool] = None,
    postprocess_w: Optional[int] = None,
    postprocess_h: Optional[int] = None,
    quality: Optional[str] = None,
    provider_size: Optional[str] = None,
    output_format: Optional[str] = None,
    output_compression: Optional[int] = None,
) -> Dict[str, Any]:
    cfg = load_config()
    gen_cfg = cfg.get("generation", {}) or {}
    eval_cfg = cfg.get("evaluation", {}) or {}
    reports_cfg = cfg.get("reports", {}) or {}
    parallel_cfg = cfg.get("parallel_vibe", {}) or {}
    mode = resolve_mode(cfg, mode_name)

    max_rounds = int(max_rounds or gen_cfg.get("default_max_rounds", 3))
    canvas_w = int(canvas_w or mode.canvas_width or gen_cfg.get("default_canvas_width", 1600))
    canvas_h = int(canvas_h or mode.canvas_height or gen_cfg.get("default_canvas_height", 900))
    effective_postprocess_resize = (
        bool(gen_cfg.get("postprocess_resize_default", False))
        if postprocess_resize is None
        else bool(postprocess_resize)
    )

    manifest = init_workspace(
        project_root=project_root,
        workspace_base=workspace_base,
        output_png=output_png,
        run_id=None,
        allow_outside_project=allow_outside_project,
    )
    run_dir = Path(manifest["run_dir"])
    request_file = Path(manifest["request_file"])
    analysis_json = Path(manifest["analysis_json"])
    result_json = Path(manifest["result_json"])

    write_text(request_file, request_text.strip() + "\n")
    provider_cfg = health_check_image_provider(
        remote_env_path=remote_env,
        provider_name=provider_name,
        timeout_s=int((cfg.get("api", {}) or {}).get("healthcheck_timeout_s", 30)),
    )
    api_cfg = cfg.get("api", {}) if isinstance(cfg.get("api"), dict) else {}
    async_job_cfg = api_cfg.get("async_image_job", {}) if isinstance(api_cfg.get("async_image_job"), dict) else {}
    current_provider_cfg = provider_cfg
    manifest["mode"] = {
        "name": mode.name,
        "label": mode.label,
        "canvas": {"width": canvas_w, "height": canvas_h},
        "accept_score": mode.accept_score,
    }
    manifest["image_provider"] = {
        "provider": provider_cfg.provider,
        "model": provider_cfg.model,
        "base_url": provider_cfg.base_url,
        "source": provider_cfg.source,
        "config_path": str(provider_cfg.config_path) if provider_cfg.config_path else None,
        "auth_path": str(provider_cfg.auth_path) if provider_cfg.auth_path else None,
        "api_key_fingerprint": provider_cfg.api_key_fingerprint,
        "config_conflicts": list(provider_cfg.config_conflicts),
        "preflight_status": provider_cfg.preflight_status,
        "preflight_scope": provider_cfg.preflight_scope,
        "generation_eligibility": provider_cfg.generation_eligibility,
        "selection": str(provider_name or "auto"),
        "allow_provider_fallback": bool(allow_provider_fallback),
        "async_image_job_enabled": bool(async_job_cfg.get("enabled", True)),
        "async_image_job_submit_mode": str(async_job_cfg.get("submit_mode") or "sub2api_job_endpoint"),
        "async_image_job_fallback_to_sync_on_unsupported": bool(
            async_job_cfg.get("fallback_to_sync_on_unsupported", True)
        ),
    }
    manifest["generation"] = {
        "postprocess_resize": effective_postprocess_resize,
        "postprocess_width": int(postprocess_w) if postprocess_w else None,
        "postprocess_height": int(postprocess_h) if postprocess_h else None,
        "quality": quality or gen_cfg.get("quality") or "low",
        "provider_size": provider_size or gen_cfg.get("provider_size") or "1024x1024",
        "output_format": output_format or gen_cfg.get("output_format") or "jpeg",
        "output_compression": output_compression if output_compression is not None else gen_cfg.get("output_compression"),
    }
    write_json(run_dir / str(reports_cfg.get("run_manifest", "run-manifest.json")), manifest)

    rounds_dir = Path(manifest["rounds_dir"])
    exports_dir = Path(manifest["exports_dir"])
    parallel_dir = ensure_dir(Path(manifest["parallel_vibe_dir"]))
    history: List[Dict[str, Any]] = []
    best_round: Optional[Dict[str, Any]] = None
    plateau_hits = 0
    repeated_hash_hits = 0
    last_hash: Optional[str] = None
    stop_reason = "max_rounds"

    for round_index in range(1, max_rounds + 1):
        round_dir = ensure_dir(rounds_dir / f"{gen_cfg.get('round_dir_prefix', 'round-')}{round_index:02d}")
        parallel_plan_path = prepare_parallel_plan(
            run_dir=run_dir,
            request_file=request_file,
            parallel_dir=parallel_dir,
            parallel_cfg=parallel_cfg,
            round_index=round_index,
            round_dir=round_dir,
        )
        prompt_data = build_round_prompt(
            request_text=request_text,
            round_index=round_index,
            max_rounds=max_rounds,
            prior_rounds=history,
            remote_env=remote_env,
            prompt_dir=round_dir,
            mode=mode,
        )
        reference_meta = build_round_reference_images(
            user_reference_images=reference_images,
            prior_rounds=history,
            max_reference_images=int(gen_cfg.get("max_reference_images", 4)),
        )
        round_reference_images = [Path(item) for item in reference_meta["reference_images"]]
        image_output = round_dir / ("output.jpg" if current_provider_cfg.provider == "gpt-image-2" else "output.png")
        image_meta = generate_image(
            prompt=prompt_data["full_prompt"],
            output_png=image_output,
            remote_env=remote_env,
            canvas_w=canvas_w,
            canvas_h=canvas_h,
            debug_dir=round_dir / "image-debug",
            reference_images=round_reference_images,
            provider_cfg=current_provider_cfg,
            provider_name=provider_name,
            allow_provider_fallback=allow_provider_fallback,
            require_reference_images=bool(round_reference_images),
            postprocess_resize=effective_postprocess_resize,
            postprocess_w=postprocess_w,
            postprocess_h=postprocess_h,
            quality=quality,
            provider_size=provider_size,
            output_format=output_format,
            output_compression=output_compression,
        )
        if image_meta.get("provider") and image_meta.get("provider") != current_provider_cfg.provider:
            current_provider_cfg = ImageProviderConfig(
                provider=str(image_meta.get("provider")),
                base_url=str(image_meta.get("base_url") or ""),
                api_key="",
                model=str(image_meta.get("model") or ""),
                env_path=remote_env,
                source=str(image_meta.get("provider_source") or "fallback"),
            )
        try:
            evaluation = evaluate_image(
                image_path=image_output,
                request_text=request_text,
                image_prompt=prompt_data["full_prompt"],
                remote_env=remote_env,
                output_json=round_dir / "evaluation.json",
                debug_dir=round_dir / "evaluation-debug",
                mode=mode,
            )
        except Exception:
            evaluation = heuristic_evaluation(image_output)
            write_json(round_dir / "evaluation.json", evaluation)
        evaluation["mode"] = mode.name
        evaluation["mode_accept_score"] = mode.accept_score
        evaluation["passed"] = bool(float(evaluation.get("score", 0.0)) >= float(mode.accept_score))

        image_hash = sha256_file(image_output)
        round_record = {
            "round": round_index,
            "round_dir": str(round_dir),
            "parallel_plan": str(parallel_plan_path),
            "prompt": prompt_data,
            "reference_strategy": reference_meta,
            "image": image_meta,
            "evaluation": evaluation,
            "image_sha256": image_hash,
        }
        history.append(round_record)
        write_json(analysis_json, {"run": manifest, "rounds": history})

        if best_round is None or float(evaluation.get("score", 0.0)) > float(
            (best_round.get("evaluation", {}) or {}).get("score", 0.0)
        ):
            best_round = round_record
            plateau_hits = 0
        else:
            delta = abs(
                float(evaluation.get("score", 0.0))
                - float((best_round.get("evaluation", {}) or {}).get("score", 0.0))
            )
            if delta <= float(eval_cfg.get("plateau_delta", 0.3)):
                plateau_hits += 1
            else:
                plateau_hits = 0

        if last_hash == image_hash:
            repeated_hash_hits += 1
        else:
            repeated_hash_hits = 0
        last_hash = image_hash

        if bool(evaluation.get("passed")):
            stop_reason = "accepted"
            break
        if repeated_hash_hits >= int(eval_cfg.get("repeated_hash_rounds", 2)) - 1:
            stop_reason = "repeated_hash"
            break
        if plateau_hits >= int(eval_cfg.get("plateau_rounds", 2)):
            stop_reason = "plateau"
            break

    if best_round is None:
        raise RuntimeError("未生成任何可用图片。")

    final_image = Path(manifest.get("public_output_image") or manifest["public_output_png"])
    best_image = Path(best_round["image"].get("output_file") or best_round["image"]["output_png"])
    write_image_with_format_contract(final_image, best_image.read_bytes())
    copy_file(best_image, exports_dir / ("best" + best_image.suffix.lower()))

    summary = {
        "run": manifest,
        "best_round": best_round["round"],
        "best_round_dir": best_round["round_dir"],
        "final_output_image": str(final_image),
        "final_output_png": str(final_image),
        "stop_reason": stop_reason,
        "round_count": len(history),
        "best_score": float((best_round.get("evaluation", {}) or {}).get("score", 0.0)),
        "mode": manifest["mode"],
        "image_provider": manifest["image_provider"],
        "providers_used": [
            {
                "round": item["round"],
                "provider": (item.get("image", {}) or {}).get("provider"),
                "model": (item.get("image", {}) or {}).get("model"),
                "base_url": (item.get("image", {}) or {}).get("base_url"),
                "source": (item.get("image", {}) or {}).get("provider_source"),
                "reference_mode": (item.get("reference_strategy", {}) or {}).get("mode"),
                "source_round": (item.get("reference_strategy", {}) or {}).get("source_round"),
            }
            for item in history
        ],
    }
    write_json(result_json, summary)
    write_json(analysis_json, {"run": manifest, "rounds": history, "summary": summary})
    return summary


def prepare_parallel_plan(
    *,
    run_dir: Path,
    request_file: Path,
    parallel_dir: Path,
    parallel_cfg: Dict[str, Any],
    round_index: int,
    round_dir: Path,
) -> Path:
    round_plan_prefix = str(parallel_cfg.get("round_plan_prefix", "parallel-plan.round-"))
    round_plan_path = parallel_dir / f"{round_plan_prefix}{round_index:02d}.json"
    plan = create_parallel_plan(
        run_dir=run_dir,
        request_file=request_file,
        round_index=round_index,
        output_plan=round_plan_path,
    )
    latest_plan_path = parallel_dir / str(parallel_cfg.get("plan_filename", "parallel-plan.json"))
    if latest_plan_path != round_plan_path:
        write_json(latest_plan_path, plan)
    write_json(round_dir / "parallel-plan.json", plan)
    return round_plan_path


def _normalize_text_list(value: Any) -> List[str]:
    if isinstance(value, list):
        return [str(item).strip() for item in value if str(item).strip()]
    if isinstance(value, str) and value.strip():
        return [value.strip()]
    return []


def main() -> None:
    parser = argparse.ArgumentParser(description="auto-draw-plot 多轮图片生成主入口。")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--request-file", help="用户需求 Markdown/TXT")
    group.add_argument("--request-text", help="用户需求文本")
    parser.add_argument("--project-root", default=".")
    parser.add_argument("--workspace-base", default="", help="自定义隐藏工作区根目录")
    parser.add_argument("--output-image", "--output-png", dest="output_image", default="", help="最终输出图片路径；--output-png 为兼容别名")
    parser.add_argument("--max-rounds", type=int, default=0)
    parser.add_argument("--canvas-width", type=int, default=0, help="期望布局宽度/宽高比参考，不承诺最终图片像素")
    parser.add_argument("--canvas-height", type=int, default=0, help="期望布局高度/宽高比参考，不承诺最终图片像素")
    parser.add_argument("--postprocess-resize", action="store_true", default=None, help="显式启用后处理尺寸对齐；默认保留 provider 原生输出")
    parser.add_argument("--postprocess-width", type=int, default=0, help="后处理目标宽度；需配合 --postprocess-resize")
    parser.add_argument("--postprocess-height", type=int, default=0, help="后处理目标高度；需配合 --postprocess-resize")
    parser.add_argument("--quality", default="", help="OpenAI 图片模型 quality：low/medium/high/auto")
    parser.add_argument("--provider-size", default="", help="OpenAI 图片模型原生尺寸枚举，默认 1024x1024")
    parser.add_argument("--output-format", default="", help="OpenAI 图片模型输出格式：jpeg/png/webp")
    parser.add_argument("--output-compression", type=int, default=-1, help="输出压缩 0-100；默认使用配置值")
    parser.add_argument("--api-env", default="", help="remote.env 路径")
    parser.add_argument("--mode", default="", help="绘图模式：general（默认）/ roadmap / schematic")
    parser.add_argument("--provider", default="auto", help="图片 provider/model：auto（默认）/ gpt-image-2.5-flare / gpt-image-2.5-sunburst / gpt-image-2 / nano_banana")
    parser.add_argument(
        "--allow-provider-fallback",
        action="store_true",
        help="provider 故障时允许从 OpenAI 图片模型切到 Nano Banana/Gemini；计费、权限与客户端策略错误仍不回退",
    )
    parser.add_argument("--allow-outside-project", action="store_true", help="允许 workspace/output 写到 project_root 外部")
    parser.add_argument("--reference-image", action="append", default=[], help="可重复传入参考图")
    args = parser.parse_args()

    request_text = read_text(Path(args.request_file)) if args.request_file else str(args.request_text or "")
    summary = run_draw_plot(
        request_text=request_text,
        project_root=expand_path(args.project_root, base=Path.cwd()),
        output_png=args.output_image or None,
        workspace_base=args.workspace_base or None,
        max_rounds=(args.max_rounds or None),
        canvas_w=(args.canvas_width or None),
        canvas_h=(args.canvas_height or None),
        remote_env=expand_path(args.api_env, base=Path.cwd()) if args.api_env else None,
        reference_images=[expand_path(item, base=Path.cwd()) for item in (args.reference_image or [])],
        mode_name=args.mode or None,
        provider_name=str(args.provider or "auto"),
        allow_provider_fallback=bool(args.allow_provider_fallback),
        allow_outside_project=bool(args.allow_outside_project),
        postprocess_resize=args.postprocess_resize,
        postprocess_w=int(args.postprocess_width) or None,
        postprocess_h=int(args.postprocess_height) or None,
        quality=args.quality or None,
        provider_size=args.provider_size or None,
        output_format=args.output_format or None,
        output_compression=args.output_compression if args.output_compression >= 0 else None,
    )
    print(summary.get("final_output_image") or summary["final_output_png"])


if __name__ == "__main__":
    main()