Ctrl K
ComunidadGuías…

scripts/prepare_e2e_data.py

Rama main · réplica de lectura del repositorio original.

Archivo original ↗Preguntar a KiBot
"""Create the deterministic save used by clean-data browser tests."""

from __future__ import annotations

import argparse
from copy import deepcopy
import json
import os
import secrets
import sqlite3
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from src.adventures import AdventureBundleLoader, AdventureResolver
from src.adventures.graph_v2 import ADVENTURE_GRAPH_FORMAT_V2
from src.adventures.progress import new_progress
from src.engine.game_instance import GameInstance, GameState
from src.engine.modules import adventure_runtime_state, room_access
from src.rulesets.dnd2024.runtime import Dnd2024Runtime
from src.webui.services.legal import bundled_documents
from scripts.e2e_coc_fixture import E2E_COC_GAME_KEYS, write_coc_fixture
from scripts.e2e_choice_fixture import E2E_CHOICE_GAME_KEYS, write_choice_fixtures
from scripts.e2e_dnd_window_fixture import (
    E2E_DND_WINDOW_GAME_KEYS,
    write_window_fixtures,
)
from scripts.e2e_recovery_fixture import E2E_RECOVERY_GAME_KEYS, write_recovery_fixtures


E2E_GAME_KEY = ("web", "e2e-room", "web_bot")
E2E_DND_GAME_KEY = ("web", "e2e-dnd2024", "web_bot")
E2E_ADVENTURE_GAME_KEY = ("web", "e2e-adventure", "web_bot")

# FIX-00:Play 页的 AdventurePanel 必须能读到投影。这个 data-only v2 冒险同时
# 承载“GM 看得到秘密节点 / 分享玩家看不到”的浏览器验收。
E2E_ADVENTURE_DIRECTORY = "e2e_quest"
E2E_ADVENTURE_ID = f"user:{E2E_ADVENTURE_DIRECTORY}"
E2E_ADVENTURE_PUBLIC_NODE = "E2E Public Gate"
E2E_ADVENTURE_SECRET_NODE = "E2E Secret Ritual"

# FIX-06:模组库(ModulesView / ModuleDetailView)的浏览器验收需要一个真实的
# data-only content-pack。它带自己的冒险 id(不与上面的 user 冒险重名,避免制造
# 来源冲突),也不被任何存档绑定,因此详情页的三个受保护按钮都应为可用。
# Lorebook Golden:真实 v4 老库迁移 + 真实检索链路的专用世界。
E2E_LORE_WORLD_ID = "e2e_lore_golden"
# Activation Inspector 走的是真实 retriever,需要一个绑定到该世界的存档。
E2E_LORE_GAME_KEY = ("web", "e2e-lore-golden", "web_bot")

E2E_PARTY_GAME_KEYS = tuple(
    ("web", f"e2e-{journey}-{viewport}", "web_bot")
    for journey in ("party-actions", "party-control-links")
    for viewport in ("desktop", "mobile")
)
# Fixture writers use these identities too; additions must be declared here or
# in the corresponding fixture module so completeness checks stay meaningful.
E2E_GAME_KEYS = (
    E2E_GAME_KEY, E2E_DND_GAME_KEY, E2E_ADVENTURE_GAME_KEY, E2E_LORE_GAME_KEY,
    *E2E_PARTY_GAME_KEYS, *E2E_CHOICE_GAME_KEYS, *E2E_DND_WINDOW_GAME_KEYS,
    *E2E_COC_GAME_KEYS.values(), *E2E_RECOVERY_GAME_KEYS,
)

E2E_MODULE_ID = "e2e-module"
E2E_MODULE_NAME = "E2E Module"
E2E_MODULE_ADVENTURE_DIRECTORY = "module_quest"
E2E_MODULE_ADVENTURE_ID = f"plugin:{E2E_MODULE_ADVENTURE_DIRECTORY}"


