novel-skill/scripts/seed_extractor.py

#!/usr/bin/env python3
"""
seed_extractor.py — InkOS 题材感知种子提取器

从 InkOS 题材配置文件（YAML frontmatter + Markdown body）提取
题材特定的初始提示词种子，映射到 novel-write 的 5 岛结构。

回退链：指定题材 → other.md → 硬编码通用种子（SEED_TEMPLATES）
"""

import sys, os, re, json, yaml
from pathlib import Path
from typing import Any

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from utils import (
    save_genre_evolution, find_project_genre, map_genre_to_inkos_id,
    resolve_genre_from_project, log_evolution_event,
)

# ── 通用回退种子（当 InkOS 不可用时） ─────────────────

FALLBACK_SEEDS: dict[str, list[str]] = {
    "writing": [
        "核心规则（写作时内化不列出）：Show, don't tell——用细节堆出真实。价值观盐溶于汤——靠行为证明不靠口号。任何行动立于三条腿：过往经历+当前利益+性格底色。配角有自己账本和利益诉求。每章结尾必有小悬念或情绪缺口。全员智商在线。日常场景七成成后续伏笔。关系改变事件驱动。每一行字要么推动剧情要么塑造人物。",
        "看点密集度（硬尺）：每300字至少1个爽点。每500字至少1个钩子。每1000-1500字至少1个完整悬念。连续300字以上只是环境/回忆/议论/心理独白而无推进=水文须删改。叙事段≥40字，目标40-120字。短段(<40字)只允许：开场反转金句/章末钩子/≤3个爆点短段，合计≤5个。",
        "去AI味铁律：叙述者永不替读者下结论。禁止分析报告式语言入正文。转折/惊讶标记词（仿佛/忽然/竟/猛地/不禁/宛如）全篇≤每3000字1次。硬性禁令：全文禁止'不是…而是…'句式；全文禁止破折号'——'。",
    ],
    "revision": [
        "修稿分流规则：PATCHES处理局部文字问题——每个PATCH引用原文段落给出替换文本。REVISED_CONTENT处理全章级问题——Critical包含字数或结构性问题时必须用REVISED_CONTENT。修稿原则：修根因不做表面润色；伏笔状态须与伏笔池同步；不改变剧情走向和核心冲突。",
        "润色边界（硬约束）：只改文字层——句式/段落/排版/用词/五感/对话自然度。禁止增删情节/改变人设/调整主线。文字层硬规约：段落3-5行/段；句式多样化；动词>形容词；每场景1-2种感官。修改后总长变化≤±15%。",
    ],
    "quality_check": [
        "自修正循环（不可跳过门控）：Loop最多5轮——综合评分≥85通过→60-84针对性修订→40-59重写问题段落→<40废弃从细纲重新写。每轮记录本轮评分/发现问题/修复内容/下一轮重点。",
        "五遍编辑评估（按顺序不可跳过）：P1结构(35%)→P2角色(25%)→P3原创性(15%)→P4对话(15%)→P5文本(10%)。综合≥85通过/60-84修订/40-59重写/<40废弃。P1不过不进P2。",
    ],
    "outline": [
        "大纲共创三层决策树：第一层全书结构——分几卷？每卷核心任务？不可逆变化？全书最高潮在哪卷？第二层卷纲细化——高潮事件？需要哪些铺垫（≥3个伏笔）？主角情感弧线？卷末钩子须全书级别。第三层细纲关键节点——前10章每章核心事件/章末钩子类型/爽点类型/伏笔推进计划。",
        "章节备忘七段格式：当前任务→读者此刻在等什么→该兑现的/暂不掀的→日常/过渡承担任务→关键抉择过三连问→章尾必须发生的改变→本章hook账→不要做。",
    ],
    "hook_management": [
        "伏笔全生命周期：plant（种植）→advance（推进）→payoff（兑现）→defer（推迟）→stale（过期）。每章前读钩子账本→列出高优活跃钩子→确认推进/兑现机会。每章后更新：兑现移走→推进追加→新种写入→检查stale。",
        "钩子准入控制：不重复——指向同一事件合并不新增。不超量——每卷最多3-5个high优先级，活跃钩子总数≤12。有计划——无兑现计划的钩子不种。新钩子候选：正文中出现会延续到后续且有具体回收方向的未解问题才新增。",
    ],
}


