novel-skill/scripts/pattern_mine.py

#!/usr/bin/env python3
"""
pattern_mine.py — 写作模式挖掘器
MAP-Elites 质量-多样性搜索，结果写入 evolution/patterns/。
"""

import sys, os, re, json, random, hashlib
from pathlib import Path
from collections import defaultdict
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, parse_chapters,
    log_evolution_event,
)


class WritingPattern:
    def __init__(self, name: str, category: str, template: str,
                 complexity: float = 0.5, diversity: float = 0.5, quality_score: float = 0.0):
        self.name = name
        self.category = category
        self.template = template
        self.complexity = complexity
        self.diversity = diversity
        self.quality_score = quality_score
        self.usage_count = 0
        self.success_count = 0
        self.fingerprint = hashlib.md5(f"{category}:{template[:100]}".encode()).hexdigest()[:8]

    def to_dict(self) -> dict:
        return {"name": self.name, "category": self.category, "template": self.template,
                "complexity": self.complexity, "diversity": self.diversity,
                "quality_score": self.quality_score, "usage_count": self.usage_count,
                "success_count": self.success_count, "fingerprint": self.fingerprint}

    @classmethod
    def from_dict(cls, d: dict) -> "WritingPattern":
        p = cls(d["name"], d["category"], d["template"],
                d.get("complexity", 0.5), d.get("diversity", 0.5), d.get("quality_score", 0.0))
        p.usage_count = d.get("usage_count", 0)
        p.success_count = d.get("success_count", 0)
        p.fingerprint = d.get("fingerprint", p.fingerprint)
        return p


class PatternExtractor:
    HOOK_PATTERNS = [
        (r"就在这时[,，](.+?)(?:[。！?]|\n)", "危机降临"),
        (r"他[忽然突然].*?发现(.+?)(?:[。！?]|\n)", "信息揭示"),
        (r"(?:但是|然而|可是)(.+?)(?:[。！?]|\n)", "反转"),
        (r'[\"\'""「].+?[\"\'""」]\s*$', "对话悬念"),
        (r"(?:明天|明日|天亮|第二).*?(?:就会|将要|会)(.+?)", "时间压力"),
        (r"(?:他|她).*?(?:决定|选择|下决心)(.+?)", "决定时刻"),
        (r"(?:远处|门外|身后).*?(?:传来|响起)(.+?)", "威胁到来"),
    ]

    PAYOFF_PATTERNS = [
        (r"(?:众人|所有人|在场).*?(?:震惊|愕然|呆住)", "打脸爽"),
        (r"(?:突破|晋升|觉醒|进化|升级).*?(?:成功|完成)", "升级爽"),
        (r"(?:获得|得到|继承|传承|吸收).+?(?:力量|能力|宝物)", "获得爽"),
        (r"(?:守护|保护|挡住|拦下).+?(?:身后|背后)", "守护爽"),
    ]

    @classmethod
    def extract_all(cls, text: str, chapter_num: int, score: float = 0) -> list[WritingPattern]:
        patterns = []
        text_clean = text.replace("\n", " ")
        last_500 = text_clean[-500:]

        for pattern_re, name in cls.HOOK_PATTERNS:
            matches = re.findall(pattern_re, last_500)
            if matches:
                p = WritingPattern(
                    name=f"钩子_{name}", category="hook", template=matches[0][:200],
                    complexity=min(1.0, len(matches[0]) / 80.0),
                    diversity=random.uniform(0.3, 0.9), quality_score=score,
                )
                p.usage_count = 1
                p.success_count = 1 if score >= 75 else 0
                patterns.append(p)

        for pattern_re, name in cls.PAYOFF_PATTERNS:
            matches = re.findall(pattern_re, text_clean)
            if matches:
                p = WritingPattern(
                    name=f"爽点_{name}", category="payoff", template=matches[0][:200],
                    complexity=min(1.0, len(matches) / 5.0),
                    diversity=0.3 + (hash(name) % 70) / 100.0, quality_score=score,
                )
                p.usage_count = len(matches)
                p.success_count = len(matches) if score >= 75 else 0
                patterns.append(p)

        return patterns


class MAPElitesGrid:
    def __init__(self, bins: int = 5):
        self.bins = bins
        self.grid: dict[tuple[int, int], list[WritingPattern]] = defaultdict(list)

    def _bin_coord(self, complexity: float, diversity: float) -> tuple[int, int]:
        c = min(self.bins - 1, int(complexity * self.bins))
        d = min(self.bins - 1, int(diversity * self.bins))
        return (c, d)

    def insert(self, pattern: WritingPattern) -> bool:
        coord = self._bin_coord(pattern.complexity, pattern.diversity)
        for existing in self.grid[coord]:
            if existing.fingerprint == pattern.fingerprint:
                existing.usage_count += pattern.usage_count
                existing.success_count += pattern.success_count
                if pattern.quality_score > existing.quality_score:
                    existing.quality_score = pattern.quality_score
                    existing.template = pattern.template
                return False
        self.grid[coord].append(pattern)
        return True

    def get_elite(self, n: int = 3) -> list[WritingPattern]:
        elites = []
        for cell in self.grid.values():
            if cell:
                cell.sort(key=lambda p: p.quality_score, reverse=True)
                elites.append(cell[0])
        elites.sort(key=lambda p: p.quality_score, reverse=True)
        return elites[:n]

    def cell_count(self) -> int:
        return sum(1 for cell in self.grid.values() if cell)


