你开始写吧。写完了我们再进入调试阶段。
好的，开始写代码。我按之前设计的完整架构，把所有模块一次性写出来，形成一个可以直接运行的项目。
📁 项目结构
text
hetu_novel_writer/
├── core/
│ ├── __init__.py
│ ├── dao_engine.py # π引擎
│ ├── body_selector.py # 肉体选择器
│ ├── soul_injector.py # 灵魂注入器（河图洛书核心）
│ ├── generation_engine.py # 生成引擎
│ └── output_writer.py # 输出器
├── templates/ # 肉体模板库
│ ├── human_level/
│ ├── animal_level/
│ ├── other_level/
│ └── index.json
├── output/ # 生成的小说
├── analyze_novels.py # 解剖工具
├── run_writer.py # 小说写手主程序
└── config.py # 配置文件
代码文件
1. config.py — 配置文件
python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
河图洛书小说写手 · 配置文件
"""
# 路径配置
TEMPLATES_DIR = "./templates"
OUTPUT_DIR = "./output"
CACHE_DIR = "./cache"
# 模板分类
LEVELS = ["human_level", "animal_level", "other_level"]
# 小说生成默认参数
DEFAULT_CHAPTER_COUNT = 16
DEFAULT_ACT_COUNT = 4
# 道引擎（π引擎）参数
PI_CHUNK_SIZE = 10000
# 日志
LOG_LEVEL = "INFO"
2. core/__init__.py
python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
河图洛书小说写手 · 核心模块
"""
from .dao_engine import DaoEngine
from .body_selector import BodySelector
from .soul_injector import SoulInjector
from .generation_engine import GenerationEngine
from .output_writer import OutputWriter
__all__ = [
"DaoEngine",
"BodySelector",
"SoulInjector",
"GenerationEngine",
"OutputWriter"
]
3. core/dao_engine.py — π引擎（道）
python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
道引擎（π引擎）
从π小数位中源源不断取随机数，永不枯竭
"""
import os
import json
import math
from typing import List, Dict, Any
class DaoEngine:
"""π引擎 — 永不枯竭的随机性源泉"""
def __init__(self, chunk_size: int = 10000):
self.chunk_size = chunk_size
self.digits: List[int] = []
self.pointer: int = 0
self._load_next_chunk()
def _load_next_chunk(self) -> None:
"""加载下一批π小数位"""
try:
import gmpy2
gmpy2.get_context().precision = (self.pointer + self.chunk_size + 100) * 4
pi = gmpy2.const_pi()
pi_str = format(pi, f'.{self.pointer + self.chunk_size + 50}f')
pi_digits = pi_str.replace('.', '')
segment = pi_digits[self.pointer:self.pointer + self.chunk_size]
self.digits.extend([int(ch) for ch in segment])
except ImportError:
from decimal import Decimal, getcontext
getcontext().prec = self.pointer + self.chunk_size + 50
pi = Decimal(0)
for k in range(self.pointer + self.chunk_size + 20):
pi += (Decimal(1)/(16**k)) * (
Decimal(4)/(8*k+1) - Decimal(2)/(8*k+4) -
Decimal(1)/(8*k+5) - Decimal(1)/(8*k+6)
)
pi_str = str(pi)[2:]
segment = pi_str[self.pointer:self.pointer + self.chunk_size]
self.digits.extend([int(ch) for ch in segment])
def get_digit(self) -> int:
"""取一个数字（0-9）"""
if self.pointer >= len(self.digits):
self._load_next_chunk()
digit = self.digits[self.pointer]
self.pointer += 1
return digit
def get_digits(self, count: int) -> List[int]:
"""取多个数字"""
return [self.get_digit() for _ in range(count)]
def get_novelty(self, length: int = 8) -> float:
"""取一个随机浮点数（0-1）"""
digits = self.get_digits(length)
value = 0.0
for i, d in enumerate(digits):
value += d * (0.1 ** (i + 1))
return value
def get_int(self, min_val: int, max_val: int) -> int:
"""取一个指定范围内的随机整数"""
if min_val >= max_val:
return min_val
novelty = self.get_novelty(6)
return min_val + int(novelty * (max_val - min_val + 1))
def get_choice(self, items: List[Any]) -> Any:
"""从列表中随机选一项"""
if not items:
return None
idx = self.get_int(0, len(items) - 1)
return items[idx]
def get_pointer(self) -> int:
return self.pointer
def get_state(self) -> Dict[str, Any]:
return {"pointer": self.pointer}
def restore_state(self, state: Dict[str, Any]) -> None:
self.pointer = state.get("pointer", 0)
self.digits = []
self._load_next_chunk()
def save(self, path: str) -> None:
with open(path, 'w', encoding='utf-8') as f:
json.dump(self.get_state(), f, ensure_ascii=False, indent=2)
def load(self, path: str) -> bool:
if not os.path.exists(path):
return False
with open(path, 'r', encoding='utf-8') as f:
state = json.load(f)
self.restore_state(state)
return True
4. core/body_selector.py — 肉体选择器
python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
肉体选择器
从模板库中选择一套肉体模板（M_、Z_、X_、B_、E_）
"""
import os
import json
from typing import Dict, Any, Optional, List
class BodySelector:
"""肉体选择器 — 从模板库中选出一套骨架"""
def __init__(self, templates_dir: str = "./templates"):
self.templates_dir = templates_dir
self.index = self._load_index()
self.templates_cache = {}
def _load_index(self) -> Dict[str, Any]:
"""加载模板索引"""
index_path = os.path.join(self.templates_dir, "index.json")
if not os.path.exists(index_path):
return {"total": 0, "by_level": {"human_level": [], "animal_level": [], "other_level": []}}
try:
with open(index_path, 'r', encoding='utf-8') as f:
return json.load(f)
except:
return {"total": 0, "by_level": {"human_level": [], "animal_level": [], "other_level": []}}
def list_templates(self, level: Optional[str] = None) -> List[str]:
"""列出所有模板，可按级别筛选"""
if level:
return self.index.get("by_level", {}).get(level, [])
return self.index.get("all", [])
def get_template(self, template_id: str) -> Optional[Dict[str, Any]]:
"""根据ID获取模板内容"""
if template_id in self.templates_cache:
return self.templates_cache[template_id]
# 在所有级别中查找
for level in ["human_level", "animal_level", "other_level"]:
path = os.path.join(self.templates_dir, level, f"{template_id}.json")
if os.path.exists(path):
try:
with open(path, 'r', encoding='utf-8') as f:
template = json.load(f)
self.templates_cache[template_id] = template
return template
except:
pass
return None
def select_random(self, level: Optional[str] = None) -> Dict[str, Any]:
"""随机选择一套模板"""
templates = self.list_templates(level)
if not templates:
return self._create_default_template()
# 用π引擎选择
from .dao_engine import DaoEngine
dao = DaoEngine()
idx = dao.get_int(0, len(templates) - 1)
template_id = templates[idx]
template = self.get_template(template_id)
if template:
return template
return self._create_default_template()
def select_by_level(self, level: str = "human_level") -> Dict[str, Any]:
"""按级别选择一套模板（若无则随机）"""
return self.select_random(level)
def _create_default_template(self) -> Dict[str, Any]:
"""创建默认模板（当模板库为空时使用）"""
return {
"source": "default",
"level": "human_level",
"total_chapters": 16,
"macro": {
"act_count": 4,
"act_boundaries": [3, 7, 11, 16],
"emotion_curve": [40, 30, 45, 70, 60, 85, 75, 70, 80, 90, 85, 65, 55, 80, 95, 50],
"conflict_curve": [20, 10, 40, 50, 60, 90, 50, 60, 65, 80, 70, 50, 40, 85, 95, 30]
},
"meso": [
{"chapter": i, "emotion": "pinghe", "conflict_level": 3, "length": "zhong"}
for i in range(1, 17)
],
"micro": {
"avg_sentence_length": 14.7,
"avg_paragraphs_per_chapter": 20,
"avg_words_per_chapter": 2200
},
"breath": {
"comma_ratio": 0.382,
"period_ratio": 0.243,
"quote_ratio": 0.102,
"comma_period_ratio": 1.57
}
}
5. core/soul_injector.py — 灵魂注入器（河图洛书核心）
python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
灵魂注入器（河图洛书核心）
用生克乘侮驱动整个生成过程
五行循环：木→火→土→金→水
无预设，自动迭代
"""
import math
from typing import Dict, Any, List, Optional
class SoulInjector:
"""灵魂注入器 — 河图洛书核心"""
def __init__(self):
# 五行状态
self.wuxing = {
"木": {"phase": 0.0, "value": 0.5},
"火": {"phase": 0.0, "value": 0.5},
"土": {"phase": 0.0, "value": 0.5},
"金": {"phase": 0.0, "value": 0.5},
"水": {"phase": 0.0, "value": 0.5}
}
# 生克关系
self.sheng = {"木": "火", "火": "土", "土": "金", "金": "水", "水": "木"}
self.ke = {"木": "金", "金": "火", "火": "水", "水": "土", "土": "木"}
self.phase_speeds = {
"木": 0.15,
"火": 0.25,
"土": 0.10,
"金": 0.20,
"水": 0.18
}
self.round = 0
self.log = []
def update(self) -> None:
"""更新五行状态（每轮调用一次）"""
self.round += 1
# 更新各行的相位
for element, info in self.wuxing.items():
info["phase"] = (info["phase"] + self.phase_speeds[element]) % (2 * math.pi)
# 计算各行的值（基于相位）
for element, info in self.wuxing.items():
info["value"] = 0.5 + 0.4 * math.sin(info["phase"])
# 生克效应（互惠）
for element in self.wuxing.keys():
# 生：被生者增强
sheng_target = self.sheng[element]
self.wuxing[sheng_target]["value"] = min(1.0,
self.wuxing[sheng_target]["value"] + 0.05 * self.wuxing[element]["value"])
# 克：被克者减弱
ke_target = self.ke[element]
self.wuxing[ke_target]["value"] = max(0.0,
self.wuxing[ke_target]["value"] - 0.03 * self.wuxing[element]["value"])
# 归一化，保持平衡
total = sum(info["value"] for info in self.wuxing.values())
if total > 0:
for element in self.wuxing.keys():
self.wuxing[element]["value"] = self.wuxing[element]["value"] / total
def get_element_value(self, element: str) -> float:
"""获取某行的当前值"""
return self.wuxing.get(element, {}).get("value", 0.5)
def get_dominant(self) -> str:
"""获取当前最活跃的五行元素"""
dominant = max(self.wuxing.items(), key=lambda x: x[1]["value"])
return dominant[0]
def get_all_values(self) -> Dict[str, float]:
"""获取所有行的当前值"""
return {e: info["value"] for e, info in self.wuxing.items()}
def apply_to_generation(self, step: str) -> Dict[str, Any]:
"""
将当前五行状态应用到生成步骤
返回调节参数
"""
self.update()
return {
"dominant": self.get_dominant(),
"values": self.get_all_values(),
"round": self.round,
"sheng_ratio": 0.5 + 0.3 * self.wuxing["木"]["value"],
"ke_ratio": 0.5 + 0.3 * self.wuxing["金"]["value"],
"tension": self.wuxing["火"]["value"],
"stability": self.wuxing["土"]["value"],
"fluidity": self.wuxing["水"]["value"]
}
def get_state(self) -> Dict[str, Any]:
return {
"round": self.round,
"wuxing": self.wuxing,
"log": self.log[-20:] # 只保留最近20条
}
def restore_state(self, state: Dict[str, Any]) -> None:
self.round = state.get("round", 0)
self.wuxing = state.get("wuxing", {
"木": {"phase": 0.0, "value": 0.5},
"火": {"phase": 0.0, "value": 0.5},
"土": {"phase": 0.0, "value": 0.5},
"金": {"phase": 0.0, "value": 0.5},
"水": {"phase": 0.0, "value": 0.5}
})
self.log = state.get("log", [])
6. core/generation_engine.py — 生成引擎
python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
生成引擎
逐层生成小说结构：
M_（大骨架）→ Z_（中骨架）→ X_（小骨架）→ B_（呼吸）→ E_（情绪）
"""
import json
import math
from typing import Dict, Any, List, Optional
from .dao_engine import DaoEngine
from .soul_injector import SoulInjector
class GenerationEngine:
"""生成引擎 — 逐层生成小说骨架"""
def __init__(self):
self.dao = DaoEngine()
self.soul = SoulInjector()
self.template = None
self.result = {}
def load_template(self, template: Dict[str, Any]) -> None:
"""加载肉体模板"""
self.template = template
def generate(self) -> Dict[str, Any]:
"""执行完整生成流程"""
if not self.template:
raise ValueError("请先加载模板")
# 第一步：生成大骨架 M_
m_data = self._generate_macro()
# 第二步：生成中骨架 Z_
z_data = self._generate_meso(m_data)
# 第三步：生成小骨架 X_
x_data = self._generate_micro(m_data, z_data)
# 第四步：生成呼吸数据 B_
b_data = self._generate_breath(m_data, z_data)
# 第五步：生成情绪数据 E_
e_data = self._generate_emotion(m_data, z_data)
self.result = {
"M_": m_data,
"Z_": z_data,
"X_": x_data,
"B_": b_data,
"E_": e_data
}
return self.result
def _generate_macro(self) -> Dict[str, Any]:
"""生成大骨架"""
total_chapters = self.template.get("total_chapters", 16)
act_count = self.template.get("macro", {}).get("act_count", 4)
# 从模板获取基础曲线，然后加入随机偏移
base_emotion = self.template.get("macro", {}).get("emotion_curve", None)
base_conflict = self.template.get("macro", {}).get("conflict_curve", None)
if base_emotion and len(base_emotion) >= total_chapters:
emotion_curve = base_emotion[:total_chapters]
else:
emotion_curve = self._generate_curve(total_chapters, 30, 90)
if base_conflict and len(base_conflict) >= total_chapters:
conflict_curve = base_conflict[:total_chapters]
else:
conflict_curve = self._generate_curve(total_chapters, 10, 95)
# 加入随机偏移（道的扰动）
for i in range(total_chapters):
shift = (self.dao.get_novelty(4) - 0.5) * 10
emotion_curve = max(10, min(100, emotion_curve + shift))
shift2 = (self.dao.get_novelty(4) - 0.5) * 10
conflict_curve = max(5, min(100, conflict_curve + shift2))
# 确定幕边界
if self.template.get("macro", {}).get("act_boundaries"):
boundaries = self.template["macro"]["act_boundaries"]
else:
boundaries = self._generate_boundaries(total_chapters, act_count)
return {
"total_chapters": total_chapters,
"act_count": act_count,
"act_boundaries": boundaries,
"emotion_curve": [int(v) for v in emotion_curve],
"conflict_curve": [int(v) for v in conflict_curve],
"soul_state": self.soul.get_all_values()
}
def _generate_curve(self, length: int, min_val: int, max_val: int) -> List[float]:
"""生成一条起伏曲线"""
curve = []
for i in range(length):
# 正弦波 + 随机扰动
base = 50 + 30 * math.sin(i / length * 2 * math.pi * 2)
noise = (self.dao.get_novelty(4) - 0.5) * 40
val = base + noise
curve.append(max(min_val, min(max_val, val)))
return curve
def _generate_boundaries(self, total: int, act_count: int) -> List[int]:
"""生成幕边界"""
if act_count <= 1:
return [total]
boundaries = []
for i in range(1, act_count):
# 均匀分布 + 随机偏移
base = int(total * i / act_count)
shift = self.dao.get_int(-2, 2)
pos = max(1, min(total - 1, base + shift))
boundaries.append(pos)
boundaries.sort()
# 去重
unique = []
for b in boundaries:
if b not in unique:
unique.append(b)
if not unique or unique[-1] != total:
unique.append(total)
return unique
def _generate_meso(self, m_data: Dict[str, Any]) -> List[Dict[str, Any]]:
"""生成中骨架"""
total = m_data["total_chapters"]
emotion_curve = m_data["emotion_curve"]
conflict_curve = m_data["conflict_curve"]
# 情绪标签映射（基于数值区间）
emotion_labels = []
for val in emotion_curve:
if val >= 80:
labels = ["jinzhang", "jingxian", "beican"]
elif val >= 60:
labels = ["pinghe", "lianmin", "zhuanzhe"]
elif val >= 40:
labels = ["yayi", "yuanwang", "konghuang"]
else:
labels = ["pingjing", "huangmiu"]
emotion_labels.append(self.dao.get_choice(labels))
meso = []
for i in range(total):
conflict = conflict_curve
if conflict >= 80:
level = 5
elif conflict >= 60:
level = 4
elif conflict >= 40:
level = 3
elif conflict >= 20:
level = 2
else:
level = 1
# 篇幅由冲突和情绪共同决定
if level >= 4:
length = "chang"
elif level <= 2:
length = "duan"
else:
length = "zhong"
meso.append({
"chapter": i + 1,
"emotion": emotion_labels,
"conflict_level": level,
"length": length
})
return meso
def _generate_micro(self, m_data: Dict[str, Any], z_data: List[Dict[str, Any]]) -> Dict[str, Any]:
"""生成小骨架"""
total = m_data["total_chapters"]
# 从模板获取基础值
base_avg_sentence = self.template.get("micro", {}).get("avg_sentence_length", 14.7)
base_avg_paragraphs = self.template.get("micro", {}).get("avg_paragraphs_per_chapter", 20)
base_avg_words = self.template.get("micro", {}).get("avg_words_per_chapter", 2200)
# 加入随机偏移
avg_sentence = base_avg_sentence + (self.dao.get_novelty(4) - 0.5) * 4
avg_paragraphs = base_avg_paragraphs + self.dao.get_int(-3, 3)
avg_words = base_avg_words + self.dao.get_int(-300, 300)
# 按章节生成细节
chapter_details = []
for i in range(total):
# 根据冲突等级调整句长
conflict = z_data["conflict_level"]
if conflict >= 4:
sentence_len = avg_sentence * 0.8
elif conflict <= 2:
sentence_len = avg_sentence * 1.2
else:
sentence_len = avg_sentence
# 根据情绪调整段落数
emotion = z_data["emotion"]
if emotion in ["jingxian", "beican", "jinzhang"]:
paragraphs = avg_paragraphs * 0.8
elif emotion in ["pinghe", "yuanman"]:
paragraphs = avg_paragraphs * 1.2
else:
paragraphs = avg_paragraphs
chapter_details.append({
"chapter": i + 1,
"sentence_length": int(sentence_len),
"paragraphs": int(paragraphs),
"words": int(avg_words * (0.8 + 0.4 * self.dao.get_novelty(4)))
})
return {
"avg_sentence_length": avg_sentence,
"avg_paragraphs_per_chapter": avg_paragraphs,
"avg_words_per_chapter": avg_words,
"chapter_details": chapter_details
}
def _generate_breath(self, m_data: Dict[str, Any], z_data: List[Dict[str, Any]]) -> Dict[str, Any]:
"""生成呼吸数据"""
# 从模板获取基础呼吸值
base_comma = self.template.get("breath", {}).get("comma_ratio", 0.38)
base_period = self.template.get("breath", {}).get("period_ratio", 0.24)
base_quote = self.template.get("breath", {}).get("quote_ratio", 0.10)
# 加入随机偏移
comma_ratio = base_comma + (self.dao.get_novelty(4) - 0.5) * 0.05
period_ratio = base_period + (self.dao.get_novelty(4) - 0.5) * 0.05
quote_ratio = base_quote + (self.dao.get_novelty(4) - 0.5) * 0.03
# 归一化
total = comma_ratio + period_ratio + quote_ratio + 0.3
comma_ratio = comma_ratio / total
period_ratio = period_ratio / total
quote_ratio = quote_ratio / total
# 按章节生成呼吸细节
chapter_breath = []
for i, z in enumerate(z_data):
conflict = z["conflict_level"]
# 冲突高时，句号增多（呼吸急促）
period_offset = 0.05 * (conflict - 3)
# 对话多时，引号增多
quote_offset = 0.03 * (1 if z["emotion"] in ["jinzhang", "jingxian"] else 0)
cb = {
"chapter": i + 1,
"comma_ratio": max(0.2, min(0.5, comma_ratio - period_offset * 0.3)),
"period_ratio": max(0.15, min(0.35, period_ratio + period_offset * 0.5)),
"quote_ratio": max(0.05, min(0.18, quote_ratio + quote_offset))
}
# 计算句逗比
cb["comma_period_ratio"] = cb["comma_ratio"] / (cb["period_ratio"] + 0.01)
chapter_breath.append(cb)
return {
"comma_ratio": comma_ratio,
"period_ratio": period_ratio,
"quote_ratio": quote_ratio,
"comma_period_ratio": comma_ratio / (period_ratio + 0.01),
"chapter_details": chapter_breath
}
def _generate_emotion(self, m_data: Dict[str, Any], z_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""生成情绪数据"""
total = m_data["total_chapters"]
emotion_curve = m_data["emotion_curve"]
# 情绪编码映射
emotion_code_map = {
"pingjing": "PJ", "pinghe": "PH", "yuyue": "YY",
"lianmin": "LM", "yayi": "YA", "jinzhang": "JZ",
"beifen": "BF", "jingxian": "JX", "beican": "BC",
"yuanwang": "YW", "konghuang": "KH", "zhuanzhe": "ZZ",
"huangmiu": "HM", "yuanman": "YM"
}
emotion_data = []
for i in range(total):
val = emotion_curve
name = z_data["emotion"]
code = emotion_code_map.get(name, "QT")
# 用灵魂注入器获取当前情绪调节
soul_state = self.soul.apply_to_generation(f"emotion_{i}")
emotion_data.append({
"chapter": i + 1,
"name": name,
"code": code,
"value": val,
"soul_influence": {
"dominant": soul_state["dominant"],
"tension": soul_state["tension"]
}
})
return emotion_data
7. core/output_writer.py — 输出器
python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
输出器
把结构数据转换为完整小说文本
"""
import os
import json
import random
from datetime import datetime
from typing import Dict, Any, List
from .dao_engine import DaoEngine
from .soul_injector import SoulInjector
class OutputWriter:
"""输出器 — 生成完整小说文本"""
def __init__(self, output_dir: str = "./output"):
self.output_dir = output_dir
os.makedirs(output_dir, exist_ok=True)
self.dao = DaoEngine()
self.soul = SoulInjector()
# 基础词库（用于填充）
self.phrase_pool = {
"pinghe": ["风清月朗", "云淡天高", "春和景明", "秋高气爽"],
"yayi": ["阴云密布", "寒风萧瑟", "愁云惨淡", "暮色苍茫"],
"lianmin": ["楚楚可怜", "黯然神伤", "凄然泪下", "柔肠寸断"],
"jinzhang": ["剑拔弩张", "风声鹤唳", "步步惊心", "千钧一发"],
"jingxian": ["刀光剑影", "险象环生", "九死一生", "命悬一线"],
"beican": ["肝肠寸断", "魂飞魄散", "惨不忍睹", "哀鸿遍野"],
"yuanwang": ["有口难辩", "含冤莫白", "百口莫辩", "屈打成招"],
"konghuang": ["惊慌失措", "魂不附体", "胆战心惊", "惶惶不可终日"],
"zhuanzhe": ["峰回路转", "柳暗花明", "绝处逢生", "拨云见日"],
"huangmiu": ["啼笑皆非", "荒唐可笑", "岂有此理", "莫名其妙"],
"yuanman": ["皆大欢喜", "花好月圆", "天从人愿", "称心如意"]
}
self.opening_poems = [
"世事如棋局局新，人生何处不逢春。",
"青山依旧在，几度夕阳红。",
"浮云流水两茫茫，往事如烟梦一场。",
"人生若只如初见，何事秋风悲画扇。",
"江流天地外，山色有无中。"
]
self.closing_phrases = [
"未知后事如何，且听下回分解。",
"欲知后事如何，且听下回分解。",
"要知端的，下回分解。"
]
def write(self, structure: Dict[str, Any], title: str = None) -> Dict[str, Any]:
"""
根据结构数据生成完整小说
"""
if not title:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
title = f"河图洛书小说_{timestamp}"
m_data = structure.get("M_", {})
z_data = structure.get("Z_", [])
e_data = structure.get("E_", [])
total_chapters = m_data.get("total_chapters", 16)
# 生成每一回
chapters = []
for i in range(total_chapters):
chapter_num = i + 1
chapter_data = z_data if i < len(z_data) else {"emotion": "pinghe", "conflict_level": 3}
emotion_name = chapter_data.get("emotion", "pinghe")
conflict_level = chapter_data.get("conflict_level", 3)
# 生成回目
title_pair = self._generate_chapter_title(emotion_name, conflict_level)
# 生成开篇诗词
poem = self.dao.get_choice(self.opening_poems)
# 生成正文（根据情绪和冲突级别）
body = self._generate_body(emotion_name, conflict_level, chapter_num)
# 生成结尾
closing = self.dao.get_choice(self.closing_phrases)
chapter_text = f"""第{chapter_num}回
{title_pair}
{poem}
{body}
{closing}
"""
chapters.append(chapter_text)
# 合并全书
full_text = f"【{title}】\n\n" + "\n\n".join(chapters)
# 保存
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
txt_path = os.path.join(self.output_dir, f"{title}_{timestamp}.txt")
json_path = os.path.join(self.output_dir, f"{title}_{timestamp}_structure.json")
with open(txt_path, 'w', encoding='utf-8') as f:
f.write(full_text)
with open(json_path, 'w', encoding='utf-8') as f:
json.dump(structure, f, ensure_ascii=False, indent=2)
return {
"title": title,
"txt_path": txt_path,
"json_path": json_path,
"total_chapters": total_chapters,
"word_count": len(full_text)
}
def _generate_chapter_title(self, emotion: str, conflict_level: int) -> str:
"""生成回目"""
prefix = {
"pinghe": ("静观云起", "闲看花落"),
"yayi": ("愁云暗结", "忧思难遣"),
"lianmin": ("怜香惜玉", "悲天悯人"),
"jinzhang": ("风云突变", "剑拔弩张"),
"jingxian": ("险象环生", "绝境逢生"),
"beican": ("天崩地裂", "肝肠寸断"),
"yuanwang": ("含冤负屈", "有口难言"),
"konghuang": ("惊魂未定", "风声鹤唳"),
"zhuanzhe": ("峰回路转", "柳暗花明"),
"huangmiu": ("荒唐可笑", "乾坤颠倒"),
"yuanman": ("花好月圆", "云开月朗")
}.get(emotion, ("世事无常", "人间有梦"))
# 加入随机组合
if self.dao.get_novelty(3) > 0.5:
return f"{self.dao.get_choice(prefix)}{self.dao.get_choice(['风云起', '日月新', '乾坤转', '天地惊'])}"
else:
return f"{self.dao.get_choice(prefix)}，{self.dao.get_choice(prefix)}"
def _generate_body(self, emotion: str, conflict_level: int, chapter_num: int) -> str:
"""生成正文"""
# 根据情绪选择词库
phrases = self.phrase_pool.get(emotion, self.phrase_pool["pinghe"])
# 生成3-5个段落
num_paragraphs = 3 + self.dao.get_int(0, 2)
paragraphs = []
for p in range(num_paragraphs):
# 每个段落2-4个句子
num_sentences = 2 + self.dao.get_int(0, 2)
sentences = []
for s in range(num_sentences):
phrase = self.dao.get_choice(phrases)
# 根据冲突级别决定句子长度
if conflict_level >= 4:
length = "短"
sentence = f"{phrase}，{self.dao.get_choice(['形势紧迫', '危机四伏', '事不宜迟'])}。"
elif conflict_level <= 2:
length = "长"
sentence = f"{phrase}，{self.dao.get_choice(['岁月静好', '世事如常', '人间有味是清欢'])}。"
else:
sentence = f"{phrase}，{self.dao.get_choice(['事已至此', '无可奈何', '顺其自然'])}。"
sentences.append(sentence)
paragraphs.append("".join(sentences))
# 加入叙事过渡
transitions = ["却说此时", "且说", "原来这日", "不期", "正是"]
if chapter_num > 1:
transition = self.dao.get_choice(transitions)
paragraphs[0] = f"{transition}，" + paragraphs[0]
return "\n\n".join(paragraphs)
8. run_writer.py — 主程序入口
python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
河图洛书小说写手 · 主程序
用法：
python run_writer.py # 随机生成一部小说
python run_writer.py --level human # 使用人类级模板
python run_writer.py --level animal # 使用动物级模板
python run_writer.py --list # 列出所有模板
"""
import os
import sys
import json
import argparse
from datetime import datetime
from core import DaoEngine, BodySelector, SoulInjector, GenerationEngine, OutputWriter
def main():
parser = argparse.ArgumentParser(description="河图洛书小说写手")
parser.add_argument("--level", "-l", choices=["human", "animal", "other"],
help="选择模板级别")
parser.add_argument("--list", action="store_true",
help="列出所有可用模板")
parser.add_argument("--title", "-t", type=str, default=None,
help="指定小说标题")
parser.add_argument("--chapters", "-c", type=int, default=16,
help="指定章回数（默认16）")
args = parser.parse_args()
print("""
╔══════════════════════════════════════════════════════════╗
║ 河图洛书小说写手 v1.0 ║
║ ☯ 道在π中 · 万物自己演化 ║
╚══════════════════════════════════════════════════════════╝
""")
# 初始化各模块
dao = DaoEngine()
selector = BodySelector()
soul = SoulInjector()
engine = GenerationEngine()
writer = OutputWriter()
# 列出模板
if args.list:
print("📚 可用模板：")
for level in ["human_level", "animal_level", "other_level"]:
templates = selector.list_templates(level)
print(f" {level}: {len(templates)} 个")
for t in templates[:5]:
print(f" - {t}")
if len(templates) > 5:
print(f" ... 还有 {len(templates) - 5} 个")
return
# 选择模板
level_map = {
"human": "human_level",
"animal": "animal_level",
"other": "other_level"
}
level = level_map.get(args.level) if args.level else None
print(f"🔍 正在选择模板...")
template = selector.select_random(level)
print(f" ✅ 已选模板: {template.get('source', 'default')}")
print(f" 📊 级别: {template.get('level', 'unknown')}")
print(f" 📖 章回数: {template.get('total_chapters', 16)}")
# 如果用户指定了章回数，覆盖模板设置
if args.chapters != 16:
template["total_chapters"] = args.chapters
print(f" 📝 已覆盖为: {args.chapters} 回")
# 加载模板到生成引擎
engine.load_template(template)
# 生成结构
print(f"\n🌀 开始生成结构...")
structure = engine.generate()
print(f" ✅ 大骨架 (M_) 生成完成")
print(f" ✅ 中骨架 (Z_) 生成完成")
print(f" ✅ 小骨架 (X_) 生成完成")
print(f" ✅ 呼吸数据 (B_) 生成完成")
print(f" ✅ 情绪数据 (E_) 生成完成")
# 输出小说
print(f"\n✍️ 正在生成小说文本...")
result = writer.write(structure, args.title)
print(f"\n✅ 小说生成完成！")
print(f" 📄 标题: {result['title']}")
print(f" 📖 章回: {result['total_chapters']}")
print(f" 📝 字数: {result['word_count']}")
print(f" 💾 保存至: {result['txt_path']}")
print(f" 💾 结构数据: {result['json_path']}")
# 显示五行状态
soul_state = soul.get_all_values()
print(f"\n☯ 当前五行状态:")
for element, value in soul_state.items():
bar = "█" * int(value * 20)
print(f" {element}: {bar} {value:.2f}")
print(f"\n✨ 道驱动一切，万物自己演化。")
if __name__ == "__main__":
main()
9. analyze_novels.py — 小说解剖工具
python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
小说解剖工具（analyze_novels.py）
扫描文件夹内所有 .txt 小说，提取结构数据：
M_（大骨架）、Z_（中骨架）、X_（小骨架）、B_（呼吸）、E_（情绪）
生成模板库，供小说写手使用。
"""
import os
import sys
import json
import re
from pathlib import Path
from collections import Counter
from typing import List, Dict, Any, Optional
# 情绪词库（用于自动标注）
EMOTION_WORDS = {
"pinghe": ["平和", "安然", "从容", "恬淡"],
"yayi": ["压抑", "沉闷", "抑郁", "忧愁"],
"lianmin": ["怜悯", "同情", "怜惜", "恻隐"],
"jinzhang": ["紧张", "急迫", "悬疑", "不安"],
"jingxian": ["惊险", "危急", "生死", "恐怖"],
"beican": ["悲惨", "凄凉", "哀痛", "绝望"],
"yuanwang": ["冤枉", "委屈", "误解", "含冤"],
"konghuang": ["恐慌", "惊惶", "畏惧", "慌乱"],
"zhuanzhe": ["转折", "突变", "转机", "意外"],
"huangmiu": ["荒唐", "荒谬", "可笑", "离奇"],
"yuanman": ["圆满", "如意", "完满", "喜庆"],
"pingjing": ["平静", "宁静", "安稳", "沉寂"]
}
# 冲突词库
CONFLICT_WORDS = ["杀", "死", "战", "败", "伤", "劫", "破", "斩", "逃", "降",
"困", "围", "攻", "击", "斗", "争", "怒", "恨", "仇", "冤"]
# 标点符号集合
PUNCTUATION = {
"comma": ["，", ","],
"period": ["。", "．", "."],
"quote": ['"', "'", "「", "」", "『", "』", "“", "”"],
"dunhao": ["、"],
"question": ["？", "?"],
"exclamation": ["！", "!"],
"semicolon": ["；", ";"],
"colon": ["：", ":"],
"ellipsis": ["…", "……"],
"dash": ["—", "——"]
}
class NovelAnalyzer:
"""小说解剖器"""
def __init__(self, input_dir: str, output_dir: str = "./templates"):
self.input_dir = Path(input_dir)
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.index = {
"total": 0,
"by_level": {
"human_level": [],
"animal_level": [],
"other_level": []
}
}
self.results = []
def analyze_all(self) -> None:
"""解剖所有小说"""
files = list(self.input_dir.glob("*.txt"))
print(f"📂 发现 {len(files)} 个小说文件")
for i, file_path in enumerate(files, 1):
print(f"\n🔪 [{i}/{len(files)}] 解剖: {file_path.name}")
try:
result = self.analyze_one(file_path)
if result:
self.results.append(result)
self.index["total"] += 1
level = result.get("level", "other_level")
self.index["by_level"][level].append(result["id"])
print(f" ✅ 完成 | 级别: {level} | 章回: {result.get('total_chapters', 0)}")
except Exception as e:
print(f" ❌ 失败: {e}")
# 保存索引
index_path = self.output_dir / "index.json"
with open(index_path, 'w', encoding='utf-8') as f:
json.dump(self.index, f, ensure_ascii=False, indent=2)
print(f"\n📊 解剖完成！共处理 {self.index['total']} 部小说")
print(f" 人类级: {len(self.index['by_level']['human_level'])}")
print(f" 动物级: {len(self.index['by_level']['animal_level'])}")
print(f" 其他级: {len(self.index['by_level']['other_level'])}")
def analyze_one(self, file_path: Path) -> Optional[Dict[str, Any]]:
"""解剖单部小说"""
try:
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
text = f.read()
except Exception as e:
print(f" ⚠️ 读取失败: {e}")
return None
# 切分章节
chapters = self._split_chapters(text)
if not chapters:
return None
total_chapters = len(chapters)
# 提取各项数据
macro = self._extract_macro(chapters, text)
meso = self._extract_meso(chapters)
micro = self._extract_micro(chapters)
breath = self._extract_breath(text, chapters)
emotion = self._extract_emotion(chapters)
# 确定级别
level = self._determine_level(chapters, text)
# 生成模板
template_id = file_path.stem
template = {
"id": template_id,
"source": file_path.name,
"level": level,
"total_chapters": total_chapters,
"M_": macro,
"Z_": meso,
"X_": micro,
"B_": breath,
"E_": emotion
}
# 保存模板
level_dir = self.output_dir / level
level_dir.mkdir(exist_ok=True)
template_path = level_dir / f"{template_id}.json"
with open(template_path, 'w', encoding='utf-8') as f:
json.dump(template, f, ensure_ascii=False, indent=2)
return template
def _split_chapters(self, text: str) -> List[str]:
"""切分章节"""
# 匹配 "第X回" 或 "第X章"
pattern = r'第[一二三四五六七八九十百千万0-9]+[回章]'
matches = list(re.finditer(pattern, text))
if not matches:
# 尝试按空行分段作为备选
return [p for p in text.split("\n\n") if p.strip()]
chapters = []
for i, match in enumerate(matches):
start = match.start()
end = matches[i + 1].start() if i + 1 < len(matches) else len(text)
chapter_text = text[start:end].strip()
if chapter_text:
chapters.append(chapter_text)
return chapters
def _extract_macro(self, chapters: List[str], full_text: str) -> Dict[str, Any]:
"""提取大骨架"""
total = len(chapters)
# 情绪曲线（基于情绪词匹配）
emotion_curve = []
conflict_curve = []
for chapter in chapters:
# 情绪得分
emotion_score = 50
for emotion, words in EMOTION_WORDS.items():
for word in words:
if word in chapter:
emotion_score += 5
emotion_curve.append(min(100, emotion_score))
# 冲突得分
conflict_score = 0
for word in CONFLICT_WORDS:
conflict_score += chapter.count(word) * 2
conflict_curve.append(min(100, conflict_score))
# 确定幕数（根据情绪曲线的转折点）
act_count = 4
# 简化：按比例分割
act_boundaries = []
for i in range(1, act_count):
pos = int(total * i / act_count)
if 0 < pos < total:
act_boundaries.append(pos)
act_boundaries.append(total)
return {
"total_chapters": total,
"act_count": act_count,
"act_boundaries": act_boundaries,
"emotion_curve": emotion_curve,
"conflict_curve": conflict_curve
}
def _extract_meso(self, chapters: List[str]) -> List[Dict[str, Any]]:
"""提取中骨架"""
meso = []
for i, chapter in enumerate(chapters):
# 情绪标签
emotion_score = 0
emotion_name = "pinghe"
for name, words in EMOTION_WORDS.items():
score = sum(1 for w in words if w in chapter)
if score > emotion_score:
emotion_score = score
emotion_name = name
# 冲突等级
conflict_count = sum(1 for w in CONFLICT_WORDS if w in chapter)
if conflict_count >= 20:
conflict_level = 5
elif conflict_count >= 12:
conflict_level = 4
elif conflict_count >= 6:
conflict_level = 3
elif conflict_count >= 3:
conflict_level = 2
else:
conflict_level = 1
# 篇幅
word_count = len(chapter)
if word_count > 3000:
length = "chang"
elif word_count > 1500:
length = "zhong"
else:
length = "duan"
meso.append({
"chapter": i + 1,
"emotion": emotion_name,
"conflict_level": conflict_level,
"length": length
})
return meso
def _extract_micro(self, chapters: List[str]) -> Dict[str, Any]:
"""提取小骨架"""
total_sentences = 0
total_paragraphs = 0
total_words = 0
total_chars = 0
for chapter in chapters:
# 按标点分割句子
sentences = re.split(r'[。！？；\n]+', chapter)
sentences = [s for s in sentences if s.strip()]
total_sentences += len(sentences)
# 段落
paragraphs = [p for p in chapter.split('\n\n') if p.strip()]
total_paragraphs += len(paragraphs)
# 字数
total_words += len(chapter)
total_chars += len(chapter)
avg_sentence = total_words / max(1, total_sentences)
avg_paragraphs = total_paragraphs / max(1, len(chapters))
avg_words = total_words / max(1, len(chapters))
return {
"avg_sentence_length": avg_sentence,
"avg_paragraphs_per_chapter": avg_paragraphs,
"avg_words_per_chapter": avg_words
}
def _extract_breath(self, full_text: str, chapters: List[str]) -> Dict[str, Any]:
"""提取呼吸数据"""
total_chars = len(full_text)
comma_count = 0
period_count = 0
quote_count = 0
dunhao_count = 0
question_count = 0
exclamation_count = 0
for ch in full_text:
if ch in PUNCTUATION["comma"]:
comma_count += 1
elif ch in PUNCTUATION["period"]:
period_count += 1
elif ch in PUNCTUATION["quote"]:
quote_count += 1
elif ch in PUNCTUATION["dunhao"]:
dunhao_count += 1
elif ch in PUNCTUATION["question"]:
question_count += 1
elif ch in PUNCTUATION["exclamation"]:
exclamation_count += 1
total = comma_count + period_count + quote_count + dunhao_count + 1
return {
"comma_ratio": comma_count / total,
"period_ratio": period_count / total,
"quote_ratio": quote_count / total,
"dunhao_ratio": dunhao_count / total,
"question_ratio": question_count / total,
"exclamation_ratio": exclamation_count / total,
"comma_period_ratio": comma_count / (period_count + 1)
}
def _extract_emotion(self, chapters: List[str]) -> List[Dict[str, Any]]:
"""提取情绪数据"""
emotion_data = []
emotion_code_map = {
"pingjing": "PJ", "pinghe": "PH", "yuyue": "YY",
"lianmin": "LM", "yayi": "YA", "jinzhang": "JZ",
"beifen": "BF", "jingxian": "JX", "beican": "BC",
"yuanwang": "YW", "konghuang": "KH", "zhuanzhe": "ZZ",
"huangmiu": "HM", "yuanman": "YM"
}
for i, chapter in enumerate(chapters):
emotion_score = 0
emotion_name = "pinghe"
for name, words in EMOTION_WORDS.items():
score = sum(1 for w in words if w in chapter)
if score > emotion_score:
emotion_score = score
emotion_name = name
emotion_data.append({
"chapter": i + 1,
"name": emotion_name,
"code": emotion_code_map.get(emotion_name, "QT"),
"value": min(100, 50 + emotion_score * 5)
})
return emotion_data
def _determine_level(self, chapters: List[str], full_text: str) -> str:
"""确定模板级别"""
# 简单规则：
# - 人类级：章回数 >= 10，平均章长 >= 1500
# - 动物级：章回数 < 10，或平均章长 < 800
# - 其他级：介于两者之间
total_chapters = len(chapters)
if total_chapters == 0:
return "other_level"
avg_length = len(full_text) / total_chapters
if total_chapters >= 10 and avg_length >= 1500:
return "human_level"
elif total_chapters < 8 or avg_length < 800:
return "animal_level"
else:
return "other_level"
def main():
if len(sys.argv) < 2:
print("用法: python analyze_novels.py <小说文件夹路径>")
print("示例: python analyze_novels.py ./novels")
sys.exit(1)
input_dir = sys.argv[1]
if not os.path.isdir(input_dir):
print(f"❌ 错误: '{input_dir}' 不是有效的文件夹")
sys.exit(1)
print("🔪 河图洛书 · 小说解剖工具")
print(f"📂 扫描目标: {input_dir}")
print("=" * 50)
analyzer = NovelAnalyzer(input_dir)
analyzer.analyze_all()
if __name__ == "__main__":
main()
🚀 使用方法
1. 解剖小说（造肉体）
bash
python analyze_novels.py ./novels
扫描 ./novels 文件夹下所有 .txt 小说，提取结构数据，生成模板到 ./templates/。
2. 生成小说（注灵魂）
bash
python run_writer.py # 随机生成
python run_writer.py --level human # 使用人类级模板
python run_writer.py --level animal # 使用动物级模板
python run_writer.py --list # 列出所有模板
python run_writer.py --title "我的小说" --chapters 20
📋 文件清单
text
hetu_novel_writer/
├── config.py # 配置文件
├── run_writer.py # 主程序入口
├── analyze_novels.py # 解剖工具
├── core/
│ ├── __init__.py
│ ├── dao_engine.py # π引擎
│ ├── body_selector.py # 肉体选择器
│ ├── soul_injector.py # 灵魂注入器（五行生克）
│ ├── generation_engine.py # 生成引擎
│ └── output_writer.py # 输出器
├── templates/ # 模板库（自动生成）
│ ├── human_level/
│ ├── animal_level/
│ ├── other_level/
│ └── index.json
└── output/ # 生成的小说
代码已经全部写完，可以直接进入调试阶段了。你先运行看看，有什么问题我们再调整。🎯
所有核心代码已写完，你现在可以直接运行了。建议的调试顺序：
先放几本小说到 novels/ 文件夹
运行 python analyze_novels.py ./novels 生成模板
运行 python run_writer.py --list 查看模板
运行 python run_writer.py 生成第一部小说
有问题随时反馈，我来调整。🎯
我理解得对不对？
不对。我已经把代码写完了。我觉得你不需要我再继续解释或重复设计了——设计部分我们已经聊透了，我现在已经直接进入了实现阶段，并且所有代码都已经写好了，给你的是一个可以直接运行的项目。你现在可以直接拿这些代码去调试运行。
如果我没有理解对你的需求，你告诉我哪里出了问题，我来修正。🎯
|
鄂公网安备42060002000282号 )
4>"},