| from __future__ import annotations |
|
|
| import io |
| import json |
| import os |
| import hashlib |
| import shutil |
| import time |
| from contextlib import contextmanager |
| from dataclasses import dataclass |
| from pathlib import Path |
| from typing import Any, Callable |
|
|
| from PIL import Image |
|
|
| from src.character_spike.assets import make_thumbnail_grid, normalize_to_stage_canvas, write_report |
| from src.character_spike.schema import EXPRESSIONS, default_character_package, slugify_identifier |
| from src.character_spike.tavern_import import convert_tavern_card, load_tavern_json |
|
|
|
|
| PROJECT_ROOT = Path(__file__).resolve().parents[1] |
| WORKSHOP_ROOT = PROJECT_ROOT / "assets" / "generated" / "character_workshop" |
| INSTALLED_CHARACTER_ROOT = PROJECT_ROOT / "assets" / "characters" |
| INSTALLED_BACKGROUND_ROOT = PROJECT_ROOT / "assets" / "backgrounds" |
| INSTALLED_CHARACTER_JSON_ROOT = PROJECT_ROOT / "characters" |
| REMBG_MODEL_ROOT = PROJECT_ROOT / ".tmp" / "rembg-models" |
| QWEN_CANDIDATE = "qwen_image" |
|
|
| EXPRESSION_PROMPTS = { |
| "idle": "neutral idle expression, relaxed posture", |
| "listening": "listening expression, attentive eyes, slight head tilt", |
| "thinking": "thinking expression, one hand near chin, focused eyes", |
| "worried": "worried expression, gentle concern, tense shoulders", |
| "smile": "small warm smile, calm and friendly", |
| "happy": "bright happy expression, open eyes, energetic", |
| "talk": "speaking expression, mouth open naturally, conversational pose", |
| "focus": "focused determined expression, stable confident posture", |
| } |
|
|
| REMOTE_PROBE: Callable[..., dict[str, Any]] | None = None |
| _REMBG_SESSION: Any = None |
| LOGIN_REQUIRED_MESSAGE = "请先使用 Hugging Face 登录，登录后会保存你的角色生成进度。" |
|
|
|
|
| @dataclass(frozen=True) |
| class UserContext: |
| username: str |
| display_name: str |
| authenticated: bool |
| user_id_hash: str |
| storage_key: str |
|
|
|
|
| def get_current_user(profile: Any | None) -> UserContext: |
| username = _profile_value(profile, "username") |
| name = _profile_value(profile, "name") or username |
| if username: |
| storage_key = slugify_identifier(username, fallback="hf_user") |
| return UserContext( |
| username=username, |
| display_name=name or username, |
| authenticated=True, |
| user_id_hash=_hash_user_id(username), |
| storage_key=storage_key, |
| ) |
| return UserContext( |
| username="anonymous", |
| display_name="未登录用户", |
| authenticated=False, |
| user_id_hash=_hash_user_id("anonymous"), |
| storage_key="legacy", |
| ) |
|
|
|
|
| def require_login_for_generation(profile: Any | None) -> UserContext: |
| user = get_current_user(profile) |
| if not user.authenticated: |
| raise ValueError(LOGIN_REQUIRED_MESSAGE) |
| return user |
|
|
|
|
| def resolve_data_root() -> Path: |
| configured = os.environ.get("VC_DATA_ROOT") |
| if configured: |
| return Path(configured) |
| data_mount = Path("/data") |
| if data_mount.exists() and os.access(data_mount, os.W_OK): |
| return data_mount / "virtual_characters" |
| return WORKSHOP_ROOT |
|
|
|
|
| def user_workshop_root(user: UserContext) -> Path: |
| return resolve_data_root() / "users" / user.storage_key / "character_workshop" |
|
|
|
|
| def create_draft_from_form( |
| *, |
| display_name: str, |
| description: str, |
| personality: str, |
| scenario: str, |
| first_mes: str, |
| tags: str = "", |
| ) -> dict[str, Any]: |
| name = (display_name or "新角色").strip() |
| character_id = slugify_identifier(name, fallback="workshop_character") |
| package = default_character_package(character_id, name) |
| if description.strip(): |
| package["description"] = description.strip() |
| package["summary"] = _summary(description.strip()) |
| package["profile"]["identity"] = _first_line(description.strip()) |
| if personality.strip(): |
| package["personality"] = personality.strip() |
| package["profile"]["core_traits"] = _split_tags(personality)[:8] or package["profile"]["core_traits"] |
| if scenario.strip(): |
| package["scenario"] = scenario.strip() |
| package["profile"]["relationship_to_user"] = _first_line(scenario.strip()) |
| if first_mes.strip(): |
| package["first_mes"] = first_mes.strip() |
| parsed_tags = _split_tags(tags) |
| if parsed_tags: |
| package["tags"] = parsed_tags |
| package["metadata"]["source"] = "character_workshop_draft" |
| return package |
|
|
|
|
| def create_draft_from_tavern_json(file: Any) -> dict[str, Any]: |
| path = _file_path(file) |
| if not path: |
| raise ValueError("请先上传 Tavern JSON 角色卡。") |
| package = convert_tavern_card(load_tavern_json(path)) |
| package["metadata"]["source"] = "character_workshop_tavern_json" |
| return package |
|
|
|
|
| def create_initial_state( |
| draft: dict[str, Any], |
| *, |
| user: UserContext | None = None, |
| persist: bool = True, |
| ) -> dict[str, Any]: |
| character_id = slugify_identifier(str(draft.get("id") or draft.get("display_name") or "workshop_character")) |
| draft["id"] = character_id |
| draft["visual"]["avatar"] = character_id |
| draft["visual"]["background_image"] = f"{character_id}_background" |
| state = { |
| "draft": draft, |
| "character_id": character_id, |
| "run_dir": None, |
| "main_candidates": [], |
| "selected_candidate_index": None, |
| "expression_assets": {}, |
| "background_asset": None, |
| "manifest_path": None, |
| "installed_character_id": None, |
| } |
| if not persist: |
| return state |
|
|
| run_dir = _unique_run_dir(character_id, user=user) |
| _initialize_run_dir(run_dir, draft, user=user) |
| state["run_dir"] = str(run_dir) |
| state["manifest_path"] = str(run_dir / "manifest.json") |
| return state |
|
|
|
|
| def ensure_user_workshop_run(state: dict[str, Any], user: UserContext) -> dict[str, Any]: |
| state = _ensure_draft_state(state) |
| if state.get("run_dir"): |
| return load_workshop_run(state["run_dir"], user=user) |
| draft = dict(state["draft"]) |
| character_id = slugify_identifier(str(draft.get("id") or draft.get("display_name") or "workshop_character")) |
| run_dir = _unique_run_dir(character_id, user=user) |
| _initialize_run_dir(run_dir, draft, user=user) |
| return load_workshop_run(run_dir, user=user) |
|
|
|
|
| def list_user_workshop_runs(user: UserContext, *, limit: int = 50) -> list[tuple[str, str]]: |
| root = user_workshop_root(user) if user.authenticated else WORKSHOP_ROOT |
| runs = [] |
| for manifest_path in root.glob("*/*/manifest.json"): |
| run_dir = manifest_path.parent |
| try: |
| summary = summarize_workshop_run(run_dir) |
| except Exception: |
| continue |
| runs.append(summary) |
| runs.sort(key=lambda item: item["updated_at_unix"], reverse=True) |
| return [(item["label"], item["run_dir"]) for item in runs[:limit]] |
|
|
|
|
| def summarize_workshop_run(run_dir: str | Path) -> dict[str, Any]: |
| run_dir = Path(run_dir) |
| manifest = _read_manifest(run_dir) |
| state = _state_from_manifest(run_dir, manifest) |
| stage = _workflow_stage(state, manifest) |
| updated = int(manifest.get("updated_at_unix") or manifest.get("created_at_unix") or run_dir.stat().st_mtime) |
| timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(updated)) |
| display_name = str(manifest.get("display_name") or state.get("draft", {}).get("display_name") or state.get("character_id") or "未命名角色") |
| next_step = _next_step_label(stage) |
| return { |
| "label": f"{display_name} / {timestamp} / {stage} / {next_step}", |
| "run_dir": str(run_dir), |
| "stage": stage, |
| "updated_at_unix": updated, |
| "character_id": state.get("character_id"), |
| "display_name": display_name, |
| } |
|
|
|
|
| def load_workshop_run(run_dir: str | Path, *, user: UserContext | None = None) -> dict[str, Any]: |
| run_dir = Path(run_dir).resolve() |
| if user is not None: |
| allowed_root = (user_workshop_root(user) if user.authenticated else WORKSHOP_ROOT).resolve() |
| if not _is_relative_to(run_dir, allowed_root): |
| raise ValueError("不能加载其他用户的角色生成任务。") |
| manifest = _read_manifest(run_dir) |
| state = _state_from_manifest(run_dir, manifest) |
| _sync_manifest_state(run_dir, state, manifest) |
| return state |
|
|
|
|
| def record_workshop_event(user: UserContext, event: str, payload: dict[str, Any] | None = None) -> None: |
| try: |
| payload = dict(payload or {}) |
| data_root = resolve_data_root() |
| stats_dir = data_root / "stats" |
| stats_dir.mkdir(parents=True, exist_ok=True) |
| record = { |
| "time_unix": int(time.time()), |
| "user_id_hash": user.user_id_hash, |
| "authenticated": user.authenticated, |
| "event": event, |
| "stage": payload.get("stage"), |
| "character_id": payload.get("character_id"), |
| "duration_seconds": payload.get("duration_seconds"), |
| "success": payload.get("success"), |
| "failure_reason": payload.get("failure_reason"), |
| "image_count": payload.get("image_count"), |
| "modal_state": payload.get("modal_state"), |
| } |
| with (stats_dir / "events.jsonl").open("a", encoding="utf-8") as handle: |
| handle.write(json.dumps(record, ensure_ascii=False) + "\n") |
| except OSError: |
| return |
|
|
|
|
| def summarize_workshop_stats() -> dict[str, Any]: |
| stats_path = resolve_data_root() / "stats" / "events.jsonl" |
| if not stats_path.exists(): |
| return {"events": 0, "users": 0, "tasks": 0, "success_rate": None, "failures_by_stage": {}} |
| events = [] |
| for line in stats_path.read_text(encoding="utf-8").splitlines(): |
| if not line.strip(): |
| continue |
| try: |
| events.append(json.loads(line)) |
| except json.JSONDecodeError: |
| continue |
| successes = [event for event in events if event.get("success") is True] |
| completed = [event for event in events if event.get("success") is not None] |
| durations = [float(event["duration_seconds"]) for event in events if event.get("duration_seconds") is not None] |
| failures_by_stage: dict[str, int] = {} |
| for event in events: |
| if event.get("success") is False: |
| stage = str(event.get("stage") or "unknown") |
| failures_by_stage[stage] = failures_by_stage.get(stage, 0) + 1 |
| return { |
| "events": len(events), |
| "users": len({event.get("user_id_hash") for event in events if event.get("user_id_hash")}), |
| "tasks": len({event.get("character_id") for event in events if event.get("character_id")}), |
| "success_rate": round(len(successes) / len(completed), 4) if completed else None, |
| "average_duration_seconds": round(sum(durations) / len(durations), 3) if durations else None, |
| "failures_by_stage": failures_by_stage, |
| } |
|
|
|
|
| def generate_main_candidates(state: dict[str, Any], *, seed: int = 42) -> dict[str, Any]: |
| state = _ensure_state(state) |
| draft = state["draft"] |
| run_dir = Path(state["run_dir"]) |
| manifest = _read_manifest(run_dir) |
| existing = _existing_main_candidates(run_dir, manifest) |
| if len(existing) >= 4: |
| state["main_candidates"] = [str(path) for path in existing[:4]] |
| if state.get("selected_candidate_index") is None: |
| state["selected_candidate_index"] = int(manifest.get("selected_candidate_index") or 0) |
| _sync_manifest_state(run_dir, state, manifest) |
| return state |
|
|
| started = time.perf_counter() |
| prompt = _main_visual_prompt(draft) |
| result = _remote_probe( |
| candidate_id=QWEN_CANDIDATE, |
| prompt=prompt, |
| batch_size=4, |
| seed=seed, |
| width=768, |
| height=1024, |
| ) |
| _raise_for_probe_failure(result) |
| paths = _write_image_payloads(run_dir / "main_candidates", result.get("images", []), expected=4) |
| state["main_candidates"] = [str(path) for path in paths] |
| state["selected_candidate_index"] = 0 |
| manifest["main_candidates"] = [str(path.relative_to(run_dir)) for path in paths] |
| manifest["selected_candidate_index"] = 0 |
| manifest["model_results"].append(_probe_record(result, "main_candidates", prompt, started)) |
| manifest["workflow_stage"] = "main_candidates" |
| _touch_manifest(manifest) |
| _write_manifest(manifest, run_dir) |
| return state |
|
|
|
|
| def select_main_candidate(state: dict[str, Any], index: int) -> dict[str, Any]: |
| state = _ensure_state(state) |
| candidates = state.get("main_candidates") or [] |
| if not candidates: |
| raise ValueError("请先生成主视觉候选。") |
| clamped = max(0, min(int(index), len(candidates) - 1)) |
| state["selected_candidate_index"] = clamped |
| run_dir = Path(state["run_dir"]) |
| manifest = _read_manifest(run_dir) |
| manifest["selected_candidate_index"] = clamped |
| _touch_manifest(manifest) |
| _write_manifest(manifest, run_dir) |
| return state |
|
|
|
|
| def generate_expression_pack(state: dict[str, Any], *, seed: int = 100) -> dict[str, Any]: |
| state = _ensure_state(state) |
| draft = state["draft"] |
| run_dir = Path(state["run_dir"]) |
| expression_dir = run_dir / "expressions_raw" |
| expression_dir.mkdir(parents=True, exist_ok=True) |
| manifest = _read_manifest(run_dir) |
| expression_assets: dict[str, str] = _existing_expression_assets(run_dir, manifest) |
|
|
| with _remote_probe_context() as probe: |
| for index, expression in enumerate(EXPRESSIONS): |
| if expression in expression_assets and Path(expression_assets[expression]).exists(): |
| continue |
| started = time.perf_counter() |
| prompt = f"{_main_visual_prompt(draft)}, {EXPRESSION_PROMPTS[expression]}" |
| result = probe( |
| candidate_id=QWEN_CANDIDATE, |
| prompt=prompt, |
| batch_size=1, |
| seed=seed + index, |
| width=768, |
| height=1024, |
| ) |
| _raise_for_probe_failure(result) |
| path = _write_image_payloads(expression_dir / expression, result.get("images", []), expected=1)[0] |
| expression_assets[expression] = str(path) |
| manifest["model_results"].append(_probe_record(result, f"expression_{expression}", prompt, started)) |
| manifest["expression_assets_raw"] = { |
| key: str(Path(value).relative_to(run_dir)) for key, value in expression_assets.items() |
| } |
| manifest["workflow_stage"] = "expressions_complete" if set(EXPRESSIONS) <= set(expression_assets) else "expressions_partial" |
| _touch_manifest(manifest) |
| _write_manifest(manifest, run_dir) |
|
|
| state["expression_assets"] = expression_assets |
| manifest["expression_assets_raw"] = { |
| key: str(Path(value).relative_to(run_dir)) for key, value in expression_assets.items() |
| } |
| manifest["workflow_stage"] = "expressions_complete" if set(EXPRESSIONS) <= set(expression_assets) else "expressions_partial" |
| _touch_manifest(manifest) |
| _write_manifest(manifest, run_dir) |
| return state |
|
|
|
|
| def generate_background(state: dict[str, Any], *, seed: int = 900) -> dict[str, Any]: |
| state = _ensure_state(state) |
| draft = state["draft"] |
| run_dir = Path(state["run_dir"]) |
| manifest = _read_manifest(run_dir) |
| existing = _existing_background(run_dir, manifest) |
| if existing is not None: |
| state["background_asset"] = str(existing) |
| _sync_manifest_state(run_dir, state, manifest) |
| return state |
|
|
| started = time.perf_counter() |
| prompt = _background_prompt(draft) |
| result = _remote_probe( |
| candidate_id=QWEN_CANDIDATE, |
| prompt=prompt, |
| batch_size=1, |
| seed=seed, |
| width=1344, |
| height=768, |
| ) |
| _raise_for_probe_failure(result) |
| path = _write_image_payloads(run_dir / "background_raw", result.get("images", []), expected=1)[0] |
| state["background_asset"] = str(path) |
| manifest["background_raw"] = str(path.relative_to(run_dir)) |
| manifest["model_results"].append(_probe_record(result, "background", prompt, started)) |
| manifest["workflow_stage"] = "assets_ready" if set(EXPRESSIONS) <= set(state.get("expression_assets") or {}) else "background_ready" |
| _touch_manifest(manifest) |
| _write_manifest(manifest, run_dir) |
| return state |
|
|
|
|
| def matte_and_package_assets(state: dict[str, Any], *, mode: str = "rembg_isnet_anime") -> dict[str, Any]: |
| state = _ensure_state(state) |
| run_dir = Path(state["run_dir"]) |
| character_id = state["character_id"] |
| draft = state["draft"] |
| package_root = run_dir / "package" |
| manifest = _read_manifest(run_dir) |
| if _package_complete(package_root, character_id): |
| state["package_dir"] = str(package_root) |
| state["manifest_path"] = str(run_dir / "manifest.json") |
| manifest["workflow_stage"] = "packaged" |
| manifest["ready_for_chat"] = True |
| _touch_manifest(manifest) |
| _write_manifest(manifest, run_dir) |
| return state |
|
|
| character_dir = package_root / "assets" / "characters" / character_id |
| background_dir = package_root / "assets" / "backgrounds" |
| generated_dir = package_root / "generated" |
| character_dir.mkdir(parents=True, exist_ok=True) |
| background_dir.mkdir(parents=True, exist_ok=True) |
| generated_dir.mkdir(parents=True, exist_ok=True) |
|
|
| expression_assets = state.get("expression_assets") or {} |
| if set(EXPRESSIONS) - set(expression_assets): |
| raise ValueError("请先生成完整 8 表情资产。") |
|
|
| packaged_assets = [] |
| for expression in EXPRESSIONS: |
| source = Path(expression_assets[expression]) |
| image = Image.open(source).convert("RGBA") |
| processed, matting_status = _matte_image(image, mode=mode) |
| normalized = normalize_to_stage_canvas(processed) |
| target = character_dir / f"{expression}.png" |
| normalized.save(target) |
| packaged_assets.append( |
| { |
| "slot": expression, |
| "path": str(target.relative_to(package_root)), |
| "source_path": str(source.relative_to(run_dir)), |
| "matting_status": matting_status, |
| "bytes": target.stat().st_size, |
| "usable": True, |
| } |
| ) |
|
|
| background_source = Path(state["background_asset"]) if state.get("background_asset") else None |
| background_target = background_dir / f"{character_id}_background.png" |
| if background_source and background_source.exists(): |
| _fit_background(Image.open(background_source).convert("RGB")).save(background_target) |
| else: |
| _fallback_background(str(draft.get("display_name") or draft.get("name") or character_id)).save(background_target) |
|
|
| package = dict(draft) |
| package["metadata"] = {**package.get("metadata", {}), "source": "character_workshop", "generation_manifest": str(run_dir / "manifest.json")} |
| package["visual"] = {**package.get("visual", {}), "avatar": character_id, "background_image": f"{character_id}_background"} |
| character_json = package_root / "characters" / f"{character_id}.json" |
| character_json.parent.mkdir(parents=True, exist_ok=True) |
| _write_json(character_json, package) |
|
|
| grid = generated_dir / "asset_grid.png" |
| make_thumbnail_grid([character_dir / f"{slot}.png" for slot in EXPRESSIONS], grid) |
| smoke = generated_dir / "stage_smoke.html" |
| _write_stage_smoke(package, package_root, smoke) |
|
|
| manifest["packaged_assets"] = packaged_assets |
| manifest["background_asset"] = str(background_target.relative_to(package_root)) |
| manifest["character_package"] = str(character_json.relative_to(package_root)) |
| manifest["package_run_dir"] = str(package_root) |
| manifest["asset_grid"] = str(grid.relative_to(package_root)) |
| manifest["stage_smoke"] = str(smoke.relative_to(package_root)) |
| manifest["ready_for_chat"] = True |
| manifest["workflow_stage"] = "packaged" |
| _touch_manifest(manifest) |
| _write_manifest(manifest, run_dir) |
| write_report(manifest, package_root / "generated" / "report.md") |
| state["package_dir"] = str(package_root) |
| state["manifest_path"] = str(run_dir / "manifest.json") |
| return state |
|
|
|
|
| def install_character_package(state: dict[str, Any]) -> dict[str, Any]: |
| state = _ensure_state(state) |
| package_dir = Path(state.get("package_dir") or Path(state["run_dir"]) / "package") |
| character_id = state["character_id"] |
| source_character_dir = package_dir / "assets" / "characters" / character_id |
| source_background = package_dir / "assets" / "backgrounds" / f"{character_id}_background.png" |
| source_json = package_dir / "characters" / f"{character_id}.json" |
| if not source_character_dir.exists() or not source_json.exists(): |
| raise ValueError("请先完成打包预览。") |
|
|
| target_character_dir = INSTALLED_CHARACTER_ROOT / character_id |
| target_background = INSTALLED_BACKGROUND_ROOT / f"{character_id}_background.png" |
| target_json = INSTALLED_CHARACTER_JSON_ROOT / f"{character_id}.json" |
| shutil.copytree(source_character_dir, target_character_dir, dirs_exist_ok=True) |
| target_background.parent.mkdir(parents=True, exist_ok=True) |
| if source_background.exists(): |
| shutil.copy2(source_background, target_background) |
| target_json.parent.mkdir(parents=True, exist_ok=True) |
| shutil.copy2(source_json, target_json) |
|
|
| manifest = _read_manifest(Path(state["run_dir"])) |
| manifest["install_paths"] = { |
| "character_assets": str(target_character_dir), |
| "background": str(target_background), |
| "character_json": str(target_json), |
| } |
| manifest["installed_at_unix"] = int(time.time()) |
| manifest["workflow_stage"] = "installed" |
| _touch_manifest(manifest) |
| _write_manifest(manifest, Path(state["run_dir"])) |
| _clear_stage_caches() |
| state["installed_character_id"] = character_id |
| return state |
|
|
|
|
| def render_packaged_stage_preview(state: dict[str, Any]) -> str: |
| state = _ensure_state(state) |
| package_dir = Path(state.get("package_dir") or Path(state["run_dir"]) / "package") |
| character_path = package_dir / "characters" / f"{state['character_id']}.json" |
| if not character_path.exists(): |
| return "" |
| character = json.loads(character_path.read_text(encoding="utf-8")) |
| from src import stage_driver |
|
|
| old_asset_root = stage_driver.ASSET_ROOT |
| old_background_root = stage_driver.BACKGROUND_ROOT |
| try: |
| stage_driver.ASSET_ROOT = package_dir / "assets" / "characters" |
| stage_driver.BACKGROUND_ROOT = package_dir / "assets" / "backgrounds" |
| stage_driver._asset_data_uri.cache_clear() |
| stage_driver._background_data_uri.cache_clear() |
| return stage_driver.render_character_stage( |
| character, |
| {"expression": "smile", "motion": "talk", "intensity": 0.72}, |
| ) |
| finally: |
| stage_driver.ASSET_ROOT = old_asset_root |
| stage_driver.BACKGROUND_ROOT = old_background_root |
| stage_driver._asset_data_uri.cache_clear() |
| stage_driver._background_data_uri.cache_clear() |
|
|
|
|
| def set_remote_probe_for_tests(fn: Callable[..., dict[str, Any]] | None) -> None: |
| global REMOTE_PROBE |
| REMOTE_PROBE = fn |
|
|
|
|
| def _remote_probe(**kwargs: Any) -> dict[str, Any]: |
| with _remote_probe_context() as probe: |
| return probe(**kwargs) |
|
|
|
|
| @contextmanager |
| def _remote_probe_context(): |
| if REMOTE_PROBE is not None: |
| yield REMOTE_PROBE |
| return |
| from modal_apps.modal_character_spike import CharacterGenerationSpike, app |
|
|
| with app.run(): |
| yield CharacterGenerationSpike().probe.remote |
|
|
|
|
| def _write_image_payloads(output_dir: Path, images: list[bytes], *, expected: int) -> list[Path]: |
| if len(images) < expected: |
| raise ValueError(f"模型返回 {len(images)} 张图，期望 {expected} 张。") |
| output_dir.mkdir(parents=True, exist_ok=True) |
| paths = [] |
| for index, payload in enumerate(images[:expected]): |
| path = output_dir / f"{index:02d}.png" |
| path.write_bytes(payload) |
| paths.append(path) |
| return paths |
|
|
|
|
| def _matte_image(image: Image.Image, *, mode: str) -> tuple[Image.Image, str]: |
| if mode != "rembg_isnet_anime": |
| return image.convert("RGBA"), "fallback_keep_background" |
| try: |
| REMBG_MODEL_ROOT.mkdir(parents=True, exist_ok=True) |
| os.environ.setdefault("U2NET_HOME", str(REMBG_MODEL_ROOT)) |
| from rembg import new_session, remove |
|
|
| global _REMBG_SESSION |
| if _REMBG_SESSION is None: |
| _REMBG_SESSION = new_session("isnet-anime") |
| buffer = io.BytesIO() |
| image.convert("RGB").save(buffer, format="PNG") |
| output = remove(buffer.getvalue(), session=_REMBG_SESSION) |
| matte = Image.open(io.BytesIO(output)).convert("RGBA") |
| if not _valid_alpha(matte): |
| return image.convert("RGBA"), "fallback_keep_background" |
| return matte, "rembg_isnet_anime" |
| except Exception: |
| return image.convert("RGBA"), "fallback_keep_background" |
|
|
|
|
| def _valid_alpha(image: Image.Image) -> bool: |
| alpha = image.getchannel("A") |
| bbox = alpha.getbbox() |
| if bbox is None: |
| return False |
| non_zero = sum(1 for value in alpha.getdata() if value > 8) |
| ratio = non_zero / (image.width * image.height) |
| return 0.05 <= ratio <= 0.92 |
|
|
|
|
| def _main_visual_prompt(draft: dict[str, Any]) -> str: |
| name = str(draft.get("display_name") or draft.get("name") or "original character") |
| description = " ".join(str(draft.get(key) or "").strip() for key in ("description", "personality", "scenario")) |
| return ( |
| f"single original anime-style virtual character portrait for {name}, {description}, " |
| "young adult original character, half body character sprite, centered composition, " |
| "plain light background, clean silhouette, no text, no logo, no poster, not a landscape, " |
| "not an existing commercial character" |
| ) |
|
|
|
|
| def _background_prompt(draft: dict[str, Any]) -> str: |
| scenario = str(draft.get("scenario") or "quiet virtual communication room") |
| return ( |
| f"empty background for a virtual character conversation stage, {scenario}, " |
| "wide landscape image, no people, no character, no text, no logo, clean atmospheric sci-fi interior" |
| ) |
|
|
|
|
| def _new_manifest(draft: dict[str, Any], run_dir: Path, *, user: UserContext | None = None) -> dict[str, Any]: |
| now = int(time.time()) |
| owner = { |
| "username": user.username if user and user.authenticated else None, |
| "authenticated": bool(user and user.authenticated), |
| } |
| return { |
| "schema_version": 1, |
| "run_type": "character_workshop", |
| "character_id": draft["id"], |
| "display_name": draft.get("display_name") or draft.get("name"), |
| "created_at_unix": now, |
| "updated_at_unix": now, |
| "run_dir": str(run_dir), |
| "draft_character": "draft_character.json", |
| "model": QWEN_CANDIDATE, |
| "model_results": [], |
| "owner": owner, |
| "workflow_stage": "draft", |
| "stats_run_id": _hash_user_id(f"{run_dir}:{now}")[:16], |
| "ready_for_chat": False, |
| } |
|
|
|
|
| def _probe_record(result: dict[str, Any], case: str, prompt: str, started: float) -> dict[str, Any]: |
| return { |
| "candidate_id": result.get("candidate_id"), |
| "benchmark_case": case, |
| "status": result.get("status"), |
| "image_count": result.get("image_count"), |
| "duration_seconds": result.get("duration_seconds"), |
| "client_wall_seconds": round(time.perf_counter() - started, 3), |
| "prompt": prompt, |
| "seed": result.get("seed"), |
| "failure_reason": result.get("failure_reason"), |
| } |
|
|
|
|
| def _raise_for_probe_failure(result: dict[str, Any]) -> None: |
| if result.get("status") != "ok": |
| reason = result.get("failure_reason") or "未知错误" |
| raise RuntimeError(f"图像生成失败：{reason}") |
|
|
|
|
| def _ensure_state(state: dict[str, Any] | None) -> dict[str, Any]: |
| if not state or not state.get("draft") or not state.get("run_dir"): |
| raise ValueError("请先创建角色草案。") |
| return dict(state) |
|
|
|
|
| def _ensure_draft_state(state: dict[str, Any] | None) -> dict[str, Any]: |
| if not state or not state.get("draft"): |
| raise ValueError("请先创建角色草案。") |
| return dict(state) |
|
|
|
|
| def _unique_run_dir(character_id: str, *, user: UserContext | None = None) -> Path: |
| root = user_workshop_root(user) if user and user.authenticated else WORKSHOP_ROOT |
| return root / character_id / time.strftime("%Y%m%d_%H%M%S") |
|
|
|
|
| def _initialize_run_dir(run_dir: Path, draft: dict[str, Any], *, user: UserContext | None = None) -> None: |
| run_dir.mkdir(parents=True, exist_ok=True) |
| _write_json(run_dir / "draft_character.json", draft) |
| manifest = _new_manifest(draft, run_dir, user=user) |
| _write_manifest(manifest, run_dir) |
|
|
|
|
| def _read_manifest(run_dir: Path) -> dict[str, Any]: |
| return json.loads((run_dir / "manifest.json").read_text(encoding="utf-8")) |
|
|
|
|
| def _write_manifest(manifest: dict[str, Any], run_dir: Path) -> None: |
| _write_json(run_dir / "manifest.json", manifest) |
|
|
|
|
| def _touch_manifest(manifest: dict[str, Any]) -> None: |
| manifest["updated_at_unix"] = int(time.time()) |
|
|
|
|
| def _write_json(path: Path, payload: dict[str, Any]) -> None: |
| path.parent.mkdir(parents=True, exist_ok=True) |
| path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") |
|
|
|
|
| def _state_from_manifest(run_dir: Path, manifest: dict[str, Any]) -> dict[str, Any]: |
| draft_path = run_dir / str(manifest.get("draft_character") or "draft_character.json") |
| draft = json.loads(draft_path.read_text(encoding="utf-8")) if draft_path.exists() else default_character_package( |
| str(manifest.get("character_id") or "workshop_character"), |
| str(manifest.get("display_name") or "工坊角色"), |
| ) |
| character_id = slugify_identifier(str(draft.get("id") or manifest.get("character_id") or "workshop_character")) |
| selected_index = int(manifest.get("selected_candidate_index") or 0) |
| main_candidates = _existing_main_candidates(run_dir, manifest) |
| expression_assets = _existing_expression_assets(run_dir, manifest) |
| background = _existing_background(run_dir, manifest) |
| package_dir = Path(str(manifest.get("package_run_dir") or run_dir / "package")) |
| state = { |
| "draft": draft, |
| "character_id": character_id, |
| "run_dir": str(run_dir), |
| "main_candidates": [str(path) for path in main_candidates], |
| "selected_candidate_index": selected_index if main_candidates else None, |
| "expression_assets": expression_assets, |
| "background_asset": str(background) if background else None, |
| "manifest_path": str(run_dir / "manifest.json"), |
| "installed_character_id": character_id if manifest.get("installed_at_unix") else None, |
| } |
| if package_dir.exists(): |
| state["package_dir"] = str(package_dir) |
| return state |
|
|
|
|
| def _sync_manifest_state(run_dir: Path, state: dict[str, Any], manifest: dict[str, Any]) -> None: |
| main_candidates = [Path(path) for path in state.get("main_candidates") or [] if Path(path).exists()] |
| if main_candidates: |
| manifest["main_candidates"] = [str(path.relative_to(run_dir)) for path in main_candidates] |
| manifest["selected_candidate_index"] = int(state.get("selected_candidate_index") or 0) |
| expression_assets = {key: value for key, value in (state.get("expression_assets") or {}).items() if Path(value).exists()} |
| if expression_assets: |
| manifest["expression_assets_raw"] = { |
| key: str(Path(value).relative_to(run_dir)) for key, value in expression_assets.items() |
| } |
| if state.get("background_asset") and Path(state["background_asset"]).exists(): |
| manifest["background_raw"] = str(Path(state["background_asset"]).relative_to(run_dir)) |
| manifest["workflow_stage"] = _workflow_stage(state, manifest) |
| _touch_manifest(manifest) |
| _write_manifest(manifest, run_dir) |
|
|
|
|
| def _existing_main_candidates(run_dir: Path, manifest: dict[str, Any]) -> list[Path]: |
| paths = [] |
| for value in manifest.get("main_candidates") or []: |
| path = _resolve_run_path(run_dir, value) |
| if path.exists(): |
| paths.append(path) |
| if paths: |
| return paths |
| return sorted((run_dir / "main_candidates").glob("*.png")) |
|
|
|
|
| def _existing_expression_assets(run_dir: Path, manifest: dict[str, Any]) -> dict[str, str]: |
| assets = {} |
| raw = manifest.get("expression_assets_raw") or {} |
| for expression in EXPRESSIONS: |
| value = raw.get(expression) |
| path = _resolve_run_path(run_dir, value) if value else run_dir / "expressions_raw" / expression / "00.png" |
| if path.exists(): |
| assets[expression] = str(path) |
| return assets |
|
|
|
|
| def _existing_background(run_dir: Path, manifest: dict[str, Any]) -> Path | None: |
| value = manifest.get("background_raw") |
| path = _resolve_run_path(run_dir, value) if value else run_dir / "background_raw" / "00.png" |
| return path if path.exists() else None |
|
|
|
|
| def _package_complete(package_root: Path, character_id: str) -> bool: |
| character_dir = package_root / "assets" / "characters" / character_id |
| background = package_root / "assets" / "backgrounds" / f"{character_id}_background.png" |
| character_json = package_root / "characters" / f"{character_id}.json" |
| return ( |
| character_json.exists() |
| and background.exists() |
| and all((character_dir / f"{expression}.png").exists() for expression in EXPRESSIONS) |
| ) |
|
|
|
|
| def _workflow_stage(state: dict[str, Any], manifest: dict[str, Any]) -> str: |
| if manifest.get("installed_at_unix"): |
| return "installed" |
| character_id = str(state.get("character_id") or manifest.get("character_id") or "") |
| package_root = Path(str(state.get("package_dir") or manifest.get("package_run_dir") or Path(state["run_dir"]) / "package")) |
| if character_id and _package_complete(package_root, character_id): |
| return "packaged" |
| expressions = state.get("expression_assets") or {} |
| background = state.get("background_asset") |
| if set(EXPRESSIONS) <= set(expressions) and background and Path(background).exists(): |
| return "assets_ready" |
| if expressions: |
| return "expressions_partial" |
| if len(state.get("main_candidates") or []) >= 4: |
| return "main_candidates" |
| return "draft" |
|
|
|
|
| def _next_step_label(stage: str) -> str: |
| return { |
| "draft": "生成主视觉", |
| "main_candidates": "生成8表情", |
| "expressions_partial": "补齐表情", |
| "assets_ready": "打包预览", |
| "packaged": "安装角色", |
| "installed": "已安装", |
| }.get(stage, "继续") |
|
|
|
|
| def _resolve_run_path(run_dir: Path, value: Any) -> Path: |
| if not value: |
| return run_dir / "__missing__" |
| path = Path(str(value)) |
| return path if path.is_absolute() else run_dir / path |
|
|
|
|
| def _is_relative_to(path: Path, root: Path) -> bool: |
| try: |
| path.relative_to(root) |
| return True |
| except ValueError: |
| return False |
|
|
|
|
| def _profile_value(profile: Any, key: str) -> str: |
| if profile is None: |
| return "" |
| if isinstance(profile, dict): |
| return str(profile.get(key) or "").strip() |
| return str(getattr(profile, key, "") or "").strip() |
|
|
|
|
| def _hash_user_id(value: str) -> str: |
| return hashlib.sha256(str(value).encode("utf-8")).hexdigest() |
|
|
|
|
| def _file_path(file: Any) -> Path | None: |
| if file is None: |
| return None |
| if isinstance(file, (str, Path)): |
| return Path(file) |
| if isinstance(file, dict): |
| value = file.get("path") or file.get("name") |
| return Path(value) if value else None |
| value = getattr(file, "path", None) or getattr(file, "name", None) |
| return Path(value) if value else None |
|
|
|
|
| def _split_tags(value: str) -> list[str]: |
| text = str(value or "") |
| for separator in ("，", "、", ";", "；", "\n"): |
| text = text.replace(separator, ",") |
| return [item.strip() for item in text.split(",") if item.strip()] |
|
|
|
|
| def _summary(text: str, limit: int = 120) -> str: |
| compact = " ".join(text.split()) |
| return compact if len(compact) <= limit else compact[: limit - 1].rstrip() + "…" |
|
|
|
|
| def _first_line(text: str) -> str: |
| for line in text.splitlines(): |
| if line.strip(): |
| return line.strip() |
| return text.strip() |
|
|
|
|
| def _fit_background(image: Image.Image) -> Image.Image: |
| image = image.convert("RGB") |
| target = (1600, 900) |
| image.thumbnail(target, Image.Resampling.LANCZOS) |
| canvas = Image.new("RGB", target, (12, 18, 31)) |
| canvas.paste(image, ((target[0] - image.width) // 2, (target[1] - image.height) // 2)) |
| return canvas |
|
|
|
|
| def _fallback_background(display_name: str) -> Image.Image: |
| from src.character_spike.assets import _draw_mock_background |
|
|
| return _draw_mock_background(display_name, __import__("random").Random(42)) |
|
|
|
|
| def _write_stage_smoke(character: dict[str, Any], package_root: Path, output: Path) -> None: |
| from src import stage_driver |
|
|
| old_asset_root = stage_driver.ASSET_ROOT |
| old_background_root = stage_driver.BACKGROUND_ROOT |
| try: |
| stage_driver.ASSET_ROOT = package_root / "assets" / "characters" |
| stage_driver.BACKGROUND_ROOT = package_root / "assets" / "backgrounds" |
| stage_driver._asset_data_uri.cache_clear() |
| stage_driver._background_data_uri.cache_clear() |
| states = [ |
| {"expression": "idle", "motion": "breathe", "intensity": 0.35}, |
| {"expression": "listening", "motion": "gentle_blink", "intensity": 0.35}, |
| {"expression": "smile", "motion": "talk", "intensity": 0.72}, |
| {"expression": "thinking", "motion": "focus", "intensity": 0.7}, |
| ] |
| html = "<!doctype html><meta charset='utf-8'><body style='margin:0;background:#020617'>" |
| html += "\n".join(stage_driver.render_character_stage(character, state) for state in states) |
| html += "</body>" |
| output.parent.mkdir(parents=True, exist_ok=True) |
| output.write_text(html, encoding="utf-8") |
| finally: |
| stage_driver.ASSET_ROOT = old_asset_root |
| stage_driver.BACKGROUND_ROOT = old_background_root |
| stage_driver._asset_data_uri.cache_clear() |
| stage_driver._background_data_uri.cache_clear() |
|
|
|
|
| def _clear_stage_caches() -> None: |
| try: |
| from src import stage_driver |
|
|
| stage_driver._asset_data_uri.cache_clear() |
| stage_driver._background_data_uri.cache_clear() |
| except Exception: |
| pass |
|
|
