#!/usr/bin/env python3 """ feedback_collector.py — 结构化反馈收集器 反馈数据写入 evolution/session_state.json 和项目数据库。 """ import sys, os, re, json from pathlib import Path from datetime import datetime from collections import Counter from typing import Any sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from utils import ( find_project_root, get_evolution_dir, load_evolution, save_evolution, load_project_db, save_project_db, log_evolution_event, ) def collect_chapter_feedback(project: Path, chapter_num: int, scores: dict[str, int], prompt_used: str = "", issues: list[str] | None = None, notes: str = "") -> dict: db = load_project_db(project) entry = { "timestamp": datetime.now().isoformat(), "chapter": chapter_num, "scores": scores, "composite": sum(scores.values()) / len(scores) if scores else 0, "prompt_used_hash": hash(prompt_used) % 100000 if prompt_used else None, "prompt_used": prompt_used[:200] if prompt_used else "", "issues": issues or [], "notes": notes, } db.setdefault("feedback", []).append(entry) db.setdefault("meta", {}).setdefault("quality_scores", {})[str(chapter_num)] = entry["composite"] save_project_db(project, db) # 更新会话状态 state = load_evolution("session_state.json") or {} state.setdefault("recent_chapters", []).append({ "chapter": chapter_num, "composite": entry["composite"], "timestamp": entry["timestamp"], }) # 只保留最近 20 章 state["recent_chapters"] = state["recent_chapters"][-20:] state["total_feedback_count"] = state.get("total_feedback_count", 0) + 1 save_evolution("session_state.json", state) # 写入事件日志（Event Sourcing — git 共享无冲突） if prompt_used: _write_feedback_event(project, chapter_num, scores, prompt_used) log_evolution_event(project, "章节反馈收集", { "章节": chapter_num, "综合评分": f"{entry['composite']:.0f}", "各维评分": ", ".join(f"{k}:{v}" for k, v in scores.items()), "问题数": len(issues or []), }) return entry def _write_feedback_event(project: Path, chapter_num: int, scores: dict[str, int], prompt_used: str) -> None: """将反馈写入事件日志（追加写入，天然无 git 冲突）。 不再直接修改 population.json —— 改为写入不可变事件文件。 population.json 通过 `evolve.sh --compile` 从事件编译生成。 """ from hashlib import md5 as _md5 from utils import (find_project_genre, map_genre_to_inkos_id, load_genre_evolution) from event_store import write_events genre_name = find_project_genre(project) genre_id = map_genre_to_inkos_id(genre_name) # 匹配变体：从当前编译后的 population 中查找 pop = load_genre_evolution(genre_id, "prompts/population.json") prompt_hash = _md5(prompt_used.encode()).hexdigest()[:12] prompt_key = prompt_used[:40] if len(prompt_used) > 40 else prompt_used matched_island = "writing" matched_id = prompt_hash if pop: for cat, pop_data in pop.items(): for variant in pop_data.get("variants", []): tmpl = variant.get("template", "") if (variant.get("id") == prompt_hash or (prompt_key and prompt_key in tmpl) or (prompt_used and len(tmpl) >= 40 and tmpl[:40] in prompt_used)): matched_island = cat matched_id = variant.get("id", prompt_hash) break else: continue break event = { "type": "feedback", "chapter": chapter_num, "island": matched_island, "variant_id": matched_id, "variant_template": prompt_used, # 携带模板文本，编译时可重建缺失变体 "scores": {k.upper(): v for k, v in scores.items()}, } write_events(genre_id, [event]) def parse_user_feedback(text: str) -> dict: feedback = {"sentiment": "neutral", "mentioned_dimensions": [], "mentioned_chapters": [], "action_items": []} pos_words = ["好", "不错", "很好", "精彩", "满意", "perfect", "great"] neg_words = ["差", "不好", "弱", "无聊", "平淡", "不行", "重写", "废弃"] pos_count = sum(1 for w in pos_words if w in text) neg_count = sum(1 for w in neg_words if w in text) if pos_count > neg_count: feedback["sentiment"] = "positive" elif neg_count > pos_count: feedback["sentiment"] = "negative" dim_keywords = { "结构": ["结构", "节奏", "场景", "布局"], "角色": ["角色", "人物", "主角", "配角", "人设"], "原创性": ["套路", "新鲜", "创新", "模板"], "对话": ["对话", "对白", "台词"], "文本": ["文字", "措辞", "句式", "描写"], "钩子": ["钩子", "悬念", "伏笔", "期待"], "爽点": ["爽点", "高潮", "打脸", "爽感"], "情感": ["情感", "情绪", "心理"], } for dim, keywords in dim_keywords.items(): if any(kw in text for kw in keywords): feedback["mentioned_dimensions"].append(dim) chapter_refs = re.findall(r"第(\d+)章", text) feedback["mentioned_chapters"] = [int(c) for c in chapter_refs] for m in re.finditer(r"(?:需要|应该|必须|要)\s*(.+?)(?:[。！，,.]|$)", text): feedback["action_items"].append({"type": "required", "content": m.group(1).strip()}) return feedback if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="反馈收集器") parser.add_argument("action", choices=["collect", "parse", "state"]) parser.add_argument("--project", "-p", default=".") parser.add_argument("--chapter", "-c", type=int) parser.add_argument("--scores", "-s", type=str, default="{}") parser.add_argument("--prompt", type=str, default="") parser.add_argument("--issues", "-i", nargs="*", default=[]) parser.add_argument("--notes", "-n", type=str, default="") parser.add_argument("--text", "-t", type=str, default="") args = parser.parse_args() if args.action == "collect": root = find_project_root(Path(args.project).resolve()) if not root: print("错误: 未找到小说项目根目录") sys.exit(1) if not args.chapter: print("错误: 需要 --chapter") sys.exit(1) try: scores = json.loads(args.scores) except json.JSONDecodeError: print("错误: --scores 需要 JSON 格式") sys.exit(1) result = collect_chapter_feedback(root, args.chapter, scores, prompt_used=args.prompt, issues=args.issues, notes=args.notes) print(f"✓ 章节 {args.chapter} 反馈已收集 (综合: {result['composite']:.0f})") elif args.action == "parse": if not args.text: print("错误: 需要 --text") sys.exit(1) result = parse_user_feedback(args.text) print(json.dumps(result, ensure_ascii=False, indent=2)) elif args.action == "state": state = load_evolution("session_state.json") or {} print(json.dumps(state, ensure_ascii=False, indent=2))