def _write_e2e_module(data_dir: Path) -> None:
    """Install a real data-only content-pack for the module-library browser checks.

    FIX-06 §8:模组页面(已安装 / 本地导入 / 在线 + 详情页受保护按钮)必须有真实
    包可读,所以这里按 PluginHost 的包目录约定落一个 content-pack 包(不写任何
    Lorebook / 卡库:catalog 投递不 autoimport)。
    """

    package = data_dir / "plugin-packages" / E2E_MODULE_ID
    adventure_dir = package / "adventures" / E2E_MODULE_ADVENTURE_DIRECTORY
    manifest = {
        "schema_version": 1,
        "id": E2E_MODULE_ID,
        "name": E2E_MODULE_NAME,
        "version": "1.0.0",
        "plugin_type": "content-pack",
        "content_profile": "adventure-module",
        "content_delivery_mode": "catalog",
        "contributes": {"world_templates": ["content/worlds/*.json"], "character_templates": ["content/characters/*.json"]},
        "adventure_packages": [f"adventures/{E2E_MODULE_ADVENTURE_DIRECTORY}"],
        "config_schema": "config.schema.json",
    }
    files: dict[str, dict] = {
        "plugin.json": manifest,
        "config.schema.json": {"type": "object", "properties": {}},
        "content/worlds/resource_world.json": {"world_schema_version": 2, "world_id": "resource_world", "world_name": "E2E Resource World", "starter_lorebook": [{"id": "gate", "name": "Gate", "content": "Module gate prose", "keywords": ["gate"]}]},
        "content/characters/resource_hero.json": {"id": "resource_hero", "character_name": "E2E Resource Hero", "background": "Module hero background"},
        f"adventures/{E2E_MODULE_ADVENTURE_DIRECTORY}/manifest.json": {
            "schema_version": 1,
            "adventure_id": E2E_MODULE_ADVENTURE_ID,
            "version": "1.0.0",
            "format": ADVENTURE_GRAPH_FORMAT_V2,
            "world_policy": "portable",
            "recommended_world_id": "default_fantasy",
            "required_runtime": {"id": "core:dnd2024", "minimum_version": 1},
            "default_locale": "zh-CN",
            "supported_locales": ["zh-CN"],
        },
        f"adventures/{E2E_MODULE_ADVENTURE_DIRECTORY}/adventure.json": {
            "schema_version": 1,
            "kind": "adventure",
            "id": E2E_MODULE_ADVENTURE_DIRECTORY,
            "source_ref": "diceframe-e2e:module-quest",
            "recommended_world_id": "default_fantasy",
            "automation_level": "guided",
            "chapters": [{"id": "module_chapter", "name": "E2E Module Chapter"}],
            "nodes": [{
                "id": "module_gate", "type": "scene", "chapter_id": "module_chapter",
                "name": "E2E Module Gate", "transitions": [],
            }],
            "objectives": [],
            "milestones": [],
            "start_node_ids": ["module_gate"],
        },
        f"adventures/{E2E_MODULE_ADVENTURE_DIRECTORY}/locales/zh-CN/adventure.json": {
            "locale_schema_version": 1,
            "locale": "zh-CN",
            "target": {"kind": "adventure", "id": E2E_MODULE_ADVENTURE_DIRECTORY},
            "fields": {"tutorial": {"name": "E2E 模组冒险", "summary": "模组库浏览器验收。"}},
        },
    }
    for relative, payload in files.items():
        target = package / relative
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(
            json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8",
        )
    if not adventure_dir.is_dir():  # pragma: no cover - 落盘失败必须显式失败
        raise RuntimeError("failed to write the E2E module package")


def _write_save(data_dir: Path, instance: GameInstance) -> Path:
    save_file = data_dir / "saves" / "#".join(instance.game_key) / "state.json"
    save_file.parent.mkdir(parents=True, exist_ok=True)
    save_file.write_text(
        json.dumps(instance.to_dict(), ensure_ascii=False, indent=2), encoding="utf-8",
    )
    return save_file


def _e2e_players() -> dict[str, dict]:
    return {
        "e2e-gm": {
            "character_name": "E2E GM",
            "character_sheet": {
                "character_name": "E2E GM",
                "attributes": {"str": 12, "dex": 10},
                "skills": [],
                "hp": 10,
                "max_hp": 10,
                "portrait": {"kind": "builtin", "id": "warrior"},
                "equipment": [
                    {"name": "Longsword", "type": "weapon", "damage": "1d8", "slot": "main_hand"},
                    {"name": "Shield", "type": "armor", "slot": "off_hand"},
                ],
                "inventory": [{"name": "Healing Potion", "quantity": 2, "effect": "Restore health"}],
                "key_items": [{"name": "Town Gate Seal", "description": "Proof of passage"}],
            },
        },
        "e2e-player": {
            "character_name": "E2E Player",
            "character_sheet": {
                "character_name": "E2E Player",
                "attributes": {"str": 9, "dex": 13},
                "skills": [],
                "hp": 9,
                "max_hp": 9,
                "portrait": {"kind": "builtin", "id": "ranger"},
                "equipment": [{"name": "Shortbow", "type": "weapon", "damage": "1d6"}],
                "inventory": [{"name": "Rope", "quantity": 1}],
                "key_items": [],
            },
        },
    }


