webnovel-writer/.claude/scripts/stress_test_500chapters.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
500章写作沙盘模拟 - 数据链稳定性压力测试

测试目标：
1. state.json 增长曲线（文件大小随章节变化）
2. entities_v3 实体数量增长
3. alias_index 别名索引膨胀
4. 伏笔追踪（埋设/回收比例）
5. 原子写入性能
6. index.db 查询性能

模拟参数（基于典型网文）：
- 500章，每章约3500字
- 平均每章新增 0.8 个角色（前100章密集，后期稀疏）
- 平均每章新增 0.3 个地点
- 平均每章埋设 0.5 个伏笔，回收 0.3 个
- 主角每 10 章升级一次境界
- 每 5 章更新一次关系
"""

import json
import os
import sys
import time
import random
import shutil
import tempfile
from pathlib import Path
from datetime import datetime
from typing import Dict, Any, List

# 添加脚本目录到路径
script_dir = Path(__file__).resolve().parent
sys.path.insert(0, str(script_dir))

from security_utils import atomic_write_json, read_json_safe

# Windows 编码修复
if sys.platform == 'win32':
    import io
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')


# ============================================================================
# 模拟配置
# ============================================================================

CONFIG = {
    "total_chapters": 500,
    "words_per_chapter": 3500,

    # 实体生成概率（随章节递减）
    "new_character_base_rate": 0.8,  # 前50章
    "new_character_decay": 0.95,      # 每50章衰减
    "new_location_rate": 0.3,
    "new_item_rate": 0.2,
    "new_faction_rate": 0.1,
    "new_technique_rate": 0.15,

    # 伏笔
    "foreshadow_plant_rate": 0.5,
    "foreshadow_resolve_rate": 0.3,
    "foreshadow_tiers": ["核心", "支线", "装饰"],
    "foreshadow_tier_weights": [0.1, 0.3, 0.6],

    # 主角升级
    "protagonist_upgrade_interval": 10,
    "realms": ["练气", "筑基", "金丹", "元婴", "化神", "炼虚", "合体", "大乘", "渡劫"],
    "layers_per_realm": 9,

    # 关系更新
    "relationship_update_interval": 5,
    "relationship_types": ["ally", "enemy", "romance", "mentor", "rival", "family"],

    # 别名生成
    "alias_per_character": 2.5,  # 平均每个角色的别名数
}

# 随机名字池
SURNAME_POOL = ["林", "陈", "王", "李", "张", "刘", "赵", "黄", "周", "吴", "徐", "孙", "马", "朱", "胡", "郭", "何", "高", "罗", "郑"]
NAME_POOL = ["天", "云", "风", "雷", "火", "水", "月", "星", "龙", "凤", "虎", "鹤", "剑", "刀", "枪", "棍", "拳", "掌", "指", "心"]
LOCATION_PREFIX = ["天", "云", "龙", "凤", "青", "白", "黑", "红", "金", "玉"]
LOCATION_SUFFIX = ["山", "谷", "城", "峰", "洞", "海", "林", "湖", "殿", "宗"]


class SimulationMetrics:
    """模拟指标收集器"""

    def __init__(self):
        self.checkpoints: List[Dict] = []
        self.write_times: List[float] = []
        self.errors: List[str] = []

    def record_checkpoint(self, chapter: int, state: Dict, state_file: Path):
        """记录检查点"""
        file_size = state_file.stat().st_size if state_file.exists() else 0

        entities_v3 = state.get("entities_v3", {})
        entity_counts = {
            etype: len(entities)
            for etype, entities in entities_v3.items()
        }
        total_entities = sum(entity_counts.values())

        alias_count = len(state.get("alias_index", {}))

        foreshadowing = state.get("foreshadowing", [])
        active_foreshadow = len([f for f in foreshadowing if f.get("status") == "未回收"])
        resolved_foreshadow = len([f for f in foreshadowing if f.get("status") == "已回收"])

        relationships = state.get("relationships", [])
        if isinstance(relationships, dict):
            relationships = list(relationships.values())

        self.checkpoints.append({
            "chapter": chapter,
            "file_size_kb": file_size / 1024,
            "total_entities": total_entities,
            "entity_counts": entity_counts,
            "alias_count": alias_count,
            "active_foreshadow": active_foreshadow,
            "resolved_foreshadow": resolved_foreshadow,
            "relationship_count": len(relationships) if isinstance(relationships, list) else 0,
            "avg_write_time_ms": sum(self.write_times[-10:]) / max(len(self.write_times[-10:]), 1) * 1000,
        })

    def record_write_time(self, duration: float):
        self.write_times.append(duration)

    def record_error(self, error: str):
        self.errors.append(error)

    def generate_report(self) -> str:
        """生成测试报告"""
        if not self.checkpoints:
            return "No data collected"

        final = self.checkpoints[-1]
        first = self.checkpoints[0]

        lines = [
            "=" * 60,
            "📊 500章沙盘模拟测试报告",
            "=" * 60,
            "",
            "## 基础指标",
            f"- 总章节数: {final['chapter']}",
            f"- 总字数: {final['chapter'] * CONFIG['words_per_chapter']:,}",
            "",
            "## state.json 增长",
            f"- 初始大小: {first['file_size_kb']:.2f} KB",
            f"- 最终大小: {final['file_size_kb']:.2f} KB",
            f"- 增长倍数: {final['file_size_kb'] / max(first['file_size_kb'], 0.1):.1f}x",
            "",
            "## 实体统计",
            f"- 总实体数: {final['total_entities']}",
        ]

        for etype, count in final['entity_counts'].items():
            lines.append(f"  - {etype}: {count}")

        lines.extend([
            f"- 别名索引条目: {final['alias_count']}",
            "",
            "## 伏笔统计",
            f"- 活跃伏笔: {final['active_foreshadow']}",
            f"- 已回收伏笔: {final['resolved_foreshadow']}",
            f"- 回收率: {final['resolved_foreshadow'] / max(final['active_foreshadow'] + final['resolved_foreshadow'], 1) * 100:.1f}%",
            "",
            "## 性能指标",
            f"- 平均写入时间: {sum(self.write_times) / max(len(self.write_times), 1) * 1000:.2f} ms",
            f"- 最大写入时间: {max(self.write_times) * 1000:.2f} ms" if self.write_times else "N/A",
            f"- 最小写入时间: {min(self.write_times) * 1000:.2f} ms" if self.write_times else "N/A",
            "",
            "## 错误统计",
            f"- 错误数: {len(self.errors)}",
        ])

        if self.errors:
            lines.append("- 错误详情:")
            for err in self.errors[:5]:
                lines.append(f"  - {err}")

        # 增长曲线（每100章采样）
        lines.extend([
            "",
            "## 增长曲线（每100章）",
            "| 章节 | 文件大小(KB) | 实体数 | 别名数 | 活跃伏笔 | 写入时间(ms) |",
            "|------|-------------|-------|-------|---------|-------------|",
        ])

        for cp in self.checkpoints:
            if cp['chapter'] % 100 == 0 or cp['chapter'] == final['chapter']:
                lines.append(
                    f"| {cp['chapter']} | {cp['file_size_kb']:.1f} | "
                    f"{cp['total_entities']} | {cp['alias_count']} | "
                    f"{cp['active_foreshadow']} | {cp['avg_write_time_ms']:.1f} |"
                )

        # 稳定性评估
        lines.extend([
            "",
            "## 稳定性评估",
        ])

        # 检查文件大小是否在合理范围
        if final['file_size_kb'] < 500:
            lines.append("✅ 文件大小合理 (< 500KB)")
        elif final['file_size_kb'] < 1024:
            lines.append("⚠️ 文件大小偏大 (500KB-1MB)，建议启用归档")
        else:
            lines.append("❌ 文件过大 (> 1MB)，需要优化")

        # 检查写入性能
        avg_write = sum(self.write_times) / max(len(self.write_times), 1) * 1000
        if avg_write < 50:
            lines.append("✅ 写入性能良好 (< 50ms)")
        elif avg_write < 200:
            lines.append("⚠️ 写入性能一般 (50-200ms)")
        else:
            lines.append("❌ 写入性能差 (> 200ms)")

        # 检查错误率
        if not self.errors:
            lines.append("✅ 无错误")
        else:
            lines.append(f"❌ 有 {len(self.errors)} 个错误")

        lines.append("")
        lines.append("=" * 60)

        return "\n".join(lines)


class ChapterSimulator:
    """章节模拟器"""

    def __init__(self, project_root: Path):
        self.project_root = project_root
        self.state_file = project_root / ".webnovel" / "state.json"
        self.metrics = SimulationMetrics()
        self.generated_names = set()
        self.entity_id_counter = 0

    def _generate_id(self, prefix: str) -> str:
        self.entity_id_counter += 1
        return f"{prefix}_{self.entity_id_counter:05d}"

    def _generate_character_name(self) -> str:
        for _ in range(100):
            name = random.choice(SURNAME_POOL) + random.choice(NAME_POOL) + random.choice(NAME_POOL)
            if name not in self.generated_names:
                self.generated_names.add(name)
                return name
        return f"角色_{len(self.generated_names)}"

    def _generate_location_name(self) -> str:
        return random.choice(LOCATION_PREFIX) + random.choice(LOCATION_SUFFIX)

    def _get_character_rate(self, chapter: int) -> float:
        """根据章节获取角色生成概率（递减）"""
        decay_periods = chapter // 50
        rate = CONFIG["new_character_base_rate"] * (CONFIG["new_character_decay"] ** decay_periods)
        return max(rate, 0.1)  # 最低 10%

    def init_project(self):
        """初始化模拟项目"""
        self.project_root.mkdir(parents=True, exist_ok=True)
        (self.project_root / ".webnovel").mkdir(exist_ok=True)
        (self.project_root / "正文").mkdir(exist_ok=True)

        # 初始 state.json
        initial_state = {
            "project_info": {
                "title": "模拟测试小说",
                "genre": "玄幻",
                "created_at": datetime.now().strftime("%Y-%m-%d"),
                "target_chapters": 500,
            },
            "progress": {
                "current_chapter": 0,
                "total_words": 0,
            },
            "protagonist_state": {
                "name": "林天",
                "realm": "练气",
                "layer": 1,
                "golden_finger": {"name": "混沌珠", "level": 1},
            },
            "entities_v3": {
                "角色": {},
                "地点": {},
                "物品": {},
                "势力": {},
                "招式": {},
            },
            "alias_index": {},
            "foreshadowing": [],
            "relationships": [],
        }

        # 添加主角到实体
        protagonist_id = "protagonist_lintian"
        initial_state["entities_v3"]["角色"][protagonist_id] = {
            "canonical_name": "林天",
            "desc": "主角，拥有混沌珠",
            "tier": "核心",
            "aliases": ["林天", "天哥", "林少侠"],
            "current": {"realm": "练气", "layer": 1},
            "history": [],
        }
        initial_state["alias_index"]["林天"] = [{"type": "角色", "id": protagonist_id}]
        initial_state["alias_index"]["天哥"] = [{"type": "角色", "id": protagonist_id}]

        atomic_write_json(self.state_file, initial_state, backup=False)
        return initial_state

    def simulate_chapter(self, chapter: int, state: Dict) -> Dict:
        """模拟一章的数据变化"""

        # 1. 更新进度
        state["progress"]["current_chapter"] = chapter
        state["progress"]["total_words"] += CONFIG["words_per_chapter"]

        entities_v3 = state["entities_v3"]
        alias_index = state["alias_index"]

        # 2. 新增角色（概率递减）
        if random.random() < self._get_character_rate(chapter):
            char_name = self._generate_character_name()
            char_id = self._generate_id("char")
            tier = random.choices(
                ["核心", "支线", "装饰"],
                weights=[0.1, 0.3, 0.6]
            )[0]

            entities_v3["角色"][char_id] = {
                "canonical_name": char_name,
                "desc": f"第{chapter}章出场的{tier}角色",
                "tier": tier,
                "aliases": [char_name],
                "current": {"first_appearance": chapter},
                "history": [],
            }
            alias_index[char_name] = [{"type": "角色", "id": char_id}]

            # 生成额外别名
            if random.random() < 0.5:
                alias = char_name[0] + "兄" if random.random() < 0.5 else char_name + "前辈"
                entities_v3["角色"][char_id]["aliases"].append(alias)
                if alias not in alias_index:
                    alias_index[alias] = []
                alias_index[alias].append({"type": "角色", "id": char_id})

        # 3. 新增地点
        if random.random() < CONFIG["new_location_rate"]:
            loc_name = self._generate_location_name()
            loc_id = self._generate_id("loc")
            entities_v3["地点"][loc_id] = {
                "canonical_name": loc_name,
                "desc": f"第{chapter}章出现的地点",
                "tier": "装饰",
                "aliases": [loc_name],
                "current": {},
                "history": [],
            }
            alias_index[loc_name] = [{"type": "地点", "id": loc_id}]

        # 4. 新增物品
        if random.random() < CONFIG["new_item_rate"]:
            item_name = random.choice(["灵", "仙", "神", "圣"]) + random.choice(["剑", "丹", "符", "器"])
            item_id = self._generate_id("item")
            entities_v3["物品"][item_id] = {
                "canonical_name": item_name,
                "desc": f"第{chapter}章获得的物品",
                "tier": "装饰",
                "aliases": [item_name],
                "current": {},
                "history": [],
            }
            if item_name not in alias_index:
                alias_index[item_name] = []
            alias_index[item_name].append({"type": "物品", "id": item_id})

        # 5. 埋设伏笔
        if random.random() < CONFIG["foreshadow_plant_rate"]:
            tier = random.choices(
                CONFIG["foreshadow_tiers"],
                weights=CONFIG["foreshadow_tier_weights"]
            )[0]
            target = chapter + random.randint(10, 100)

            state["foreshadowing"].append({
                "id": f"fs_{chapter}_{random.randint(1000, 9999)}",
                "content": f"第{chapter}章埋设的{tier}伏笔",
                "tier": tier,
                "status": "未回收",
                "planted_chapter": chapter,
                "target_chapter": target,
            })

        # 6. 回收伏笔
        active_foreshadows = [
            f for f in state["foreshadowing"]
            if f.get("status") == "未回收" and f.get("target_chapter", 999) <= chapter
        ]
        for fs in active_foreshadows:
            if random.random() < CONFIG["foreshadow_resolve_rate"]:
                fs["status"] = "已回收"
                fs["resolved_chapter"] = chapter

        # 7. 主角升级
        if chapter % CONFIG["protagonist_upgrade_interval"] == 0:
            ps = state["protagonist_state"]
            current_layer = ps.get("layer", 1)
            current_realm_idx = CONFIG["realms"].index(ps.get("realm", "练气"))

            if current_layer < CONFIG["layers_per_realm"]:
                ps["layer"] = current_layer + 1
            elif current_realm_idx < len(CONFIG["realms"]) - 1:
                ps["realm"] = CONFIG["realms"][current_realm_idx + 1]
                ps["layer"] = 1

        # 8. 更新关系
        if chapter % CONFIG["relationship_update_interval"] == 0:
            char_ids = list(entities_v3["角色"].keys())
            if len(char_ids) >= 2:
                char1, char2 = random.sample(char_ids, 2)
                rel_type = random.choice(CONFIG["relationship_types"])

                state["relationships"].append({
                    "char1_id": char1,
                    "char2_id": char2,
                    "type": rel_type,
                    "intensity": random.randint(30, 100),
                    "established_chapter": chapter,
                })

        return state

    def run_simulation(self, checkpoint_interval: int = 10):
        """运行完整模拟"""
        print("🚀 开始500章沙盘模拟...")
        print(f"📁 测试目录: {self.project_root}")
        print()

        state = self.init_project()
        self.metrics.record_checkpoint(0, state, self.state_file)

        start_time = time.time()

        for chapter in range(1, CONFIG["total_chapters"] + 1):
            try:
                # 模拟章节
                state = self.simulate_chapter(chapter, state)

                # 原子写入
                write_start = time.time()
                atomic_write_json(self.state_file, state, use_lock=True, backup=False)
                write_duration = time.time() - write_start
                self.metrics.record_write_time(write_duration)

                # 记录检查点
                if chapter % checkpoint_interval == 0:
                    self.metrics.record_checkpoint(chapter, state, self.state_file)
                    elapsed = time.time() - start_time
                    eta = elapsed / chapter * (CONFIG["total_chapters"] - chapter)
                    print(f"  第 {chapter:3d} 章完成 | "
                          f"文件 {self.state_file.stat().st_size / 1024:.1f}KB | "
                          f"实体 {sum(len(e) for e in state['entities_v3'].values())} | "
                          f"写入 {write_duration*1000:.1f}ms | "
                          f"ETA {eta:.0f}s")

            except Exception as e:
                self.metrics.record_error(f"Chapter {chapter}: {str(e)}")
                print(f"  ❌ 第 {chapter} 章错误: {e}")

        # 最终检查点
        self.metrics.record_checkpoint(CONFIG["total_chapters"], state, self.state_file)

        total_time = time.time() - start_time
        print()
        print(f"✅ 模拟完成！总耗时: {total_time:.1f}s")
        print()

        return self.metrics.generate_report()


def main():
    """主函数"""
    # 创建临时测试目录
    test_dir = Path(tempfile.mkdtemp(prefix="webnovel_stress_test_"))

    try:
        simulator = ChapterSimulator(test_dir)
        report = simulator.run_simulation(checkpoint_interval=10)

        print(report)

        # 保存报告
        report_file = test_dir / "stress_test_report.md"
        report_file.write_text(report, encoding="utf-8")
        print(f"\n📄 报告已保存: {report_file}")

        # 询问是否保留测试数据
        print(f"\n测试数据目录: {test_dir}")
        print("（测试完成后可手动删除）")

    except KeyboardInterrupt:
        print("\n⚠️ 测试被中断")
    except Exception as e:
        print(f"\n❌ 测试失败: {e}")
        raise


if __name__ == "__main__":
    main()