#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
提纲生成器 V9.2 · 小说写手专用版（简介附在提纲末尾）
【修改版：放行条件仅80章，简介不合规不丢弃】
【核心规则】
1. 放行条件：仅检查80章是否完整。
2. 金4：只允许润色语言和格式，严禁修改、添加或删除任何章节内容。
3. 火2：素材必须完整，不完整则重新抽取。
4. 火2：人物姓名从姓名池中随机抽取（先取姓，再取名），不硬编码默认姓名。
5. 火2：恋人与主角性别相反（硬性规定）。
6. 火2：姓名池不允许兜底。
7. 所有API全程禁止修改人物姓名。
8. 简介附在提纲末尾（不单独保存），尽量35-50字，含书名标签，但不作为放行条件。
9. 保存时只保留"小说信息"及以下内容，小说写手直接使用。
流程：木3生成初稿 → 水1补齐 → 师7打分 → 金4润色+生成简介
每章目标字数：2500-3000字
不合格（章节不足80）直接丢弃
保留道（π引擎）与河图洛书结构
"""
import os
import sys
import json
import random
import re
import hashlib
import time
import requests
import math
from typing import List, Dict, Tuple, Optional, Any
from datetime import datetime
# ==================== 配置 ====================
CONFIG = {
"api_key": os.getenv("DEEPSEEK_API_KEY", "sk-KEY"),
"api_url": "https://api.deepseek.com/v1/chat/completions",
"lib_dir": "创作库",
"masterpieces_dir": "masterpieces",
"output_dir": "output_提纲",
"cache_dir": "cache_提纲",
"logs_dir": "logs_提纲",
"checkpoint_dir": "checkpoints_提纲",
"min_chapters": 80,
"min_mysteries": 3,
"min_characters": 6,
"min_images": 8,
"min_conflicts": 6,
"min_subplots": 6,
"chapter_min_words": 2500,
"chapter_max_words": 3000,
"max_retry_fire2": 20,
}
# 创建目录
for d in [CONFIG["masterpieces_dir"], CONFIG["output_dir"],
CONFIG["cache_dir"], CONFIG["logs_dir"], CONFIG["checkpoint_dir"], CONFIG["lib_dir"]]:
os.makedirs(d, exist_ok=True)
# ==================== 缺口记录 ====================
MISSING_LOG = []
def record_missing(lib_name: str, category: str, context: str, detail: str = ""):
entry = {
"timestamp": datetime.now().isoformat(),
"lib_name": lib_name,
"category": category,
"context": context,
"detail": detail,
"round": len(MISSING_LOG) + 1
}
MISSING_LOG.append(entry)
print(f" 📝 记录缺口: {lib_name} → {category} ({context})")
def save_missing_log():
if not MISSING_LOG:
return
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{CONFIG['logs_dir']}/missing_log_{timestamp}.json"
with open(filename, 'w', encoding='utf-8') as f:
json.dump(MISSING_LOG, f, ensure_ascii=False, indent=2)
print(f" 💾 缺口记录已保存: {filename}")
# ==================== 打印辅助函数 ====================
def print_full(content: str, label: str):
"""打印完整内容，不截断"""
if not content or len(content) < 50:
print(f"\n📄 [{label}] 内容为空或过短")
return
total_len = len(content)
print(f"\n{'='*70}")
print(f"📄 [{label}] 总长度: {total_len} 字符")
print(f"{'='*70}")
print(content)
print(f"{'='*70}\n")
# ==================== API调用 ====================
def call_deepseek(prompt: str, max_tokens: int = 32768, temperature: float = 0.75, timeout: int = 180) -> str:
cache_key = hashlib.md5(prompt.encode()).hexdigest()
cache_file = f"{CONFIG['cache_dir']}/{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 {CONFIG['api_key']}", "Content-Type": "application/json"}
data = {
"model": "deepseek-chat",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens,
"temperature": temperature
}
response = requests.post(CONFIG['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:
print(f" ⚠️ API调用失败: {e}")
return ""
# ==================== 完整性检测器 ====================
class CompletenessChecker:
def __init__(self, config: dict):
self.config = config
self.targets = {
"mysteries": config.get("min_mysteries", 3),
"characters": config.get("min_characters", 6),
"images": config.get("min_images", 8),
"chapters": config.get("min_chapters", 80),
"conflicts": config.get("min_conflicts", 6),
"subplots": config.get("min_subplots", 6),
}
def check(self, outline: str) -> dict:
result = {
"mysteries": 0, "characters": 0, "images": 0,
"chapters": 0, "conflicts": 0, "subplots": 0,
"missing": [], "details": {}
}
mysteries_section = re.search(r'核心谜团[：:]\s*(.*?)(?=\n主题[：:]|\n背景[：:]|\n##|\n###|\Z)', outline, re.DOTALL)
if mysteries_section:
result["mysteries"] = len(re.findall(r'\d+\.\s*[^\n]+', mysteries_section.group(1)))
characters_section = re.search(r'###\s*第二部分：核心人物.*?(?=###\s*第三部分|\Z)', outline, re.DOTALL)
if characters_section:
char_matches = re.findall(r'\d+\.\s*\*\*[^*]+\*\*', characters_section.group(0))
if not char_matches:
char_matches = re.findall(r'-\s*\*\*[^*]+\*\*', characters_section.group(0))
result["characters"] = len(char_matches) if char_matches else 0
images_section = re.search(r'###\s*第三部分：关键意象.*?(?=###\s*第四部分|\Z)', outline, re.DOTALL)
if images_section:
table_rows = re.findall(r'\|\s*[^|]+\s*\|\s*[^|]+\s*\|\s*[^|]+\s*\|', images_section.group(0))
header_count = sum(1 for row in table_rows if '---' in row or ('：' in row and '象征' in row))
result["images"] = max(0, len(table_rows) - header_count)
result["chapters"] = len(re.findall(r'---第\d+章：', outline))
conflicts_section = re.search(r'###\s*第五部分：核心冲突总表.*?(?=###\s*第六部分|\Z)', outline, re.DOTALL)
if conflicts_section:
result["conflicts"] = len(re.findall(r'\d+\.\s*[^\n]+', conflicts_section.group(0)))
subplots_section = re.search(r'###\s*第六部分：伏线总表.*?(?=\Z)', outline, re.DOTALL)
if subplots_section:
result["subplots"] = len(re.findall(r'\d+\.\s*[^\n]+', subplots_section.group(0)))
for key, target in self.targets.items():
if result[key] < target:
result["missing"].append(f"{key} ({result[key]}/{target})")
result["is_complete"] = len(result["missing"]) == 0
return result
# ==================== 道（π引擎） ====================
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])
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
return sum(d * (0.1 ** (i+1)) for i, d in enumerate(segment))
def get_rhythm(self) -> Tuple[float, float]:
phase1 = self.get_novelty(4) * 2 * math.pi
phase2 = self.get_novelty(4) * 2 * math.pi
return phase1, phase2
# ==================== 河图中心 ====================
class HeTuCenter:
def __init__(self):
self.scores = {"shi9": [], "shi7": [], "shi6": [], "shi8": []}
self.balance = 0.0
self.sheng = {}
self.cheng = {}
def add_score(self, teacher: str, value: float):
if teacher in self.scores:
self.scores[teacher].append(value)
self._update_balance()
def _update_balance(self):
all_vals = [v for vals in self.scores.values() for v in vals if v > 0]
if all_vals:
self.balance = sum(all_vals) / len(all_vals)
def update_sheng(self, idx: int, value: float):
self.sheng[str(idx)] = value
self._update_balance()
def update_cheng(self, idx: int, value: float):
self.cheng[str(idx)] = value
self._update_balance()
def get_average(self, teacher: str) -> float:
vals = self.scores.get(teacher, [])
return sum(vals) / len(vals) if vals else 0.0
# ==================== 库加载器 ====================
class LibraryLoader:
def __init__(self, lib_dir: str = "创作库"):
self.lib_dir = lib_dir
self.libraries = {}
self._load_all()
def _load_all(self):
print(f"\n📚 从 '{self.lib_dir}' 加载所有库...")
if not os.path.exists(self.lib_dir):
raise FileNotFoundError(f"库文件夹 '{self.lib_dir}' 不存在")
lib_files = [f for f in os.listdir(self.lib_dir) if f.endswith('.json')]
if not lib_files:
raise FileNotFoundError(f"库文件夹 '{self.lib_dir}' 中没有找到任何 JSON 文件")
for fname in sorted(lib_files):
path = os.path.join(self.lib_dir, fname)
try:
with open(path, 'r', encoding='utf-8') as f:
self.libraries[fname] = json.load(f)
print(f" ✅ {fname}")
except Exception as e:
print(f" ⚠️ {fname} 加载失败: {e}")
if not self.libraries:
raise RuntimeError("未加载任何库文件")
def get(self, fname: str) -> dict:
return self.libraries.get(fname)
def get_data(self, fname: str, key_path: str) -> list:
lib = self.get(fname)
if lib is None:
raise KeyError(f"库 {fname} 未加载")
keys = key_path.split('.')
data = lib
found = True
for k in keys:
if isinstance(data, dict) and k in data:
data = data[k]
else:
found = False
break
if found and isinstance(data, list):
return data
lib_name = fname.replace('.json', '')
full_key_path = f"{lib_name}.{key_path}"
full_keys = full_key_path.split('.')
data = lib
found = True
for k in full_keys:
if isinstance(data, dict) and k in data:
data = data[k]
else:
found = False
break
if found and isinstance(data, list):
return data
if len(lib) == 1:
top_key = list(lib.keys())[0]
if isinstance(lib[top_key], dict):
test_keys = f"{top_key}.{key_path}".split('.')
data = lib
found = True
for k in test_keys:
if isinstance(data, dict) and k in data:
data = data[k]
else:
found = False
break
if found and isinstance(data, list):
return data
raise KeyError(f"在 {fname} 中找不到路径: {key_path}")
def get_conflicts(self) -> list:
return self.get_data("库1_冲突模式池.json", "冲突模式")
def get_motivations(self) -> list:
return self.get_data("库2_人物动机池.json", "人物动机")
def get_event_skeletons(self) -> list:
lib = self.get("库3_核心事件骨架池.json")
if not lib:
raise KeyError("库3_核心事件骨架池.json 未加载")
data = lib
for k in ["库3_核心事件骨架池", "骨架分类"]:
if isinstance(data, dict) and k in data:
data = data[k]
else:
break
if isinstance(data, dict):
result = []
for sub_key, sub_value in data.items():
if isinstance(sub_value, list):
result.extend(sub_value)
elif isinstance(sub_value, dict):
for v in sub_value.values():
if isinstance(v, list):
result.extend(v)
if result:
return result
try:
return self.get_data("库3_核心事件骨架池.json", "骨架分类")
except:
pass
raise ValueError("库3中未找到任何事件骨架")
def get_devices(self) -> list:
lib = self.get("库4_转折装置池.json")
result = []
for value in lib.values():
if isinstance(value, list):
result.extend(value)
elif isinstance(value, dict):
for v in value.values():
if isinstance(v, list):
result.extend(v)
if not result:
raise ValueError("库4中未找到任何转折装置")
return result
def get_images(self) -> list:
lib = self.get("库5_意象库_叙事版.json")
result = []
for value in lib.values():
if isinstance(value, list):
result.extend(value)
elif isinstance(value, dict):
for v in value.values():
if isinstance(v, list):
result.extend(v)
if not result:
raise ValueError("库5中未找到任何意象")
return result
def get_authors(self) -> list:
return self.get_data("库11_作者及风格库.json", "作者列表")
def get_eras(self) -> list:
return self.get_data("库16_时代背景池.json", "时代条目")
def get_opponents(self) -> list:
return self.get_data("库17_对手阻力池.json", "对手类型")
def get_themes(self) -> list:
lib = self.get("库15_主题句池.json")
result = []
for path in ["主题句分类", "库15_主题句池.主题句分类", "库15_主题句池"]:
try:
data = lib
for k in path.split('.'):
if isinstance(data, dict) and k in data:
data = data[k]
else:
break
if isinstance(data, list):
for item in data:
if isinstance(item, dict) and "条目" in item:
result.extend(item["条目"])
else:
result.append(item)
if result:
return result
except:
continue
if not result:
raise ValueError("库15中未找到任何主题句")
return result
def get_genres(self) -> list:
return self.get_data("库27_小说类型模板池.json", "类型模板")
def get_name_pools(self) -> dict:
lib = self.get("库7_姓名库.json")
if lib:
for path in ["子池", "库7_姓名库.子池", "库7_姓名库"]:
data = lib
found = True
for k in path.split('.'):
if isinstance(data, dict) and k in data:
data = data[k]
else:
found = False
break
if found and isinstance(data, dict):
return data
raise ValueError("库7中未找到任何姓名子池")
def get_title_prefixes(self) -> list:
lib = self.get("库6_章名元素池.json")
if lib:
for k in ["前7字", "库6_章名元素池.前7字"]:
parts = k.split('.')
d = lib
ok = True
for p in parts:
if isinstance(d, dict) and p in d:
d = d[p]
else:
ok = False
break
if ok and isinstance(d, list):
return [item.get("内容", "") for item in d if item.get("内容")]
return []
def get_title_suffixes(self) -> list:
lib = self.get("库6_章名元素池.json")
if lib:
for k in ["后7字", "库6_章名元素池.后7字"]:
parts = k.split('.')
d = lib
ok = True
for p in parts:
if isinstance(d, dict) and p in d:
d = d[p]
else:
ok = False
break
if ok and isinstance(d, list):
return [item.get("内容", "") for item in d if item.get("内容")]
return []
def safe_random_pick(self, items, lib_name, category):
if not items:
return {"名称": f"临时条目_{category}", "描述": "待补充"}
return random.choice(items)
# ==================== 火2：随机抽取 ====================
class Fire2:
def __init__(self, loader: LibraryLoader):
self.loader = loader
self.completeness_rules = {
"conflict": ["名称", "描述"],
"motivation": ["名称", "描述"],
"event": ["名称", "描述"],
"device": ["名称", "描述"],
"image": ["名称", "描述"],
"author": ["name", "风格"],
"era": ["名称", "描述"],
"opponent": ["名称", "描述"],
"theme": ["内容"],
"genre": ["名称", "描述"],
}
def _is_complete(self, item: dict, rules: List[str]) -> bool:
if not item or not isinstance(item, dict):
return False
for field in rules:
value = item.get(field)
if value is None or (isinstance(value, str) and not value.strip()):
return False
return True
def _pick_with_retry(self, items: list, rules: List[str], lib_name: str, category: str, max_retry: int = 20) -> dict:
if not items:
print(f" ⚠️ {category} 列表为空，使用临时条目")
return {"名称": f"临时条目_{category}", "描述": "待补充"}
for attempt in range(max_retry):
candidate = random.choice(items)
if self._is_complete(candidate, rules):
return candidate
if attempt < 3:
missing = [f for f in rules if not candidate.get(f) or (isinstance(candidate.get(f), str) and not candidate.get(f).strip())]
print(f" 🔄 {category} 第{attempt+1}次抽取不完整（缺: {missing}），重新抽取...")
print(f" ⚠️ {category} 达到最大重试次数({max_retry})，返回第一个可用条目")
for item in items:
if item and isinstance(item, dict):
for rule in rules:
if rule not in item or not item.get(rule):
item[rule] = f"待补充_{rule}"
return item
return {"名称": f"临时条目_{category}", "描述": "待补充"}
def _find_valid_name_pool(self, name_pools: dict) -> Tuple[dict, str]:
if not name_pools:
raise ValueError("姓名池为空，无法抽取")
pool_names = list(name_pools.keys())
attempt = 0
while True:
attempt += 1
random.shuffle(pool_names)
for pool_name in pool_names:
pool = name_pools[pool_name]
if not isinstance(pool, dict):
continue
surnames = pool.get("surnames") or pool.get("姓氏", [])
male_names = pool.get("male") or pool.get("男名", [])
female_names = pool.get("female") or pool.get("女名", [])
if surnames and male_names and female_names:
normalized_pool = {
"surnames": surnames,
"male": male_names,
"female": female_names,
}
return normalized_pool, pool_name
print(f" ⚠️ 第{attempt}次遍历：所有姓名子池均缺少必要字段，重新尝试...")
def generate_seed(self) -> dict:
print("\n 🔥 火2：随机抽取素材（完整性校验开启）...")
name_pools = self.loader.get_name_pools()
while True:
try:
selected_pool, pool_name = self._find_valid_name_pool(name_pools)
print(f" 🏷️ 选中姓名子池: {pool_name}")
break
except ValueError as e:
print(f" ❌ 致命错误: {e}")
raise RuntimeError(f"无法获取姓名池: {e}")
surnames = selected_pool.get("surnames", [])
male_names = selected_pool.get("male", [])
female_names = selected_pool.get("female", [])
protagonist_is_male = random.choice([True, False])
love_is_male = not protagonist_is_male
rival_is_male = random.choice([True, False])
p_surname = random.choice(surnames)
p_given = random.choice(male_names if protagonist_is_male else female_names)
protagonist = p_surname + p_given
available_surnames_l = [s for s in surnames if s != p_surname]
if not available_surnames_l:
available_surnames_l = surnames
l_surname = random.choice(available_surnames_l)
l_given = random.choice(male_names if love_is_male else female_names)
love_interest = l_surname + l_given
available_surnames_r = [s for s in surnames if s not in [p_surname, l_surname]]
if not available_surnames_r:
available_surnames_r = [s for s in surnames if s != p_surname]
r_surname = random.choice(available_surnames_r)
r_given = random.choice(male_names if rival_is_male else female_names)
rival = r_surname + r_given
side_names = []
used_surnames = {p_surname, l_surname, r_surname}
for _ in range(3):
available_s = [s for s in surnames if s not in used_surnames]
if not available_s:
available_s = [s for s in surnames if s not in [p_surname]]
s = random.choice(available_s)
used_surnames.add(s)
g = random.choice(male_names if random.choice([True, False]) else female_names)
side_names.append(s + g)
print(f" 👤 姓名已固定（先取姓，再取名，组合）:")
print(f" 主角: {protagonist} ({'男' if protagonist_is_male else '女'})")
print(f" 恋人: {love_interest} ({'男' if love_is_male else '女'}) ← 硬性与主角相反")
print(f" 对手: {rival} ({'男' if rival_is_male else '女'})")
print(f" 配角: {side_names}")
print(" 📦 正在抽取素材...")
conflict = self._pick_with_retry(
self.loader.get_conflicts(),
self.completeness_rules["conflict"],
"库1", "冲突"
)
motivation = self._pick_with_retry(
self.loader.get_motivations(),
self.completeness_rules["motivation"],
"库2", "动机"
)
event = self._pick_with_retry(
self.loader.get_event_skeletons(),
self.completeness_rules["event"],
"库3", "事件"
)
device = self._pick_with_retry(
self.loader.get_devices(),
self.completeness_rules["device"],
"库4", "装置"
)
image = self._pick_with_retry(
self.loader.get_images(),
self.completeness_rules["image"],
"库5", "意象"
)
author = self._pick_with_retry(
self.loader.get_authors(),
self.completeness_rules["author"],
"库11", "作者"
)
era = self._pick_with_retry(
self.loader.get_eras(),
self.completeness_rules["era"],
"库16", "时代"
)
opponent = self._pick_with_retry(
self.loader.get_opponents(),
self.completeness_rules["opponent"],
"库17", "对手"
)
theme = self._pick_with_retry(
self.loader.get_themes(),
self.completeness_rules["theme"],
"库15", "主题"
)
genre = self._pick_with_retry(
self.loader.get_genres(),
self.completeness_rules["genre"],
"库27", "类型"
)
print(" ✅ 所有素材抽取完成")
return {
"conflict": conflict,
"motivation": motivation,
"event": event,
"device": device,
"image": image,
"author": author,
"era": era,
"opponent": opponent,
"theme": theme,
"genre": genre,
"name_pools": name_pools,
"title_prefixes": self.loader.get_title_prefixes(),
"title_suffixes": self.loader.get_title_suffixes(),
"protagonist": protagonist,
"love_interest": love_interest,
"rival": rival,
"side_names": side_names,
"gender_info": {
"protagonist": "男" if protagonist_is_male else "女",
"love_interest": "男" if love_is_male else "女",
"rival": "男" if rival_is_male else "女",
}
}
# ==================== 提取书名辅助函数 ====================
def extract_book_title_from_outline(outline: str) -> str:
"""从提纲内容中提取书名，支持多种格式"""
patterns = [
r'书名[：:]\s*[《「""]?([^》」""\n]+)[》」""]?',
r'小说名称[：:]\s*[《「""]?([^》」""\n]+)[》」""]?',
r'小说名[：:]\s*[《「""]?([^》」""\n]+)[》」""]?',
r'---小说信息---.*?书名[：:]\s*[《「""]?([^》」""\n]+)[》」""]?',
r'---小说信息---.*?小说名称[：:]\s*[《「""]?([^》」""\n]+)[》」""]?',
]
for pattern in patterns:
match = re.search(pattern, outline, re.DOTALL)
if match:
title = match.group(1).strip()
title = re.sub(r'[《》""''「」]', '', title)
if title and title != '未命名':
return title
return "未命名"
# ==================== 提取“小说信息”位置 ====================
def extract_from_book_info(content: str) -> str:
"""从"小说信息"开始截取，去除之前的所有内容"""
match = re.search(r'(###\s*第一部分：小说信息|---小说信息---)', content, re.DOTALL)
if match:
return content[match.start():]
return content
# ==================== 硬性约束：禁止省略 ====================
NO_SKIP_CONSTRAINT = """
【硬性约束 - 禁止省略】（必须100%执行，违反即不合格）
1. 你必须逐章列出全部80章的内容，从第1章到第80章，一章不能少。
2. 禁止使用任何形式的省略表述，包括但不限于：
- "其余章节按此结构"
- "第X章至第Y章类似"
- "详见上文"
- "后续章节同理"
- "按此模式"
- 任何形式的"注"、"说明"、"省略"等跳过章节的表述。
3. 每一章都必须单独、完整地写出，不得省略任何一章。
4. 如果输出中缺失任何一章，或使用任何形式的省略，则视为不合格。
"""
# ==================== 师9 + 木3 ====================
def call_shi9_mu3(seed: dict, dao_novelty: float, sheng_ratio: float) -> Tuple[str, float, str]:
print("\n 📞 师9 + 木3：生成初稿...")
p = seed["protagonist"]
l = seed["love_interest"]
r = seed["rival"]
sides = ", ".join(seed["side_names"])
prefixes = ", ".join(seed["title_prefixes"][:10])
suffixes = ", ".join(seed["title_suffixes"][:10])
min_words = CONFIG.get("chapter_min_words", 2500)
max_words = CONFIG.get("chapter_max_words", 3000)
name_constraint = f"""
【人物姓名约束 - 绝对禁止修改】
以下人物姓名由火2固定生成，在整个提纲生成过程中【严禁任何修改、替换、增删】：
- 主角：{p}
- 恋人：{l}
- 对手：{r}
- 配角：{seed['side_names'][0] if len(seed['side_names']) > 0 else '待定'}、{seed['side_names'][1] if len(seed['side_names']) > 1 else '待定'}、{seed['side_names'][2] if len(seed['side_names']) > 2 else '待定'}
你必须严格使用以上姓名，不得创造新名字或更改已有名字。即使剧情需要，也只能扩展这些角色的背景和关系，不能改变其姓名。
"""
prompt = f"""
请完成两个任务：
### 任务1：师9评种子（0-1分，20字评语）
评价种子素材的主题契合度、素材质量、多样性。
种子素材：
- 冲突：{seed['conflict'].get('名称', '未知')}
- 动机：{seed['motivation'].get('名称', '未知')}
- 主题：{seed['theme'].get('内容', '未知')}
- 风格：{seed['author'].get('name', '未知')}
- 时代：{seed['era'].get('名称', '未知')}
输出：师9评分：X.XX 师9评语：（评语）
### 任务2：木3生成80章提纲
【种子素材】
- 核心冲突：{seed['conflict'].get('名称', '身世之谜')}
- 主角动机：{seed['motivation'].get('名称', '寻找真相')}
- 关键事件：{seed['event'].get('名称', '发现旧物')}
- 转折装置：{seed['device'].get('名称', '旧玉佩')}
- 核心意象：{seed['image'].get('名称', '月光')}
- 作者风格：{seed['author'].get('name', '沈从文')}
- 时代背景：{seed['era'].get('名称', '民国')}
- 对手类型：{seed['opponent'].get('名称', '过去的自己')}
- 主题句：{seed['theme'].get('内容', '')}
- 小说类型：{seed['genre'].get('名称', '家族传奇')}
{name_constraint}
{NO_SKIP_CONSTRAINT}
【章名素材】前7字：{prefixes}；后7字：{suffixes}
【字数要求】每章目标字数：{min_words}-{max_words}字
【输出】完整80章提纲（四个部分：小说信息、核心人物、关键意象、章节大纲）
注意：不需要输出核心冲突总表和伏线总表，小说写手不需要这两个部分。
每章格式：---第X章：七字标题--- 后跟 核心事件、出场人物、关键意象、伏笔、目标字数（{min_words}-{max_words}字）
"""
result = call_deepseek(prompt, max_tokens=32768, temperature=0.6 + sheng_ratio * 0.4)
shi9_score = 0.0
shi9_comment = ""
if result:
m = re.search(r'师9评分[：:]\s*([0-9.]+)', result)
if m:
try: shi9_score = float(m.group(1))
except: pass
cm = re.search(r'师9评语[：:]\s*(.*?)(?=\n---|\n###|$)', result, re.DOTALL)
if cm:
shi9_comment = cm.group(1).strip()[:50]
if "小说信息" in result and len(result) > 500:
return result, shi9_score, shi9_comment
return "", shi9_score, shi9_comment
# ==================== 水1 + 师7 ====================
def call_shi7_shui1(outline: str, seed: dict, completeness: dict) -> Tuple[str, float, str]:
print("\n 📞 师7 + 水1（补齐一次 + 结构评分）...")
min_words = CONFIG.get("chapter_min_words", 2500)
max_words = CONFIG.get("chapter_max_words", 3000)
p = seed["protagonist"]
l = seed["love_interest"]
r = seed["rival"]
side_names = ", ".join(seed["side_names"])
name_constraint = f"""
【人物姓名约束 - 绝对禁止修改】
以下人物姓名由火2固定生成，在整个提纲生成过程中【严禁任何修改、替换、增删】：
- 主角：{p}
- 恋人：{l}
- 对手：{r}
- 配角：{side_names}
你必须严格使用以上姓名，不得创造新名字或更改已有名字。补齐章节时，继续使用这些名字。
"""
if completeness.get("is_complete"):
prompt = f"""
请以师7身份评价以下提纲的结构完整性、伏笔密度、逻辑连贯性。
评分0-1分，20字评语。
{name_constraint}
输出：师7评分：X.XX 师7评语：（评语）
【提纲摘要】
{outline[:1500] if outline else '（空）'}
"""
result = call_deepseek(prompt, max_tokens=500, temperature=0.3)
shi7_score = 0.0
shi7_comment = ""
if result:
m = re.search(r'师7评分[：:]\s*([0-9.]+)', result)
if m:
try: shi7_score = float(m.group(1))
except: pass
cm = re.search(r'师7评语[：:]\s*(.*?)(?=\n|$)', result)
if cm:
shi7_comment = cm.group(1).strip()[:50]
return outline, shi7_score, shi7_comment
missing_desc = "\n".join([f" - {m}" for m in completeness["missing"]])
targets = {
"mysteries": CONFIG["min_mysteries"],
"characters": CONFIG["min_characters"],
"images": CONFIG["min_images"],
"chapters": CONFIG["min_chapters"],
"conflicts": CONFIG["min_conflicts"],
"subplots": CONFIG["min_subplots"],
}
current_chapters = completeness.get("chapters", 0)
chapters_ok = current_chapters >= CONFIG["min_chapters"]
chapter_instruction = ""
if chapters_ok:
chapter_instruction = "\n【重要】章节数已达80章，不要补写新章节，只补齐其他缺失项（谜团、人物、意象）。"
else:
chapter_instruction = f"\n需要补写缺失章节（从第{current_chapters+1}章到第{CONFIG['min_chapters']}章）。"
prompt = f"""
请完成两个任务：
### 任务1：水1强制补齐（只做一次）
修复以下不完整的提纲，**补齐所有缺失项**。
**缺失项：**
{missing_desc}
**目标：**
- 核心谜团：{targets['mysteries']}个 | 核心人物：{targets['characters']}人
- 关键意象：{targets['images']}个 | 章节：{targets['chapters']}章（必须）
- 核心冲突：{targets['conflicts']}种 | 伏线：{targets['subplots']}条
**当前章节数：{current_chapters}**
{chapter_instruction}
{name_constraint}
{NO_SKIP_CONSTRAINT}
**规则：**
- 已有章节一字不改（核心事件、出场人物、关键意象、伏笔、目标字数均保留）。
- 只补写缺失的章节（若章节不足80）或只补齐其他缺失项（若章节已达80）。
- 补写章节时，每章必须包含核心事件、出场人物、关键意象、伏笔、目标字数（{min_words}-{max_words}字）。
- 补全其他缺失项（谜团、人物、意象）。
- 不需要补写核心冲突总表和伏线总表。
### 任务2：师7结构评分
补齐完成后，评价提纲的结构完整性、伏笔密度、逻辑连贯性。
评分0-1分，20字评语。
输出：
师7评分：X.XX
师7评语：（评语）
【修复后的完整提纲】
（直接输出完整提纲，不要任何解释）
"""
result = call_deepseek(prompt, max_tokens=65536, temperature=0.55)
shi7_score = 0.0
shi7_comment = ""
new_outline = outline
if result:
m = re.search(r'师7评分[：:]\s*([0-9.]+)', result)
if m:
try: shi7_score = float(m.group(1))
except: pass
cm = re.search(r'师7评语[：:]\s*(.*?)(?=\n---|\n###|\n师7|$)', result, re.DOTALL)
if cm:
shi7_comment = cm.group(1).strip()[:50]
patterns = [
r'【修复后的完整提纲】\s*\n(.*?)(?=\n师7评分|$)',
r'---小说信息---(.*?)(?=\n师7评分|$)',
]
for pattern in patterns:
m2 = re.search(pattern, result, re.DOTALL)
if m2:
candidate = m2.group(1).strip()
if "小说信息" in candidate and len(candidate) > 500:
new_outline = candidate
break
if new_outline == outline and "小说信息" in result and len(result) > 2000:
new_outline = result
return new_outline, shi7_score, shi7_comment
# ==================== 师6 + 金4 ====================
def call_shi6_jin4(outline: str, seed: dict, shi7_score: float) -> Tuple[str, float, str, str]:
print("\n 📞 师6 + 金4：润色 + 生成简介...")
min_words = CONFIG.get("chapter_min_words", 2500)
max_words = CONFIG.get("chapter_max_words", 3000)
p = seed["protagonist"]
l = seed["love_interest"]
r = seed["rival"]
side_names = ", ".join(seed["side_names"])
book_title = extract_book_title_from_outline(outline)
name_constraint = f"""
【人物姓名约束 - 绝对禁止修改】
以下人物姓名由火2固定生成，在整个提纲生成过程中【严禁任何修改、替换、增删】：
- 主角：{p}
- 恋人：{l}
- 对手：{r}
- 配角：{side_names}
金4润色时，只能润色语言和格式，绝对不得修改任何人物姓名。
"""
prompt = f"""
请完成两个任务：
### 任务1：师6语言评分（0-1分，20字评语）
评价提纲的语言质量、格式规范性、文学性。
### 任务2：金4润色 + 生成简介
**重要约束：**
- 金4只允许润色语言和修正格式问题。
- **严禁修改、添加或删除任何章节内容**。
- 不能改变章节顺序，不能增删章节。
- 检查格式是否统一（如是否用1. 2. 3. 编号核心事件，出场人物是否用顿号分隔等）。
{name_constraint}
**操作：**
- 修正明显的语法错误和错别字。
- 优化句式，使语言更流畅。
- 统一格式（如编号、标点）。
- 生成一段小说简介，**必须包含书名「{book_title}」**，**字数尽量控制在35-50字之间**（但若超出或不足也接受，不影响提纲保存）。
- 保持每章字数在 {min_words}-{max_words} 字。
- **如果无法生成简介，返回空，不要强行凑数。**
【原始提纲】
{outline}
【输出格式】
师6评分：X.XX
师6评语：（评语）
最终提纲：
（润色后的完整提纲，仅语言和格式改动，内容不变）
小说简介：
（尽量35-50字，必须包含书名「{book_title}」。若无法生成，留空不写。）
"""
result = call_deepseek(prompt, max_tokens=32768, temperature=0.5)
final_outline = outline
shi6_score = 0.0
shi6_comment = ""
intro = ""
if result:
m = re.search(r'师6评分[：:]\s*([0-9.]+)', result)
if m:
try: shi6_score = float(m.group(1))
except: pass
cm = re.search(r'师6评语[：:]\s*(.*?)(?=\n最终提纲|\n---|$)', result, re.DOTALL)
if cm:
shi6_comment = cm.group(1).strip()[:50]
fm = re.search(r'最终提纲[：:]\s*\n(.*?)(?=\n小说简介|$)', result, re.DOTALL)
if fm:
final_outline = fm.group(1).strip()
else:
alt_match = re.search(r'### 最终提纲\s*\n(.*?)(?=\n### 小说简介|$)', result, re.DOTALL)
if alt_match:
final_outline = alt_match.group(1).strip()
im = re.search(r'小说简介[：:]\s*(.*?)(?=$)', result, re.DOTALL)
if im:
intro = im.group(1).strip()
intro = re.sub(r'^(小说简介[：:])?', '', intro).strip()
if intro:
# 确保包含书名
if f"《{book_title}》" not in intro and book_title not in intro:
intro = f"《{book_title}》{intro}"
# 不校验长度，只做简单截断（最长保留50字，避免过长）
if len(intro) > 50:
intro = intro[:47] + "..."
else:
intro = ""
# 不强制校验长度，只要非空即保留
if not intro or len(intro) < 1:
intro = ""
if len(final_outline) < 1000 or "小说信息" not in final_outline:
final_outline = outline
return final_outline, shi6_score, shi6_comment, intro
# ==================== 师8 ====================
def call_shi8_summary(outline: str, seed: dict, scores: dict, intro: str, completeness: dict) -> Tuple[float, str]:
print("\n 📞 师8：全景汇总...")
p = seed["protagonist"]
l = seed["love_interest"]
r = seed["rival"]
prompt = f"""
师8评分（0-1分，20字评语）：
评价最终提纲的人物塑造深度、主题一致性、整体完成度。
【人物姓名确认】
主角：{p}，恋人：{l}，对手：{r}
【参考数据】
- 师9（种子质量）：{scores.get('shi9', 0):.2f}
- 师7（结构质量）：{scores.get('shi7', 0):.2f}
- 师6（语言质量）：{scores.get('shi6', 0):.2f}
- 完整性：{'✅ 全部达标' if completeness.get('is_complete') else '⚠️ 存在缺失'}
- 简介：{intro if intro else '（未生成）'}
输出格式：
师8评分：X.XX
师8评语：（评语）
全景观察报告（200字内）：
（分析各环节质量趋势、优缺点、改进建议）
"""
result = call_deepseek(prompt, max_tokens=2000, temperature=0.3)
shi8_score = 0.0
report = ""
if result:
m = re.search(r'师8评分[：:]\s*([0-9.]+)', result)
if m:
try: shi8_score = float(m.group(1))
except: pass
rm = re.search(r'全景观察报告[：:]\s*\n(.*?)(?=$)', result, re.DOTALL)
if rm:
report = rm.group(1).strip()[:500]
return shi8_score, report
# ==================== 主程序 ====================
class OutlineGenerator:
def __init__(self):
print("\n" + "=" * 70)
print("🧠 提纲生成器 V9.2 · 小说写手专用版（简介附在提纲末尾）")
print(" 【修改：放行仅80章，简介不合规不丢弃】")
print("")
print(" 【核心规则】")
print(" ✅ 放行条件：仅80章完整")
print(" ✅ 金4：只润色不增删章节")
print(" ✅ 火2：强制取完整素材（无兜底）")
print(" ✅ 全程：禁止修改人物姓名")
print(" ✅ 姓名池：先取姓，再取名，组合成完整姓名")
print(" ✅ 恋人：性别与主角相反（硬性规定）")
print(" ✅ 简介：附在提纲末尾，不校验长度，不影响放行")
print(" ✅ 保存：只保留'小说信息'及以下内容")
print(" ✅ 禁止省略：强制逐章输出全部80章")
print(" ✅ 输出精简：无核心冲突总表、无伏线总表")
print("")
print(" 流程：木3 → 水1（一次） → 师7 → 金4（润色+简介）")
print(" 每章目标字数：2500-3000字")
print(" 不合格（章节不足80）直接丢弃")
print(" 道（π引擎）+ 河图洛书结构保留")
print("=" * 70)
try:
self.loader = LibraryLoader(CONFIG["lib_dir"])
except Exception as e:
print(f"\n❌ {e}")
sys.exit(1)
self.dao = DaoEngine()
self.hetu = HeTuCenter()
self.fire2 = Fire2(self.loader)
self.checker = CompletenessChecker(CONFIG)
if CONFIG["api_key"] == "sk-你的KEY":
print("\n⚠️ 警告：未配置 API Key")
def run_forever(self):
success_count = 0
round_num = 0
total_attempts = 0
try:
while True:
round_num += 1
total_attempts += 1
print(f"\n{'─'*70}")
print(f"第 {round_num} 轮（总尝试 {total_attempts} 次）")
print(f"{'─'*70}")
dao_novelty = self.dao.get_novelty(6)
phase1, phase2 = self.dao.get_rhythm()
sheng_ratio = 0.5 + 0.3 * math.sin(phase1)
seed = self.fire2.generate_seed()
outline, shi9_score, shi9_comment = call_shi9_mu3(seed, dao_novelty, sheng_ratio)
if not outline:
print(" ❌ 木3未生成有效内容，跳过本轮")
continue
self.hetu.add_score("shi9", shi9_score)
print_full(outline, "木3 初稿")
comp_initial = self.checker.check(outline)
print(f"\n 📊 木3生成: 章节 {comp_initial['chapters']}/{CONFIG['min_chapters']} | "
f"谜团 {comp_initial['mysteries']}/{CONFIG['min_mysteries']} | "
f"人物 {comp_initial['characters']}/{CONFIG['min_characters']} | "
f"意象 {comp_initial['images']}/{CONFIG['min_images']} | "
f"冲突 {comp_initial['conflicts']}/{CONFIG['min_conflicts']} | "
f"伏线 {comp_initial['subplots']}/{CONFIG['min_subplots']}")
outline, shi7_score, shi7_comment = call_shi7_shui1(
outline, seed, comp_initial
)
self.hetu.add_score("shi7", shi7_score)
print(f" 📊 师7: {shi7_score:.2f} | 评语: {shi7_comment}")
print_full(outline, "水1 补齐后")
outline, shi6_score, shi6_comment, intro = call_shi6_jin4(
outline, seed, shi7_score
)
self.hetu.add_score("shi6", shi6_score)
print(f" 📊 师6: {shi6_score:.2f} | 评语: {shi6_comment}")
if intro:
print(f" 📝 简介: {intro} ({len(intro)}字)")
else:
print(f" 📝 简介: （未生成）")
print_full(outline, "金4 润色后")
final_comp = self.checker.check(outline)
chapter_count = final_comp["chapters"]
chap_ok = (chapter_count >= CONFIG["min_chapters"])
print(f"\n 📊 最终验收状态:")
print(f" 章节: {'✅' if chap_ok else '❌'} ({chapter_count}/{CONFIG['min_chapters']})")
if chap_ok:
success_count += 1
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
# 从"小说信息"开始截取
content_to_save = extract_from_book_info(outline)
# 如果有简介，附在末尾（不校验长度）
if intro and len(intro) >= 1:
final_content = content_to_save + f"\n\n### 小说简介\n{intro}"
else:
final_content = content_to_save
fname = f"{CONFIG['output_dir']}/提纲_{timestamp}_第{success_count}部.txt"
with open(fname, 'w', encoding='utf-8') as f:
f.write(final_content)
print(f"\n 🎉 验收通过！已保存：{fname}")
if intro:
print(f" 📝 简介: {intro}")
scores = {"shi9": shi9_score, "shi7": shi7_score, "shi6": shi6_score}
shi8_score, report = call_shi8_summary(outline, seed, scores, intro, final_comp)
self.hetu.add_score("shi8", shi8_score)
print(f" 📊 师8: {shi8_score:.2f}")
if report:
print(f"\n 📋 {report}")
print(f"\n ⚖️ 河图平衡: {self.hetu.balance:.3f}")
print(f" 📊 各师平均: 师9={self.hetu.get_average('shi9'):.2f} | "
f"师7={self.hetu.get_average('shi7'):.2f} | "
f"师6={self.hetu.get_average('shi6'):.2f} | "
f"师8={self.hetu.get_average('shi8'):.2f}")
save_missing_log()
else:
print(f" ❌ 不合格，丢弃: 章节不足80({chapter_count})")
print(f" 🔄 进入下一轮...")
except KeyboardInterrupt:
print(f"\n\n⏸️ 中断，共 {round_num} 轮，尝试 {total_attempts} 次，成功 {success_count} 部")
save_missing_log()
print(" 道已行，万物已演。")
sys.exit(0)
if __name__ == "__main__":
try:
generator = OutlineGenerator()
generator.run_forever()
except Exception as e:
print(f"\n❌ 致命错误: {e}")
import traceback
traceback.print_exc()
save_missing_log()
|
鄂公网安备42060002000282号 )