def _write_e2e_adventure(data_dir: Path) -> dict:
    """Install the data-only v2 adventure used by the browser play checks."""

    package = data_dir / "templates" / "adventures" / E2E_ADVENTURE_DIRECTORY
    files = {
        "manifest.json": {
            "schema_version": 1,
            "adventure_id": E2E_ADVENTURE_ID,
            "version": "1.0.0",
            "format": ADVENTURE_GRAPH_FORMAT_V2,
            "world_policy": "portable",
            "recommended_world_id": "default_fantasy",
            "required_runtime": {"id": "core:dnd2024", "minimum_version": 1},
            "default_locale": "zh-CN",
            "supported_locales": ["zh-CN"],
        },
        "adventure.json": {
            "schema_version": 1,
            "kind": "adventure",
            "id": E2E_ADVENTURE_DIRECTORY,
            "source_ref": "diceframe-e2e:e2e-quest",
            "recommended_world_id": "default_fantasy",
            "automation_level": "guided",
            "estimated_minutes": 15,
            "visibility": "public",
            "chapters": [
                {"id": "public_chapter", "name": "E2E Public Chapter"},
                {"id": "secret_chapter", "name": "E2E Secret Chapter", "visibility": "gm"},
            ],
            "nodes": [
                {
                    "id": "public_gate", "type": "scene", "chapter_id": "public_chapter",
                    "name": E2E_ADVENTURE_PUBLIC_NODE,
                    "transitions": [{"to": "secret_ritual"}],
                },
                {
                    "id": "secret_ritual", "type": "scene", "chapter_id": "secret_chapter",
                    "visibility": "gm", "name": E2E_ADVENTURE_SECRET_NODE, "transitions": [],
                },
            ],
            "objectives": [
                {"id": "reach_gate", "name": "Reach the gate", "node_ids": ["public_gate"]},
            ],
            "milestones": [],
            "start_node_ids": ["public_gate"],
        },
        "locales/zh-CN/adventure.json": {
            "locale_schema_version": 1,
            "locale": "zh-CN",
            "target": {"kind": "adventure", "id": E2E_ADVENTURE_DIRECTORY},
            "fields": {"tutorial": {"name": "E2E 冒险", "summary": "浏览器验收用冒险。"}},
        },
    }
    for relative, payload in files.items():
        target = package / relative
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(
            json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8",
        )
    bundle_loader_dir = package.parent
    resolver = AdventureResolver.single_directory(bundle_loader_dir, source_kind="user")
    resolution = resolver.resolve_with_source(E2E_ADVENTURE_ID, "zh-CN")
    if resolution.bundle.content_digest != AdventureBundleLoader(
        bundle_loader_dir,
    ).resolve(E2E_ADVENTURE_ID, "zh-CN").content_digest:
        raise RuntimeError("E2E adventure digest is not stable")
    return resolution.binding("default_fantasy")


