Zhihu MCP Server

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import logging import os from typing import Optional, Any, Dict # 修正导入路径，确保它们相对于项目结构是正确的 # 假设 main.py 位于 zhihu_mcp_server/ 目录下 # 如果 main.py 在 zhihu_mcp_server/zhihu_mcp_server/ 内部，则需要调整 from zhihu_mcp_server.auth import ZhihuAuth # <--- 没有点，使用 ZhihuAuth from zhihu_mcp_server.publisher import ZhihuPublisher, ArticleData as PublisherArticleData # 重命名导入的 ArticleData from zhihu_mcp_server.utils import setup_logging # 日志设置 setup_logging() logger = logging.getLogger(__name__) # Pydantic模型定义移到前面，并且在 publisher.py 中的 ArticleData 基础上扩展或保持一致 class ArticleDataApi(PublisherArticleData): # 继承自 publisher 中的 ArticleData # ArticleDataApi 现在拥有 title, content, image_paths, tags # 如果需要API特定字段，可以在这里添加，或者直接使用 PublisherArticleData # 为了接收来自客户端的 cover_image，我们在这里确保它存在 cover_image: Optional[str] = None app = FastAPI() # 全局实例 auth_manager_instance: Optional[ZhihuAuth] = None # Ensure it's Optional for robustness publisher_instance: Optional[ZhihuPublisher] = None def initialize_service_components(): """Initializes or re-initializes service components.""" # 修正文档字符串 global auth_manager_instance, publisher_instance logger.info("开始服务组件初始化...") if auth_manager_instance: logger.info("检测到现有AuthManager实例，正在关闭其WebDriver...") auth_manager_instance.close_driver() # Setting to None 필수, to allow re-creation auth_manager_instance = None try: # 假设 main.py 在项目的根目录 zhihu_mcp_server/ 下 project_root_dir = os.path.dirname(os.path.abspath(__file__)) # cookies.json 存储在 项目根目录/zhihu_cookies.json cookies_file_path = os.path.join(project_root_dir, 'zhihu_cookies.json') logger.info(f"AuthManager将使用Cookie文件路径: {cookies_file_path}") # 确保包含cookie文件的目录存在（如果您的cookie文件在子目录中，则需要此步骤） # 如果cookie文件直接在项目根目录，则下面这部分可以省略或调整 # cookie_parent_dir = os.path.dirname(cookies_file_path) # if not os.path.exists(cookie_parent_dir): # os.makedirs(cookie_parent_dir, exist_ok=True) # logger.info(f"为Cookie文件创建了父目录: {cookie_parent_dir}") auth_manager_instance = ZhihuAuth(cookies_file_path=cookies_file_path) publisher_instance = ZhihuPublisher(auth_manager=auth_manager_instance) logger.info("服务组件初始化成功。") # Initial check, can be light or full based on preference. # For startup, a light check is often sufficient. initial_login_status = auth_manager_instance.check_login_status(require_browser_session=False) if initial_login_status is True: logger.info("用户已登录 (通过启动时检查)。") elif initial_login_status is False: logger.warning("用户未登录 (通过启动时检查)。") else: # None logger.info("用户登录状态未知 (通过启动时轻量级检查)。") except Exception as e: logger.error(f"服务组件初始化过程中发生严重错误: {e}", exc_info=True) # Reset instances on failure to prevent using partially initialized objects auth_manager_instance = None publisher_instance = None # We might not want to raise HTTPException here, as it could stop the server from starting # Or, if these components are critical, then it's appropriate. # For now, let's log and allow server to start in a degraded state if this fails. # raise HTTPException(status_code=500, detail=f"关键服务初始化失败: {e}") @app.on_event("startup") async def startup_event(): logger.info("MCP服务器正在启动生命周期事件...") initialize_service_components() logger.info("MCP服务器已启动并完成组件初始化尝试。") # No longer trying to close driver immediately after startup here # Let the /health or other endpoints manage browser session if needed @app.on_event("shutdown") async def shutdown_event(): global auth_manager_instance logger.info("MCP服务器正在关闭生命周期事件...") if auth_manager_instance: logger.info("正在关闭AuthManager持有的WebDriver...") auth_manager_instance.close_driver() logger.info("MCP服务器已成功关闭。") class ReinitializeResponse(BaseModel): status: str message: str login_status: Optional[bool] # Can be True, False, or None details: Optional[Dict[str, Any]] = None class HealthResponse(BaseModel): status: str login_status: Optional[bool] message: Optional[str] = None service_initialized: bool @app.get("/zhihu-mcp-server/health", response_model=HealthResponse) async def health_check(): """轻量级健康检查，主要检查服务是否初始化以及缓存的登录状态。不启动浏览器。""" logger.info("收到 /health (轻量级) 健康检查请求...") global auth_manager_instance service_initialized = False login_s = None msg = "" if auth_manager_instance and publisher_instance: service_initialized = True try: login_s = auth_manager_instance.check_login_status(require_browser_session=False) if login_s is True: msg = "服务已初始化，缓存状态为已登录。" elif login_s is False: msg = "服务已初始化，缓存状态为未登录。" else: # None msg = "服务已初始化，缓存的登录状态未知或已过期。" except Exception as e: logger.error(f"/health 检查登录状态时出错: {e}", exc_info=True) msg = f"检查登录状态时出错: {e}" # login_s remains None or its last value else: msg = "核心服务组件 (AuthManager 或 Publisher) 未初始化。" return HealthResponse( status="ok" if service_initialized else "error", login_status=login_s, message=msg, service_initialized=service_initialized ) @app.get("/zhihu-mcp-server/reinitialize", response_model=ReinitializeResponse) async def reinitialize_service_endpoint(): """强制重新初始化服务组件，并执行完整的登录状态检查 (可能启动浏览器)。""" logger.info("收到 /reinitialize (强制刷新与检查) 请求...") try: initialize_service_components() login_s: Optional[bool] = None message = "服务实例已成功重新初始化。" details_dict = {} if auth_manager_instance: # This is a full check, will start browser if needed login_s = auth_manager_instance.check_login_status(require_browser_session=True) if login_s is True: message += " 用户已登录。" details_dict = auth_manager_instance.get_user_info() # Get user info if logged in elif login_s is False: message += " 用户未登录。可能需要手动登录。" else: # login_status is None, should not happen with require_browser_session=True unless error message += " 登录状态检查未能明确确定 (可能发生错误)。" else: message = "服务实例重新初始化后，AuthManager未能成功创建。" logger.error(message) # Return 503 as service is not usable return ReinitializeResponse(status="error", message=message, login_status=None, details={"error_detail": "AuthManager is None after init"}) return ReinitializeResponse(status="success", message=message, login_status=login_s, details=details_dict) except Exception as e: logger.error(f"服务重新初始化过程中发生未知错误: {e}", exc_info=True) return ReinitializeResponse(status="error", message=f"重新初始化过程失败: {e}", login_status=None, details={"error_detail": str(e)}) class BrowserControlResponse(BaseModel): status: str message: str @app.post("/zhihu-mcp-server/close_browser", response_model=BrowserControlResponse) async def close_browser_session(): """尝试关闭由AuthManager管理的任何活动的浏览器会话。""" logger.info("收到 /close_browser 请求...") global auth_manager_instance if auth_manager_instance: try: auth_manager_instance.close_driver() logger.info("通过API请求关闭浏览器会话成功。") return BrowserControlResponse(status="success", message="浏览器会话已尝试关闭。") except Exception as e: logger.error(f"关闭浏览器会话时发生错误: {e}", exc_info=True) return BrowserControlResponse(status="error", message=f"关闭浏览器时出错: {e}") else: logger.warning("/close_browser 请求，但AuthManager未初始化。") return BrowserControlResponse(status="error", message="AuthManager未初始化，无法关闭浏览器。") class ArticleResponse(BaseModel): status: str message: str data: Optional[dict] = None # Made data optional @app.post("/zhihu-mcp-server/create_article", response_model=ArticleResponse) async def create_article_endpoint(article_data: ArticleDataApi): global publisher_instance, auth_manager_instance logger.info(f"收到文章发布请求: 标题: {article_data.title}, 内容长度: {len(article_data.content or '')}, 图片数量: {len(article_data.image_paths or [])}, 封面图: {article_data.cover_image}, 标签: {article_data.tags}") if not auth_manager_instance or not publisher_instance: logger.error("发布文章：核心服务未初始化。") # Try to reinitialize once if critical components are missing try: logger.info("发布文章：尝试自动重新初始化核心服务...") initialize_service_components() if not auth_manager_instance or not publisher_instance: logger.error("发布文章：自动重新初始化后核心服务仍然不可用。") raise HTTPException(status_code=503, detail="核心服务在自动重新初始化后仍不可用。") except Exception as e_init: logger.error(f"发布文章：自动重新初始化核心服务失败: {e_init}", exc_info=True) raise HTTPException(status_code=503, detail=f"核心服务自动重新初始化失败: {e_init}") try: # Perform a full login check (will use browser if needed) before attempting to publish current_login_status = auth_manager_instance.check_login_status(require_browser_session=True) if current_login_status is not True: logger.error("发布文章：用户未登录或登录状态无法确认。") return ArticleResponse(status="error", message="用户未登录或登录状态无法确认，请先通过 /reinitialize 确保登录。", data={"success": False}) except Exception as e_check: logger.error(f"发布文章：检查登录状态时发生错误: {e_check}", exc_info=True) return ArticleResponse(status="error", message=f"检查登录状态时出错: {e_check}", data={"success": False}) try: result = publisher_instance.create_article( title=article_data.title, content=article_data.content, image_paths=article_data.image_paths, tags=article_data.tags, cover_image=article_data.cover_image ) if result.get("success"): logger.info(f"文章 '{article_data.title}' 发布成功。") return ArticleResponse(status="success", message="文章发布成功", data=result) else: logger.error(f"文章 '{article_data.title}' 发布失败: {result.get('message')}") return ArticleResponse(status="error", message=result.get("message", "文章发布失败，未知原因。"), data=result) except Exception as e_publish: logger.error(f"发布文章 '{article_data.title}' 过程中发生未处理的异常: {e_publish}", exc_info=True) return ArticleResponse(status="error", message=f"发布文章时发生服务器内部错误: {e_publish}", data={"success": False, "detail": str(e_publish)}) @app.get("/zhihu-mcp-server/status") # This can be deprecated in favor of /health async def get_status_deprecated(): logger.warning("接口 /zhihu-mcp-server/status 已被弃用，请使用 /zhihu-mcp-server/health。") global auth_manager_instance, publisher_instance # Added publisher_instance login_s = None # Default to None service_initialized_properly = False if auth_manager_instance and publisher_instance: # Using light check for this deprecated status endpoint login_s = auth_manager_instance.check_login_status(require_browser_session=False) service_initialized_properly = True return { "service_status": "running" if service_initialized_properly else "degraded_or_uninitialized", "login_status": login_s, "service_components_initialized": service_initialized_properly, "deprecation_warning": "This endpoint is deprecated. Please use /zhihu-mcp-server/health for lightweight status or /zhihu-mcp-server/reinitialize for a full check and refresh." } if __name__ == "__main__": import uvicorn logger.info("以直接脚本方式启动Uvicorn服务器 (主要用于开发测试)...") # 推荐从命令行运行: uvicorn zhihu_mcp_server.main:app --reload --port 8000 uvicorn.run("main:app", host="0.0.0.0", port=8005, reload=True) # reload=True 方便开发