games-creater/server.py

"""Anim Studio 网站后端（仅用 Python 标准库，无 web 框架依赖）。

启动:
    pip install Pillow        # 唯一必需第三方库
    python server.py
浏览器打开 http://127.0.0.1:7861

路由:
  GET  /                      可视化网站
  GET  /api/manifest          默认 manifest
  GET  /api/games             已生成的 game 列表
  GET  /api/library?game=..   某 game 的资源/动画库
  GET  /assets/<game>/<path>  资源文件
  POST /api/generate          运行生成管线
  POST /api/export            把某 game 打包成 Cocos 整合包
  POST /api/delete            删除某 game 的资源库
"""

import json
import mimetypes
import os
import posixpath
import shutil
import threading
import time
import traceback
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs, unquote

import pipeline
import exporter
import slot_workflow
import providers
import config

HERE = os.path.dirname(os.path.abspath(__file__))
OUT_ROOT = os.path.join(HERE, "out")
WEB_DIR = os.path.join(HERE, "web")
DEFAULT_MANIFEST = os.path.join(HERE, "animation_manifest.json")
PORT = 7861
DEFAULT_BASE_URL = config.get("ANIM_STUDIO_BASE_URL", "https://x.long.bid/v1")
DEFAULT_API_KEY = config.get("ANIM_STUDIO_API_KEY", "")
DEFAULT_IMAGE_MODEL = config.get("ANIM_STUDIO_IMAGE_MODEL", "gpt-image-2")
DEFAULT_TEXT_MODEL = config.get("ANIM_STUDIO_TEXT_MODEL", "gpt-5.4-mini")
JOBS = {}
JOBS_LOCK = threading.Lock()


def list_games():
    if not os.path.isdir(OUT_ROOT):
        return []
    return [n for n in sorted(os.listdir(OUT_ROOT))
            if os.path.isfile(os.path.join(OUT_ROOT, n, "library.json"))]


def safe_join(root, *parts):
    p = posixpath.normpath("/".join(parts)).lstrip("/")
    full = os.path.join(root, *p.split("/"))
    if not os.path.abspath(full).startswith(os.path.abspath(root)):
        raise ValueError("path traversal")
    return full


def _job_snapshot(job_id):
    with JOBS_LOCK:
        job = JOBS.get(job_id)
        return dict(job) if job else None


def _set_job(job_id, **fields):
    with JOBS_LOCK:
        job = JOBS.setdefault(job_id, {})
        job.update(fields)


def _append_job_log(job_id, msg):
    with JOBS_LOCK:
        job = JOBS.setdefault(job_id, {})
        job.setdefault("logs", []).append(msg)
        job["updatedAt"] = time.time()


def _run_generate_job(job_id, manifest, creds):
    _set_job(job_id, status="running", logs=[], game=manifest.get("game"), startedAt=time.time(), updatedAt=time.time())
    try:
        lib, _ = pipeline.run(manifest, OUT_ROOT, creds=creds, log=lambda m: _append_job_log(job_id, m))
        _set_job(job_id, status="done", ok=True, game=lib["game"], updatedAt=time.time())
    except Exception as e:
        traceback.print_exc()
        _append_job_log(job_id, f"❌ 生成任务失败: {e}")
        _set_job(job_id, status="error", ok=False, error=str(e), updatedAt=time.time())


def _split_lines(value):
    if isinstance(value, list):
        return [str(x).strip() for x in value if str(x).strip()]
    return [x.strip() for x in str(value or "").splitlines() if x.strip()]


