gouki
cbb32ea3b0
get_expiring/get_overview 去掉 expiry_date>=today 下界（仅排除已注销）并新增 overdue_count；renewal_service 续费提醒纳入已过期、文案区分已过期N天/今天到期/N天后；dashboard 用到期日历逐条显示具体到期日与倒计时；api.js 新增 Fmt.countdownText 统一文案。
91 lines
3.1 KiB
Python
91 lines
3.1 KiB
Python
"""续费检查服务
|
||
|
||
检查即将到期的资产，生成提醒消息并通过通知渠道发送。
|
||
"""
|
||
|
||
from datetime import date, timedelta
|
||
|
||
from sqlmodel import Session, select
|
||
|
||
from app.core.config import settings
|
||
from app.models.asset import Asset, AssetStatus
|
||
from app.services import notification_service
|
||
|
||
# 已注销的资产不参与续费提醒
|
||
_SKIP_STATUS = {AssetStatus.CANCELLED}
|
||
|
||
|
||
def get_expiring_assets(session: Session, threshold_days: int) -> list:
|
||
"""返回 threshold_days 天内到期（含已过期未处理）的资产列表 [(asset, days), ...]，按剩余天数升序
|
||
|
||
到期日不设下界：过了续费日仍未注销/续费的资产，提醒不能就此消失。
|
||
性能优化：日期范围与状态过滤均下推到 SQL 层。
|
||
"""
|
||
today = date.today()
|
||
deadline = today + timedelta(days=threshold_days)
|
||
assets = session.exec(
|
||
select(Asset).where(
|
||
Asset.expiry_date.is_not(None),
|
||
Asset.expiry_date <= deadline,
|
||
Asset.status.notin_(_SKIP_STATUS),
|
||
)
|
||
).all()
|
||
expiring = [(asset, (asset.expiry_date - today).days) for asset in assets]
|
||
expiring.sort(key=lambda x: x[1])
|
||
return expiring
|
||
|
||
|
||
def build_renewal_message(expiring: list, threshold_days: int) -> str:
|
||
"""生成续费提醒消息（纯文本，兼容 Telegram/邮件）"""
|
||
if not expiring:
|
||
return ""
|
||
overdue = sum(1 for _, d in expiring if d < 0)
|
||
if overdue:
|
||
head = f"⏰ 资产续费提醒（已过期 {overdue} 项 + {threshold_days} 天内到期 {len(expiring) - overdue} 项）"
|
||
else:
|
||
head = f"⏰ 资产续费提醒（{threshold_days} 天内到期，共 {len(expiring)} 项）"
|
||
lines = [head, ""]
|
||
for asset, days in expiring:
|
||
if days < 0:
|
||
tag = f"（已过期 {-days} 天）"
|
||
elif days == 0:
|
||
tag = "（今天到期）"
|
||
else:
|
||
tag = f"（{days} 天后）"
|
||
lines.append(f"• {asset.name} [{asset.provider}] {tag} 到期 {asset.expiry_date}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def check_renewals(session: Session, threshold_days: int = None, send: bool = True) -> dict:
|
||
"""检查续费并（可选）发送提醒
|
||
|
||
threshold_days: 提醒阈值天数，默认取配置 RENEWAL_THRESHOLD_DAYS
|
||
send: 是否实际发送通知（False 时仅返回预览）
|
||
"""
|
||
if threshold_days is None:
|
||
threshold_days = settings.RENEWAL_THRESHOLD_DAYS
|
||
|
||
expiring = get_expiring_assets(session, threshold_days)
|
||
message = build_renewal_message(expiring, threshold_days)
|
||
|
||
result = {
|
||
"threshold_days": threshold_days,
|
||
"expiring_count": len(expiring),
|
||
"expiring": [
|
||
{
|
||
"name": asset.name,
|
||
"provider": asset.provider,
|
||
"days": days,
|
||
"expiry_date": str(asset.expiry_date),
|
||
}
|
||
for asset, days in expiring
|
||
],
|
||
"message": message,
|
||
}
|
||
|
||
if send and message:
|
||
result["notifications"] = notification_service.notify(message)
|
||
else:
|
||
result["notifications"] = []
|
||
return result
|