138 lines
4.4 KiB
Python
138 lines
4.4 KiB
Python
# ///
|
|
# webhook.py
|
|
# 描述：Webhook管理API路由
|
|
# 作者：AI Generated
|
|
# 创建日期：2026-04-06
|
|
# ///
|
|
|
|
from fastapi import APIRouter, HTTPException, Depends
|
|
from typing import List, Dict, Any
|
|
from pydantic import BaseModel
|
|
|
|
from database_service import get_db_service
|
|
from utils import require_api_key
|
|
|
|
|
|
router = APIRouter(prefix="/webhook", tags=["Webhook管理"])
|
|
|
|
|
|
class WebhookAddRequest(BaseModel):
|
|
name: str
|
|
url: str
|
|
format: str = "bark"
|
|
event_types: List[str] = []
|
|
|
|
|
|
class WebhookUpdateRequest(BaseModel):
|
|
name: str = None
|
|
url: str = None
|
|
format: str = None
|
|
enabled: bool = None
|
|
event_types: List[str] = None
|
|
|
|
|
|
@router.get("/", summary="获取所有Webhook")
|
|
async def get_webhooks(_=Depends(require_api_key)) -> Dict[str, Any]:
|
|
db_service = get_db_service()
|
|
if not db_service:
|
|
raise HTTPException(status_code=500, detail="Database not available")
|
|
|
|
webhooks = db_service.get_all_webhooks()
|
|
return {
|
|
"success": True,
|
|
"data": [w.to_dict() for w in webhooks],
|
|
"count": len(webhooks)
|
|
}
|
|
|
|
|
|
@router.get("/{webhook_id}", summary="获取指定Webhook")
|
|
async def get_webhook(webhook_id: int, _=Depends(require_api_key)) -> Dict[str, Any]:
|
|
db_service = get_db_service()
|
|
if not db_service:
|
|
raise HTTPException(status_code=500, detail="Database not available")
|
|
|
|
webhook = db_service.get_webhook(webhook_id)
|
|
if not webhook:
|
|
raise HTTPException(status_code=404, detail=f"Webhook {webhook_id} not found")
|
|
|
|
return {"success": True, "data": webhook.to_dict()}
|
|
|
|
|
|
@router.post("/", summary="添加Webhook")
|
|
async def add_webhook(request: WebhookAddRequest,
|
|
_=Depends(require_api_key)) -> Dict[str, Any]:
|
|
db_service = get_db_service()
|
|
if not db_service:
|
|
raise HTTPException(status_code=500, detail="Database not available")
|
|
|
|
success = db_service.add_webhook(
|
|
name=request.name,
|
|
url=request.url,
|
|
format=request.format,
|
|
event_types=request.event_types
|
|
)
|
|
|
|
if not success:
|
|
raise HTTPException(status_code=400, detail="Failed to add webhook")
|
|
|
|
return {"success": True, "message": "Webhook added successfully"}
|
|
|
|
|
|
@router.put("/{webhook_id}", summary="更新Webhook")
|
|
async def update_webhook(webhook_id: int,
|
|
request: WebhookUpdateRequest,
|
|
_=Depends(require_api_key)) -> Dict[str, Any]:
|
|
db_service = get_db_service()
|
|
if not db_service:
|
|
raise HTTPException(status_code=500, detail="Database not available")
|
|
|
|
update_data = request.model_dump(exclude_unset=True)
|
|
success = db_service.update_webhook(webhook_id, **update_data)
|
|
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail=f"Webhook {webhook_id} not found")
|
|
|
|
return {"success": True, "message": "Webhook updated successfully"}
|
|
|
|
|
|
@router.delete("/{webhook_id}", summary="删除Webhook")
|
|
async def delete_webhook(webhook_id: int,
|
|
_=Depends(require_api_key)) -> Dict[str, Any]:
|
|
db_service = get_db_service()
|
|
if not db_service:
|
|
raise HTTPException(status_code=500, detail="Database not available")
|
|
|
|
success = db_service.delete_webhook(webhook_id)
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail=f"Webhook {webhook_id} not found")
|
|
|
|
return {"success": True, "message": "Webhook deleted successfully"}
|
|
|
|
|
|
@router.post("/{webhook_id}/enable", summary="启用Webhook")
|
|
async def enable_webhook(webhook_id: int,
|
|
_=Depends(require_api_key)) -> Dict[str, Any]:
|
|
db_service = get_db_service()
|
|
if not db_service:
|
|
raise HTTPException(status_code=500, detail="Database not available")
|
|
|
|
success = db_service.update_webhook(webhook_id, enabled=True)
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail=f"Webhook {webhook_id} not found")
|
|
|
|
return {"success": True, "message": "Webhook enabled"}
|
|
|
|
|
|
@router.post("/{webhook_id}/disable", summary="禁用Webhook")
|
|
async def disable_webhook(webhook_id: int,
|
|
_=Depends(require_api_key)) -> Dict[str, Any]:
|
|
db_service = get_db_service()
|
|
if not db_service:
|
|
raise HTTPException(status_code=500, detail="Database not available")
|
|
|
|
success = db_service.update_webhook(webhook_id, enabled=False)
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail=f"Webhook {webhook_id} not found")
|
|
|
|
return {"success": True, "message": "Webhook disabled"}
|