Novel

import os
import json
from typing import List, Dict, Any
from dataclasses import dataclass, field
import dashscope
from http import HTTPStatus


# ========== 数据结构（同上） ==========
@dataclass
class StoryConfig:
title: str = "未命名小说"
genre: str = "奇幻"
synopsis: str = ""
world_setting: str = ""
characters: List[Dict] = field(default_factory=list)
outline: List[str] = field(default_factory=list)
style_instructions: str = ""


@dataclass
class StoryState:
current_chapter: int = 0
chapter_texts: List[str] = field(default_factory=list)
chapter_summaries: List[str] = field(default_factory=list)
open_foreshadowings: List[str] = field(default_factory=list)
character_states: Dict[str, Dict] = field(default_factory=dict)


# ========== 小说写作Agent（DashScope版） ==========
class NovelWritingAgent:
def __init__(self, api_key: str = None, model: str = "qwen-max"):
dashscope.api_key = api_key or os.getenv("DASHSCOPE_API_KEY")
self.model = model # 如 qwen-max, qwen-plus, qwen-turbo
self.config: StoryConfig = None
self.state: StoryState = None

def _call_llm(self, system_prompt: str, user_content: str = None, temperature: float = 0.7,
response_format: str = None) -> str:
"""统一调用通义千问，支持json格式输出（非强制，但可通过prompt引导）"""
messages = [{"role": "system", "content": system_prompt}]
if user_content:
messages.append({"role": "user", "content": user_content})

# 注意：dashscope 的 JSON 模式需要通过 result_format 或者 prompt 引导
response = dashscope.Generation.call(
model=self.model,
messages=messages,
temperature=temperature,
result_format='message' # 返回完整消息
)
if response.status_code == HTTPStatus.OK:
return response.output.choices[0].message.content
else:
raise Exception(f"API调用失败: {response.code} - {response.message}")

def _call_llm_json(self, system_prompt: str, user_content: str = None, temperature: float = 0.7) -> dict:
"""强制要求输出JSON，并通过额外提示词保证"""
prompt = system_prompt + "\n请确保输出是合法的JSON格式，不要添加任何额外解释。"
text = self._call_llm(prompt, user_content, temperature)
# 尝试提取JSON
try:
return json.loads(text)
except:
# 如果失败，尝试截取 {} 部分
import re
match = re.search(r'\{.*\}', text, re.DOTALL)
if match:
return json.loads(match.group())
raise ValueError("模型未返回有效的JSON")

# ---------- 规划阶段 ----------
def plan_story(self, user_prompt: str) -> StoryConfig:
system_prompt = """你是一个专业的小说规划师。根据用户提供的创作想法，生成完整的故事设定：
- 小说标题（吸引人）
- 体裁（从用户描述中推断）
- 详细的世界观设定（200字以内）
- 主要角色列表（每个角色包含：姓名、身份、性格、核心欲望/目标）
- 分章大纲（10~20章，每章一句话梗概）
请用JSON格式输出，键名为：title, genre, world_setting, characters, outline。
characters是列表，每个元素包含name, role, personality, goal。
outline是字符串列表。
"""
result = self._call_llm_json(system_prompt, user_prompt, temperature=0.7)
self.config = StoryConfig(
title=result["title"],
genre=result["genre"],
synopsis=user_prompt,
world_setting=result["world_setting"],
characters=result["characters"],
outline=result["outline"],
style_instructions=""
)
self.state = StoryState()
for c in self.config.characters:
self.state.character_states[c["name"]] = {"personality": c["personality"], "current_goal": c["goal"]}
return self.config

# ---------- 正文生成 ----------
def write_chapter(self, chapter_idx: int) -> str:
if chapter_idx >= len(self.config.outline):
raise IndexError("章节索引超出大纲范围")
chapter_summary = self.config.outline[chapter_idx]