def _creative_fallback(data):
    brief = (data.get("brief") or "").lower()
    text = " ".join([brief, data.get("styleNotes", "").lower(), data.get("avoidNotes", "").lower()])
    allowed_themes = {"jelly", "fruit", "egypt", "pirate", "pirate_jelly", "cyber"}
    requested_theme = data.get("theme") if data.get("theme") in allowed_themes else ""
    theme = requested_theme or "jelly"
    pirate_hint = any(k in text for k in ("pirate", "海盗", "treasure", "宝藏", "金币", "船长"))
    jelly_hint = any(k in text for k in ("jelly", "果冻", "candy", "糖果", "gummy", "软糖"))
    if not requested_theme or requested_theme == "jelly":
        if pirate_hint and jelly_hint:
            theme = "pirate_jelly"
        elif any(k in text for k in ("egypt", "埃及", "金字塔", "法老")):
            theme = "egypt"
        elif pirate_hint:
            theme = "pirate"
        elif any(k in text for k in ("cyber", "赛博", "neon", "霓虹")):
            theme = "cyber"
        elif any(k in text for k in ("fruit", "水果", "cherry", "樱桃")):
            theme = "fruit"
    reel_mode = data.get("reelMode") or ("cluster" if any(k in text for k in ("消除", "cluster", "连通", "match")) else "ways")
    volatility = data.get("volatility") or ("high" if any(k in text for k in ("刺激", "大奖", "高波动", "big win")) else "medium")
    features = list(data.get("features") or ["cascades", "free_spins", "wilds"])
    if any(k in text for k in ("金币", "jackpot", "hold", "respin", "大奖池")) and "hold_win" not in features:
        features.append("hold_win")
    if any(k in text for k in ("倍率", "multiplier", "连锁")) and "multipliers" not in features:
        features.append("multipliers")
    if theme == "pirate_jelly":
        for feature in ("hold_win", "multipliers"):
            if feature not in features:
                features.append(feature)
    title = data.get("title") or "AI Custom Slot"
    style = data.get("styleNotes") or data.get("brief") or ""
    return {
        "gameId": data.get("gameId") or title,
        "title": title,
        "theme": theme,
        "style": style,
        "reelMode": reel_mode,
        "volatility": volatility,
        "targetRtp": data.get("targetRtp", 96),
        "characterCount": data.get("characterCount", 10),
        "uiCompleteness": data.get("uiCompleteness", "full"),
        "feedbackIntensity": data.get("feedbackIntensity", "standard"),
        "enableMathModel": bool(data.get("enableMathModel", True)),
        "features": features,
    }


def build_game_plan(slot_request, creative_payload, source):
    features = slot_request.get("features") or []
    hook_parts = []
    if "hold_win" in features:
        hook_parts.append("金币锁格 respin 大奖目标")
    if "cascades" in features:
        hook_parts.append("连锁下落爽感")
    if "free_spins" in features:
        hook_parts.append("Scatter 免费旋转期待")
    if "multipliers" in features:
        hook_parts.append("递增倍率爆点")
    core_hook = " + ".join(hook_parts[:2]) or "轻量现代老虎机循环"
    vision = creative_payload.get("visionStyleAnalysis") or {}
    art_direction = {
        "theme": slot_request.get("theme"),
        "style": slot_request.get("style", ""),
        "reference_style_analysis": vision,
        "avoid": creative_payload.get("avoidNotes", ""),
    }
    return {
        "source": source,
        "creative": {
            "brief": creative_payload.get("brief", ""),
            "references": creative_payload.get("references", []),
            "uploadedReferenceImages": creative_payload.get("uploadedReferenceImages", 0),
            "visionStyleAnalysis": vision,
            "visionStyleAnalysisError": creative_payload.get("visionStyleAnalysisError", ""),
        },
        "gameDesign": {
            "title": slot_request.get("title"),
            "coreHook": core_hook,
            "artDirection": art_direction,
            "reelExperience": slot_request.get("reelMode"),
            "volatility": slot_request.get("volatility"),
            "differentiators": hook_parts,
            "assetStrategy": {
                "symbolCount": slot_request.get("characterCount"),
                "uiCompleteness": slot_request.get("uiCompleteness"),
                "feedbackIntensity": slot_request.get("feedbackIntensity"),
            },
        },
    }


