novel-skill/scripts/event_store.py

#!/usr/bin/env python3
"""
event_store.py — 事件溯源引擎（Event Sourcing）

解决多用户 git 共享进化数据的冲突问题：
- 每个用户写入不可变事件日志文件（追加写入，天然无冲突）
- population.json / thresholds.json 从事件日志编译生成（可复现）
- git pull → 获得他人事件 → recompile → 最新状态

事件文件格式:
  evolution/genres/{genre_id}/events/{timestamp}_{user}.json
  内容: [event, event, ...]

事件类型:
  - feedback: 章节反馈（多维评分写入）
  - mutate: 变异产生新变体 / 移除旧变体
  - extinction: 灭绝重播
  - threshold_adjust: 阈值调优
"""

import json, os, re, hashlib, subprocess
from pathlib import Path
from datetime import datetime
from typing import Any

import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from utils import (
    get_evolution_dir, get_genre_evolution_dir,
    load_genre_evolution, save_genre_evolution,
    load_evolution, save_evolution,
)


def _get_user() -> str:
    """获取当前用户标识（git config → 系统用户名 → hostname）"""
    try:
        result = subprocess.run(
            ["git", "config", "user.name"], capture_output=True, text=True
        )
        if result.returncode == 0 and result.stdout.strip():
            return result.stdout.strip()
    except Exception:
        pass
    try:
        result = subprocess.run(
            ["git", "config", "user.email"], capture_output=True, text=True
        )
        if result.returncode == 0 and result.stdout.strip():
            return result.stdout.strip().split("@")[0]
    except Exception:
        pass
    return os.environ.get("USER", os.environ.get("USERNAME", "unknown"))


def _events_dir(genre_id: str) -> Path:
    d = get_genre_evolution_dir(genre_id) / "events"
    d.mkdir(parents=True, exist_ok=True)
    return d


def _event_filename(user: str | None = None) -> str:
    """生成事件文件名: {timestamp}_{user}.json"""
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    u = user or _get_user()
    return f"{ts}_{u}.json"


def write_events(genre_id: str, events: list[dict],
                 user: str | None = None) -> Path:
    """追加写入事件文件（一次会话一个文件，追加写入天然无冲突）。

    返回写入的文件路径。
    """
    if not events:
        return None

    u = user or _get_user()
    fname = _event_filename(u)
    fpath = _events_dir(genre_id) / fname

    with open(fpath, "w", encoding="utf-8") as f:
        json.dump({
            "user": u,
            "timestamp": datetime.now().isoformat(),
            "events": events,
        }, f, ensure_ascii=False, indent=2)

    return fpath


def read_all_events(genre_id: str) -> list[dict]:
    """按时间顺序读取所有事件文件，返回展开后的事件列表。"""
    d = _events_dir(genre_id)
    if not d.exists():
        return []

    all_events = []
    for fpath in sorted(d.glob("*.json")):
        try:
            data = json.loads(fpath.read_text(encoding="utf-8"))
            for evt in data.get("events", []):
                evt["_source_user"] = data.get("user", "unknown")
                evt["_source_file"] = fpath.name
                all_events.append(evt)
        except (json.JSONDecodeError, KeyError):
            continue

    return all_events


def compile_population(genre_id: str) -> dict:
    """从事件日志编译 population.json。

    回放所有事件，重建 5 岛种群状态。
    """
    from prompt_evolver import PromptVariant, PromptPopulation, IslandModel

    events = read_all_events(genre_id)

    # 从种子初始化
    from seed_extractor import SeedExtractor
    extractor = SeedExtractor()
    seeds = extractor.extract_seeds(genre_id=genre_id)

    model = IslandModel()
    model.initialize(seeds)
    for cat, pop in model.islands.items():
        pop.generation = 0

    gen_counters: dict[str, int] = {}

    for evt in events:
        etype = evt.get("type", "")

        if etype == "feedback":
            island = evt.get("island", "writing")
            variant_id = evt.get("variant_id", "")
            scores = evt.get("scores", {})
            variant_template = evt.get("variant_template", "")

            if island in model.islands:
                pop = model.islands[island]
                target = None
                for v in pop.variants:
                    if v.id == variant_id:
                        target = v
                        break

                # 变体不存在 → 从事件携带的模板文本重建
                if target is None and variant_template:
                    from prompt_evolver import PromptVariant
                    target = PromptVariant(variant_template, source="event:feedback")
                    target.id = variant_id
                    pop.add(target)

                if target:
                    for dim, score in scores.items():
                        dim_key = dim.upper()
                        count = target.trials_per_dim.get(dim_key, 0)
                        if count > 0:
                            target.scores[dim_key] = (
                                target.scores.get(dim_key, 0) * count + score
                            ) / (count + 1)
                        else:
                            target.scores[dim_key] = float(score)
                        target.trials_per_dim[dim_key] = count + 1
                    target.trials += 1

        elif etype == "mutate":
            island = evt.get("island", "")
            new_variants = evt.get("new_variants", [])
            removed_ids = set(evt.get("removed_variants", []))
            if island in model.islands:
                pop = model.islands[island]
                # Remove old variants
                pop.variants = [v for v in pop.variants if v.id not in removed_ids]
                # Add new ones
                for vd in new_variants:
                    v = PromptVariant.from_dict(vd)
                    pop.add(v)
                gen_counters[island] = gen_counters.get(island, 0) + 1
                pop.generation = gen_counters[island]

        elif etype == "extinction":
            island = evt.get("island", "")
            removed_ids = set(evt.get("removed_variants", []))
            added = evt.get("added_variants", [])
            if island in model.islands:
                pop = model.islands[island]
                pop.variants = [v for v in pop.variants if v.id not in removed_ids]
                for vd in added:
                    v = PromptVariant.from_dict(vd)
                    pop.add(v)

        elif etype == "threshold_adjust":
            # Handled separately — see compile_thresholds
            pass

    return {
        cat: {
            "generation": pop.generation,
            "variants": [v.to_dict() for v in pop.variants],
        }
        for cat, pop in model.islands.items()
    }


