#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
通用小说生成器 V1.0
设计哲学:
- 程序本身是空的、通用的,不包含任何具体小说数据
- 所有内容来自外部的提纲文件(.txt)
- 换一本书只需要换提纲文件,不需要改代码
- 永不停止,写完一本自动随机选下一本
- 保留头脑风暴V3.0的进化机制(π引擎、河图、老师评分)
数据流:
启动 → 初始化:随机选一个提纲加载到内存
↓
循环(永不停止):
火2:取10个素材 → 合并成1个文本 → 告诉木3写第几章
↓
木3:素材 + 提纲章节信息 → 生成章节 → 打印初稿
↓
水1:润色 + 增加对话 → 打印润色稿
↓
金4:评分 → 低分改写 → 保存 → 打印最终稿
↓
老师评分 → 河图更新
↓
进度更新 → 如果全书完成,重新初始化
"""
import os
import sys
import json
import random
import re
import math
import hashlib
import time
import pickle
import shutil
import requests
from typing import List, Dict, Tuple, Optional
from datetime import datetime
# ==================== API配置 ====================
DEEPSEEK_API_KEY = "sk-09cd932d54934ac0abad9c27b6f8e686"
DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions"
# ==================== 文件夹结构 ====================
DIRS = {
"outline": "提纲",
"masterpieces": "masterpieces",
"saved_chapters": "保存章节",
"cache": "cache_小说",
"logs": "logs_小说",
"checkpoints": "checkpoints_小说",
"recovery": "recovery_小说"
}
for d in DIRS.values():
os.makedirs(d, exist_ok=True)
# ==================== 调用计数 ====================
CALL_COUNT_FILE = os.path.join(DIRS["checkpoints"], "call_count.json")
def get_call_count() -> int:
if os.path.exists(CALL_COUNT_FILE):
try:
with open(CALL_COUNT_FILE, 'r', encoding='utf-8') as f:
return json.load(f).get("count", 0)
except:
return 0
return 0
def increment_call_count() -> int:
count = get_call_count() + 1
with open(CALL_COUNT_FILE, 'w', encoding='utf-8') as f:
json.dump({"count": count, "last_update": datetime.now().isoformat()}, f, ensure_ascii=False, indent=2)
return count
# ==================== API调用 ====================
def call_deepseek(prompt: str, max_tokens: int = 4096, temperature: float = 0.75, timeout: int = 120) -> str:
cache_key = hashlib.md5(prompt.encode()).hexdigest()
cache_file = f"{DIRS['cache']}/{cache_key}.json"
if os.path.exists(cache_file):
try:
with open(cache_file, 'r', encoding='utf-8') as f:
return json.load(f)["response"]
except:
pass
try:
headers = {"Authorization": f"Bearer {DEEPSEEK_API_KEY}", "Content-Type": "application/json"}
data = {
"model": "deepseek-chat",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"temperature": temperature
}
response = requests.post(DEEPSEEK_API_URL, json=data, headers=headers, timeout=timeout)
if response.status_code == 200:
result = response.json()["choices"][0]["message"]["content"]
with open(cache_file, 'w', encoding='utf-8') as f:
json.dump({"prompt": prompt, "response": result}, f, ensure_ascii=False)
return result
return ""
except Exception as e:
return ""
# ==================== π引擎 ====================
class DaoEngine:
def __init__(self, chunk_size=10000):
self.chunk_size = chunk_size
self.digits = []
self.pointer = 0
self._load_next_chunk()
def _load_next_chunk(self):
print(f" 🔄 正在加载π第 {self.pointer//self.chunk_size + 1} 批小数位...")
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])
return
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_novelty(self, length=8) -> float:
while self.pointer + length >= len(self.digits):
self._load_next_chunk()
segment = self.digits[self.pointer:self.pointer+length]
self.pointer += length
value = 0
for i, d in enumerate(segment):
value += d * (0.1 ** (i+1))
return value
def get_state(self) -> dict:
return {"pointer": self.pointer}
def restore_state(self, state: dict):
self.pointer = state.get("pointer", 0)
self.digits = []
self._load_next_chunk()
# ==================== 河图中心 ====================
class HeTuCenter:
def __init__(self):
self.sheng_info = {"1": 0.0, "2": 0.0, "3": 0.0, "4": 0.0}
self.cheng_info = {"6": 0.0, "7": 0.0, "8": 0.0, "9": 0.0}
self.global_state = {"sheng": 0.0, "cheng": 0.0, "balance": 0.0}
def update_sheng(self, idx: int, value: float):
self.sheng_info[str(idx)] = value
self._update_global()
def update_cheng(self, idx: int, value: float):
self.cheng_info[str(idx)] = value
self._update_global()
def _update_global(self):
self.global_state["sheng"] = sum(self.sheng_info.values()) / 4
self.global_state["cheng"] = sum(self.cheng_info.values()) / 4
self.global_state["balance"] = self.global_state["sheng"] / (self.global_state["cheng"] + 0.01)
def get_full_state(self):
return {"sheng": self.sheng_info.copy(), "cheng": self.cheng_info.copy(), "global": self.global_state.copy()}
def get_save_state(self):
return {"sheng_info": self.sheng_info, "cheng_info": self.cheng_info, "global_state": self.global_state}
def restore_state(self, state: dict):
self.sheng_info = state.get("sheng_info", {"1": 0.0, "2": 0.0, "3": 0.0, "4": 0.0})
self.cheng_info = state.get("cheng_info", {"6": 0.0, "7": 0.0, "8": 0.0, "9": 0.0})
self.global_state = state.get("global_state", {"sheng": 0.0, "cheng": 0.0, "balance": 0.0})
# ==================== 节奏控制器 ====================
class RhythmController:
def __init__(self):
self.sheng_phase = 0
self.bian_phase = 0
self.sheng_speed = 0.2 * 2 * math.pi / 5
self.bian_speed = 2 * math.pi / 1
def update(self):
self.sheng_phase = (self.sheng_phase + self.sheng_speed) % (2 * math.pi)
self.bian_phase = (self.bian_phase + self.bian_speed) % (2 * math.pi)
def get_sheng_ratio(self):
return 0.55 + 0.25 * math.sin(self.sheng_phase)
def get_bian_ratio(self):
return 0.55 + 0.35 * math.sin(self.bian_phase)
def get_state(self) -> dict:
return {"sheng_phase": self.sheng_phase, "bian_phase": self.bian_phase}
def restore_state(self, state: dict):
self.sheng_phase = state.get("sheng_phase", 0)
self.bian_phase = state.get("bian_phase", 0)
# ============================================================
# 提纲解析(从 .txt 读取)
# ============================================================
def parse_outline(filepath: str) -> dict:
"""解析提纲文件,返回结构化数据"""
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
outline = {
"name": os.path.basename(filepath).replace('.txt', ''),
"title": "",
"type": "",
"core_mystery": "",
"theme": "",
"setting": "",
"total_chapters": 0,
"characters": {"main": [], "secondary": []},
"key_images": [],
"chapters": []
}
lines = content.split('\n')
i = 0
current_chapter = None
in_events = False
current_char = None
current_section = None
while i < len(lines):
line = lines[i].strip()
if not line:
i += 1
continue
# 段落标记
if line == '---小说信息---':
current_section = 'book'
i += 1
continue
if line == '---人物---':
current_section = 'main_characters'
current_char = None
i += 1
continue
if line == '---次要人物---':
current_section = 'secondary_characters'
current_char = None
i += 1
continue
if line == '---关键意象---':
current_section = 'key_images'
i += 1
continue
# 幕标记(跳过,只取章节)
if line.startswith('---第') and '幕' in line and '---' in line:
i += 1
continue
# 章节标题:---第1章:深秋的红疮---
chapter_match = re.match(r'---第(\d+)章[::]\s*(.+?)---', line)
if chapter_match:
ch_num = int(chapter_match.group(1))
ch_title = chapter_match.group(2).strip()
current_chapter = {
"chapter": ch_num,
"title": ch_title,
"characters": [],
"key_images": [],
"events": [],
"foreshadowing": [],
"target_word_count": 3000
}
outline["chapters"].append(current_chapter)
in_events = False
i += 1
continue
# 人物块
char_match = re.match(r'【(.+?)】', line)
if char_match and current_section in ['main_characters', 'secondary_characters']:
current_char = {'name': char_match.group(1)}
i += 1
continue
# 键值对
if ':' in line or ':' in line:
sep = ':' if ':' in line else ':'
key, val = line.split(sep, 1)
key = key.strip()
val = val.strip()
if current_section == 'book':
if key == '书名':
outline["title"] = val
elif key == '类型':
outline["type"] = val
elif key == '核心谜团':
outline["core_mystery"] = val
elif key == '主题':
outline["theme"] = val
elif key == '背景':
outline["setting"] = val
elif key == '总章数':
try:
outline["total_chapters"] = int(val)
except:
pass
elif current_section in ['main_characters', 'secondary_characters'] and current_char:
if key in ['年龄', '身份', '外貌', '性格', '内心冲突', '动机', '关键台词', '关键碎片', '核心秘密', '贡献']:
if key == '关键台词':
if 'key_lines' not in current_char:
current_char['key_lines'] = []
current_char['key_lines'].append(val)
else:
current_char[key] = val
elif current_section == 'key_images':
parts = val.split('|')
chapters = []
meaning = ""
if parts:
ch_part = parts[0].strip()
try:
if ',' in ch_part:
chapters = [int(x.strip()) for x in ch_part.split(',')]
else:
chapters = [int(ch_part)]
except:
pass
if len(parts) > 1:
meaning = parts[1].strip()
outline["key_images"].append({
"name": key,
"chapters": chapters,
"meaning": meaning
})
# 章节内部字段
if current_chapter is not None:
if line.startswith('核心事件:') or line.startswith('核心事件:'):
in_events = True
i += 1
continue
elif line.startswith('出场人物:') or line.startswith('出场人物:'):
sep = ':' if ':' in line else ':'
_, val = line.split(sep, 1)
current_chapter["characters"] = [c.strip() for c in val.split(',') if c.strip()]
i += 1
continue
elif line.startswith('关键意象:') or line.startswith('关键意象:'):
sep = ':' if ':' in line else ':'
_, val = line.split(sep, 1)
current_chapter["key_images"] = [c.strip() for c in val.split(',') if c.strip()]
i += 1
continue
elif line.startswith('伏笔:') or line.startswith('伏笔:'):
sep = ':' if ':' in line else ':'
_, val = line.split(sep, 1)
current_chapter["foreshadowing"] = [c.strip() for c in val.split(',') if c.strip()]
i += 1
continue
elif line.startswith('目标字数:') or line.startswith('目标字数:'):
sep = ':' if ':' in line else ':'
_, val = line.split(sep, 1)
try:
current_chapter["target_word_count"] = int(val)
except:
pass
i += 1
continue
# 核心事件列表项
if in_events and line and line[0].isdigit():
if '. ' in line:
event = line.split('. ', 1)[1].strip()
elif '.' in line:
event = line.split('.', 1)[1].strip()
else:
event = line
if event and len(event) > 3:
current_chapter["events"].append(event)
i += 1
continue
# 结束事件模式
if in_events and line and not line[0].isdigit():
if not any(line.startswith(x) for x in ['核心事件', '出场人物', '关键意象', '伏笔', '目标字数']):
in_events = False
i += 1
# 如果没解析到总章数,从章节数推导
if outline["total_chapters"] == 0 and outline["chapters"]:
outline["total_chapters"] = len(outline["chapters"])
return outline
# ============================================================
# 状态管理(从保存目录推导)
# ============================================================
def get_state(outline: dict) -> dict:
"""从保存目录推导当前进度"""
title = outline["title"] if outline["title"] else outline["name"]
safe_title = re.sub(r'[\\/*?:"<>|]', '', title)
base_dir = os.path.join(DIRS["saved_chapters"], safe_title)
total = outline["total_chapters"]
if not os.path.exists(base_dir):
return {"round": 1, "chapter": 1, "total": total}
# 找最大的轮次
max_round = 0
for d in os.listdir(base_dir):
match = re.match(r'第(\d+)轮', d)
if match:
max_round = max(max_round, int(match.group(1)))
if max_round == 0:
return {"round": 1, "chapter": 1, "total": total}
# 检查最新轮次已有多少章
latest_dir = os.path.join(base_dir, f"第{max_round:02d}轮")
if os.path.exists(latest_dir):
chapter_files = []
for f in os.listdir(latest_dir):
if f.endswith('.txt') and re.match(r'第\d+章', f):
chapter_files.append(f)
generated = len(chapter_files)
if generated >= total:
return {"round": max_round + 1, "chapter": 1, "total": total}
return {"round": max_round, "chapter": generated + 1, "total": total}
return {"round": max_round, "chapter": 1, "total": total}
def get_save_path(outline: dict, chapter_num: int, round_num: int, title: str) -> str:
"""生成保存路径"""
book_title = outline["title"] if outline["title"] else outline["name"]
safe_title = re.sub(r'[\\/*?:"<>|]', '', book_title)
round_dir = f"第{round_num:02d}轮"
save_dir = os.path.join(DIRS["saved_chapters"], safe_title, round_dir)
os.makedirs(save_dir, exist_ok=True)
filename = f"第{chapter_num:02d}章_{title}.txt"
return os.path.join(save_dir, filename)
# ============================================================
# 初始化:随机选一个提纲加载
# ============================================================
def init_outline() -> dict:
"""初始化:从提纲文件夹随机选一个加载"""
outline_files = []
if os.path.exists(DIRS["outline"]):
for fname in os.listdir(DIRS["outline"]):
if fname.endswith('.txt'):
outline_files.append(os.path.join(DIRS["outline"], fname))
if not outline_files:
raise FileNotFoundError(f"提纲文件夹为空: {DIRS['outline']}")
path = random.choice(outline_files)
outline = parse_outline(path)
outline["_path"] = path
# 获取当前状态
state = get_state(outline)
outline["_state"] = state
print(f"\n📖 初始化: {os.path.basename(path)}")
print(f" 书名: {outline['title'] or outline['name']}")
print(f" 总章数: {outline['total_chapters']}")
print(f" 当前轮次: {state['round']}")
print(f" 当前章节: {state['chapter']}")
return outline
# ============================================================
# 火2:取素材 + 合并
# ============================================================
def get_materials(count: int = 10) -> str:
"""随机取10个素材,合并成一个文本字符串返回"""
all_files = []
# 从 masterpieces 取
if os.path.exists(DIRS["masterpieces"]):
for fname in os.listdir(DIRS["masterpieces"]):
if fname.endswith('.txt'):
all_files.append(os.path.join(DIRS["masterpieces"], fname))
# 如果 masterpieces 不够,从 logs 补
if len(all_files) < count:
if os.path.exists(DIRS["logs"]):
for fname in os.listdir(DIRS["logs"]):
if fname.endswith('.txt'):
all_files.append(os.path.join(DIRS["logs"], fname))
if len(all_files) >= count * 2:
break
# 如果还不够,从 saved_chapters 补
if len(all_files) < count:
if os.path.exists(DIRS["saved_chapters"]):
for root, dirs, files in os.walk(DIRS["saved_chapters"]):
for fname in files:
if fname.endswith('.txt'):
all_files.append(os.path.join(root, fname))
if len(all_files) >= count * 2:
break
if len(all_files) >= count * 2:
break
if not all_files:
return ""
# 去重
all_files = list(set(all_files))
selected = random.sample(all_files, min(count, len(all_files)))
merged = []
for path in selected:
try:
with open(path, 'r', encoding='utf-8', errors='ignore') as f:
text = f.read()
if text and len(text) > 50:
merged.append(f"【素材来源:{os.path.basename(path)}】\n{text[:2000]}")
except:
continue
return "\n\n---\n\n".join(merged)
# ============================================================
# 木3:生成章节
# ============================================================
def generate_chapter(outline: dict, chapter_info: dict, materials: str, temperature: float) -> str:
"""生成章节正文"""
chapter_num = chapter_info["chapter"]
chapter_title = chapter_info["title"]
events = chapter_info.get("events", [])
characters = chapter_info.get("characters", [])
key_images = chapter_info.get("key_images", [])
target_word_count = chapter_info.get("target_word_count", 3000)
# 构建人物信息(从提纲中查找)
char_info = []
for name in characters:
found = False
for mc in outline["characters"]["main"]:
if mc.get("name") == name:
char_info.append(f"- {name}: {mc.get('性格', '')} | {mc.get('身份', '')}")
found = True
break
if not found:
for sec in outline["characters"]["secondary"]:
if sec.get("name") == name:
char_info.append(f"- {name}: {sec.get('身份', '')}")
found = True
break
char_text = "\n".join(char_info) if char_info else "参考人物设定"
events_text = "\n".join([f"{i+1}. {e}" for i, e in enumerate(events)])
images_text = "、".join(key_images) if key_images else "无"
prompt = f"""请根据以下提纲,创作小说的第{chapter_num}章。
【小说信息】
- 书名:{outline['title'] or outline['name']}
- 主题:{outline.get('theme', '')}
- 背景:{outline.get('setting', '')}
【本章信息】
- 章节:第{chapter_num}章
- 标题:{chapter_title}
- 目标字数:{target_word_count}字左右
- 核心情节(必须全部写入):
{events_text}
- 出场人物:
{char_text}
- 关键意象:{images_text}
【风格参考素材】
{materials[:3000]}
【要求】
1. 严格按照核心情节推进,不遗漏任何事件
2. 人物对话符合性格设定
3. 关键意象自然出现
4. 有场景描写、人物心理、对话互动
5. 输出格式:第{chapter_num}章:{chapter_title}
只输出正文,不要任何解释:"""
result = call_deepseek(prompt, max_tokens=4096, temperature=temperature)
if result and len(result) > 200:
if not result.strip().startswith(f"第{chapter_num}章"):
result = f"第{chapter_num}章:{chapter_title}\n\n{result}"
if len(result) > 3500:
result = result[:3500]
return result
return f"第{chapter_num}章:{chapter_title}\n\n(本章生成失败)"
# ============================================================
# 水1:润色 + 增加对话
# ============================================================
def polish_chapter(text: str, chapter_info: dict, outline: dict, temperature: float) -> str:
"""润色章节,增加对话"""
if not text or len(text) < 100:
return text
chapter_num = chapter_info["chapter"]
chapter_title = chapter_info["title"]
char_names = chapter_info.get("characters", [])
char_personalities = []
for name in char_names:
for mc in outline["characters"]["main"]:
if mc.get("name") == name:
char_personalities.append(f"- {name}: {mc.get('性格', '')}")
break
char_text = "\n".join(char_personalities) if char_personalities else "参考人物设定"
prompt = f"""润色第{chapter_num}章,增加对话。
【章节】{chapter_title}
【人物性格】
{char_text}
【原文】
{text}
【要求】
1. 情节不变
2. 增加对话(用引号)
3. 对话符合性格
4. 保持章节整体风格
润色后的正文:"""
result = call_deepseek(prompt, max_tokens=4096, temperature=temperature)
if result and len(result) > 100:
if not result.strip().startswith(f"第{chapter_num}章"):
result = f"第{chapter_num}章:{chapter_title}\n\n{result}"
if len(result) > 3500:
result = result[:3500]
return result
return text
# ============================================================
# 金4:评分 + 改写 + 保存
# ============================================================
def check_rewrite_save(text: str, chapter_info: dict, outline: dict, save_path: str, max_rewrites: int = 2) -> Tuple[str, float, int]:
"""
金4:评分 → 低分改写 → 保存
返回:(最终文本, 最终评分, 改写次数)
永不失败,一定会保存
"""
chapter_num = chapter_info["chapter"]
chapter_title = chapter_info["title"]
core_events = chapter_info.get("events", [])
key_images = chapter_info.get("key_images", [])
events_check = "\n".join([f"- {e}" for e in core_events])
images_check = "、".join(key_images) if key_images else "无"
current_text = text
rewrite_count = 0
final_score = 0.0
for attempt in range(max_rewrites + 1):
# ---- 评分 ----
prompt = f"""请对第{chapter_num}章进行评分(0-1分)并给出简要评语。
【核心情节要求】
{events_check}
【关键意象】{images_check}
【章节内容】
{current_text[:2000]}
【输出格式】
评分:X.XX
评语:一句话说明优点和缺点(20字内)"""
result = call_deepseek(prompt, max_tokens=200, temperature=0.3)
score = 0.7
comment = ""
if result:
score_match = re.search(r'评分[::]\s*([0-9.]+)', result)
if score_match:
score = float(score_match.group(1))
comment_match = re.search(r'评语[::]\s*(.+?)(?:\n|$)', result)
if comment_match:
comment = comment_match.group(1).strip()[:50]
score = min(1.0, max(0.0, score))
final_score = score
print(f" 金4评分(第{attempt+1}次): {score:.2f} | {comment}")
# ---- 如果评分达标,直接保存 ----
if score >= 0.7:
with open(save_path, 'w', encoding='utf-8') as f:
f.write(current_text)
return current_text, score, rewrite_count
# ---- 评分不达标,改写 ----
if attempt < max_rewrites:
print(f" ⚠️ 评分 {score:.2f} < 0.7,第 {attempt+1} 次改写...")
prompt = f"""请根据以下评语,对第{chapter_num}章进行改写优化。
【评语】{comment}
【原章节】
{current_text}
【改写要求】
1. 保留核心情节和人物
2. 根据评语指出的问题修改
3. 保持对话自然、符合人物性格
4. 直接输出改写后的完整章节,格式:第{chapter_num}章:{chapter_title}
改写后的章节:"""
rewrite_result = call_deepseek(prompt, max_tokens=4096, temperature=0.7)
if rewrite_result and len(rewrite_result) > 200:
if not rewrite_result.strip().startswith(f"第{chapter_num}章"):
rewrite_result = f"第{chapter_num}章:{chapter_title}\n\n{rewrite_result}"
if len(rewrite_result) > 3500:
rewrite_result = rewrite_result[:3500]
current_text = rewrite_result
rewrite_count += 1
# 所有改写都尝试完了,无论如何都保存
with open(save_path, 'w', encoding='utf-8') as f:
f.write(current_text)
return current_text, final_score, rewrite_count
# ============================================================
# 老师评分
# ============================================================
def evaluate_teachers(text: str) -> Dict[str, Tuple[float, str]]:
"""四个老师评分"""
results = {}
teacher_configs = [
(6, "语言质量", "语句通顺、修辞美感、节奏感"),
(7, "情节结构", "情节推进、转折、章节独立性"),
(8, "人物对话", "言行符合设定、对话自然"),
(9, "整体主题", "体现核心主题、完成度")
]
for tid, name, criteria in teacher_configs:
if not text or len(text) < 100:
results[name] = (0.5, "文本太短")
continue
text_slice = text[:600] + "..." + text[-200:] if len(text) > 800 else text
prompt = f"""你是老师{tid},专攻「{name}」。
【标准】{criteria}
【文章】
{text_slice}
输出:分数|评语(分数0-1之间,评语20字内)"""
result = call_deepseek(prompt, max_tokens=100, temperature=0.3)
score = 0.6
comment = ""
if result and '|' in result:
parts = result.split('|')
try:
score = float(parts[0].strip())
comment = parts[1].strip()[:30]
except:
pass
results[name] = (min(1.0, max(0.0, score)), comment)
return results
# ============================================================
# 打印分隔线函数
# ============================================================
def print_section(title: str, content: str, max_len: int = 3500):
"""打印带分隔线的章节内容"""
print(f"\n{'='*40} {title} {'='*40}")
if len(content) > max_len:
print(content[:max_len])
print(f"\n... (内容过长,截断,共 {len(content)} 字)")
else:
print(content)
print(f"{'='*90}\n")
# ============================================================
# 主循环
# ============================================================
def main():
print("\n" + "=" * 70)
print("📚 通用小说生成器 V1.0")
print(" 程序本身是空的、通用的")
print(" 所有数据来自外部的提纲文件")
print(" 换一本书只需要换提纲文件")
print(" 永不停止,自动循环")
print("=" * 70)
# 初始化
dao = DaoEngine()
hetu = HeTuCenter()
rhythm = RhythmController()
call_count = get_call_count()
print(f"\n📊 累计调用次数: {call_count}")
# 尝试恢复检查点
checkpoint_path = os.path.join(DIRS["checkpoints"], "checkpoint.pkl")
if os.path.exists(checkpoint_path):
try:
with open(checkpoint_path, 'rb') as f:
cp = pickle.load(f)
print("📂 恢复检查点...")
outline = cp.get("outline")
state = cp.get("state")
dao.restore_state(cp.get("dao_pointer", {}))
hetu.restore_state(cp.get("hetu", {}))
rhythm.restore_state(cp.get("rhythm", {}))
call_count = cp.get("call_count", call_count)
print(f" 恢复进度: 第 {state['round']} 轮 第 {state['chapter']} 章")
except Exception as e:
print(f" ⚠️ 恢复检查点失败: {e}")
outline = init_outline()
state = outline["_state"]
else:
outline = init_outline()
state = outline["_state"]
print("\n🚀 启动主循环...\n")
try:
while True:
# ===== 检查当前书是否完成 =====
call_count = increment_call_count()
current_chapter_num = state["chapter"]
total_chapters = outline["total_chapters"]
if current_chapter_num > total_chapters:
print(f"\n🎉 全书 {outline['title'] or outline['name']} 第 {state['round']} 轮已完成!")
# 重置进度,进入下一轮
state["chapter"] = 1
state["round"] = state["round"] + 1
print(f" 进入第 {state['round']} 轮")
continue
# ===== 获取章节信息 =====
chapter_info = None
for ch in outline["chapters"]:
if ch["chapter"] == current_chapter_num:
chapter_info = ch
break
if chapter_info is None:
print(f"⚠️ 找不到第 {current_chapter_num} 章的信息,跳过")
state["chapter"] = current_chapter_num + 1
continue
dao_novelty = dao.get_novelty(6)
rhythm.update()
sheng_ratio = rhythm.get_sheng_ratio()
bian_ratio = rhythm.get_bian_ratio()
print(f"\n{'─'*70}")
print(f"轮次: 第 {state['round']} 轮")
print(f"章节: 第 {current_chapter_num} 章 - {chapter_info['title']}")
print(f"道新奇度: {dao_novelty:.4f}")
# ===== 火2:取素材 =====
print(f" 🔥 火2: 取素材中...")
materials = get_materials(10)
print(f" 素材大小: {len(materials)} 字符")
if len(materials) > 0:
print(f" 素材预览: {materials[:200]}...")
# ===== 木3:生成 =====
temp = 0.6 + sheng_ratio * 0.4
print(f" 🌳 木3: 生成中 (温度: {temp:.2f})...")
draft = generate_chapter(outline, chapter_info, materials, temp)
print(f" 生成 {len(draft)} 字")
# ---- 打印木3初稿 ----
print_section("木3初稿", draft)
if len(draft) < 100:
print(f" ⚠️ 生成失败,字数不足,跳过本章")
state["chapter"] = current_chapter_num + 1
continue
# ===== 水1:润色 =====
temp = 0.5 + bian_ratio * 0.3
print(f" 💧 水1: 润色中 (温度: {temp:.2f})...")
polished = polish_chapter(draft, chapter_info, outline, temp)
print(f" 润色后 {len(polished)} 字")
# ---- 打印水1润色稿 ----
print_section("水1润色稿", polished)
if len(polished) < 100:
polished = draft
# ===== 金4:评分 + 改写 + 保存 =====
save_path = get_save_path(outline, current_chapter_num, state["round"], chapter_info["title"])
print(f" 💎 金4: 评分+改写中...")
final_text, final_score, rewrite_count = check_rewrite_save(
polished, chapter_info, outline, save_path, max_rewrites=2
)
print(f" 最终评分: {final_score:.2f} | 改写次数: {rewrite_count}")
print(f" 保存到: {save_path}")
# ---- 打印金4最终稿 ----
print_section("金4最终稿", final_text)
# ===== 老师评分 =====
print(f" 📊 老师评分中...")
scores = evaluate_teachers(final_text)
avg_score = 0
for name, (s, c) in scores.items():
print(f" {name}: {s:.2f} | {c}")
avg_score += s
avg_score = avg_score / len(scores) if scores else 0
# ===== 更新河图 =====
hetu.update_sheng(1, avg_score)
hetu.update_cheng(7, avg_score)
print(f" 📈 平均分: {avg_score:.3f}")
# ===== 更新进度 =====
state["chapter"] = current_chapter_num + 1
# ===== 保存状态到检查点 =====
checkpoint_data = {
"call_count": call_count,
"outline": outline,
"state": state,
"dao_pointer": dao.get_state(),
"hetu": hetu.get_save_state(),
"rhythm": rhythm.get_state()
}
with open(os.path.join(DIRS["checkpoints"], "checkpoint.pkl"), 'wb') as f:
pickle.dump(checkpoint_data, f)
except KeyboardInterrupt:
print(f"\n\n⏸️ 停止")
print(f" 累计调用: {call_count}")
print(f" 当前书: {outline['title'] or outline['name']}")
print(f" 当前进度: 第 {state['round']} 轮 第 {state['chapter']} 章")
print("\n💾 状态已保存到 checkpoints/checkpoint.pkl")
print(" 下次运行继续")
except Exception as e:
print(f"\n❌ 错误: {e}")
import traceback
traceback.print_exc()
print("\n💾 状态已保存到 checkpoints/checkpoint.pkl")
if __name__ == "__main__":
main() |