def creative_to_slot_request(data):
    api_key = (data.get("api_key") or DEFAULT_API_KEY).strip()
    base_url = (data.get("base_url") or DEFAULT_BASE_URL).strip()
    text_model = (data.get("text_model") or DEFAULT_TEXT_MODEL).strip()
    references = _split_lines(data.get("references"))
    reference_images = data.get("reference_images") or []
    if not isinstance(reference_images, list):
        reference_images = []
    payload = {
        "title": data.get("title") or "AI Custom Slot",
        "gameId": data.get("gameId") or "",
        "brief": data.get("brief") or "",
        "references": references,
        "uploadedReferenceImages": len(reference_images),
        "styleNotes": data.get("styleNotes") or "",
        "avoidNotes": data.get("avoidNotes") or "",
        "theme": data.get("theme") or "",
        "reelMode": data.get("reelMode") or "",
        "volatility": data.get("volatility") or "",
        "targetRtp": data.get("targetRtp", 96),
        "characterCount": data.get("characterCount", 10),
        "uiCompleteness": data.get("uiCompleteness", "full"),
        "feedbackIntensity": data.get("feedbackIntensity", "standard"),
        "features": data.get("features") or [],
        "enableMathModel": bool(data.get("enableMathModel", True)),
    }
    if not api_key:
        fallback = _creative_fallback(payload)
        return fallback, "fallback_no_api_key", build_game_plan(fallback, payload, "fallback_no_api_key")
    vision_notes = {}
    direct_image_urls = [
        x for x in references
        if x.lower().startswith(("http://", "https://")) and x.lower().split("?")[0].endswith((".png", ".jpg", ".jpeg", ".webp"))
    ]
    if direct_image_urls or reference_images:
        try:
            vision_notes = providers.analyze_reference_images(
                reference_urls=direct_image_urls,
                image_data_urls=reference_images,
                api_key=api_key,
                base_url=base_url,
                model=text_model,
            )
            payload["visionStyleAnalysis"] = vision_notes
            if vision_notes.get("image_prompt_style"):
                payload["styleNotes"] = "；".join([
                    str(payload.get("styleNotes") or ""),
                    "视觉参考分析：" + str(vision_notes.get("image_prompt_style")),
                ]).strip("；")
            elif vision_notes:
                payload["styleNotes"] = "；".join([
                    str(payload.get("styleNotes") or ""),
                    "视觉参考分析：" + json.dumps(vision_notes, ensure_ascii=False),
                ]).strip("；")
        except Exception as e:
            payload["visionStyleAnalysisError"] = str(e)[:500]
    messages = [
        {
            "role": "system",
            "content": (
                "你是资深移动老虎机游戏策划和美术总监。根据用户的基础描述、参考图或网址、风格要求，"
                "生成一份可执行的 slot 工作流输入 JSON。必须原创，不能复刻参考图里的具体 IP、logo、角色。"
                "只输出 JSON，不要解释。"
            ),
        },
        {
            "role": "user",
            "content": json.dumps({
                "task": "把创意简报转换为 slot_workflow.build_workflow 可接受的输入",
                "allowedFields": {
                    "gameId": "string slug or title",
                    "title": "string",
                    "theme": "jelly|fruit|egypt|pirate|pirate_jelly|cyber",
                    "style": "English image-generation style prompt, include reference-derived art direction",
                    "reelMode": "ways|paylines|megaways|cluster",
                    "volatility": "low|medium|high",
                    "targetRtp": "number percent, e.g. 96",
                    "characterCount": "6|8|10|12",
                    "uiCompleteness": "basic|full",
                    "feedbackIntensity": "quiet|standard|loud",
                    "enableMathModel": "boolean",
                    "features": "array of cascades, free_spins, wilds, hold_win, multipliers"
                },
                "creativeBrief": payload,
                "outputRules": [
                    "Return JSON object with exactly these workflow fields.",
                    "Pick one clear core hook, not every feature.",
                    "Use references as style inspiration only; do not copy protected characters, logos, or exact layout.",
                    "Style must be specific enough for image generation."
                ],
            }, ensure_ascii=False),
        },
    ]
    try:
        obj = providers.chat_json_openai(messages, api_key=api_key, base_url=base_url, model=text_model)
    except Exception:
        fallback = _creative_fallback(payload)
        return fallback, "fallback_text_model_failed", build_game_plan(fallback, payload, "fallback_text_model_failed")
    fallback = _creative_fallback(payload)
    out = dict(fallback)
    for key in ("gameId", "title", "theme", "style", "reelMode", "volatility", "targetRtp",
                "characterCount", "uiCompleteness", "feedbackIntensity", "enableMathModel"):
        if key in obj and obj[key] not in (None, ""):
            out[key] = obj[key]
    if isinstance(obj.get("features"), list) and obj["features"]:
        allowed = {"cascades", "free_spins", "wilds", "hold_win", "multipliers"}
        out["features"] = [x for x in obj["features"] if x in allowed]
    source = "vision_text_model" if vision_notes else "text_model"
    return out, source, build_game_plan(out, payload, source)


