diff --git a/CoreRouters/private.py b/CoreRouters/private.py deleted file mode 100644 index 6dfffa7..0000000 --- a/CoreRouters/private.py +++ /dev/null @@ -1,111 +0,0 @@ -"""私聊相关API路由""" -import logging -from typing import List, Dict -from fastapi import APIRouter, HTTPException -from fastapi.responses import JSONResponse - -from core.database import get_db -from core.models import PrivateMessageRequest, CheckBatchRequest, CheckBatchResponse -from utils.message import send_private_message - -logger = logging.getLogger(__name__) - -router = APIRouter() - - -@router.post("/private/send") -async def send_private(request: PrivateMessageRequest): - """发送私聊消息 - - 请求体: - { - "user_id": 123456, - "content": "消息内容", - "msg_type": "text" // 可选,默认为"text" - } - """ - try: - # 验证msg_type - if request.msg_type not in ['text', 'markdown']: - raise HTTPException( - status_code=400, - detail="msg_type必须是'text'或'markdown'" - ) - - # 调用send_private_message - success = await send_private_message( - user_id=request.user_id, - content=request.content, - msg_type=request.msg_type - ) - - if not success: - # 检查用户是否有个人URL - db = get_db() - has_url = db.has_webhook_url(request.user_id) - if not has_url: - raise HTTPException( - status_code=400, - detail=f"用户 {request.user_id} 没有注册个人webhook URL" - ) - else: - raise HTTPException( - status_code=500, - detail="消息发送失败,请稍后重试" - ) - - return JSONResponse({ - "success": True, - "message": "消息发送成功" - }) - - except HTTPException: - raise - except Exception as e: - logger.error(f"发送私聊消息API错误: {e}", exc_info=True) - raise HTTPException( - status_code=500, - detail=f"服务器内部错误: {str(e)}" - ) - - -@router.get("/private/check/{user_id}") -async def check_user_webhook(user_id: int): - """检查用户是否有个人webhook URL""" - try: - db = get_db() - has_webhook_url = db.has_webhook_url(user_id) - - return JSONResponse({ - "user_id": user_id, - "has_webhook_url": has_webhook_url - }) - except Exception as e: - logger.error(f"检查用户webhook URL错误: {e}", exc_info=True) - raise HTTPException( - status_code=500, - detail=f"服务器内部错误: {str(e)}" - ) - - -@router.post("/private/check-batch") -async def check_users_webhook_batch(request: CheckBatchRequest): - """批量检查用户是否有个人webhook URL - - 请求体: - { - "user_ids": [123456, 789012, ...] - } - """ - try: - db = get_db() - results = db.check_users_webhook_urls(request.user_ids) - - return CheckBatchResponse(results=results) - except Exception as e: - logger.error(f"批量检查用户webhook URL错误: {e}", exc_info=True) - raise HTTPException( - status_code=500, - detail=f"服务器内部错误: {str(e)}" - ) -