"""FastMCP 服务器 - 小说处理服务"""
from config import (
MCP_APP_NAME,
BACKGROUND_CHECK_INTERVAL,
TASK_TIMEOUT_MINUTES,
PROMPT_FILE_EXTENSIONS,
ensure_directories,
)
import json
import asyncio
from pathlib import Path
from typing import Any, Dict
from datetime import datetime
from fastmcp import FastMCP
from starlette.staticfiles import StaticFiles
from starlette.responses import RedirectResponse
from .task_manager import TaskManager
from .novel_processor import NovelProcessor
from .logger import get_logger
from .workspace_manager import workspace_manager
# 导入配置
import sys
sys.path.insert(0, str(Path(__file__).parent.parent))
# 确保所有必需的目录存在
ensure_directories()
# 获取日志记录器
logger = get_logger(__name__)
# 创建 FastMCP 应用
mcp = FastMCP(MCP_APP_NAME)
# 创建任务管理器
task_manager = TaskManager()
# 后台任务引用
background_task = None
# ==================== 核心任务管理工具 (MCP) ====================
@mcp.tool()
def run_task(project_name: str) -> str:
"""
获取并执行一个待处理的任务
Args:
project_name: 项目名称（对应输出目录下的文件夹名）
Returns:
任务详情，包含原文和提示词
"""
logger.info(f"运行项目 {project_name} 的下一个待处理任务")
# 获取项目目录
project_dir = workspace_manager.get_output_dir() / project_name
if not project_dir.exists():
return f"❌ 项目不存在: {project_name}"
# 获取当前提示词文件
default_prompt_file = workspace_manager.get_default_prompt_file()
task_info = task_manager.get_next_task(project_name, default_prompt_file)
if not task_info:
logger.info("没有待处理的任务")
return "📭 没有待处理的任务"
logger.info(
f"获取到任务: {task_info['task']['id']}, 文件: {task_info['task']['file_name']}")
task = task_info["task"]
content = task_info["content"]
prompt = task_info["prompt"]
min_words = task_info["min_words"]
# 构造完整的输出路径
output_path = task['output_path']
result = f"""🎯 任务: {task['id']}
📄 文件: {task['file_name']}
📊 原始字数: {task['word_count_original']} 字
⚠️ 最低要求: {min_words} 字 (≥80%)
---
‼️ 重要：你必须按照以下步骤完成任务
## 第一步：改写小说
按照下方提示词的要求，对原文进行改写：
{prompt}
## 第二步：写入文件
**你不能直接在对话中输出改写内容！**
改写完成后，使用 Cursor 的 `write` 工具将改写内容直接写入到目标文件：
**目标文件路径**: `{output_path}`
示例：
```
write(
file_path: "{output_path}",
contents: [你改写后的完整内容]
)
```
## 第三步：更新任务状态
文件写入成功后，调用 `complete_task` 工具更新任务状态：
```
complete_task(
task_id: "{task['id']}",
project_name: "{project_name}"
)
```
**注意事项：**
1. 不要在对话中粘贴改写内容
2. 必须先使用 `write` 工具写入文件
3. 然后调用 `complete_task` 更新任务状态
4. complete_task 会自动从文件中读取字数统计
---
原文：
<novel>
{content}
</novel>
---
现在请开始改写，完成后按步骤写入文件并更新状态。
"""
return result
@mcp.tool()
def complete_task(task_id: str, project_name: str) -> str:
"""
标记任务为已完成（文件已由AI直接写入）
Args:
task_id: 任务ID
project_name: 项目名称
Returns:
完成结果
"""
logger.info(f"标记任务完成: {task_id}, 项目: {project_name}")
result = task_manager.complete_task(
task_id=task_id,
project_name=project_name
)
if result.get("success"):
logger.info(f"任务已完成: {task_id}")
else:
logger.error(f"标记任务完成失败: {task_id}, {result.get('message', '')}")
return result["message"]
@mcp.tool()
def fail_task(task_id: str, project_name: str, error_message: str) -> str:
"""
标记任务为失败
Args:
task_id: 任务ID
project_name: 项目名称
error_message: 错误信息
Returns:
失败结果
"""
logger.warning(
f"标记任务失败: {task_id}, 项目: {project_name}, 错误: {error_message}")
result = task_manager.fail_task(task_id, project_name, error_message)
return result["message"]
@mcp.tool()
def show_web_usage() -> str:
"""
显示如何使用前端网页进行任务管理
Returns:
使用说明和访问地址
"""
from config import SERVER_HOST, SERVER_PORT, BATCH_TASK_DIR_NAME
# 获取当前工作目录信息
work_dir = workspace_manager.get_work_dir()
batch_dir = workspace_manager.get_batch_dir()
prompts_dir = workspace_manager.get_prompts_dir()
output_dir = workspace_manager.get_output_dir()
tasks_file = workspace_manager.get_tasks_file()
# 检查目录是否存在
batch_exists = "✅ 存在" if batch_dir.exists() else "❌ 不存在"
tasks_exists = "✅ 存在" if tasks_file.exists() else "❌ 不存在"
result = f"""
🌐 前端网页使用指南
{'=' * 60}
📁 当前工作目录配置:
工作目录: {work_dir}
批处理目录: {batch_dir} ({batch_exists})
提示词目录: {prompts_dir}
输出目录: {output_dir}
任务文件: {tasks_file} ({tasks_exists})
{'=' * 60}
🚀 启动 Web 服务器:
在命令行中执行以下命令:
cd {work_dir.parent / 'fastmcp'}
uv run app.py
或者使用快捷脚本:
{work_dir.parent / 'fastmcp' / 'run-server.bat'}
如果需要指定工作目录，可以使用:
uv run app.py --work-dir "{work_dir}"
{'=' * 60}
🌍 访问前端网页:
服务启动后，打开浏览器访问:
主页: http://{SERVER_HOST}:{SERVER_PORT}/
页面功能:
📄 index.html - 主控制台（任务监控和管理）
✂️ split.html - 小说分割工具
📝 prompt.html - 提示词管理
📊 tasks.html - 任务列表查看
{'=' * 60}
📖 使用步骤:
1️⃣ 准备小说文件
将小说 txt 文件放在任意位置
2️⃣ 分割小说
访问分割页面，上传小说文件进行分割
分割后的文件会保存在: {output_dir}/[项目名称]/
3️⃣ 编辑提示词
访问提示词管理页面，编辑改写提示词
默认提示词位置: {prompts_dir}/改写提示词.md
4️⃣ 监控任务
回到主控制台页面，查看任务进度
任务会自动保存在: {tasks_file}
5️⃣ 使用 MCP 工具处理任务
在 Cursor 中调用 run_task() 执行任务
改写完成后调用 complete_task() 标记完成
{'=' * 60}
💡 提示:
- MCP 工具会自动读取当前工作目录的配置
- 所有任务数据都存储在 {BATCH_TASK_DIR_NAME} 目录中
- 可以在 Web 界面实时查看任务进度和状态
{'=' * 60}
"""
return result
# ==================== 后台任务 ====================
async def background_check_loop():
"""后台定时检查超时任务"""
logger.info(
f"后台检查已启动，间隔 {BACKGROUND_CHECK_INTERVAL} 秒，超时阈值 {TASK_TIMEOUT_MINUTES} 分钟")
while True:
try:
# 检查所有项目的超时任务
result = task_manager.check_all_projects_timeout(
timeout_minutes=TASK_TIMEOUT_MINUTES
)
if result["checked_count"] > 0:
logger.info(f"后台检查: 检查了 {result['projects_count']} 个项目, "
f"发现 {result['checked_count']} 个超时任务, "
f"完成 {result['completed_count']} 个, "
f"重置 {result['recovered_count']} 个")
await asyncio.sleep(BACKGROUND_CHECK_INTERVAL)
except asyncio.CancelledError:
logger.info("后台检查已停止")
break
except Exception as e:
logger.error(f"后台检查错误: {e}", exc_info=True)
await asyncio.sleep(BACKGROUND_CHECK_INTERVAL)
# 导出用于其他模块
__all__ = ['mcp', 'task_manager', 'background_check_loop']
if __name__ == "__main__":
logger.info("请使用以下命令启动服务:")
logger.info("")
logger.info(" Web 界面服务器: uv run web_server.py")
logger.info(" MCP 工具服务器: uv run mcp_server.py")
logger.info("")
logger.info("详见 README.md 和 QUICKSTART.md")
