novel-skill/scripts/utils.py

#!/usr/bin/env python3
"""
novel-write 自优化系统 — 共享工具模块
OpenEvolve 启发的进化优化框架

进化输出到 skill 目录下的 evolution/ 文件夹，不修改 SKILL.md。
"""

import json
import re
import os
import sys
import hashlib
import time
from pathlib import Path
from datetime import datetime
from collections import defaultdict
from typing import Any

# ── 路径解析 ───────────────────────────────────────────

def get_skill_dir() -> Path:
    """获取 skill 安装目录（symlink 目标）"""
    # 优先使用环境变量，否则推断
    env = os.environ.get("NOVEL_SKILL_DIR", "")
    if env:
        return Path(env)
    # 从脚本位置推断: scripts/utils.py → 上级 = skill 目录
    return Path(__file__).resolve().parent.parent


def get_evolution_dir() -> Path:
    """获取进化输出目录"""
    return get_skill_dir() / "evolution"


def find_project_root(start: Path | None = None) -> Path | None:
    """向上查找小说项目根目录（包含 项目铁律.md 的目录）"""
    d = Path(start or os.getcwd()).resolve()
    for _ in range(10):
        if (d / "项目铁律.md").exists():
            return d
        if d.parent == d:
            return None
        d = d.parent
    return None


def find_all_projects(base: str = os.path.expanduser("~/Documents")) -> list[Path]:
    """查找 base 下所有小说项目"""
    projects = []
    base = Path(base)
    if not base.exists():
        return projects
    for d in base.iterdir():
        if d.is_dir() and (d / "项目铁律.md").exists():
            projects.append(d)
    return projects


# ── 章节解析 ───────────────────────────────────────────

def parse_chapters(project: Path) -> dict[int, Path]:
    """解析项目中的章节文件，返回 {章号: 路径}"""
    chapters = {}
    body_dir = project / "正文"
    if not body_dir.exists():
        return chapters
    for f in sorted(body_dir.glob("*第*章*.md")):
        m = re.search(r"第(\d+)章", f.name)
        if m:
            chapters[int(m.group(1))] = f
    return chapters


def count_words(text: str) -> int:
    """中文字数统计"""
    chinese = len(re.findall(r"[一-鿿]", text))
    return chinese


def extract_chapter_meta(content: str) -> dict:
    """从章节 markdown 提取元数据"""
    meta = {}
    patterns = {
        "核心事件": r"## 核心事件[：:]\s*(.+?)(?:\n|$)",
        "章末钩子类型": r"章末钩子.*?类型[：:]\s*(.+?)(?:\n|$)",
        "期待感等级": r"期待感等级[：:]\s*(.+?)(?:\n|$)",
        "爽点类型": r"爽点.*?类型[：:]\s*(.+?)(?:\n|$)",
        "爽点模式": r"爽点.*?模式[：:]\s*(.+?)(?:\n|$)",
    }
    for key, pat in patterns.items():
        m = re.search(pat, content)
        if m:
            meta[key] = m.group(1).strip()
    return meta


# ── 质量数据解析 ────────────────────────────────────────

def parse_quality_log(project: Path) -> list[dict]:
    """解析优化日志，返回历史质检记录"""
    log_file = project / "追踪" / "优化日志.md"
    if not log_file.exists():
        return []

    records = []
    text = log_file.read_text(encoding="utf-8")
    blocks = re.split(r"\n## \d{4}-\d{2}-\d{2}", text)
    for block in blocks[1:]:
        record = {}
        m = re.search(r"(?:第\d+章)?综合评分\*{0,2}[：:]\s*(\d+)", block)
        if m:
            record["score"] = int(m.group(1))
        for p in ["P1", "P2", "P3", "P4", "P5"]:
            m = re.search(rf"\*{{0,2}}{p}\*{{0,2}}[：:]\s*(\d+)", block)
            if m:
                record[f"score_{p.lower()}"] = int(m.group(1))
        m = re.search(r"第(\d+)章", block)
        if m:
            record["chapter"] = int(m.group(1))
        record["s1_count"] = len(re.findall(r"S1[：:]", block))
        record["s2_count"] = len(re.findall(r"S2[：:]", block))
        records.append(record)
    return records


def parse_hook_ledger(project: Path) -> dict:
    """解析钩子账本，返回钩子健康数据"""
    ledger = project / "追踪" / "钩子账本.md"
    if not ledger.exists():
        return {}
    text = ledger.read_text(encoding="utf-8")
    return {
        "total_active": len(re.findall(r"状态[：:]\s*(?:活跃|active)", text)),
        "stale": len(re.findall(r"stale|过期", text)),
        "high_priority": len(re.findall(r"优先级[：:]\s*high|高", text)),
        "paid_off": len(re.findall(r"已兑现|payoff", text)),
    }


# ── 特征提取 ───────────────────────────────────────────