def prepare_e2e_data(data_dir: Path) -> Path:
    data_dir = data_dir.resolve()
    documents = bundled_documents()
    config_file = data_dir / "config.json"
    config_file.parent.mkdir(parents=True, exist_ok=True)
    config_file.write_text(
        json.dumps(
            {
                "ai_providers": [{
                    "id": "e2e-provider",
                    "name": "E2E Provider",
                    "base_url": "http://127.0.0.1:9/v1",
                    "api_format": "openai",
                    "models": ["e2e-chat", "e2e-embedding"],
                }],
                "llm_provider_ref": "e2e-provider",
                "model": "e2e-chat",
                "hub_telemetry_enabled": False,
                "hub_telemetry_choice_made": True,
                "legal_terms_accepted_updated_at": documents["terms"]["updated_at"],
                "legal_privacy_accepted_updated_at": documents["privacy"]["updated_at"],
                "legal_accepted_at": "2026-08-11T00:00:00+00:00",
            },
            ensure_ascii=False,
            indent=2,
        ),
        encoding="utf-8",
    )
    (data_dir / "secrets.json").write_text(
        json.dumps(
            {"ai_provider_key_e2e-provider": secrets.token_urlsafe(24)},
            ensure_ascii=False,
            indent=2,
        ),
        encoding="utf-8",
    )
    instance = GameInstance(
        game_key=E2E_GAME_KEY,
        world_id="default_fantasy",
        world_name="E2E Adventure",
        group_name="Browser Tests",
        state=GameState.ACTIVE_ACTION,
        gm_uid="e2e-gm",
    )
    instance.set_scene("Town Gate")
    instance.round_number = 2
    instance.players = _e2e_players()
    instance.log = [{
        "round": 1,
        "actions": [
            {"user_id": "e2e-gm", "text": "Inspect the gate."},
            {"user_id": "e2e-player", "text": "Watch the road."},
        ],
        "gm_response": "The road is quiet.",
    }]
    save_file = _write_save(data_dir, instance)
    # Mutating party journeys need independent seats for each viewport; other
    # smoke tests invite extra members into the shared room.
    for game_key in E2E_PARTY_GAME_KEYS:
        party = deepcopy(instance)
        party.game_key = game_key
        _write_save(data_dir, party)

    # FIX-00:独立的一局绑定 data-only v2 冒险,供 Play 页 AdventurePanel 的
    # “GM 可读 / 分享玩家只读公开部分”浏览器验收使用,避免扰动既有布局用例。
    adventure_instance = GameInstance(
        game_key=E2E_ADVENTURE_GAME_KEY,
        world_id="default_fantasy",
        world_name="E2E Adventure Module",
        group_name="Adventure Browser Tests",
        state=GameState.ACTIVE_ACTION,
        gm_uid="e2e-gm",
    )
    adventure_instance.set_scene("Town Gate")
    adventure_instance.round_number = 1
    adventure_instance.players = _e2e_players()
    if not adventure_instance.bind_adventure(_write_e2e_adventure(data_dir)):
        raise RuntimeError("failed to bind the E2E adventure in the fixture")
    # Mirror the real v2 create lifecycle: the panel must receive an
    # authoritative active node, never infer it from the graph transition.
    adventure_bundle = AdventureBundleLoader(
        data_dir / "templates" / "adventures",
    ).resolve(E2E_ADVENTURE_ID, "zh-CN")
    adventure_runtime_state.replace_progress(adventure_instance, new_progress(adventure_bundle.adventure))
    _write_save(data_dir, adventure_instance)

    runtime = Dnd2024Runtime()
    choices = runtime.builder_choices(None, {"locale": "zh-CN"})
    preset = next(item for item in choices["quick_presets"] if item["id"] == "stalwart_guardian")
    dnd_character = runtime.finalize_character(
        None, {**preset["draft"], "locale": "zh-CN", "name": "新手守护者"},
    )
    dnd_instance = GameInstance(
        game_key=E2E_DND_GAME_KEY,
        world_id="default_fantasy",
        world_name="D&D 2024 新手桌",
        group_name="Professional Ruleset Browser Tests",
        state=GameState.ACTIVE_ACTION,
        gm_uid="e2e-gm",
        rule_id="dnd2024_srd",
        language="zh-CN",
    )
    dnd_instance.set_scene("灰沼村议事厅")
    room_access.replace_max_players(dnd_instance, 2)
    dnd_instance.players = {
        "e2e-gm": {
            "character_name": "新手守护者",
            "character_sheet": dnd_character,
        },
    }
    if not dnd_instance.bind_ruleset_runtime(dnd_character["rule_binding"]):
        raise RuntimeError("failed to bind D&D 2024 runtime in E2E fixture")
    _write_save(data_dir, dnd_instance)
    write_choice_fixtures(data_dir, dnd_instance, _write_save)
    write_window_fixtures(data_dir, runtime, dnd_character, _write_save)
    # FIX-06:模组库的浏览器验收需要真实 content-pack 包(§8 的产品面)。
    _write_e2e_module(data_dir)
    # Lorebook Golden 的真实链路验收需要一个**真的 v4 老库**:服务器启动时会跑
    # 真实 migration,浏览器随后看到的就是迁移产物,而不是测试自己捏的 JSON。
    _write_legacy_lorebook_db(data_dir)
    lore_instance = GameInstance(
        game_key=E2E_LORE_GAME_KEY,
        world_id=E2E_LORE_WORLD_ID,
        world_name="Golden Lore World",
        group_name="Lorebook Golden",
        state=GameState.ACTIVE_ACTION,
        gm_uid="e2e-gm",
        language="zh-CN",
    )
    lore_instance.set_scene("旧城门前")
    lore_instance.round_number = 2
    lore_instance.players = _e2e_players()
    _write_save(data_dir, lore_instance)
    write_coc_fixture(data_dir, _write_save)
    write_recovery_fixtures(data_dir, _write_save)
    return save_file


