feat: integrate invitation access, world NPCs, and deployment

This commit is contained in:
2026-09-08 22:22:38 +08:00
parent 2a3125075f
commit 513a3eba31
75 changed files with 9566 additions and 1070 deletions

View File

@@ -0,0 +1,24 @@
# WhaleTown 角色工坊
这是现有角色皮肤生成脚本的本地网页封装。它复用 `scripts/skin_generation/generate_skin_from_prompt.py`,不经过玩家账号、不会消耗注册生成次数。
## 启动
在后端目录执行:
```bash
export NOVAMAILIO_API_KEY="你的密钥"
python3 tools/character_maker/app.py
```
浏览器打开 <http://127.0.0.1:8765>。如生成环境使用独立 Python
```bash
python3 tools/character_maker/app.py --python /path/to/python --port 8765
```
任务和所有中间产物保存在 `generated/character_maker/<任务ID>/`。页面会恢复历史记录,并提供最终 8x4 皮肤表、总览验收图和足部动作检查图。
动作阶段每个必要格都会立即显示单角色生成预览。QA 失败或工具中断时,页面会从最后一次抠图/原始图中恢复候选预览,并提供查看大图和重新生成入口;抠图工具的双栏诊断图只保留在任务目录中,不作为用户预览。
工具只监听 `127.0.0.1`API Key 不会发给浏览器。生成过程会实际调用图片生成服务并产生费用。

View File