def extract_style_features(text: str) -> dict:
    """提取文本的风格特征向量"""
    sentences = re.split(r"[。！？.!?\n]+", text)
    sentences = [s.strip() for s in sentences if s.strip()]
    if not sentences:
        return {}
    lengths = [len(s) for s in sentences]
    dialogue_lines = len(re.findall(r"[「""'].*?[」""']", text))
    return {
        "avg_sentence_len": sum(lengths) / len(lengths),
        "sentence_len_variance": sum((l - sum(lengths) / len(lengths)) ** 2 for l in lengths) / len(lengths),
        "short_sentence_ratio": sum(1 for l in lengths if l <= 5) / len(lengths),
        "long_sentence_ratio": sum(1 for l in lengths if l >= 30) / len(lengths),
        "dialogue_ratio": dialogue_lines / max(len(sentences), 1),
        "paragraph_count": len(re.split(r"\n\n+", text)),
        "total_chinese_chars": count_words(text),
    }


def extract_quality_features(records: list[dict]) -> dict:
    """从质检记录提取质量特征"""
    if not records:
        return {}
    scores = [r.get("score", 0) for r in records if r.get("score")]
    p1_scores = [r.get("score_p1", 0) for r in records if r.get("score_p1")]
    p2_scores = [r.get("score_p2", 0) for r in records if r.get("score_p2")]
    return {
        "avg_score": sum(scores) / len(scores) if scores else 0,
        "min_score": min(scores) if scores else 0,
        "max_score": max(scores) if scores else 0,
        "score_trend": "up" if len(scores) >= 3 and scores[-3:] == sorted(scores[-3:]) else "flat",
        "weakest_pass": "P1" if (sum(p1_scores) / max(len(p1_scores), 1)) < 70 else
                        "P2" if (sum(p2_scores) / max(len(p2_scores), 1)) < 70 else "none",
        "total_s1": sum(r.get("s1_count", 0) for r in records),
        "total_s2": sum(r.get("s2_count", 0) for r in records),
    }


# ── 进化数据持久化 ─────────────────────────────────────

def load_evolution(name: str) -> dict:
    """从 evolution/ 目录加载进化数据"""
    fpath = get_evolution_dir() / name
    if not fpath.exists():
        return {}
    with open(fpath, "r", encoding="utf-8") as f:
        return json.load(f)


def save_evolution(name: str, data: dict) -> None:
    """保存进化数据到 evolution/ 目录"""
    evo_dir = get_evolution_dir()
    evo_dir.mkdir(parents=True, exist_ok=True)
    fpath = evo_dir / name
    with open(fpath, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)


def load_project_db(project: Path) -> dict:
    """加载项目进化数据库"""
    db_file = project / "追踪" / ".evolution_db.json"
    if not db_file.exists():
        return {"prompts": [], "patterns": [], "config_mutations": [], "meta": {}}
    with open(db_file, "r", encoding="utf-8") as f:
        return json.load(f)


def save_project_db(project: Path, db: dict) -> None:
    """保存项目进化数据库"""
    db_file = project / "追踪" / ".evolution_db.json"
    db["meta"]["last_updated"] = datetime.now().isoformat()
    db["meta"]["version"] = db["meta"].get("version", 0) + 1
    with open(db_file, "w", encoding="utf-8") as f:
        json.dump(db, f, ensure_ascii=False, indent=2)


def log_evolution_event(project: Path, event_type: str, data: dict) -> None:
    """记录进化事件到 evolution/evolution_log.md 和项目日志"""
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")
    entry = f"\n## {timestamp} | {event_type}\n\n"
    for k, v in data.items():
        entry += f"- **{k}**: {v}\n"
    entry += "\n---\n"

    # 写到 evolution/ 目录
    evo_log = get_evolution_dir() / "evolution_log.md"
    with open(evo_log, "a", encoding="utf-8") as f:
        f.write(entry)

    # 也写项目日志
    proj_log = project / "追踪" / "优化日志.md"
    proj_log.parent.mkdir(parents=True, exist_ok=True)
    with open(proj_log, "a", encoding="utf-8") as f:
        f.write(entry)


# ── 统计工具 ───────────────────────────────────────────

def compute_improvement(before: list[float], after: list[float]) -> float:
    if not before or not after:
        return 0.0
    avg_before = sum(before) / len(before)
    avg_after = sum(after) / len(after)
    if avg_before == 0:
        return 100.0 if avg_after > 0 else 0.0
    return ((avg_after - avg_before) / avg_before) * 100


# ── 题材级共享进化路径 ─────────────────────────────────

def get_genre_evolution_dir(genre_id: str) -> Path:
    """获取题材级共享进化目录，自动创建"""
    evo_dir = get_evolution_dir() / "genres" / genre_id
    evo_dir.mkdir(parents=True, exist_ok=True)
    return evo_dir


def load_genre_evolution(genre_id: str, name: str) -> dict:
    """从 evolution/genres/{genre_id}/ 加载进化数据"""
    fpath = get_genre_evolution_dir(genre_id) / name
    if not fpath.exists():
        return {}
    with open(fpath, "r", encoding="utf-8") as f:
        return json.load(f)


