#!/usr/bin/env python3 """Local web wrapper for the WhaleTown character generation pipeline.""" from __future__ import annotations import argparse import json import mimetypes import os import re import subprocess import threading import time import uuid import io import shutil from PIL import Image from http import HTTPStatus from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import unquote, urlparse TOOL_DIR = Path(__file__).resolve().parent BACKEND_DIR = TOOL_DIR.parent.parent WORKER = BACKEND_DIR / "scripts" / "skin_generation" / "generate_skin_from_prompt.py" DEFAULT_OUTPUT = BACKEND_DIR / "generated" / "character_maker" DEFAULT_SECRET_FILE = BACKEND_DIR.parent / "whale-town-front-v2" / ".secrets" / "novamailio.env" MAX_UPLOAD_BYTES = 8 * 1024 * 1024 ALLOWED_MIME = {"image/png": ".png", "image/jpeg": ".jpg", "image/webp": ".webp"} def load_local_secret() -> None: if os.getenv("NOVAMAILIO_API_KEY") or not DEFAULT_SECRET_FILE.is_file(): return for line in DEFAULT_SECRET_FILE.read_text("utf-8").splitlines(): if line.startswith("NOVAMAILIO_API_KEY="): value = line.split("=", 1)[1].strip().strip('"').strip("'") if value: os.environ["NOVAMAILIO_API_KEY"] = value return class CharacterMaker: def __init__(self, output_root: Path, python: str) -> None: self.output_root = output_root.resolve() self.python = python self.jobs: dict[str, dict[str, object]] = {} self.lock = threading.Lock() self.output_root.mkdir(parents=True, exist_ok=True) self._load_jobs() @staticmethod def _safe_name(value: str) -> str: value = re.sub(r"[^a-zA-Z0-9_\u4e00-\u9fff]+", "_", value.strip()) return re.sub(r"_+", "_", value).strip("_")[:40] or "custom_character" @staticmethod def _pose_plan() -> list[dict[str, object]]: plan = [] for direction in ("down", "up", "right", "left"): plan.extend( [ {"direction": direction, "pose": "neutral", "frame": 1, "depends_on": []}, {"direction": direction, "pose": "opposite_leg", "frame": 4, "depends_on": [f"{direction}:neutral"]}, {"direction": direction, "pose": "first_leg", "frame": 2, "depends_on": [f"{direction}:opposite_leg"]}, ] ) return plan def _load_jobs(self) -> None: for path in sorted(self.output_root.glob("*/job.json"), reverse=True): try: job = json.loads(path.read_text("utf-8")) if job.get("status") == "running": job.update(status="interrupted", message="工具曾退出，可重新开始生成") if job.get("poses"): for pose_state in job["poses"].values(): if pose_state.get("status") == "running": pose_state.update(status="interrupted", error="工具曾退出，可重新生成") self.jobs[str(job["id"])] = job except (OSError, ValueError, KeyError): continue def list_jobs(self) -> list[dict[str, object]]: with self.lock: jobs = list(self.jobs.values()) return sorted((self._public_job(job) for job in jobs), key=lambda x: str(x["created_at"]), reverse=True) def template_prompts(self) -> list[list[str]]: code = "import json; from pathlib import Path; p=Path(%r); ns={'__file__':str(p),'__name__':'prompt_templates'}; exec(compile(p.read_text(),str(p),'exec'),ns); poses=(0,1,0,2,0,1,0,2); print(json.dumps({'identity':ns['_identity_prompt'](),'frames':[[ns['_single_pose_prompt'](d,x) if x==0 else ns['_single_pose_from_sibling_prompt'](d,x) for x in poses] for d in ('down','up','right','left')]},ensure_ascii=False))" % str(WORKER) result = subprocess.run([self.python, "-c", code], cwd=BACKEND_DIR, capture_output=True, text=True) if result.returncode != 0: raise RuntimeError(result.stderr.strip() or "无法读取生成脚本提示词") return json.loads(result.stdout) def get_job(self, job_id: str) -> dict[str, object] | None: with self.lock: job = self.jobs.get(job_id) return self._public_job(job) if job else None def create_job(self, name: str, quality: str, mime: str, image: bytes, phase: str = "identity") -> dict[str, object]: if mime not in ALLOWED_MIME: raise ValueError("只支持 PNG、JPG 和 WebP 图片") if not 512 <= len(image) <= MAX_UPLOAD_BYTES: raise ValueError("参考图大小须在 512B 到 8MB 之间") if quality not in {"low", "medium", "high"}: raise ValueError("无效的质量档位") job_id = f"{time.strftime('%Y%m%d-%H%M%S')}-{uuid.uuid4().hex[:6]}" job_dir = self.output_root / job_id job_dir.mkdir(parents=True) source = job_dir / f"source{ALLOWED_MIME[mime]}" source.write_bytes(image) job: dict[str, object] = { "id": job_id, "name": self._safe_name(name), "quality": quality, "status": "queued", "stage": "queued", "message": "任务已创建", "created_at": time.strftime("%Y-%m-%d %H:%M:%S"), "source": source.name, "log": [], "phase": phase, "pose_plan": self._pose_plan(), } with self.lock: self.jobs[job_id] = job self._save(job) threading.Thread(target=self._run, args=(job_id,), daemon=True).start() return self._public_job(job) def _run(self, job_id: str) -> None: job = self.jobs[job_id] job_dir = self.output_root / job_id status_path = job_dir / "status.json" result_path = job_dir / "result.json" source = job_dir / str(job["source"]) cmd = [ self.python, str(WORKER), "--source-image", str(source), "--out-dir", str(job_dir), "--name", str(job["name"]), "--quality", str(job["quality"]), "--status-json", str(status_path), "--result-json", str(result_path), "--phase", str(job.get("phase", "all")), ] self._update(job, status="running", stage="start", message="正在启动角色生成流程") try: process = subprocess.Popen( cmd, cwd=BACKEND_DIR, env=os.environ.copy(), text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1, ) assert process.stdout is not None for line in process.stdout: line = line.strip() if line: logs = list(job.get("log", []))[-79:] + [line] updates: dict[str, object] = {"message": line, "log": logs} if status_path.exists(): try: status = json.loads(status_path.read_text("utf-8")) updates.update(stage=status.get("stage", job["stage"]), message=status.get("message", line)) except (OSError, ValueError): pass self._update(job, **updates) code = process.wait() result = json.loads(result_path.read_text("utf-8")) if result_path.exists() else {} if code != 0 or not result.get("ok"): raise RuntimeError(str(result.get("error") or f"生成进程退出码 {code}")) files = {} if result.get("phase") == "identity": identity_path = Path(str(result.get("identity_path", ""))) if identity_path.exists() and identity_path.is_relative_to(job_dir): files["identity_path"] = str(identity_path.relative_to(job_dir)) self._update(job, status="awaiting_confirmation", stage="identity_done", message="身份母版已生成，请确认后继续动作", files=files) return for key in ("spritesheet_path", "review_path", "feet_zoom_path"): path = Path(str(result.get(key, ""))) if path.exists() and path.is_relative_to(job_dir): files[key] = str(path.relative_to(job_dir)) self._update(job, status="completed", stage="done", message="角色生成完成", files=files) except Exception as exc: # Keep worker failures visible in the local UI. self._update(job, status="failed", stage="failed", message=f"生成失败：{exc}") def continue_job(self, job_id: str) -> dict[str, object] | None: job = self.jobs.get(job_id) if not job or job.get("status") == "running": return self._public_job(job) if job else None job["phase"] = "actions"; job["status"] = "action_ready"; job["stage"] = "action_ready"; job["message"] = "身份母版已确认，请逐个生成必要动作"; job["poses"] = job.get("poses", {}) self._save(job); return self._public_job(job) def generate_pose(self, job_id: str, direction: str, pose: str) -> dict[str, object] | None: job = self.jobs.get(job_id) if not job: return None if direction not in ("down", "up", "right", "left") or pose not in ("neutral", "opposite_leg", "first_leg"): raise ValueError("无效动作") poses = dict(job.get("poses", {})); key = f"{direction}:{pose}" dependencies = {"opposite_leg": f"{direction}:neutral", "first_leg": f"{direction}:opposite_leg"} dependency = dependencies.get(pose) if dependency and poses.get(dependency, {}).get("status") != "completed": raise ValueError(f"请先完成前置动作：{dependency}") if poses.get(key, {}).get("status") == "running": return self._public_job(job) poses[key] = {"status": "running", "attempt": 0}; job["poses"] = poses; self._save(job) threading.Thread(target=self._run_pose, args=(job_id, direction, pose), daemon=True).start(); return self._public_job(job) def _run_pose(self, job_id: str, direction: str, pose: str) -> None: job = self.jobs[job_id]; job_dir = self.output_root / job_id; key = f"{direction}:{pose}" result_path = job_dir / f"result_{direction}_{pose}.json"; status_path = job_dir / f"status_{direction}_{pose}.json" base_cmd = [self.python, str(WORKER), "--source-image", str(job_dir / str(job["source"])), "--out-dir", str(job_dir), "--name", str(job["name"]), "--quality", str(job["quality"]), "--result-json", str(result_path), "--status-json", str(status_path), "--phase", "pose", "--direction", direction, "--pose", pose] try: payload = {}; last_error = ""; archived_attempts: list[list[str]] = [] for attempt in range(1, 4): if attempt > 1: archived_attempts.append(self._archive_pose_attempt(job_dir, str(job["name"]), direction, pose, attempt - 1)) cmd = list(base_cmd) if last_error: cmd.extend(["--qa-feedback", last_error[:1200]]) result = subprocess.run(cmd, cwd=BACKEND_DIR, env=os.environ.copy(), capture_output=True, text=True) payload = json.loads(result_path.read_text("utf-8")) if result_path.exists() else {} if result.returncode == 0 and payload.get("ok"): break last_error = str(payload.get("error") or result.stderr[-1000:] or "动作生成失败") poses = dict(job.get("poses", {})); poses[key] = {"status": "running", "attempt": attempt, "last_error": last_error, "archived_attempts": archived_attempts}; job["poses"] = poses; self._save(job) else: raise RuntimeError(last_error) display = self._pose_display_path(job_id, str(job["name"]), direction, pose) preview = display or Path(str(payload["preview_path"])); inputs = {} for field in ("pose_reference_path", "layout_guide_path", "prompt_path"): path = Path(str(payload.get(field, ""))) if path.exists() and path.is_relative_to(job_dir): inputs[field] = str(path.relative_to(job_dir)) poses = dict(job.get("poses", {})); poses[key] = {"status": "completed", "attempt": attempt, "preview_url": f"/files/{job_id}/{preview.relative_to(job_dir)}", "qa": payload.get("qa", {}), "inputs": inputs, "archived_attempts": archived_attempts}; job["poses"] = poses; job["message"] = f"{direction} {pose} 已生成并通过逐格检查"; self._save(job) except Exception as exc: raw = job_dir / "raw" / f"{job['name']}_{direction}_{pose}_source.png" preview = job_dir / "cutout" / f"{job['name']}_{direction}_{pose}_preview.png" candidate = {"status": "failed", "error": str(exc), "candidate": True} display = self._pose_display_path(job_id, str(job["name"]), direction, pose) if display: candidate["preview_url"] = f"/files/{job_id}/{display.relative_to(job_dir)}" elif preview.exists(): candidate["preview_url"] = f"/files/{job_id}/{preview.relative_to(job_dir)}" if raw.exists(): candidate["raw_url"] = f"/files/{job_id}/{raw.relative_to(job_dir)}" if "archived_attempts" not in candidate: candidate["archived_attempts"] = locals().get("archived_attempts", []) poses = dict(job.get("poses", {})); poses[key] = candidate; job["poses"] = poses; job["message"] = f"动作生成失败，已保留候选预览：{exc}"; self._save(job) @staticmethod def _archive_pose_attempt(job_dir: Path, skin_name: str, direction: str, pose: str, attempt: int) -> list[str]: archive_dir = job_dir / "attempts" / f"{direction}_{pose}" / f"attempt_{attempt}" archived: list[str] = [] for folder, suffix in (("raw", "_source.png"), ("cutout", "_cutout.png"), ("cutout", "_mask.png"), ("cutout", "_preview.png"), ("cutout", "_display.png")): source = job_dir / folder / f"{skin_name}_{direction}_{pose}{suffix}" if not source.is_file(): continue archive_dir.mkdir(parents=True, exist_ok=True) target = archive_dir / source.name shutil.copy2(source, target) archived.append(str(target.relative_to(job_dir))) return archived def _pose_display_path(self, job_id: str, skin_name: str, direction: str, pose: str) -> Path | None: """Create a single-character preview from the transparent cutout. The cutout tool's preview is a two-panel diagnostic image, so it is kept for QA but never shown as the user's generated pose preview. """ job_dir = self.output_root / job_id cutout = job_dir / "cutout" / f"{skin_name}_{direction}_{pose}_cutout.png" display = job_dir / "cutout" / f"{skin_name}_{direction}_{pose}_display.png" if cutout.is_file() and ( not display.is_file() or display.stat().st_mtime < cutout.stat().st_mtime ): try: with Image.open(cutout).convert("RGBA") as source: canvas = Image.new("RGBA", source.size, (255, 0, 255, 255)) canvas.alpha_composite(source) canvas.convert("RGB").save(display, format="PNG") except (OSError, ValueError): return None if display.is_file(): return display raw = job_dir / "raw" / f"{skin_name}_{direction}_{pose}_source.png" return raw if raw.is_file() else None def _update(self, job: dict[str, object], **updates: object) -> None: with self.lock: job.update(updates) self._save(job) def _save(self, job: dict[str, object]) -> None: path = self.output_root / str(job["id"]) / "job.json" path.write_text(json.dumps(job, ensure_ascii=False, indent=2), "utf-8") def _public_job(self, job: dict[str, object]) -> dict[str, object]: data = dict(job) job_id = str(job["id"]) data["source_url"] = f"/files/{job_id}/{job['source']}" data["file_urls"] = { key: f"/files/{job_id}/{value}" for key, value in dict(job.get("files", {})).items() } data["prompts"] = job.get("prompts", []) data["pose_plan"] = job.get("pose_plan") or self._pose_plan() # Recover previews from disk for jobs interrupted after image generation. # Older jobs may have raw/cutout files but no pose URL persisted in job.json. poses = {str(key): dict(value) for key, value in dict(job.get("poses") or {}).items()} skin_name = str(job.get("name", "custom_character")) for direction in ("down", "up", "right", "left"): for pose in ("neutral", "first_leg", "opposite_leg"): key = f"{direction}:{pose}" state = poses.setdefault(key, {}) preview = self._pose_display_path(job_id, skin_name, direction, pose) raw = self.output_root / job_id / "raw" / f"{skin_name}_{direction}_{pose}_source.png" if preview and preview.is_file(): state.update(preview_url=f"/files/{job_id}/{preview.relative_to(self.output_root / job_id)}", candidate=state.get("status") != "completed") elif state.get("preview_url") or state.get("raw_url"): continue elif raw.is_file(): state.update(raw_url=f"/files/{job_id}/{raw.relative_to(self.output_root / job_id)}", candidate=state.get("status") != "completed") if poses: data["poses"] = poses data["frame_plan"] = [{"frame": i + 1, "source_frame": (1, 2, 1, 4, 1, 2, 1, 4)[i], "generated": i in (0, 1, 3)} for i in range(8)] data["reference_urls"] = [f"/references/human_whale_reference_{d}.png" for d in ("down", "up", "right", "left")] data["reference_cell_urls"] = [[f"/references/cell/{d}/{f}.png" for f in (1, 2, 4)] for d in ("down", "up", "right", "left")] return data class Handler(SimpleHTTPRequestHandler): maker: CharacterMaker def do_GET(self) -> None: path = urlparse(self.path).path if path == "/api/health": self._json({"ok": True, "worker": WORKER.exists(), "api_key": bool(os.getenv("NOVAMAILIO_API_KEY"))}) return if path == "/api/jobs": self._json({"jobs": self.maker.list_jobs()}) return if path == "/api/prompt-templates": try: templates = self.maker.template_prompts() self._json({"identity_prompt": templates["identity"], "prompts": templates["frames"], "identity_reference_url": "/references/whaleboy_identity_single.png"}) except Exception as exc: self._json({"error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) return match = re.fullmatch(r"/api/jobs/([\w-]+)", path) if match: job = self.maker.get_job(match.group(1)) self._json(job or {"error": "任务不存在"}, HTTPStatus.OK if job else HTTPStatus.NOT_FOUND) return if path.startswith("/files/"): self._serve_output(path) return if path.startswith("/references/"): cell_match = re.fullmatch(r"/references/cell/(down|up|right|left)/(1|2|4)\.png", path) if cell_match: direction, frame = cell_match.groups() source = BACKEND_DIR / "scripts" / "skin_generation" / "references" / f"human_whale_reference_{direction}.png" with Image.open(source).convert("RGB") as image: cell_width = image.width // 8 cell = image.crop(((int(frame) - 1) * cell_width, 0, int(frame) * cell_width, image.height)) pixels = cell.load(); xs=[]; ys=[] for y in range(cell.height): for x in range(cell.width): red, green, blue = pixels[x, y] if not (red > 220 and green < 80 and blue > 180): xs.append(x); ys.append(y) box = (max(0, min(xs)-8), max(0, min(ys)-8), min(cell.width, max(xs)+9), min(cell.height, max(ys)+9)) cropped = cell.crop(box); scale = 700 / max(1, max(ys)-min(ys)+1); cropped = cropped.resize((round(cropped.width*scale), round(cropped.height*scale)), Image.Resampling.LANCZOS) canvas = Image.new("RGB", (1024, 1024), (255, 0, 255)); canvas.paste(cropped, ((1024-cropped.width)//2, (1024-cropped.height)//2)); output = io.BytesIO(); canvas.save(output, format="PNG"); data = output.getvalue() self.send_response(200); self.send_header("Content-Type", "image/png"); self.send_header("Content-Length", str(len(data))); self.send_header("Cache-Control", "no-store"); self.end_headers(); self.wfile.write(data); return name = Path(unquote(path).split("/")[-1]).name target = BACKEND_DIR / "scripts" / "skin_generation" / "references" / name if target.is_file(): data = target.read_bytes(); self.send_response(200); self.send_header("Content-Type", "image/png"); self.send_header("Content-Length", str(len(data))); self.end_headers(); self.wfile.write(data); return self.send_error(404); return if path in {"/", "/index.html"}: self.path = "/index.html" return super().do_GET() def do_POST(self) -> None: request_path = urlparse(self.path).path pose_match = re.fullmatch(r"/api/jobs/([\w-]+)/poses", request_path) if pose_match: try: length = int(self.headers.get("Content-Length", "0")); payload = json.loads(self.rfile.read(length)); job = self.maker.generate_pose(pose_match.group(1), str(payload.get("direction", "")), str(payload.get("pose", ""))); self._json(job or {"error": "任务不存在"}, HTTPStatus.ACCEPTED if job else HTTPStatus.NOT_FOUND) except (ValueError, json.JSONDecodeError) as exc: self._json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) return if request_path != "/api/jobs": self._json({"error": "接口不存在"}, HTTPStatus.NOT_FOUND) return try: length = int(self.headers.get("Content-Length", "0")) if length > MAX_UPLOAD_BYTES + 4096: raise ValueError("上传内容不能超过 8MB") payload = json.loads(self.rfile.read(length)) import base64 image = base64.b64decode(str(payload.get("image", "")), validate=True) job = self.maker.create_job( str(payload.get("name", "")), str(payload.get("quality", "medium")), str(payload.get("mime", "")), image, str(payload.get("phase", "identity")), ) self._json(job, HTTPStatus.CREATED) except (ValueError, json.JSONDecodeError) as exc: self._json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) def do_PUT(self) -> None: match = re.fullmatch(r"/api/jobs/([\w-]+)/continue", urlparse(self.path).path) if match: job = self.maker.continue_job(match.group(1)); self._json(job or {"error": "任务不存在"}, HTTPStatus.OK if job else HTTPStatus.NOT_FOUND); return self._json({"error": "接口不存在"}, HTTPStatus.NOT_FOUND) def translate_path(self, path: str) -> str: relative = urlparse(path).path.lstrip("/") return str((TOOL_DIR / relative).resolve()) def _serve_output(self, request_path: str) -> None: relative = unquote(request_path.removeprefix("/files/")) target = (self.maker.output_root / relative).resolve() if not target.is_relative_to(self.maker.output_root) or not target.is_file(): self.send_error(HTTPStatus.NOT_FOUND) return data = target.read_bytes() self.send_response(HTTPStatus.OK) self.send_header("Content-Type", mimetypes.guess_type(target.name)[0] or "application/octet-stream") self.send_header("Content-Length", str(len(data))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(data) def _json(self, payload: object, status: HTTPStatus = HTTPStatus.OK) -> None: data = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(data) def log_message(self, fmt: str, *args: object) -> None: print(f"[character-maker] {self.address_string()} {fmt % args}") def main() -> None: load_local_secret() parser = argparse.ArgumentParser(description="WhaleTown local character maker") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8765) parser.add_argument("--python", default=os.getenv("SKIN_GENERATION_PYTHON", "python3")) parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT) args = parser.parse_args() Handler.maker = CharacterMaker(args.output, args.python) server = ThreadingHTTPServer((args.host, args.port), Handler) print(f"角色工坊已启动：http://{args.host}:{args.port}") server.serve_forever() if __name__ == "__main__": main()