# ── 主流程 ─────────────────────────────────────────────

def mine_project(project: Path, min_score: int = 70, top_k: int = 20) -> dict:
    print(f"⛏  挖掘项目: {project.name}")
    chapters = parse_chapters(project)
    if not chapters:
        print("  ⚠ 未找到章节文件")
        return {"status": "no_chapters"}

    grid = MAPElitesGrid(bins=5)
    total_patterns = 0
    high_quality = 0

    # 加载项目数据库获取质量评分
    from utils import load_project_db
    db = load_project_db(project)
    quality_scores = db.get("meta", {}).get("quality_scores", {})

    for ch_num, ch_path in sorted(chapters.items()):
        try:
            text = ch_path.read_text(encoding="utf-8")
        except Exception:
            continue
        score = quality_scores.get(str(ch_num), 75)
        patterns = PatternExtractor.extract_all(text, ch_num, score)
        for p in patterns:
            total_patterns += 1
            if grid.insert(p) and p.quality_score >= min_score:
                high_quality += 1

    elites = grid.get_elite(top_k)

    # 合并到 evolution/patterns/pattern_library.json
    existing = load_evolution("patterns/pattern_library.json") or {"patterns": [], "grid_cells": 0}
    all_patterns = {p.get("fingerprint", ""): p for p in existing.get("patterns", [])}
    for p in elites:
        fp = p.fingerprint
        if fp not in all_patterns or p.quality_score > all_patterns[fp].get("quality_score", 0):
            all_patterns[fp] = p.to_dict()

    pattern_list = sorted(all_patterns.values(), key=lambda x: x.get("quality_score", 0), reverse=True)
    # 保留前 50 个
    pattern_list = pattern_list[:50]

    output = {
        "patterns": pattern_list,
        "grid_cells": grid.cell_count(),
        "last_mined_from": project.name,
        "total_chapters": len(chapters),
    }
    save_evolution("patterns/pattern_library.json", output)

    print(f"  ✓ 分析了 {len(chapters)} 章")
    print(f"  ✓ 发现 {total_patterns} 个模式（高质量 {high_quality}）")
    print(f"  ✓ MAP-Elites 网格: {grid.cell_count()}/25 格")
    print(f"  ✓ 模式库保有 {len(pattern_list)} 个模式")
    print(f"  ✓ 已写入 evolution/patterns/pattern_library.json")

    return {"status": "ok", "chapters": len(chapters),
            "patterns_found": total_patterns, "library_size": len(pattern_list)}


def show_patterns(category: str | None = None, top_n: int = 10) -> None:
    data = load_evolution("patterns/pattern_library.json")
    patterns = data.get("patterns", [])
    if category:
        patterns = [p for p in patterns if p.get("category") == category]
    patterns.sort(key=lambda p: p.get("quality_score", 0), reverse=True)

    print(f"\n═══ 写作模式库 (evolution/patterns/) ═══")
    print(f"总计: {len(patterns)} 个模式\n")
    for i, p in enumerate(patterns[:top_n]):
        print(f"{'★' if p.get('quality_score', 0) >= 80 else '●'} "
              f"[{p.get('category', '?')}] {p.get('name', '?')}")
        print(f"  质量: {p.get('quality_score', 0):.0f} | "
              f"complexity={p.get('complexity', 0):.2f} diversity={p.get('diversity', 0):.2f}")
        preview = p.get("template", "")[:80].replace("\n", " ")
        print(f"  {preview}...")
        print()


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="写作模式挖掘器")
    parser.add_argument("action", choices=["mine", "show"],
                        help="mine=挖掘 | show=展示")
    parser.add_argument("--project", "-p", default=".")
    parser.add_argument("--min-score", type=int, default=70)
    parser.add_argument("--top", type=int, default=10)
    parser.add_argument("--category", "-c",
                        choices=["hook", "payoff", "dialogue", "pacing", "description", "emotion"])
    args = parser.parse_args()

    if args.action == "mine":
        root = find_project_root(Path(args.project).resolve())
        if not root:
            print("错误: 未找到小说项目根目录")
            sys.exit(1)
        result = mine_project(root, min_score=args.min_score, top_k=args.top)
        sys.exit(0 if result["status"] == "ok" else 1)
    elif args.action == "show":
        show_patterns(category=args.category, top_n=args.top)