def save_genre_evolution(genre_id: str, name: str, data: dict) -> None:
    """保存进化数据到 evolution/genres/{genre_id}/"""
    evo_dir = get_genre_evolution_dir(genre_id)
    fpath = evo_dir / name
    fpath.parent.mkdir(parents=True, exist_ok=True)
    with open(fpath, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)


def find_project_genre(project_path: Path) -> str:
    """从 项目铁律.md frontmatter 提取 genre 字段"""
    iron_law = project_path / "项目铁律.md"
    if not iron_law.exists():
        return "other"
    text = iron_law.read_text(encoding="utf-8")
    m = re.search(r'genre[:\s]+["\']?([^"\'\n]+)', text)
    if m:
        return m.group(1).strip()
    m = re.search(r'#\s*(.+?)(?:\n|$)', text)
    return m.group(1).strip() if m else "other"


def find_style_reports(project_path: Path) -> list[Path]:
    """查找对标/ 目录下所有风格报告"""
    reports_dir = project_path / "对标"
    if not reports_dir.exists():
        return []
    return sorted(reports_dir.glob("*/风格报告.md"))


def resolve_genre_from_project(project_path: Path) -> str:
    """从项目路径解析题材 ID，确保题材进化目录已初始化"""
    genre_name = find_project_genre(project_path)
    genre_id = map_genre_to_inkos_id(genre_name)
    return genre_id


def ensure_genre_evolution(genre_id: str) -> bool:
    """确保题材进化目录存在并已初始化；返回 True 表示已存在"""
    evo_dir = get_genre_evolution_dir(genre_id)
    if evo_dir.exists() and (evo_dir / "prompts" / "population.json").exists():
        return True
    from seed_extractor import SeedExtractor
    extractor = SeedExtractor()
    extractor.initialize_genre(genre_id)
    log_evolution_event(Path("."), "题材进化初始化", {
        "题材ID": genre_id,
        "类型": "genre-level evolution init",
    })
    return False


# ── 中文题材→InkOS ID 映射 ─────────────────────────────

GENRE_NAME_TO_ID = {
    # 仙侠 — 修炼/悟道/天劫/因果
    "仙侠": "xianxia", "修仙": "xianxia", "修真": "xianxia",
    # 玄幻 — 异界/战力/血脉
    "玄幻": "xuanhuan",
    # 都市 — 商战/社交/年代考据
    "都市": "urban", "现代": "urban", "都市异能": "urban", "职场": "urban",
    # 恐怖 — 灵异/惊悚/恐惧层级
    "恐怖": "horror", "惊悚": "horror", "灵异": "horror", "悬疑": "horror",
    # 科幻 — 科技/未来/星际
    "科幻": "sci-fi", "星际": "sci-fi", "未来": "sci-fi",
    # 末世/废土 — System Apocalypse 规则并入 other（中文通用回退）
    "末世": "system-apocalypse", "废土": "system-apocalypse", "末日": "system-apocalypse",
    # 无限流/塔爬 — 层级递进
    "无限流": "tower-climber", "无限": "tower-climber", "轮回": "tower-climber",
    # LitRPG/游戏 — 数值系统/面板/属性
    "游戏": "litrpg", "网游": "litrpg", "系统": "litrpg",
    # 奇幻/西幻 — 魔法/种族/世界构建
    "奇幻": "progression", "西幻": "progression", "魔法": "progression",
    # 言情/浪漫奇幻 — 关系驱动
    "言情": "romantasy", "古言": "romantasy", "现言": "romantasy", "恋爱": "romantasy",
    # 异世界/轻小说 — 穿越/转生
    "轻小说": "isekai", "异世界": "isekai", "穿越": "isekai", "转生": "isekai",
    # 种田/日常/温馨 — 慢节奏/生活流
    "种田": "cozy", "日常": "cozy", "温馨": "cozy", "治愈": "cozy",
    # 地下城核心 — 地城建造/管理
    "地下城": "dungeon-core", "地城": "dungeon-core",
    # 东方修真（English Cultivation）
    "东方玄幻": "cultivation", "修真English": "cultivation",
}


def map_genre_to_inkos_id(genre_name: str) -> str:
    """将中文题材名映射到 InkOS 题材 ID"""
    if not genre_name:
        return "other"
    # 精确匹配
    if genre_name.lower() in GENRE_NAME_TO_ID:
        return GENRE_NAME_TO_ID[genre_name.lower()]
    # 子串匹配（处理"种田+诡异"这种复合题材）
    for cn_name, inkos_id in GENRE_NAME_TO_ID.items():
        if cn_name in genre_name:
            return inkos_id
    return "other"


if __name__ == "__main__":
    print(f"Skill 目录: {get_skill_dir()}")
    print(f"进化目录: {get_evolution_dir()}")
    print("utils.py 模块加载正常")