class Handler(BaseHTTPRequestHandler):
    def _send(self, code, body, ctype="application/json; charset=utf-8"):
        if isinstance(body, (dict, list)):
            body = json.dumps(body, ensure_ascii=False).encode("utf-8")
        elif isinstance(body, str):
            body = body.encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _file(self, path):
        if not os.path.isfile(path):
            return self._send(404, {"error": "not found"})
        ctype = mimetypes.guess_type(path)[0] or "application/octet-stream"
        with open(path, "rb") as f:
            data = f.read()
        self.send_response(200)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def log_message(self, *a):
        pass  # 静音

    def do_GET(self):
        u = urlparse(self.path)
        path, qs = u.path, parse_qs(u.query)
        if path == "/" or path == "/index.html":
            return self._file(os.path.join(WEB_DIR, "index.html"))
        if path == "/api/manifest":
            with open(DEFAULT_MANIFEST, encoding="utf-8") as f:
                return self._send(200, f.read())
        if path == "/api/games":
            return self._send(200, {"games": list_games()})
        if path == "/api/job":
            job_id = qs.get("id", [""])[0]
            job = _job_snapshot(job_id)
            if not job:
                return self._send(404, {"ok": False, "error": "job not found"})
            job["id"] = job_id
            return self._send(200, job)
        if path == "/api/library":
            games = list_games()
            game = (qs.get("game", [None])[0]) or (games[-1] if games else None)
            if not game:
                return self._send(200, {"game": None, "games": games,
                                        "characters": [], "vfx": [], "ui": []})
            library_path = os.path.join(OUT_ROOT, game, "library.json")
            if not os.path.isfile(library_path):
                return self._send(200, {"game": game, "games": games, "assetBase": f"/assets/{game}/",
                                        "characters": [], "vfx": [], "ui": []})
            with open(library_path, encoding="utf-8") as f:
                lib = json.load(f)
            lib["games"] = games
            lib["assetBase"] = f"/assets/{game}/"
            return self._send(200, lib)
        if path.startswith("/assets/"):
            rel = unquote(path[len("/assets/"):])
            try:
                return self._file(safe_join(OUT_ROOT, rel))
            except ValueError:
                return self._send(400, {"error": "bad path"})
        return self._send(404, {"error": "not found"})

    def do_POST(self):
        try:
            return self._do_POST()
        except Exception as e:
            traceback.print_exc()
            return self._send(500, {"ok": False, "error": str(e)})

    def _read_json_body(self):
        length = int(self.headers.get("Content-Length", 0))
        return json.loads(self.rfile.read(length) or b"{}")

    def _do_POST(self):
        route = urlparse(self.path).path
        if route == "/api/export":
            return self._post_export()
        if route == "/api/delete":
            return self._post_delete()
        if route == "/api/slot-workflow":
            return self._post_slot_workflow()
        if route == "/api/creative-manifest":
            return self._post_creative_manifest()
        if route != "/api/generate":
            return self._send(404, {"error": "not found"})
        try:
            data = self._read_json_body()
        except Exception as e:
            return self._send(400, {"ok": False, "error": f"请求体不是合法 JSON: {e}"})
        try:
            manifest = json.loads(data["manifest"]) if isinstance(data.get("manifest"), str) \
                else data.get("manifest")
        except Exception as e:
            return self._send(400, {"ok": False, "error": f"manifest 非法 JSON: {e}"})
        creds = {
            "provider": data.get("provider", "OpenAI 兼容接口"),
            "api_key": (data.get("api_key") or DEFAULT_API_KEY).strip(),
            "base_url": (data.get("base_url") or DEFAULT_BASE_URL).strip(),
            "model": (data.get("model") or DEFAULT_IMAGE_MODEL).strip(),
            "size": data.get("size", "1024x1024"),
            "remove_bg": bool(data.get("remove_bg", False)),
        }
        logs = []
        if data.get("async", True):
            job_id = uuid.uuid4().hex
            with JOBS_LOCK:
                JOBS[job_id] = {"status": "queued", "ok": None, "logs": ["任务已创建，等待开始…"],
                                "game": manifest.get("game"), "createdAt": time.time(), "updatedAt": time.time()}
            threading.Thread(target=_run_generate_job, args=(job_id, manifest, creds), daemon=True).start()
            return self._send(200, {"ok": True, "jobId": job_id, "game": manifest.get("game")})
        try:
            lib, _ = pipeline.run(manifest, OUT_ROOT, creds=creds, log=lambda m: logs.append(m))
        except Exception as e:
            return self._send(500, {"ok": False, "error": str(e), "logs": logs})
        return self._send(200, {"ok": True, "logs": logs, "game": lib["game"]})

    def _post_slot_workflow(self):
        try:
            data = self._read_json_body()
        except Exception as e:
            return self._send(400, {"ok": False, "error": f"请求体不是合法 JSON: {e}"})
        try:
            result = slot_workflow.build_workflow(data)
        except Exception as e:
            traceback.print_exc()
            return self._send(500, {"ok": False, "error": str(e)})
        return self._send(200, {"ok": True, **result})

    def _post_creative_manifest(self):
        try:
            data = self._read_json_body()
        except Exception as e:
            return self._send(400, {"ok": False, "error": f"请求体不是合法 JSON: {e}"})
        try:
            normalized, source, game_plan = creative_to_slot_request(data)
            normalized["creative"] = game_plan.get("creative", {})
            normalized["gameDesign"] = game_plan.get("gameDesign", {})
            result = slot_workflow.build_workflow(normalized)
            return self._send(200, {"ok": True, "source": source, "game_plan": game_plan,
                                    "creative_request": normalized, **result})
        except Exception as e:
            traceback.print_exc()
            return self._send(500, {"ok": False, "error": str(e)})

    def _post_export(self):
        try:
            data = self._read_json_body()
        except Exception as e:
            return self._send(400, {"ok": False, "error": f"请求体不是合法 JSON: {e}"})
        game = (data.get("game") or "").strip()
        if not game or game not in list_games():
            return self._send(400, {"ok": False, "error": f"无效的 game: {game!r}"})
        logs = []
        try:
            pack = exporter.export(game, OUT_ROOT, log=lambda m: logs.append(m))
        except Exception as e:
            traceback.print_exc()
            return self._send(500, {"ok": False, "error": str(e), "logs": logs})
        return self._send(200, {"ok": True, "logs": logs, "game": game,
                                "pack": os.path.abspath(pack)})

    def _post_delete(self):
        try:
            data = self._read_json_body()
        except Exception as e:
            return self._send(400, {"ok": False, "error": f"请求体不是合法 JSON: {e}"})
        game = (data.get("game") or "").strip()
        if not game or game not in list_games():
            return self._send(400, {"ok": False, "error": f"无效的 game: {game!r}"})
        target = os.path.join(OUT_ROOT, game)
        # 安全校验：必须落在 OUT_ROOT 内
        if not os.path.abspath(target).startswith(os.path.abspath(OUT_ROOT) + os.sep):
            return self._send(400, {"ok": False, "error": "非法路径"})
        shutil.rmtree(target)
        return self._send(200, {"ok": True, "deleted": game, "games": list_games()})


if __name__ == "__main__":
    os.makedirs(OUT_ROOT, exist_ok=True)
    print(f"Anim Studio 网站: http://127.0.0.1:{PORT}")
    ThreadingHTTPServer(("127.0.0.1", PORT), Handler).serve_forever()