novel-skill/scripts/feedback_collector.py

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