@@ -0,0 +1,324 @@
const $ = (selector) => document.querySelector(selector);
let selectedFile = null;
let currentJob = null;
let currentJobData = null;
let row = 0;
let frame = 0;
let sheet = new Image();
let poller = null;
let currentPrompts = [];
let currentRefs = [
'/references/human_whale_reference_down.png',
'/references/human_whale_reference_up.png',
'/references/human_whale_reference_right.png',
'/references/human_whale_reference_left.png',
];
const directions = ['down', 'up', 'right', 'left'];
const directionLabels = ['正面', '背面', '右侧', '左侧'];
const poseItems = [
{ frame: 1, pose: 'neutral', label: '中立' },
{ frame: 2, pose: 'first_leg', label: '动作 A' },
{ frame: 4, pose: 'opposite_leg', label: '动作 B' },
];
const stages = { queued: 4, start: 8, identity: 18, generate: 34, cutout: 54, expand: 68, assemble: 82, normalize: 90, qa: 96, done: 100, failed: 100 };
async function api(path, options) {
const response = await fetch(path, options);
const data = await response.json();
if (!response.ok) throw new Error(data.error || '请求失败');
return data;
}
function statusText(status) {
return ({ queued: '排队中', running: '生成中', completed: '已完成', failed: '失败', interrupted: '已中断', awaiting_confirmation: '待确认身份', action_ready: '动作待生成' })[status] || status;
}
function escapeHtml(value) {
const node = document.createElement('div');
node.textContent = value == null ? '' : String(value);
return node.innerHTML;
}
async function health() {
try {
const state = await api('/api/health');
$('#health').className = `health ${state.worker && state.api_key ? 'ok' : 'bad'}`;
$('#health span').textContent = !state.worker ? '生成脚本缺失' : !state.api_key ? '未配置 API Key' : '产线就绪';
} catch {
$('#health').className = 'health bad';
$('#health span').textContent = '工具离线';
}
}
function setFile(file) {
if (!file) return;
if (!['image/png', 'image/jpeg', 'image/webp'].includes(file.type) || file.size > 8 * 1024 * 1024) {
$('#formError').textContent = '请选择 8MB 以内的 PNG、JPG 或 WebP';
return;
}
selectedFile = file;
$('#formError').textContent = '';
$('#sourcePreview').src = URL.createObjectURL(file);
$('#dropzone').classList.add('has-image');
$('#generate').disabled = false;
if (!$('#name').value) $('#name').value = file.name.replace(/\.[^.]+$/, '').slice(0, 40);
}
function fileToBase64(file) {
return new Promise((resolve, reject) => {
const reader = new FileReader();
reader.onload = () => resolve(reader.result.split(',')[1]);
reader.onerror = reject;
reader.readAsDataURL(file);
});
}
async function createJob() {
if (!selectedFile) return;
const button = $('#generate');
button.disabled = true;
$('#formError').textContent = '';
try {
const job = await api('/api/jobs', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ name: $('#name').value, quality: $('input[name=quality]:checked').value, mime: selectedFile.type, image: await fileToBase64(selectedFile) }),
});
currentJob = job.id;
await loadJobs();
show(job);
startPoll();
} catch (error) {
$('#formError').textContent = error.message;
} finally {
button.disabled = false;
}
}
async function loadJobs() {
const { jobs } = await api('/api/jobs');
$('#jobList').innerHTML = jobs.length ? jobs.map((job) => `<button class="job-item ${job.id === currentJob ? 'active' : ''}" data-id="${escapeHtml(job.id)}"><strong>${escapeHtml(job.name)}</strong><span>${statusText(job.status)} · ${escapeHtml(job.created_at)}</span></button>`).join('') : '<p class="error">还没有制作记录</p>';
document.querySelectorAll('.job-item').forEach((button) => {
button.onclick = async () => {
currentJob = button.dataset.id;
show(await api(`/api/jobs/${currentJob}`));
startPoll();
loadJobs();
};
});
}
function startPoll() {
clearInterval(poller);
poller = setInterval(async () => {
if (!currentJob) return;
try { show(await api(`/api/jobs/${currentJob}`)); } catch { /* health indicator handles transient outages */ }
}, 1500);
}
function clearCanvas() {
$('#canvas').getContext('2d').clearRect(0, 0, $('#canvas').width, $('#canvas').height);
$('#frames').innerHTML = '';
}
function drawFrame(canvas, directionRow, frameIndex) {
const context = canvas.getContext('2d');
context.imageSmoothingEnabled = false;
context.clearRect(0, 0, canvas.width, canvas.height);
context.drawImage(sheet, frameIndex * 160, directionRow * 160, 160, 160, 0, 0, canvas.width, canvas.height);
}
function drawAll() {
const hasSheet = sheet.complete && sheet.naturalWidth;
if (hasSheet) {
drawFrame($('#canvas'), row, frame);
$('#frames').innerHTML = '';
for (let index = 0; index < 8; index += 1) {
const button = document.createElement('button');
button.className = index === frame ? 'active' : '';
const thumbnail = document.createElement('canvas');
thumbnail.width = thumbnail.height = 80;
drawFrame(thumbnail, row, index);
button.append(thumbnail);
button.onclick = () => { frame = index; drawAll(); };
$('#frames').append(button);
}
} else {
$('#frames').innerHTML = '';
}
$('#frameLabel').textContent = `${frame + 1} / 8 帧`;
const promptText = $('#promptText');
if (promptText) promptText.value = selectedPrompt();
const poseName = { 1: 'neutral', 2: 'first_leg', 4: 'opposite_leg' }[frame + 1] || 'neutral';
const state = currentJobData?.poses?.[`${directions[row]}:${poseName}`];
const poseUrl = state?.preview_url || state?.raw_url;
const posePreview = $('#poseCanvasPreview');
if (!hasSheet && poseUrl) {
posePreview.src = `${poseUrl}?t=${Date.now()}`;
posePreview.classList.remove('hidden');
$('#canvas').classList.add('hidden');
} else {
posePreview.classList.add('hidden');
$('#canvas').classList.toggle('hidden', !hasSheet);
}
}
function selectedPrompt() {
return currentPrompts?.[row]?.[frame] || '提示词尚未生成';
}
function poseReference(direction, frameNumber, rowIndex) {
const cellIndex = [1, 2, 4].indexOf(frameNumber);
return currentJobData?.reference_cell_urls?.[rowIndex]?.[cellIndex] || `/references/cell/${direction}/${frameNumber}.png`;
}
function poseState(direction, pose) {
return currentJobData?.poses?.[`${direction}:${pose}`] || {};
}
function poseEnabled(direction, pose) {
if (pose === 'neutral') return true;
if (pose === 'opposite_leg') return poseState(direction, 'neutral').status === 'completed';
return poseState(direction, 'opposite_leg').status === 'completed';
}
async function requestPose(direction, pose) {
try {
await api(`/api/jobs/${currentJob}/poses`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ direction, pose }) });
show(await api(`/api/jobs/${currentJob}`));
startPoll();
} catch (error) {
$('#jobMessage').textContent = error.message;
}
}
function renderPromptGrid() {
const grid = $('#jobPromptGrid');
if (!grid) return;
grid.innerHTML = '';
let completed = 0;
let candidates = 0;
directions.forEach((direction, rowIndex) => poseItems.forEach((item) => {
const state = poseState(direction, item.pose);
const resultUrl = state.preview_url || state.raw_url;
const imageUrl = resultUrl || poseReference(direction, item.frame, rowIndex);
const status = state.status === 'completed' ? '通过 QA' : state.candidate ? 'QA 未通过 · 候选' : state.status === 'running' ? '生成中…' : '待生成';
if (state.status === 'completed') completed += 1;
if (state.candidate) candidates += 1;
const card = document.createElement('div');
card.className = `prompt-cell pose-cell${state.status === 'running' ? ' busy' : ''}`;
card.innerHTML = `<img class="pose-image ${resultUrl ? '' : 'reference'}" src="${imageUrl}${imageUrl.includes('?') ? '&' : '?'}t=${Date.now()}" alt="${directionLabels[rowIndex]}${item.frame}${resultUrl ? '生成预览' : '参考图'}"><span class="pose-badge ${state.candidate ? 'candidate' : state.status === 'completed' ? '' : 'failed'}">${status}</span>${resultUrl ? `<a class="pose-link" href="${resultUrl}" target="_blank" rel="noreferrer">查看大图</a>` : ''}<b>${directionLabels[rowIndex]} · 第 ${item.frame} 格 · ${item.label}</b><p>${escapeHtml(currentPrompts?.[rowIndex]?.[item.frame - 1] || '')}</p><button class="pose-generate" ${!poseEnabled(direction, item.pose) || state.status === 'running' ? 'disabled' : ''}>${state.status === 'completed' ? '重新生成' : state.status === 'running' ? '生成中…' : state.candidate ? '重新生成' : '生成这一格'}</button>${state.error ? `<small>${escapeHtml(state.error)}</small>` : ''}`;
card.onclick = (event) => {
if (event.target.closest('button,a')) return;
row = rowIndex;
frame = item.frame - 1;
document.querySelectorAll('.pose-cell').forEach((cell) => cell.classList.remove('active'));
card.classList.add('active');
drawAll();
};
card.querySelector('button').onclick = () => requestPose(direction, item.pose);
if (state.inputs?.layout_guide_path) {
const guideLink = document.createElement('a');
guideLink.className = 'pose-link';
guideLink.href = `/files/${currentJob}/${state.inputs.layout_guide_path}`;
guideLink.target = '_blank';
guideLink.rel = 'noreferrer';
guideLink.textContent = '查看布局框';
card.append(guideLink);
}
if (state.qa?.warnings?.length) {
const warning = document.createElement('small');
warning.className = 'qa-warning';
warning.textContent = `提示:${state.qa.warnings.join('')}`;
card.append(warning);
}
if (state.archived_attempts?.length) {
const attempts = document.createElement('small');
attempts.className = 'attempt-note';
attempts.textContent = `已归档 ${state.archived_attempts.length} 次候选`;
card.append(attempts);
}
grid.append(card);
}));
$('#poseSummary').textContent = `${completed}/12 已通过${candidates ? ` · ${candidates} 个候选待确认` : ''}`;
}
function show(job) {
currentJobData = job;
$('#empty').classList.add('hidden');
$('#jobView').classList.remove('hidden');
$('#jobName').textContent = job.name || '';
$('#jobStatus').textContent = statusText(job.status);
$('#jobMessage').textContent = job.message || '';
$('#jobTime').textContent = job.created_at || '';
$('#progressBar').style.width = `${stages[job.stage] ?? 12}%`;
$('#progressBar').style.background = job.status === 'failed' ? '#e7674f' : '';
$('#logs').textContent = (job.log || []).join('\n');
const identityUrl = job.file_urls?.identity_path;
$('#identityResult').classList.toggle('hidden', !identityUrl);
if (identityUrl) $('#identityImage').src = `${identityUrl}?t=${Date.now()}`;
const actionsVisible = job.phase === 'actions' || job.stage === 'done' || Object.keys(job.poses || {}).length > 0;
$('#actionTask').classList.toggle('hidden', !actionsVisible);
$('#actionPreview').classList.toggle('hidden', !actionsVisible);
const continueButton = $('#continueJob');
continueButton.classList.toggle('hidden', job.stage !== 'identity_done');
continueButton.onclick = async () => {
const next = await api(`/api/jobs/${job.id}/continue`, { method: 'PUT' });
show(next);
startPoll();
};
const sheetUrl = job.file_urls?.spritesheet_path;
if (sheetUrl) {
sheet.onload = drawAll;
sheet.src = `${sheetUrl}?t=${Date.now()}`;
$('#downloads').classList.remove('hidden');
document.querySelectorAll('#downloads a').forEach((link) => {
const url = job.file_urls?.[link.dataset.file];
link.classList.toggle('hidden', !url);
if (url) link.href = url;
});
} else {
$('#downloads').classList.add('hidden');
sheet = new Image();
clearCanvas();
}
if (actionsVisible) {
renderPromptGrid();
drawAll();
}
if (['completed', 'failed', 'interrupted', 'awaiting_confirmation', 'action_ready'].includes(job.status)) {
clearInterval(poller);
poller = null;
}
}
document.addEventListener('DOMContentLoaded', () => {
$('#file').addEventListener('change', (event) => setFile(event.target.files[0]));
['dragenter', 'dragover'].forEach((eventName) => $('#dropzone').addEventListener(eventName, (event) => { event.preventDefault(); $('#dropzone').classList.add('drag'); }));
['dragleave', 'drop'].forEach((eventName) => $('#dropzone').addEventListener(eventName, (event) => { event.preventDefault(); $('#dropzone').classList.remove('drag'); }));
$('#dropzone').addEventListener('drop', (event) => setFile(event.dataTransfer.files[0]));
$('#generate').addEventListener('click', createJob);
$('#refresh').addEventListener('click', loadJobs);
$('#copyPrompt')?.addEventListener('click', () => navigator.clipboard?.writeText(selectedPrompt()));
document.querySelectorAll('#directionTabs button').forEach((button) => button.addEventListener('click', () => {
document.querySelectorAll('#directionTabs button').forEach((tab) => tab.classList.remove('active'));
button.classList.add('active');
row = Number(button.dataset.row);
drawAll();
}));
health();
loadJobs().catch(() => {});
setInterval(health, 10000);
api('/api/prompt-templates').then((templates) => {
currentPrompts = templates.prompts || [];
const identityPrompt = $('#identityPrompt');
if (identityPrompt) identityPrompt.textContent = templates.identity_prompt || '';
if (currentJobData) { renderPromptGrid(); drawAll(); }
}).catch(() => {});
});

