"""
HTTP API 路由
提供 REST API 接口
"""
from typing import List, Optional
from fastapi import APIRouter, HTTPException, Depends
from pydantic import BaseModel
from loguru import logger
from service import XiaohongshuService
# ========== 请求模型 ==========
class PublishContentRequest(BaseModel):
"""发布图文请求"""
title: str
content: str
images: List[str]
tags: Optional[List[str]] = None
class PublishVideoRequest(BaseModel):
"""发布视频请求"""
title: str
content: str
video_path: str
cover_path: Optional[str] = None
tags: Optional[List[str]] = None
class SearchRequest(BaseModel):
"""搜索请求"""
keyword: str
filters: Optional[dict] = None
class FeedDetailRequest(BaseModel):
"""Feed 详情请求"""
feed_id: str
xsec_token: str
load_all_comments: bool = False
class CommentRequest(BaseModel):
"""评论请求"""
feed_id: str
xsec_token: str
content: str
class ReplyCommentRequest(BaseModel):
"""回复评论请求"""
feed_id: str
xsec_token: str
comment_id: str
content: str
class UserProfileRequest(BaseModel):
"""用户主页请求"""
user_id: str
xsec_token: str
# ========== 路由 ==========
def create_router(service: XiaohongshuService) -> APIRouter:
"""创建路由"""
router = APIRouter()
# 健康检查
@router.get("/health")
async def health_check():
"""健康检查"""
return {"status": "ok"}
# ========== 登录相关 ==========
@router.get("/api/v1/login/status")
async def check_login_status():
"""检查主站登录状态"""
try:
is_logged_in = await service.check_login_status()
return {"logged_in": is_logged_in}
except Exception as e:
logger.error(f"Check login status failed: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.get("/api/v1/login/creator/status")
async def check_creator_login_status():
"""检查创作中心登录状态"""
try:
is_logged_in = await service.check_creator_login_status()
return {"logged_in": is_logged_in}
except Exception as e:
logger.error(f"Check creator login status failed: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.post("/api/v1/login/creator/init")
async def init_creator_login():
"""初始化创作中心登录（如果主站已登录，会自动同步）"""
try:
result = await service.init_creator_login()
return result
except Exception as e:
logger.error(f"Init creator login failed: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.get("/api/v1/login/qrcode")
async def get_login_qrcode():
"""获取登录二维码"""
try:
result = await service.get_login_qrcode()
return result
except Exception as e:
logger.error(f"Get login qrcode failed: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.delete("/api/v1/login/cookies")
async def delete_cookies():
"""删除 cookies"""
try:
await service.delete_cookies()
return {"message": "Cookies deleted successfully"}
except Exception as e:
logger.error(f"Delete cookies failed: {e}")
raise HTTPException(status_code=500, detail=str(e))
# ========== Feed 相关 ==========
@router.get("/api/v1/feeds/list")
async def list_feeds():
"""获取 Feed 列表"""
try:
feeds = await service.list_feeds()
return {"feeds": [feed.model_dump() for feed in feeds]}
except Exception as e:
logger.error(f"List feeds failed: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.post("/api/v1/feeds/search")
async def search_feeds(request: SearchRequest):
"""搜索内容"""
try:
feeds = await service.search_feeds(request.keyword, request.filters)
return {"feeds": [feed.model_dump() for feed in feeds]}
except Exception as e:
logger.error(f"Search feeds failed: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.post("/api/v1/feeds/detail")
async def get_feed_detail(request: FeedDetailRequest):
"""获取 Feed 详情"""
try:
detail = await service.get_feed_detail(
request.feed_id,
request.xsec_token,
request.load_all_comments
)
return detail.model_dump()
except Exception as e:
logger.error(f"Get feed detail failed: {e}")
raise HTTPException(status_code=500, detail=str(e))
# ========== 评论相关 ==========
@router.post("/api/v1/feeds/comment")
async def post_comment(request: CommentRequest):
"""发表评论"""
try:
success = await service.post_comment_to_feed(
request.feed_id,
request.xsec_token,
request.content
)
return {"success": success}
except Exception as e:
logger.error(f"Post comment failed: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.post("/api/v1/feeds/comment/reply")
async def reply_comment(request: ReplyCommentRequest):
"""回复评论"""
try:
success = await service.reply_comment_to_feed(
request.feed_id,
request.xsec_token,
request.comment_id,
request.content
)
return {"success": success}
except Exception as e:
logger.error(f"Reply comment failed: {e}")
raise HTTPException(status_code=500, detail=str(e))
# ========== 用户相关 ==========
@router.post("/api/v1/user/profile")
async def user_profile(request: UserProfileRequest):
"""获取用户主页"""
try:
profile = await service.user_profile(request.user_id, request.xsec_token)
return profile.model_dump()
except Exception as e:
logger.error(f"Get user profile failed: {e}")
raise HTTPException(status_code=500, detail=str(e))
# ========== 发布相关 ==========
@router.post("/api/v1/publish")
async def publish_content(request: PublishContentRequest):
"""发布图文内容"""
try:
success = await service.publish_content(
request.title,
request.content,
request.images,
request.tags
)
return {"success": success}
except Exception as e:
logger.error(f"Publish content failed: {e}")
raise HTTPException(status_code=500, detail=str(e))
@router.post("/api/v1/publish_video")
async def publish_video(request: PublishVideoRequest):
"""发布视频内容"""
try:
success = await service.publish_video(
request.title,
request.content,
request.video_path,
request.cover_path,
request.tags
)
return {"success": success}
except Exception as e:
logger.error(f"Publish video failed: {e}")
raise HTTPException(status_code=500, detail=str(e))
return router