class SeedExtractor:
    """从 InkOS 题材配置提取 5 岛种子提示词"""

    def __init__(self, inkos_genres_dir: str | None = None):
        if inkos_genres_dir:
            self.genres_dir = Path(inkos_genres_dir)
        else:
            # 内置题材配置，无外部依赖
            self.genres_dir = Path(__file__).resolve().parent.parent / "genres"
        self._profile_cache: dict[str, dict | None] = {}

    # ── 题材配置加载 ─────────────────────────────────

    def _load_profile(self, genre_id: str) -> dict | None:
        """加载单个题材配置文件，解析 YAML frontmatter + body sections"""
        if genre_id in self._profile_cache:
            return self._profile_cache[genre_id]

        fpath = self.genres_dir / f"{genre_id}.md"
        if not fpath.exists():
            self._profile_cache[genre_id] = None
            return None

        text = fpath.read_text(encoding="utf-8")
        fm_match = re.match(r"^---\s*\n([\s\S]*?)\n---\s*\n([\s\S]*)$", text)
        if not fm_match:
            self._profile_cache[genre_id] = None
            return None

        try:
            frontmatter = yaml.safe_load(fm_match.group(1))
        except yaml.YAMLError:
            frontmatter = {}

        body = fm_match.group(2)
        sections = {}
        for sec_match in re.finditer(r"##\s+(.+?)\s*\n([\s\S]*?)(?=\n## |\Z)", body):
            title = sec_match.group(1).strip()
            content = sec_match.group(2).strip()
            sections[title] = content

        profile = {
            "frontmatter": frontmatter or {},
            "sections": sections,
            "language": frontmatter.get("language", "zh") if frontmatter else "zh",
        }
        self._profile_cache[genre_id] = profile
        return profile

    def _load_with_fallback(self, genre_id: str) -> dict:
        """加载题材配置，失败时回退到 other → None"""
        profile = self._load_profile(genre_id)
        if profile:
            return profile
        if genre_id != "other":
            return self._load_profile("other") or {"frontmatter": {}, "sections": {}, "language": "zh"}
        return {"frontmatter": {}, "sections": {}, "language": "zh"}

    # ── 字段→岛屿 映射 ───────────────────────────────

    def _build_writing_seeds(self, fm: dict, sections: dict, lang: str) -> list[str]:
        """从题材配置构建 writing 岛种子"""
        seeds = []

        # 疲劳词 → 词汇疲劳规避规则
        fatigue_words = fm.get("fatigueWords", [])
        if fatigue_words:
            word_list = "、".join(fatigue_words)
            seeds.append(
                f"本文疲劳词监控（题材特定）：以下词汇单章最多使用1次——{word_list}。"
                f"超限即标记为S2警告。这些是{lang == 'en' and 'genre fatigue words' or '题材疲劳词'}，"
                f"过度使用会让读者产生'又是这套'的审美疲劳。"
            )

        # 节奏规则
        pacing = fm.get("pacingRule", "")
        if pacing:
            seeds.append(f"节奏控制规则（题材特定）：{pacing}。偏离此节奏需在章节备忘中说明原因。")

        # 爽点类型
        satisfaction = fm.get("satisfactionTypes", [])
        if satisfaction:
            types_str = "、".join(satisfaction)
            seeds.append(
                f"本文期待感类型优先排序（题材特定）：{types_str}。"
                f"每章至少兑现其中一种类型。爽点设计优先从该清单选取，清单外类型需在细纲中标注'非标爽点'。"
            )

        # 叙事指导 section
        narrative_guidance = (
            sections.get("叙事指导")
            or sections.get("Narrative Guidance")
            or sections.get("Pacing Guidance")
            or ""
        )
        if narrative_guidance:
            # 提取关键句子（以句号分隔，取有实质内容的句子）
            key_lines = [
                s.strip() for s in re.split(r"[。.\n]", narrative_guidance)
                if len(s.strip()) > 10 and not s.strip().startswith("-")
            ][:5]
            if key_lines:
                guidance_text = "。".join(key_lines)
                seeds.append(f"题材叙事指导（注入写作规则）：{guidance_text}。")

        return seeds

    def _build_revision_seeds(self, fm: dict, sections: dict, lang: str) -> list[str]:
        """从题材配置构建 revision 岛种子"""
        seeds = []

        # 语言铁律 / Language Rules
        language_rules = sections.get("语言铁律") or sections.get("Language Rules") or ""
        if language_rules:
            # 提取具体规则行
            rule_lines = [
                line.strip().lstrip("- ").lstrip("* ")
                for line in language_rules.split("\n")
                if line.strip() and (line.strip().startswith("-") or line.strip().startswith("*")
                                     or "→" in line or "✗" in line or "✓" in line)
            ]
            if rule_lines:
                rules_text = "；".join(rule_lines[:8])
                seeds.insert(0, f"题材润色特化规则：{rules_text}。这些规则优先于通用润色规则执行。")

        # 年代考据规则
        if fm.get("eraResearch"):
            seeds.append(
                "年代考据规则（题材特定）：涉及法律/政策/商业规则/物价/机构名/行业术语必须符合设定年代。"
                "金融操作和公司运营需有基本可信度。地名和机构名不可凭空编造。"
            )

        return seeds

    def _build_quality_check_seeds(self, fm: dict, sections: dict, lang: str) -> list[str]:
        """从题材配置构建 quality_check 岛种子"""
        seeds = []

        # 题材禁忌 → S1/S2 检查项
        prohibitions = sections.get("题材禁忌") or sections.get("Genre Prohibitions") or ""
        if prohibitions:
            items = []
            for line in prohibitions.split("\n"):
                stripped = line.strip().lstrip("- ").lstrip("* ")
                if len(stripped) > 10:
                    items.append(stripped)

            if items:
                check_items = []
                for i, item in enumerate(items[:10]):
                    severity = "S1" if any(
                        kw in item for kw in ["禁止", "不可", "杜绝", "never", "must not", "prohibited"]
                    ) else "S2"
                    check_items.append(f"  {i+1}. [{severity}] {item}")
                seeds.append(
                    f"题材特定质量审查清单：\n" + "\n".join(check_items) +
                    f"\n每章完成后逐项检查，S1项不过直接退回修改。"
                )

        # 审计维度
        audit_dims = fm.get("auditDimensions", [])
        if audit_dims:
            dims_str = ",".join(str(d) for d in audit_dims)
            seeds.append(
                f"37维审查启用维度子集（题材特定）：[{dims_str}]。"
                f"未启用的维度本作不适用，跳过检查以降低误报。"
            )

        # 数值系统检查
        if fm.get("numericalSystem"):
            seeds.append(
                "数值一致性检查（题材特定）：统计面板/战力数值/资源计数必须与上下文一致。"
                "同质资源重复获取须写明衰减。增量逐笔列出，不可跳跃式增长。"
            )

        # 战力体系检查
        if fm.get("powerScaling"):
            seeds.append(
                "战力体系检查（题材特定）：境界/品阶/等级突破必须有积累过程。"
                "不可出现无铺垫跳跃式突破。跨大境界需特殊条件（天劫/机缘/丹药）。"
                "能力上限/附加代价/触发条件/成长路径——四维约束缺一不可。"
            )

        return seeds

    def _build_outline_seeds(self, fm: dict, sections: dict, lang: str) -> list[str]:
        """从题材配置构建 outline 岛种子"""
        seeds = []

        # 章节类型
        chapter_types = fm.get("chapterTypes", [])
        if chapter_types:
            types_str = "、".join(chapter_types)
            seeds.append(
                f"章节类型轮换策略（题材特定）：可用类型——{types_str}。"
                f"连续3章以上同类型触发疲劳警告。每卷类型分布应覆盖所有列出类型。"
                f"各类型比例为：战斗/推进类40%，布局/过渡类35%，回收/高潮类25%。"
            )

        # 世界观规则 section
        world_rules = (
            sections.get("修炼规则")
            or sections.get("World Rules")
            or sections.get("年代与现实约束")
            or sections.get("System Design Rules")
            or ""
        )
        if world_rules:
            key_rules = []
            for line in world_rules.split("\n"):
                stripped = line.strip().lstrip("- ").lstrip("* ")
                if len(stripped) > 15:
                    key_rules.append(stripped)
            if key_rules:
                rules_text = "；".join(key_rules[:6])
                seeds.append(f"世界观一致性规则（题材特定，大纲阶段检查）：{rules_text}。")

        return seeds

    def _build_hook_management_seeds(self, fm: dict, sections: dict, lang: str) -> list[str]:
        """从题材配置构建 hook_management 岛种子"""
        seeds = []

        # 题材特定的悬念期待类型
        satisfaction = fm.get("satisfactionTypes", [])
        if satisfaction:
            types_str = "、".join(satisfaction[:4])
            seeds.append(
                f"题材悬念期待管理：本类型读者的核心期待——{types_str}。"
                f"伏笔设计优先围绕这几类期待展开。每卷至少种下2个指向不同类型期待的伏笔。"
            )

        # 节奏规则 → 伏笔兑现节奏
        pacing = fm.get("pacingRule", "")
        if pacing:
            seeds.append(
                f"伏笔兑现节奏约束：{pacing}。伏笔的advance和payoff需与此节奏同步——"
                f"快节奏段兑现密度可提高，慢节奏段以种植和推进为主。"
            )

        # 从世界观/修炼规则提取伏笔管理特殊规则
        world_rules = (
            sections.get("修炼规则")
            or sections.get("World Rules")
            or sections.get("年代与现实约束")
            or ""
        )
        if world_rules:
            key_lines = [
                s.strip().lstrip("- ").lstrip("* ")
                for s in world_rules.split("\n")
                if len(s.strip()) > 15 and any(
                    kw in s for kw in ["规则", "约束", "限制", "禁止", "条件", "代价", "rule", "limit", "cannot", "must"]
                )
            ][:3]
            if key_lines:
                rules_text = "；".join(key_lines)
                seeds.append(
                    f"题材伏笔管理特殊规则：{rules_text}。"
                    f"违反这些规则的伏笔在plant阶段即应被拒绝或标记为'需特殊机制'。"
                )

        return seeds

    # ── 主入口 ──────────────────────────────────────

    def extract_seeds(self, genre_id: str | None = None,
                      project_path: Path | None = None) -> dict[str, list[str]]:
        """
        提取题材特定的 5 岛种子提示词。

        返回: {island_name: [prompt_strings, ...]}
        """
        # 确定题材 ID
        if not genre_id and project_path:
            genre_name = find_project_genre(project_path)
            genre_id = map_genre_to_inkos_id(genre_name)
        elif not genre_id:
            genre_id = "other"

        profile = self._load_with_fallback(genre_id)
        fm = profile.get("frontmatter", {})
        sections = profile.get("sections", {})
        lang = profile.get("language", "zh")

        # 如果 profile 是空的（InkOS 不可用），用回退种子
        if not fm and not sections:
            print(f"  ⚠ InkOS 题材配置不可用，使用通用回退种子")
            return FALLBACK_SEEDS

        seeds = {
            "writing": self._build_writing_seeds(fm, sections, lang),
            "revision": self._build_revision_seeds(fm, sections, lang),
            "quality_check": self._build_quality_check_seeds(fm, sections, lang),
            "outline": self._build_outline_seeds(fm, sections, lang),
            "hook_management": self._build_hook_management_seeds(fm, sections, lang),
        }

        # 对种子过少的岛屿，从 general pool 补充（避免重复）
        general = self._load_profile("other")
        for island, current in seeds.items():
            if len(current) < 2:
                # 先从 general profile 补
                if general and general.get("frontmatter"):
                    gf = general["frontmatter"]
                    gs = general["sections"]
                    gl = general.get("language", "zh")
                    fallback_builder = getattr(self, f"_build_{island}_seeds", None)
                    if fallback_builder:
                        fb_seeds = fallback_builder(gf, gs, gl)
                        for s in fb_seeds:
                            if s not in current:
                                current.append(s)
                                if len(current) >= 2:
                                    break
                # 还不够 → 从硬编码回退种子补
                if len(current) < 2 and island in FALLBACK_SEEDS:
                    for s in FALLBACK_SEEDS[island]:
                        if s not in current:
                            current.append(s)
                            if len(current) >= 2:
                                break

        return seeds

    def initialize_genre(self, genre_id: str) -> dict[str, list[str]]:
        """
        初始化题材级共享进化目录 evolution/genres/{genre_id}/。

        创建 population.json（含题材种子）和相关文件。
        返回提取的种子字典。
        """
        seeds = self.extract_seeds(genre_id=genre_id)

        import hashlib
        population = {}
        for island_name, templates in seeds.items():
            variants = []
            for template in templates:
                variant_id = hashlib.md5(template.encode()).hexdigest()[:12]
                variants.append({
                    "id": variant_id,
                    "template": template,
                    "source": f"seed:{genre_id}",
                    "scores": {},
                    "trials": 0,
                    "trials_per_dim": {},
                    "lineage": [],
                })
            population[island_name] = {
                "generation": 0,
                "variants": variants,
            }

        save_genre_evolution(genre_id, "prompts/population.json", population)
        save_genre_evolution(genre_id, "prompts/current_best.json",
                             {k: [v["template"] for v in pop["variants"][:3]]
                              for k, pop in population.items()})

        default_thresholds = load_evolution("thresholds.json")
        if default_thresholds:
            save_genre_evolution(genre_id, "thresholds.json", default_thresholds)
        else:
            save_genre_evolution(genre_id, "thresholds.json", {
                "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},
            })

        save_genre_evolution(genre_id, "patterns/pattern_library.json", {
            "patterns": [], "meta": {"total_mined": 0}
        })
        save_genre_evolution(genre_id, "session_state.json", {
            "recent_chapters": [], "total_feedback_count": 0
        })

        print(f"  ✓ 题材进化目录已初始化: evolution/genres/{genre_id}/")
        for island, tmpls in seeds.items():
            print(f"    [{island}]: {len(tmpls)} 条种子")

        return seeds


# ── CLI ─────────────────────────────────────────────────

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="InkOS 题材感知种子提取器")
    parser.add_argument("--project", "-p", type=str, default=None,
                        help="项目路径（自动检测题材）")
    parser.add_argument("--genre", "-g", type=str, default=None,
                        help="强制指定题材 ID")
    parser.add_argument("--show", action="store_true",
                        help="仅显示提取的种子，不初始化")
    args = parser.parse_args()

    extractor = SeedExtractor()

    if args.project:
        project = Path(args.project)
        genre_id = args.genre or resolve_genre_from_project(project)
    else:
        genre_id = args.genre or "other"

    if args.show:
        seeds = extractor.extract_seeds(genre_id=genre_id)
        for island, tmpls in seeds.items():
            print(f"\n## {island} ({len(tmpls)} 条)")
            for i, t in enumerate(tmpls):
                print(f"  {i+1}. {t[:150]}...")
    else:
        extractor.initialize_genre(genre_id)