previous_chapters_summary = "\n".join(self.state.chapter_summaries[-3:])
characters_desc = "\n".join(
[f"- {c['name']}（{c['role']}）：{c['personality']}。当前目标：{c['goal']}" for c in self.config.characters])
foreshadowing_text = "无" if not self.state.open_foreshadowings else "\n".join(self.state.open_foreshadowings)

system_prompt = f"""你是一位小说作家，正在创作{self.config.genre}题材的小说《{self.config.title}》。

## 世界观
{self.config.world_setting}

## 角色设定
{characters_desc}

## 前情回顾（最近3章摘要）
{previous_chapters_summary if previous_chapters_summary else "这是故事开篇。"}

## 未解决的伏笔
{foreshadowing_text}

## 本章大纲
{chapter_summary}

## 风格要求
{self.config.style_instructions if self.config.style_instructions else "流畅自然，适当描写环境和心理。"}

请写出本章正文，字数500~1500字。结尾处可以埋下新的伏笔或制造悬念。直接输出小说正文，不要加额外说明。
"""
chapter_text = self._call_llm(system_prompt, temperature=0.8)

summary = self._summarize_chapter(chapter_text)
self.state.chapter_texts.append(chapter_text)
self.state.chapter_summaries.append(summary)
self.state.current_chapter += 1

new_foreshadows = self._extract_foreshadowings(chapter_text)
self.state.open_foreshadowings.extend(new_foreshadows)
return chapter_text

def _summarize_chapter(self, chapter_text: str) -> str:
return self._call_llm("请用一句话（50字以内）总结下面这段小说的主要情节。", chapter_text, temperature=0.3)

def _extract_foreshadowings(self, chapter_text: str) -> List[str]:
try:
result = self._call_llm_json(
"从以下小说片段中，找出所有伏笔（未解释的悬念、预言、奇怪物品等），以{\"foreshadowings\": [\"伏笔1\", ...]}格式输出。如果没有则输出{\"foreshadowings\": []}。",
chapter_text[:2000],
temperature=0.2
)
return result.get("foreshadowings", [])
except:
return []

def review_chapter(self, chapter_idx: int) -> Dict[str, Any]:
if chapter_idx >= len(self.state.chapter_texts):
return {"error": "章节尚未生成"}
text = self.state.chapter_texts[chapter_idx]
system_prompt = f"""你是一位资深编辑。请仔细阅读以下小说章节，分析：
1. 是否有逻辑矛盾或前后不一致？
2. 角色行为是否符合其性格设定？（角色设定：{self.config.characters}）
3. 是否存在严重的文笔问题（如重复、语病）？
输出JSON格式：{{"has_issues": true/false, "issues": ["具体问题1", ...], "suggestion": "修改建议"}}
"""
return self._call_llm_json(system_prompt, text, temperature=0.4)

def generate_full_novel(self, user_idea: str, chapters_to_write: int = None) -> str:
self.plan_story(user_idea)
total = chapters_to_write if chapters_to_write else len(self.config.outline)
for i in range(min(total, len(self.config.outline))):
print(f"✍️ 正在生成第{i + 1}章...")
chapter = self.write_chapter(i)
print(f"✅ 第{i + 1}章完成，字数：{len(chapter)}")
review = self.review_chapter(i)
if review.get("has_issues"):
print(f"⚠️ 发现问题：{review['issues']}")
print(f"💡 建议：{review.get('suggestion', '无')}")
return "\n\n".join(self.state.chapter_texts)


# ========== 使用示例 ==========
if __name__ == "__main__":
# 设置 API Key: os.environ["DASHSCOPE_API_KEY"] = "sk-xxx"
# 或者直接在代码中传入
agent = NovelWritingAgent(model="qwen-max") # 可选 qwen-plus, qwen-turbo, qwen3.6 系列的具体版本名
user_idea = "一个现代女法医穿越大唐，成为仵作学徒，用现代法医学知识屡破奇案，同时与大理寺少卿展开双强较量，最终揭露前朝宝藏的秘密。"
full_text = agent.generate_full_novel(user_idea, chapters_to_write=2) # 先写2章测试
print(full_text)