def _write_legacy_lorebook_db(data_dir: Path) -> Path:
    """Materialise an authentic pre-v6 (schema v4) lorebook.db.

    The Golden browser run must exercise the real chain — old SQLite → real
    migration → backend → resolver → matcher → browser — so the fixture stops at
    the historical schema version and lets normal startup migrate it. Writing the
    already-migrated shape here would test nothing about migration, and
    hand-copying an old CREATE TABLE would drift from the real one, so the real
    migration steps are replayed up to v4 instead.
    """

    from src.lorebook.store import SCHEMA
    from src.migrations.lorebook import migrate

    db_path = data_dir / "lorebook.db"
    db_path.parent.mkdir(parents=True, exist_ok=True)
    db_path.unlink(missing_ok=True)
    conn = sqlite3.connect(db_path)
    try:
        conn.executescript(SCHEMA)
        migrate(conn, upto=4)
        conn.execute(
            "INSERT OR IGNORE INTO worlds (id, name, description, language) VALUES (?, ?, ?, ?)",
            (E2E_LORE_WORLD_ID, "Golden Lore World", "Lorebook Golden E2E", "zh-CN"),
        )
        # 内容语言选择器回归数据:ja / de 标记的世界,供语言切换用例筛选。
        for world_id, name, language in (
            ("e2e_lore_golden_ja", "ゴールデン・ロア・ワールド", "ja"),
            ("e2e_lore_golden_de", "Golden Lore Welt", "de"),
        ):
            conn.execute(
                "INSERT OR IGNORE INTO worlds (id, name, description, language) VALUES (?, ?, ?, ?)",
                (world_id, name, f"Lorebook Golden E2E ({language})", language),
            )
        for entry in _legacy_lore_entries():
            columns = ", ".join(f'"{key}"' for key in entry)
            placeholders = ", ".join("?" for _ in entry)
            conn.execute(
                f"INSERT INTO lorebook_entries ({columns}) VALUES ({placeholders})",
                tuple(entry.values()),
            )
        conn.commit()
        version = conn.execute("PRAGMA user_version").fetchone()[0]
        if version != 4:
            raise RuntimeError(f"legacy lorebook fixture must stay at schema v4, got {version}")
    finally:
        conn.close()
    return db_path


def _legacy_lore_entries() -> list[dict[str, object]]:
    """Legacy rows covering the activation features the Golden run asserts.

    ``visible_to`` uses the legacy JSON shape on purpose: the migration is what
    turns these into canonical v6 rows.
    """

    return [
        {
            "id": "legacy-gate", "world_id": E2E_LORE_WORLD_ID, "name": "旧城门",
            "type": "location", "keywords": json.dumps(["旧城门"], ensure_ascii=False),
            "content": "旧城门的门闩上刻着一枚封印纹章。", "tier": "core",
            "visible_to": json.dumps(["*"]), "order": 10,
        },
        {
            "id": "legacy-sigil", "world_id": E2E_LORE_WORLD_ID, "name": "封印纹章",
            "type": "item", "keywords": json.dumps(["封印纹章"], ensure_ascii=False),
            "content": "纹章是打开地下档案室的钥匙。", "tier": "background",
            "visible_to": json.dumps(["*"]), "order": 20,
        },
        {
            "id": "legacy-secret", "world_id": E2E_LORE_WORLD_ID, "name": "GM 密档",
            "type": "event", "keywords": json.dumps(["旧城门"], ensure_ascii=False),
            "content": "守门人其实是伪装的内应。", "tier": "core",
            "visible_to": json.dumps([]), "order": 30,
        },
    ]


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--data-dir", type=Path, default=None)
    args = parser.parse_args()
    environment_dir = os.getenv("DICEFRAME_E2E_DATA_DIR")
    raw_data_dir = args.data_dir or (Path(environment_dir) if environment_dir else None)
    if raw_data_dir is None:
        parser.error("--data-dir or DICEFRAME_E2E_DATA_DIR is required")
    print(prepare_e2e_data(raw_data_dir))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())