feat(renewal): 提醒前核对平台真实状态,消除手工数据导致的误报

容错策略:核对失败/无可用通道时标记「未核对」并继续按本地数据提醒,
避免因核对链路故障漏掉真正到期的资产;已停止资产仅在 auto_renew=true
时提醒(停止实例不等于停止扣费)。新增 WHMCS 面板只读适配器,
用于 TotHost 这类没有开放 API 的平台。
This commit is contained in:
gouki
2026-10-25 23:05:39 +00:00
parent de8a3caedd
commit abd2fa8206
12 changed files with 1229 additions and 27 deletions
+33 -5
View File
@@ -1,9 +1,13 @@
"""定时检查续费并发送提醒(供 systemd timer 调用)
用法:python scripts/check_renewals.py [threshold_days]
依赖 .env 中的通知渠道配置(TELEGRAM_*/SMTP_*)。
用法:
python scripts/check_renewals.py [threshold_days] [--dry-run] [--no-verify]
默认发送通知;--dry-run 只打印、不发送,便于核对平台真实状态。
依赖 .env 中的通知渠道配置与账号登录凭据(TELEGRAM_*/SMTP_*、credentials)。
"""
import argparse
import sys
from pathlib import Path
@@ -17,21 +21,45 @@ from app.services import renewal_service
def main() -> int:
threshold = int(sys.argv[1]) if len(sys.argv) > 1 else None
parser = argparse.ArgumentParser(description="检查资产续费并发送提醒")
parser.add_argument("threshold_days", nargs="?", type=int, default=None)
parser.add_argument("--dry-run", action="store_true", help="只打印,不发送通知")
parser.add_argument("--no-verify", action="store_true", help="跳过平台真实状态核对")
args = parser.parse_args()
init_db()
with Session(assets_engine) as session:
result = renewal_service.check_renewals(session, threshold_days=threshold)
result = renewal_service.check_renewals(
session,
threshold_days=args.threshold_days,
send=not args.dry_run,
verify=not args.no_verify,
)
# days 可为负(已过期),汇总行拆开统计,单项文案复用 day_text(与提醒消息一致)
overdue = sum(1 for item in result["expiring"] if item["days"] < 0)
print(
summary = (
f"[续费检查] 阈值 {result['threshold_days']} 天 | 已过期 {overdue} 项 | "
f"即将到期 {result['expiring_count'] - overdue} 项"
)
verification = result.get("verification")
if verification:
summary += (
f" | 已核对 {verification['verified']} 项"
f" | 未核对 {verification['unverified']} 项"
)
print(summary)
for item in result["expiring"]:
print(
f" - {item['name']} [{item['provider']}] "
f"{renewal_service.day_text(item['days'])}(到期 {item['expiry_date']})"
)
if item.get("verified") is True:
print(f" 已核对[{item['verify_source']}]: {item['verify_detail']}")
elif item.get("verified") is False:
print(f" 未核对: {item['verify_detail']}")
for n in result.get("notifications", []):
status = "成功" if n["ok"] else f"失败:{n['message']}"
print(f" 通知[{n['channel']}]: {status}")
+2 -1
View File
@@ -26,7 +26,8 @@ def main() -> int:
print(
f"[云资产同步] 账号 {result['total']} | 成功 {result['success']} | 失败 {result['failed']} "
f"| 新增 {result['created']} | 更新 {result['updated']} "
f"| 跳过(无凭证) {result['skipped_no_config']} | 跳过(无适配器) {result['skipped_no_adapter']}"
f"| 跳过(无凭证) {result['skipped_no_config']} | 跳过(无适配器) {result['skipped_no_adapter']} "
f"| 跳过(面板核对) {result.get('skipped_panel', 0)}"
)
for err in result["errors"]:
print(f" - {err}")