#!/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}")