#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Chapter Analyzer - 章节自动解构与模式提取 功能: 1. 章节内容解析(场景分割、人物提取) 2. 张力曲线分析(情绪压强跟踪) 3. 爽点模式识别(打脸、降维、闭环、僭越) 4. 套路结构提取(铺垫→压抑→爆发→反转) 5. 学习成果输出(可复用模式) 用法: python chapter_analyzer.py --chapter-file "正文/第0003章-当众打脸.md" --project-root . --learn """ import re import json import hashlib import sqlite3 from pathlib import Path from dataclasses import dataclass, field, asdict from typing import List, Dict, Any, Optional, Tuple from datetime import datetime from enum import Enum import asyncio try: from runtime_compat import enable_windows_utf8_stdio except ImportError: enable_windows_utf8_stdio = lambda: None class PatternType(Enum): """模式类型""" HOOK = "hook" # 钩子 PACING = "pacing" # 节奏 DIALOGUE = "dialogue" # 对话 PAYOFF = "payoff" # 兑现 EMOTION = "emotion" # 情绪 TENSION_BUILD = "tension_build" # 压强积累 RELEASE = "release" # 释放 TWIST = "twist" # 反转 class CatharsisModel(Enum): """爽感模型""" TABOO_TRANSGRESSION = "taboo_transgression" # 禁忌僭越 OVERKILL_REVERSAL = "overkill_reversal" # 降维打击 COGNITIVE_CLOSURE = "cognitive_closure" # 认知闭环 @dataclass class TensionPoint: """张力点""" position: int # 位置(字符偏移) tension: float # 张力值 0-1 event_type: str # 事件类型 description: str # 描述 @dataclass class HotSpot: """爽点""" position: int span: int # 持续长度 catharsis_type: str # 爽感类型 intensity: float # 强度 0-1 description: str @dataclass class ChapterStructure: """章节结构""" setup: int = 0 # 铺垫段落长度 suppress: int = 0 # 压抑段落长度 release: int = 0 # 释放段落长度 twist: int = 0 # 反转段落长度 @dataclass class LearnedPattern: """学习到的模式""" pattern_id: str pattern_type: str title: str description: str tension_curve: List[List[float]] # [[position, tension], ...] catharsis_model: str structure: Dict[str, int] hot_spots: List[List[Any]] style_tags: List[str] source_project: str source_chapter: int learned_at: str usage_count: int = 0 metadata: Dict[str, Any] = field(default_factory=dict) @dataclass class ChapterAnalysisResult: """章节分析结果""" chapter: int title: str word_count: int scene_count: int character_count: int # 张力分析 tension_curve: List[List[float]] # [[position, tension], ...] tension_avg: float tension_peak: float tension_peak_position: int # 爽点分析 hot_spots: List[HotSpot] catharsis_model: str # 结构分析 structure: ChapterStructure # 模式 detected_patterns: List[str] # 风格标签 style_tags: List[str] # 人物出场 characters: List[str] # 建议 suggestions: List[str] class ChapterAnalyzer: """ 章节分析器 自动解构章节,提取: - 张力曲线 - 爽点位置和类型 - 套路结构 - 可复用模式 """ # 爽点关键词 HOT_SPOT_KEYWORDS = { CatharsisModel.OVERKILL_REVERSAL: [ "碾压", "秒杀", "不堪一击", "一招", "跪下", "颤抖", "惊恐", "脸色大变", "难以置信", "怎么可能", "废物", "蝼蚁", "打脸", "当众", "所有人", "目瞪口呆", "鸦雀无声", "死寂" ], CatharsisModel.TABOO_TRANSGRESSION: [ "禁忌", "僭越", "突破底线", "危险", "禁忌", "邪魅", "诱惑", "堕落", "黑化", "疯狂", "失控", "暴走", "越界", "禁忌" ], CatharsisModel.COGNITIVE_CLOSURE: [ "原来", "竟然", "真相", "伏笔", "恍然大悟", "所有一切", "早该", "早就在", "铺垫", "埋下", "算计", "布局" ] } # 情绪压强关键词 TENSION_KEYWORDS = { # 压强上升 "rise": ["紧张", "危机", "危险", "困境", "难题", "冲突", "对峙", "杀意", "阴谋", "威胁", "危机", "悬念", "未知"], # 压强高峰 "peak": ["爆发", "突破", "反击", "反转", "真相", "高潮", "决战", "绝杀", "逆转", "翻盘", "打脸", "碾压"], # 压强下降 "fall": ["松了口气", "终于", "安心", "胜利", "结束", "平息", "安然"] } def __init__(self, project_root: Optional[Path] = None): self.project_root = Path(project_root) if project_root else Path.cwd() def analyze_chapter( self, chapter_file: Path, chapter_num: Optional[int] = None ) -> ChapterAnalysisResult: """ 分析章节 Args: chapter_file: 章节文件路径 chapter_num: 章节号(从文件名推断或手动指定) Returns: 章节分析结果 """ content = chapter_file.read_text(encoding="utf-8") # 解析章节号 if chapter_num is None: chapter_num = self._extract_chapter_num(chapter_file.name) # 解析标题 title = self._extract_title(chapter_file.name, content) # 统计字数 word_count = len(content) # 场景分割 scenes = self._split_scenes(content) scene_count = len(scenes) # 人物提取 characters = self._extract_characters(content, scenes) # 张力曲线分析 tension_curve = self._analyze_tension_curve(content, scenes) tension_avg = sum(p[1] for p in tension_curve) / len(tension_curve) if tension_curve else 0 peak = max(tension_curve, key=lambda x: x[1]) if tension_curve else (0, 0) tension_peak = peak[1] tension_peak_position = peak[0] # 爽点分析 hot_spots = self._detect_hot_spots(content, scenes) catharsis_model = self._detect_catharsis_model(hot_spots) # 结构分析 structure = self._analyze_structure(content, scenes, hot_spots) # 模式检测 detected_patterns = self._detect_patterns( content, scenes, tension_curve, hot_spots, structure ) # 风格标签 style_tags = self._extract_style_tags(content, tension_curve, hot_spots) # 建议 suggestions = self._generate_suggestions( tension_curve, hot_spots, structure, detected_patterns ) return ChapterAnalysisResult( chapter=chapter_num, title=title, word_count=word_count, scene_count=scene_count, character_count=len(characters), tension_curve=tension_curve, tension_avg=tension_avg, tension_peak=tension_peak, tension_peak_position=tension_peak_position, hot_spots=hot_spots, catharsis_model=catharsis_model, structure=structure, detected_patterns=detected_patterns, style_tags=style_tags, characters=characters, suggestions=suggestions ) def _extract_chapter_num(self, filename: str) -> int: """从文件名提取章节号""" match = re.search(r'第(\d+)[章节]', filename) if match: return int(match.group(1)) return 0 def _extract_title(self, filename: str, content: str) -> str: """提取标题""" # 尝试从文件名提取 match = re.search(r'第\d+章[章节]-?(.+)', filename) if match: return match.group(1).replace('.md', '').strip() # 尝试从内容第一行提取 lines = content.split('\n') for line in lines[:5]: line = line.strip() if line.startswith('#'): return line.lstrip('#').strip() return filename.replace('.md', '') def _split_scenes(self, content: str) -> List[Dict[str, Any]]: """ 分割场景 Returns: List[{"start": int, "end": int, "location": str, "characters": []}] """ scenes = [] lines = content.split('\n') current_scene = { "start": 0, "end": 0, "location": "未知", "characters": [], "lines": [] } location_patterns = [ r'^---+$', # 分割线 r'^【(.+?)】', # 【场景名】 r'^((.+?))', # (场景名) ] for i, line in enumerate(lines): # 检测场景分隔 is_divider = any(re.match(p, line.strip()) for p in location_patterns) if is_divider and current_scene["lines"]: # 保存当前场景 current_scene["end"] = sum(len(l) + 1 for l in current_scene["lines"][:-1]) scenes.append(current_scene) # 新场景 location_match = re.search(r'【(.+?)】|((.+?))', line) current_scene = { "start": sum(len(l) + 1 for l in lines[:i]) + 1, "end": 0, "location": location_match.group(1) if location_match else "未知", "characters": [], "lines": [] } current_scene["lines"].append(line) # 保存最后一个场景 if current_scene["lines"]: current_scene["end"] = len(content) scenes.append(current_scene) return scenes if scenes else [{"start": 0, "end": len(content), "location": "未知", "characters": [], "lines": content.split('\n')}] def _extract_characters( self, content: str, scenes: List[Dict[str, Any]] ) -> List[str]: """提取人物列表""" # 简单实现:提取引号内的对话人 characters = set() # 匹配 "XXX说" 格式 dialogue_pattern = re.compile(r'^"?([^"说]{2,5})"?[说问道喊叫笑骂冷哼]') for scene in scenes: for line in scene["lines"]: match = dialogue_pattern.match(line.strip()) if match: name = match.group(1).strip() if name and len(name) <= 5: characters.add(name) return list(characters) def _analyze_tension_curve( self, content: str, scenes: List[Dict[str, Any]] ) -> List[List[float]]: """ 分析张力曲线 Returns: [[position, tension], ...] - 位置(字数)和张力值(0-1) """ curve = [] total_len = len(content) segment_size = 200 # 每200字一个采样点 content_lower = content # 网文不区分大小写 for pos in range(0, total_len, segment_size): segment = content_lower[pos:pos + segment_size] tension = 0.0 # 检查压强上升关键词 for kw in self.TENSION_KEYWORDS["rise"]: if kw in segment: tension += 0.2 # 检查压强高峰关键词 for kw in self.TENSION_KEYWORDS["peak"]: if kw in segment: tension += 0.4 # 检查压强下降关键词 for kw in self.TENSION_KEYWORDS["fall"]: if kw in segment: tension -= 0.15 # 限制范围 tension = max(0.0, min(1.0, tension)) curve.append([pos, tension]) # 平滑曲线 curve = self._smooth_curve(curve) return curve def _smooth_curve(self, curve: List[List[float]]) -> List[List[float]]: """平滑张力曲线""" if len(curve) < 3: return curve smoothed = [] for i, point in enumerate(curve): if i == 0 or i == len(curve) - 1: smoothed.append(point) else: # 移动平均 avg_pos = (curve[i-1][0] + point[0] + curve[i+1][0]) / 3 avg_tension = (curve[i-1][1] + point[1] + curve[i+1][1]) / 3 smoothed.append([avg_pos, avg_tension]) return smoothed def _detect_hot_spots( self, content: str, scenes: List[Dict[str, Any]] ) -> List[HotSpot]: """检测爽点""" hot_spots = [] for model_type, keywords in self.HOT_SPOT_KEYWORDS.items(): for keyword in keywords: # 查找所有出现位置 start = 0 while True: pos = content.find(keyword, start) if pos == -1: break # 计算附近区域的强度 context_start = max(0, pos - 100) context_end = min(len(content), pos + 100) context = content[context_start:context_end] # 统计上下文中的情绪词数量 intensity = 0.5 # 基础强度 for kw_set in self.HOT_SPOT_KEYWORDS.values(): for kw in kw_set: if kw in context: intensity += 0.1 intensity = min(1.0, intensity) # 避免重叠 span = len(keyword) is_overlap = any( abs(pos - hs.position) < span for hs in hot_spots ) if not is_overlap: hot_spots.append(HotSpot( position=pos, span=span, catharsis_type=model_type.value, intensity=intensity, description=f"发现「{keyword}」" )) start = pos + 1 # 按位置排序 hot_spots.sort(key=lambda x: x.position) return hot_spots def _detect_catharsis_model( self, hot_spots: List[HotSpot] ) -> str: """检测主导爽感模型""" if not hot_spots: return "unknown" model_counts = {} model_intensity = {} for hs in hot_spots: model = hs.catharsis_type model_counts[model] = model_counts.get(model, 0) + 1 model_intensity[model] = model_intensity.get(model, 0) + hs.intensity # 综合评分:出现次数 * 0.4 + 强度 * 0.6 scores = { m: model_counts[m] * 0.4 + model_intensity[m] * 0.6 for m in model_counts } return max(scores, key=scores.get) if scores else "unknown" def _analyze_structure( self, content: str, scenes: List[Dict[str, Any]], hot_spots: List[HotSpot] ) -> ChapterStructure: """分析章节结构""" total_len = len(content) if not hot_spots: # 无爽点:均匀分布 segment = total_len // 4 return ChapterStructure( setup=segment, suppress=segment, release=segment, twist=segment ) # 找第一个爽点位置作为分界 first_hot = hot_spots[0].position last_hot = hot_spots[-1].position # 铺垫:第一个爽点之前 setup = first_hot # 压抑+释放:根据爽点密集程度判断 # 简化:中间区域前40%压抑,后40%释放 middle_start = first_hot middle_end = last_hot + 100 middle_len = middle_end - middle_start suppress = int(middle_len * 0.4) release = int(middle_len * 0.4) twist = total_len - middle_end return ChapterStructure( setup=max(0, setup), suppress=max(0, suppress), release=max(0, release), twist=max(0, twist) ) def _detect_patterns( self, content: str, scenes: List[Dict[str, Any]], tension_curve: List[List[float]], hot_spots: List[HotSpot], structure: ChapterStructure ) -> List[str]: """检测章节中的模式""" patterns = [] # 检测钩子模式 if tension_curve and tension_curve[0][1] > 0.3: patterns.append("开篇高能钩") elif "?" in content[:200] or "!" in content[:200]: patterns.append("悬念钩子") # 检测打脸模式 if any(hs.catharsis_type == "overkill_reversal" for hs in hot_spots): patterns.append("打脸反转") # 检测升级模式 if any(kw in content for kw in ["突破", "晋升", "升级", "进阶"]): patterns.append("境界突破") # 检测装逼模式 if any(kw in content for kw in ["冷笑", "不屑", "蝼蚁", "可笑"]): patterns.append("装逼打脸") # 张力曲线形状检测 if tension_curve: # 持续上升 = 压抑型 tensions = [p[1] for p in tension_curve] if tensions == sorted(tensions) and tensions[-1] - tensions[0] > 0.5: patterns.append("单线压抑") # 波动大 = 节奏快 elif max(tensions) - min(tensions) > 0.6: patterns.append("爽点密集") return patterns def _extract_style_tags( self, content: str, tension_curve: List[List[float]], hot_spots: List[HotSpot] ) -> List[str]: """提取风格标签""" tags = [] # 基于字数 word_count = len(content) if word_count < 2000: tags.append("短小精悍") elif word_count > 5000: tags.append("长篇巨制") # 基于爽点密度 if tension_curve: avg_tension = sum(p[1] for p in tension_curve) / len(tension_curve) if avg_tension > 0.5: tags.append("情绪压强高") elif avg_tension < 0.2: tags.append("节奏舒缓") # 基于爽点类型 if hot_spots: model = self._detect_catharsis_model(hot_spots) if model == "overkill_reversal": tags.append("爽点密集") elif model == "taboo_transgression": tags.append("边缘拉扯") elif model == "cognitive_closure": tags.append("多线收束") # 基于对话比例 dialogue_count = content.count('"') + content.count('"') dialogue_ratio = dialogue_count / max(word_count, 1) if dialogue_ratio > 0.3: tags.append("对话驱动") elif dialogue_ratio < 0.1: tags.append("叙事为主") return tags def _generate_suggestions( self, tension_curve: List[List[float]], hot_spots: List[HotSpot], structure: ChapterStructure, patterns: List[str] ) -> List[str]: """生成改进建议""" suggestions = [] # 张力曲线建议 if tension_curve: avg = sum(p[1] for p in tension_curve) / len(tension_curve) if avg < 0.2: suggestions.append("张力整体偏低,建议增加危机感或悬念") elif avg > 0.7: suggestions.append("张力持续偏高,建议适当释放避免读者疲劳") # 爽点建议 if not hot_spots: suggestions.append("未检测到明显爽点,建议增加打脸/反转情节") # 结构建议 if structure.setup < 500: suggestions.append("铺垫不足,建议增加背景/动机描写") # 节奏建议 if "单线压抑" in patterns: suggestions.append("压抑较长,建议穿插小高潮保持节奏") return suggestions def learn_pattern( self, analysis: ChapterAnalysisResult, layer: str = "project" ) -> LearnedPattern: """ 从分析结果提取可学习模式 Args: analysis: 章节分析结果 layer: 存储层级 ("project" 或 "system") Returns: 学习到的模式 """ pattern_id = hashlib.md5( f"{self.project_root.name}_{analysis.chapter}_{datetime.now().isoformat()}".encode() ).hexdigest()[:16] return LearnedPattern( pattern_id=pattern_id, pattern_type=self._infer_pattern_type(analysis), title=f"第{analysis.chapter}章模式: {analysis.title}", description=self._generate_description(analysis), tension_curve=analysis.tension_curve, catharsis_model=analysis.catharsis_model, structure=asdict(analysis.structure), hot_spots=[[hs.position, hs.description] for hs in analysis.hot_spots], style_tags=analysis.style_tags, source_project=self.project_root.name, source_chapter=analysis.chapter, learned_at=datetime.now().isoformat(), metadata={ "word_count": analysis.word_count, "scene_count": analysis.scene_count, "character_count": analysis.character_count, "detected_patterns": analysis.detected_patterns, "suggestions": analysis.suggestions } ) def _infer_pattern_type(self, analysis: ChapterAnalysisResult) -> str: """推断模式类型""" if "打脸反转" in analysis.detected_patterns: return PatternType.HOOK.value elif "境界突破" in analysis.detected_patterns: return PatternType.PAYOFF.value elif analysis.tension_avg > 0.5: return PatternType.TENSION_BUILD.value else: return PatternType.PACING.value def _generate_description(self, analysis: ChapterAnalysisResult) -> str: """生成模式描述""" parts = [] if analysis.detected_patterns: parts.append(f"包含模式: {', '.join(analysis.detected_patterns[:3])}") if analysis.style_tags: parts.append(f"风格标签: {', '.join(analysis.style_tags[:3])}") parts.append(f"字数: {analysis.word_count},场景: {analysis.scene_count}") if analysis.hot_spots: parts.append(f"检测到 {len(analysis.hot_spots)} 个爽点") return "; ".join(parts) def save_learned_pattern( self, pattern: LearnedPattern, layer: str = "project" ) -> bool: """保存学习到的模式""" try: if layer == "project": return self._save_to_project_db(pattern) else: return self._save_to_system(pattern) except Exception as e: print(f"Error saving pattern: {e}") return False def _init_project_learned_db(self) -> Path: """初始化项目学习库""" db_dir = self.project_root / ".noma" / "rag" db_dir.mkdir(parents=True, exist_ok=True) db_path = db_dir / "learned.db" conn = sqlite3.connect(str(db_path)) cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS learned_patterns ( pattern_id TEXT PRIMARY KEY, pattern_type TEXT NOT NULL, title TEXT NOT NULL, description TEXT, tension_curve TEXT, catharsis_model TEXT, structure TEXT, hot_spots TEXT, style_tags TEXT, source_project TEXT, source_chapter INTEGER, learned_at TEXT, usage_count INTEGER DEFAULT 0, metadata TEXT ) """) conn.commit() conn.close() return db_path def _save_to_project_db(self, pattern: LearnedPattern) -> bool: """保存到项目数据库""" db_path = self._init_project_learned_db() conn = sqlite3.connect(str(db_path)) cursor = conn.cursor() cursor.execute(""" INSERT OR REPLACE INTO learned_patterns (pattern_id, pattern_type, title, description, tension_curve, catharsis_model, structure, hot_spots, style_tags, source_project, source_chapter, learned_at, usage_count, metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( pattern.pattern_id, pattern.pattern_type, pattern.title, pattern.description, json.dumps(pattern.tension_curve), pattern.catharsis_model, json.dumps(pattern.structure), json.dumps(pattern.hot_spots), json.dumps(pattern.style_tags), pattern.source_project, pattern.source_chapter, pattern.learned_at, pattern.usage_count, json.dumps(pattern.metadata) )) conn.commit() conn.close() return True def _save_to_system(self, pattern: LearnedPattern) -> bool: """保存到系统共享库""" # 需要系统根目录 system_root = self._resolve_system_root() learned_dir = system_root / "rag" / "learned" learned_dir.mkdir(parents=True, exist_ok=True) pattern_file = learned_dir / f"{pattern.pattern_id}.json" pattern_data = asdict(pattern) pattern_file.write_text( json.dumps(pattern_data, ensure_ascii=False, indent=2), encoding="utf-8" ) return True def _resolve_system_root(self) -> Path: """解析系统根目录""" # 与 project_root 同级的 .noma sibling = self.project_root.parent / ".noma" if sibling.exists(): return sibling return self.project_root / ".noma" # ==================== CLI 接口 ==================== def main(): import argparse import sys if sys.platform == "win32": enable_windows_utf8_stdio() parser = argparse.ArgumentParser( description="Chapter Analyzer - 章节自动解构与模式提取" ) parser.add_argument("--project-root", type=str, default=".", help="项目根目录") parser.add_argument("--chapter-file", type=str, required=True, help="章节文件路径") parser.add_argument("--chapter-num", type=int, help="章节号(可选,从文件名推断)") parser.add_argument("--learn", action="store_true", help="学习并存储模式") parser.add_argument("--learn-layer", choices=["project", "system"], default="project", help="学习成果存储层级") parser.add_argument("--output", type=str, help="输出文件(JSON格式)") args = parser.parse_args() project_root = Path(args.project_root).resolve() chapter_file = project_root / args.chapter_file if not chapter_file.exists(): print(f"Error: Chapter file not found: {chapter_file}") sys.exit(1) # 分析章节 analyzer = ChapterAnalyzer(project_root) result = analyzer.analyze_chapter(chapter_file, args.chapter_num) # 输出结果 if args.output: output_path = Path(args.output) output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text( json.dumps(asdict(result), ensure_ascii=False, indent=2), encoding="utf-8" ) print(f"Analysis saved to: {output_path}") else: # 打印摘要 print(f"\n{'='*60}") print(f"章节分析报告: 第{result.chapter}章 - {result.title}") print(f"{'='*60}") print(f"\n基本信息:") print(f" 字数: {result.word_count}") print(f" 场景数: {result.scene_count}") print(f" 人物数: {result.character_count}") print(f" 人物列表: {', '.join(result.characters[:5]) or '无'}") print(f"\n张力分析:") print(f" 平均张力: {result.tension_avg:.2f}") print(f" 峰值张力: {result.tension_peak:.2f} (位置: {result.tension_peak_position})") print(f"\n爽点分析:") print(f" 爽点数量: {len(result.hot_spots)}") print(f" 主导模型: {result.catharsis_model}") if result.hot_spots: print(f" 主要爽点:") for hs in result.hot_spots[:3]: print(f" - [{hs.position}] {hs.description} ({hs.intensity:.2f})") print(f"\n结构分析:") print(f" 铺垫: {result.structure.setup}字") print(f" 压抑: {result.structure.suppress}字") print(f" 释放: {result.structure.release}字") print(f" 反转: {result.structure.twist}字") print(f"\n检测到的模式:") for p in result.detected_patterns: print(f" - {p}") print(f"\n风格标签:") for t in result.style_tags: print(f" - {t}") if result.suggestions: print(f"\n改进建议:") for s in result.suggestions: print(f" - {s}") # 学习模式 if args.learn: pattern = analyzer.learn_pattern(result, args.learn_layer) success = analyzer.save_learned_pattern(pattern, args.learn_layer) if success: print(f"\n✓ 模式已学习并存储到 {args.learn_layer} 层") print(f" Pattern ID: {pattern.pattern_id}") else: print(f"\n✗ 模式存储失败") sys.exit(1) if __name__ == "__main__": main()