37 lines
1.2 KiB
Python
37 lines
1.2 KiB
Python
"""定时检查续费并发送提醒（供 systemd timer 调用）
|
|
|
|
用法：python scripts/check_renewals.py [threshold_days]
|
|
依赖 .env 中的通知渠道配置（TELEGRAM_*/SMTP_*）。
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# 确保能导入 app 包（脚本位于 scripts/ 子目录）
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from sqlmodel import Session
|
|
|
|
from app.database import assets_engine, init_db
|
|
from app.services import renewal_service
|
|
|
|
|
|
def main() -> int:
|
|
threshold = int(sys.argv[1]) if len(sys.argv) > 1 else None
|
|
init_db()
|
|
with Session(assets_engine) as session:
|
|
result = renewal_service.check_renewals(session, threshold_days=threshold)
|
|
print(
|
|
f"[续费检查] 阈值 {result['threshold_days']} 天 | 即将到期 {result['expiring_count']} 项"
|
|
)
|
|
for item in result["expiring"]:
|
|
print(f" - {item['name']} [{item['provider']}] {item['days']} 天后到期")
|
|
for n in result.get("notifications", []):
|
|
status = "成功" if n["ok"] else f"失败：{n['message']}"
|
|
print(f" 通知[{n['channel']}]: {status}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|