43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
"""通知与续费提醒路由"""
|
|
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlmodel import Session
|
|
|
|
from app.core.security import require_api_key
|
|
from app.database import get_session
|
|
from app.services import notification_service, renewal_service
|
|
|
|
router = APIRouter(prefix="/api/notify", tags=["notify"])
|
|
|
|
|
|
@router.get("/channels", summary="已配置的通知渠道")
|
|
def channels() -> dict:
|
|
return {"channels": notification_service.configured_channels()}
|
|
|
|
|
|
@router.post(
|
|
"/test",
|
|
summary="测试通知渠道",
|
|
dependencies=[Depends(require_api_key)],
|
|
)
|
|
def test_notify() -> dict:
|
|
results = notification_service.notify(
|
|
"✅ 这是一条来自 vps-manager 的测试通知", subject="vps-manager 测试"
|
|
)
|
|
return {"notifications": results}
|
|
|
|
|
|
@router.post(
|
|
"/renewals",
|
|
summary="检查续费并发送提醒",
|
|
dependencies=[Depends(require_api_key)],
|
|
)
|
|
def check_renewals(
|
|
threshold: Optional[int] = Query(default=None, description="提醒阈值天数，默认取配置"),
|
|
send: bool = Query(default=True, description="是否实际发送（false 仅预览）"),
|
|
session: Session = Depends(get_session),
|
|
) -> dict:
|
|
return renewal_service.check_renewals(session, threshold_days=threshold, send=send)
|