Init

CoreRouters/private.py

@@ -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)}"
-        )
-