"""统一的服务器入口 - 支持 Web 界面和 MCP SSE"""
from src.server import mcp, background_check_loop
from src.routes import get_all_routes
from src.logger import setup_app_logger, get_logs_dir
import uvicorn
import sys
import os
import signal
import asyncio
from pathlib import Path
from contextlib import asynccontextmanager
from starlette.applications import Starlette
# 导入配置
from config import (
LOG_LEVEL,
LOGS_DIR,
SERVER_HOST,
SERVER_PORT,
MCP_MOUNT_PATH,
ensure_directories,
)
# 确保所有必需的目录存在
ensure_directories()
# 延迟初始化 logger（将在 create_app 或 main 中创建）
logger = None
# 添加 src 到路径
sys.path.insert(0, str(Path(__file__).parent / "src"))
# 后台任务引用
background_task = None
@asynccontextmanager
async def app_lifespan(app):
"""应用生命周期管理 - 启动和停止后台任务"""
global background_task
import logging
from config import APP_LOGGER_NAME
logger = logging.getLogger(APP_LOGGER_NAME)
logger.info("应用启动中...")
# 启动后台检查任务
background_task = asyncio.create_task(background_check_loop())
logger.info("后台检查任务已创建并启动")
# yield 之前的代码在应用启动时运行
yield
# yield 之后的代码在应用关闭时运行
logger.info("应用关闭中...")
if background_task:
background_task.cancel()
try:
await background_task
except asyncio.CancelledError:
pass
logger.info("后台检查任务已停止")
def create_app():
"""创建完整应用 - 集成 FastMCP"""
from starlette.routing import Mount, Route
from src.routes import add_middleware
import logging
# 获取已配置的 logger（在 app.py 中创建）
from config import APP_LOGGER_NAME
logger = logging.getLogger(APP_LOGGER_NAME)
logger.info("开始创建 FastMCP 应用...")
# 首先创建 MCP 的 HTTP 应用（必须先创建，以获取 lifespan）
# 关键：path='/' 表示在其内部应用的根路径，然后我们通过 Mount 挂载到 /mcp
mcp_app = mcp.http_app(path='/')
logger.info(f"MCP HTTP 应用已创建: {type(mcp_app)}")
logger.info(
f"MCP 应用路由: {mcp_app.routes if hasattr(mcp_app, 'routes') else 'N/A'}")
# 获取所有路由配置（从 routes.py）
all_routes = get_all_routes(mcp_app, MCP_MOUNT_PATH)
logger.info(f"已从路由配置加载 {len(all_routes)} 个路由")
# 创建 Starlette 应用，使用自定义的 lifespan（包含 MCP 和后台任务）
base_app = Starlette(
routes=all_routes,
lifespan=app_lifespan # 使用自定义 lifespan 管理后台任务
)
# 添加中间件
add_middleware(base_app)
logger.info("已添加中间件")
# 打印所有路由信息
logger.info("=" * 60)
logger.info("应用路由列表:")
for i, route in enumerate(base_app.routes):
if isinstance(route, Mount):
logger.info(
f" [{i}] Mount: {route.path} -> {type(route.app).__name__}")
elif isinstance(route, Route):
logger.info(f" [{i}] Route: {route.path} {route.methods}")
else:
logger.info(f" [{i}] {type(route).__name__}: {route}")
logger.info("=" * 60)
logger.info(f"MCP 端点已成功挂载到 {MCP_MOUNT_PATH}")
logger.info(
f"Cursor 配置使用: http://{SERVER_HOST}:{SERVER_PORT}{MCP_MOUNT_PATH}")
return base_app
def setup_signal_handlers():
"""设置信号处理器，确保 Ctrl+C 能正确终止程序"""
import logging
from config import APP_LOGGER_NAME
logger = logging.getLogger(APP_LOGGER_NAME)
def signal_handler(signum, frame):
logger.info("=" * 60)
logger.info("收到终止信号，正在关闭服务器...")
logger.info("=" * 60)
sys.exit(0)
# 注册信号处理器
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
logger.info("信号处理器已设置")
def run_app():
"""Web 应用入口点（用于 uvx 和命令行）"""
from src.utils import kill_port_process
from src.cli_args import (
parse_args,
setup_environment,
ensure_work_directory,
get_log_level
)
# 解析参数
args = parse_args(mode="web")
# 设置环境变量
setup_environment(args)
# 确保工作目录存在
ensure_work_directory(args)
# 清理端口
kill_port_process(args.port)
# 设置日志级别并初始化 logger（直接运行时使用控制台输出）
log_level = get_log_level(args)
logger = setup_app_logger(level=log_level, console_output=True)
# 设置信号处理器
setup_signal_handlers()
app = create_app()
logger.info("=" * 60)
logger.info("小说处理服务器已启动")
logger.info("=" * 60)
if args.work_dir:
logger.info(f"工作目录: {args.work_dir}")
logger.info(f"批处理目录: {Path(args.work_dir) / args.batch_task_dir}")
logger.info(f"Web 界面: http://{args.host}:{args.port}/")
# logger.info(
# f"MCP 端点: http://{args.host}:{args.port}{MCP_MOUNT_PATH}")
logger.info("=" * 60)
logger.info("Cursor 配置:")
logger.info(
f' "url": "http://{args.host}:{args.port}{MCP_MOUNT_PATH}"')
logger.info(f"日志目录: {get_logs_dir()}")
logger.info("按 Ctrl+C 停止服务器")
# 使用配置运行 uvicorn，确保能正确响应信号
try:
uvicorn.run(
app,
host=args.host,
port=args.port,
# 以下配置有助于改善 Windows 上的信号处理
log_level="info",
access_log=False, # 减少日志输出
)
except KeyboardInterrupt:
logger.info("=" * 60)
logger.info("服务器已停止")
logger.info("=" * 60)
except Exception as e:
logger.error(f"服务器错误: {e}")
sys.exit(1)
if __name__ == "__main__":
run_app()