View File

@@ -0,0 +1,450 @@
#!/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()

View File

@@ -0,0 +1,55 @@
<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>WhaleTown 角色工坊</title>
<link rel="icon" href="data:,">
<link rel="stylesheet" href="styles.css?v=20260829-refactor-3">
</head>
<body>
<style>.pose-generate{width:100%;height:30px;border:0;background:#147b78;color:#fff;font-size:11px;cursor:pointer}.pose-generate:disabled{background:#aeb8b5}.prompt-cell small{display:block;color:#b44736;font-size:9px;margin-top:4px}.identity-result{background:#fff;border:1px solid #dce1df;padding:12px;margin-bottom:14px}.identity-result>span{display:block;font-size:12px;font-weight:700;margin-bottom:8px}.identity-result img{display:block;width:min(520px,100%);max-height:420px;object-fit:contain;margin:auto;background:#ff00ff}</style>
<header>
<div class="brand"><span class="mark">W</span><div><strong>WhaleTown</strong><span>角色工坊</span></div></div>
<div id="health" class="health"><i></i><span>检查产线</span></div>
</header>
<main>
<aside>
<section class="creator">
<div class="section-title"><span>新建角色</span><small>8 × 4 动画皮肤</small></div>
<label id="dropzone" class="dropzone">
<input id="file" type="file" accept="image/png,image/jpeg,image/webp">
<img id="sourcePreview" alt="角色参考图预览">
<div id="dropHint"><b></b><strong>放入角色参考图</strong><span>PNG / JPG / WebP最大 8MB</span></div>
</label>
<label class="field"><span>角色名称</span><input id="name" maxlength="40" placeholder="例如:蓝鲸研究员"></label>
<fieldset><legend>生成质量</legend><div class="segments">
<label><input type="radio" name="quality" value="low"><span>草稿</span></label>
<label><input type="radio" name="quality" value="medium" checked><span>标准</span></label>
<label><input type="radio" name="quality" value="high"><span>精细</span></label>
</div></fieldset>
<button id="generate" class="primary" disabled><span>开始生成</span><b></b></button>
<p id="formError" class="error"></p>
</section>
<section class="history"><div class="section-title"><span>制作记录</span><button id="refresh" title="刷新"></button></div><div id="jobList"></div></section>
</aside>
<section class="workspace">
<div id="empty" class="empty"><div class="grid-icon"><i></i><i></i><i></i><i></i></div><h1>任务 1 · 生成身份母版</h1><p>用户图 + 海风少年单格风格参考,只生成一个正面身份角色。</p><div class="identity-inputs"><img src="/references/whaleboy_identity_single.png?v=2" alt="海风少年官方身份参考图"><div><b>本任务真实提示词</b><pre id="identityPrompt">正在加载脚本提示词…</pre></div></div></div>
<div id="jobView" class="job-view hidden">
<div class="job-head"><div><span id="jobStatus" class="status"></span><h1 id="jobName"></h1><p id="jobMessage"></p></div><button id="continueJob" class="primary hidden">确认身份母版,生成动作 →</button><span id="jobTime"></span></div>
<div class="progress"><i id="progressBar"></i></div>
<div id="identityResult" class="identity-result hidden"><span>任务 1 · 身份母版结果</span><img id="identityImage" alt="生成的身份母版"></div>
<div id="actionPreview" class="preview-panel hidden">
<div class="preview-top"><div class="tabs" id="directionTabs"><button data-row="0" class="active">正面</button><button data-row="1">背面</button><button data-row="2">右侧</button><button data-row="3">左侧</button></div><span id="frameLabel">第 1 / 8 帧</span></div>
<div class="stage"><canvas id="canvas" width="320" height="320"></canvas><img id="poseCanvasPreview" class="pose-canvas-preview hidden" alt="当前动作格生成预览"></div>
<div id="frames" class="frames"></div>
</div>
<div id="actionTask" class="prompt-box hidden"><div class="task-heading"><span>任务 2 · 四方向逐格动作</span><span id="poseSummary" class="pose-summary"></span><button id="copyPrompt" title="复制提示词">复制当前格</button></div><div class="preview-note">生成完成后立即显示候选图QA 不通过时也保留最后一次候选,确认无误后再重试。</div><div id="jobPromptGrid" class="prompt-grid"></div></div>
<div id="downloads" class="downloads hidden"><a data-file="spritesheet_path" download>下载皮肤表</a><a data-file="review_path" download>下载验收图</a><a data-file="feet_zoom_path" download>下载足部检查图</a></div>
<details><summary>运行日志</summary><pre id="logs"></pre></details>
</div>
</section>
</main>
<style>.prompt-grid{display:grid;grid-template-columns:repeat(4,1fr);gap:8px;padding:10px;background:#f7f9f8}.prompt-cell{background:#fff;border:1px solid #dce1df;padding:6px;min-width:0}.prompt-cell.active{border:2px solid #147b78}.prompt-cell img{width:100%;height:62px;object-fit:contain;background:#eef1ef}.prompt-cell b{display:block;font-size:10px;margin:4px 0}.prompt-cell p{font-size:9px;line-height:1.35;color:#6b7478;height:58px;overflow:auto;margin:0;white-space:pre-wrap}@media(max-width:760px){.prompt-grid{grid-template-columns:repeat(2,1fr)}}</style><script src="app.js?v=20260829-refactor-3"></script>
</body>
</html>

File diff suppressed because one or more lines are too long