zhihu_http_api_server.py•15.9 kB
# -*- coding: utf-8 -*-
"""
知乎发布HTTP API服务器
采用和小红书、头条相同的简单实现方式
"""
import sys
import os
import time
# 强制设置UTF-8编码
if hasattr(sys.stdout, 'reconfigure'):
sys.stdout.reconfigure(encoding='utf-8')
if hasattr(sys.stderr, 'reconfigure'):
sys.stderr.reconfigure(encoding='utf-8')
os.environ['PYTHONIOENCODING'] = 'utf-8'
from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn
from typing import List, Optional
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.common.exceptions import TimeoutException
# 导入知乎发布功能
from zhihu_mcp_server.auth import ZhihuAuth
from zhihu_mcp_server.publisher import ZhihuPublisher
app = FastAPI(title="知乎发布API", version="1.0.0")
# 全局变量存储服务实例
auth_manager = None
publisher = None
def initialize_services():
"""初始化服务实例"""
global auth_manager, publisher
try:
auth_manager = ZhihuAuth()
publisher = ZhihuPublisher(auth_manager)
print("知乎服务实例初始化成功")
# 尝试检查并预热登录环境
try:
if auth_manager.check_login_status():
print("检测到已登录状态，开始预热发布环境...")
driver = auth_manager.driver
# 预热发布环境 - 访问写文章页面并停留在此页面
print("正在访问知乎写文章页面...")
driver.get("https://zhuanlan.zhihu.com/write")
# 等待编辑器加载
try:
WebDriverWait(driver, 15).until(
lambda driver: driver.execute_script("return document.readyState") == "complete"
)
time.sleep(5) # 确保页面完全加载
# 检查是否成功加载
if "write" in driver.current_url:
print("成功进入知乎文章编辑页面，发布环境预热完成")
else:
print(f"预热发布环境可能不完整，当前URL: {driver.current_url}")
except TimeoutException:
print("编辑器加载超时，预热可能不完整")
else:
print("初始化后未检测到登录状态，需要先登录")
except Exception as e:
print(f"预热发布环境失败: {e}")
return True
except Exception as e:
print(f"知乎服务实例初始化失败: {e}")
return False
class CreateAnswerRequest(BaseModel):
"""创建回答请求模型"""
question_id: str
content: str
images: Optional[List[str]] = None
is_anonymous: bool = False
class CreateArticleRequest(BaseModel):
"""创建文章请求模型"""
title: str
content: str
images: Optional[List[str]] = None
tags: Optional[List[str]] = None
column_id: Optional[str] = None
cover_image: Optional[str] = None
publish_time: Optional[str] = None
class CreateColumnRequest(BaseModel):
"""创建专栏请求模型"""
title: str
description: str
cover_image: Optional[str] = None
@app.post("/zhihu-mcp-server/create_answer")
async def create_answer(request: CreateAnswerRequest):
"""创建知乎回答"""
try:
print(f"收到回答发布请求:")
print(f" 问题ID: {request.question_id}")
print(f" 内容长度: {len(request.content)}")
print(f" 图片数量: {len(request.images) if request.images else 0}")
print(f" 匿名回答: {request.is_anonymous}")
# 检查服务是否初始化
if not publisher or not auth_manager:
return {"status": "error", "message": "服务未初始化"}
# 检查登录状态
if not auth_manager.check_login_status():
return {"status": "error", "message": "请先登录"}
# 执行发布
print("开始执行回答发布...")
result = publisher.create_answer(
question_id=request.question_id,
content=request.content,
images=request.images,
is_anonymous=request.is_anonymous
)
print(f"回答发布结果: {result}")
if result.get("success"):
return {"status": "success", "message": result.get("message", "发布成功"), "data": result}
else:
return {"status": "error", "message": result.get("message", "发布失败"), "data": result}
except Exception as e:
import traceback
error_msg = str(e)
error_trace = traceback.format_exc()
print(f"回答发布失败: {error_msg}")
print(f"错误堆栈: {error_trace}")
return {"status": "error", "message": error_msg, "trace": error_trace}
@app.post("/zhihu-mcp-server/create_article")
async def create_article(request: CreateArticleRequest):
"""创建知乎文章"""
try:
print(f"收到文章发布请求:")
print(f" 标题: {request.title}")
print(f" 内容长度: {len(request.content)}")
print(f" 图片数量: {len(request.images) if request.images else 0}")
print(f" 标签: {request.tags}")
# 检查服务是否初始化
if not publisher or not auth_manager:
return {"status": "error", "message": "服务未初始化"}
# 检查登录状态
if not auth_manager.check_login_status():
return {"status": "error", "message": "请先登录"}
# 执行发布
print("开始执行文章发布...")
result = publisher.create_article(
title=request.title,
content=request.content,
images=request.images,
tags=request.tags,
column_id=request.column_id,
cover_image=request.cover_image,
publish_time=request.publish_time
)
print(f"文章发布结果: {result}")
if result.get("success"):
return {"status": "success", "message": result.get("message", "发布成功"), "data": result}
else:
return {"status": "error", "message": result.get("message", "发布失败"), "data": result}
except Exception as e:
import traceback
error_msg = str(e)
error_trace = traceback.format_exc()
print(f"文章发布失败: {error_msg}")
print(f"错误堆栈: {error_trace}")
return {"status": "error", "message": error_msg, "trace": error_trace}
@app.post("/zhihu-mcp-server/create_column")
async def create_column(request: CreateColumnRequest):
"""创建知乎专栏"""
try:
print(f"收到专栏创建请求:")
print(f" 标题: {request.title}")
print(f" 描述: {request.description}")
# 检查服务是否初始化
if not publisher or not auth_manager:
return {"status": "error", "message": "服务未初始化"}
# 检查登录状态
if not auth_manager.check_login_status():
return {"status": "error", "message": "请先登录"}
# 执行创建
print("开始执行专栏创建...")
result = publisher.create_column(
title=request.title,
description=request.description,
cover_image=request.cover_image
)
print(f"专栏创建结果: {result}")
if result.get("success"):
return {"status": "success", "message": result.get("message", "创建成功"), "data": result}
else:
return {"status": "error", "message": result.get("message", "创建失败"), "data": result}
except Exception as e:
import traceback
error_msg = str(e)
error_trace = traceback.format_exc()
print(f"专栏创建失败: {error_msg}")
print(f"错误堆栈: {error_trace}")
return {"status": "error", "message": error_msg, "trace": error_trace}
@app.get("/zhihu-mcp-server/health")
async def health_check():
"""健康检查接口"""
try:
# 检查服务状态
service_status = "initialized" if (publisher and auth_manager) else "not_initialized"
# 安全地检查登录状态，如果失败不中断整个健康检查
login_status = False
try:
if auth_manager:
login_status = auth_manager.check_login_status()
except Exception as login_e:
print(f"健康检查中登录状态检测失败: {login_e}")
return {
"status": "ok",
"service_status": service_status,
"login_status": login_status,
"message": "知乎发布服务运行正常"
}
except Exception as e:
return {
"status": "error",
"message": f"健康检查失败: {str(e)}"
}
@app.get("/zhihu-mcp-server/reinitialize")
async def reinitialize_services():
"""重新初始化服务实例
当登录状态检测失败或浏览器会话失效时，
可以通过这个接口重新初始化服务实例，
而不必重启整个API服务器
"""
global auth_manager, publisher
try:
print("正在重新初始化服务实例...")
# 关闭现有实例
if auth_manager:
try:
auth_manager.close()
print("现有认证管理器已关闭")
except Exception as e:
print(f"关闭认证管理器时出错: {e}")
# 重新创建实例
result = initialize_services()
# 检查初始化结果
if result:
# 尝试立即检查登录状态
login_status = False
try:
if auth_manager:
login_status = auth_manager.check_login_status()
print(f"服务重新初始化后登录状态: {'已登录' if login_status else '未登录'}")
# 如果登录成功，执行预热操作
if login_status:
print("登录成功，开始执行发布环境预热...")
try:
# 预热发布环境 - 访问写文章页面并停留在此页面
driver = auth_manager.driver
print("正在访问写文章页面预热发布环境...")
driver.get("https://zhuanlan.zhihu.com/write")
# 等待编辑器加载
try:
WebDriverWait(driver, 15).until(
lambda driver: driver.execute_script("return document.readyState") == "complete"
)
time.sleep(5) # 确保页面完全加载
# 检查是否成功加载
if "write" in driver.current_url:
print("成功进入知乎文章编辑页面，发布环境预热完成")
else:
print(f"预热发布环境可能不完整，当前URL: {driver.current_url}")
except TimeoutException:
print("编辑器加载超时，预热可能不完整")
except Exception as warm_e:
print(f"预热发布环境失败: {warm_e}")
except Exception as login_e:
print(f"重新初始化后检查登录状态失败: {login_e}")
return {
"status": "success",
"message": "服务实例重新初始化成功",
"login_status": login_status
}
else:
return {
"status": "error",
"message": "服务实例重新初始化失败"
}
except Exception as e:
import traceback
error_trace = traceback.format_exc()
print(f"重新初始化服务时发生错误: {e}")
print(f"错误堆栈: {error_trace}")
return {
"status": "error",
"message": f"重新初始化服务时发生错误: {str(e)}"
}
@app.get("/zhihu-mcp-server/warmup")
async def warmup_publishing_environment():
"""预热发布环境
在有些情况下，浏览器会话可能已经正常，但发布环境需要预热。
此接口访问文章编辑页面，确保后续发布操作可以正常进行。
"""
try:
# 检查服务状态
if not publisher or not auth_manager:
return {"status": "error", "message": "服务未初始化"}
# 检查登录状态
login_status = False
try:
if auth_manager:
login_status = auth_manager.check_login_status()
except Exception as login_e:
return {"status": "error", "message": f"检查登录状态失败: {str(login_e)}"}
if not login_status:
return {"status": "error", "message": "未登录，无法预热发布环境"}
# 执行预热
try:
driver = auth_manager.driver
# 直接访问写文章页面
print("正在访问知乎写文章页面...")
driver.get("https://zhuanlan.zhihu.com/write")
# 等待编辑器加载
try:
WebDriverWait(driver, 15).until(
lambda driver: driver.execute_script("return document.readyState") == "complete"
)
time.sleep(5) # 确保页面完全加载
except TimeoutException:
return {"status": "error", "message": "编辑器加载超时"}
# 检查是否成功加载
write_page_loaded = "write" in driver.current_url
if write_page_loaded:
print("成功进入知乎文章编辑页面")
return {
"status": "success",
"message": "发布环境预热完成，已进入写文章页面",
}
else:
print(f"可能未正确加载编辑页面，当前URL: {driver.current_url}")
return {
"status": "warning",
"message": f"写文章页面可能未正确加载，当前URL: {driver.current_url}"
}
except Exception as e:
import traceback
error_trace = traceback.format_exc()
print(f"预热发布环境失败: {e}")
print(f"错误堆栈: {error_trace}")
return {
"status": "error",
"message": f"预热发布环境失败: {str(e)}",
"trace": error_trace
}
except Exception as e:
return {"status": "error", "message": f"预热过程异常: {str(e)}"}
if __name__ == "__main__":
print("启动知乎发布HTTP API服务器...")
# 初始化服务
if initialize_services():
print("服务初始化成功，启动HTTP API服务器...")
uvicorn.run(app, host="0.0.0.0", port=8005)
else:
print("服务初始化失败，无法启动服务器")
