451 lines
25 KiB
Python
451 lines
25 KiB
Python
#!/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()
|