def compile_thresholds(genre_id: str) -> dict:
    """从事件日志编译当前阈值（应用最后一次 threshold_adjust 事件）。"""
    events = read_all_events(genre_id)
    # 从默认阈值开始
    default = load_genre_evolution(genre_id, "thresholds.json")
    if not default:
        default = load_evolution("thresholds.json") or {
            "pass": 85, "revise_low": 60, "rewrite_low": 40,
            "max_revision_rounds": 3,
            "weights": {"P1": 0.35, "P2": 0.25, "P3": 0.15, "P4": 0.15, "P5": 0.10},
        }

    current = dict(default)
    for evt in events:
        if evt.get("type") == "threshold_adjust":
            changes = evt.get("changes", {})
            for k, v in changes.items():
                if k == "weights" and isinstance(v, dict):
                    current.setdefault("weights", {}).update(v)
                else:
                    current[k] = v

    return current


def compile_all(genre_id: str) -> dict:
    """编译题材的全部进化数据并写入文件。

    返回: {"population": ..., "thresholds": ..., "event_count": N}
    """
    pop = compile_population(genre_id)
    thresh = compile_thresholds(genre_id)

    save_genre_evolution(genre_id, "prompts/population.json", pop)
    save_genre_evolution(genre_id, "thresholds.json", thresh)

    # 同步 current_best
    best = {}
    for cat, pop_data in pop.items():
        variants = [v for v in pop_data.get("variants", []) if v.get("scores")]
        variants.sort(key=lambda v: sum(v["scores"].values()) / max(len(v["scores"]), 1), reverse=True)
        if len(variants) < 3:
            unscored = [v for v in pop_data.get("variants", []) if not v.get("scores")]
            for v in unscored:
                if v not in variants:
                    variants.append(v)
                if len(variants) >= 3:
                    break
        best[cat] = [v["template"] for v in variants[:3]]

    save_genre_evolution(genre_id, "prompts/current_best.json", best)

    events = read_all_events(genre_id)
    print(f"  ✓ 编译完成: {len(events)} 条事件 → evolution/genres/{genre_id}/")
    print(f"    种群: {sum(len(p['variants']) for p in pop.values())} 变体")
    print(f"    阈值: pass={thresh.get('pass')}")

    return {
        "population": pop,
        "thresholds": thresh,
        "current_best": best,
        "event_count": len(events),
    }


def list_event_summary(genre_id: str) -> list[dict]:
    """列出事件摘要（不展开所有事件细节）。"""
    d = _events_dir(genre_id)
    if not d.exists():
        return []

    summaries = []
    for fpath in sorted(d.glob("*.json")):
        try:
            data = json.loads(fpath.read_text(encoding="utf-8"))
            summaries.append({
                "file": fpath.name,
                "user": data.get("user", "?"),
                "timestamp": data.get("timestamp", ""),
                "event_types": list(set(
                    e.get("type", "?") for e in data.get("events", [])
                )),
                "event_count": len(data.get("events", [])),
            })
        except Exception:
            summaries.append({"file": fpath.name, "error": "parse failed"})

    return summaries


# ── CLI ─────────────────────────────────────────────────

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="事件溯源引擎")
    parser.add_argument("action", choices=["compile", "list", "write"],
                        help="compile=编译全部 | list=列出事件 | write=写入事件")
    parser.add_argument("--genre", "-g", type=str, required=True)
    parser.add_argument("--events", "-e", type=str, default=None,
                        help="JSON 事件列表（write 时使用）")
    args = parser.parse_args()

    if args.action == "compile":
        compile_all(args.genre)

    elif args.action == "list":
        summaries = list_event_summary(args.genre)
        if not summaries:
            print("  暂无事件")
        else:
            for s in summaries:
                print(f"  {s['file']} | {s['user']} | {s['event_count']} events: {s['event_types']}")

    elif args.action == "write":
        if not args.events:
            print("错误: 需要 --events JSON")
            sys.exit(1)
        try:
            events = json.loads(args.events)
        except json.JSONDecodeError:
            print("错误: --events 需要 JSON 数组")
            sys.exit(1)
        fpath = write_events(args.genre, events)
        print(f"  ✓ 写入 {len(events)} 条事件 → {fpath}")