"""定时检查站点 SSL 证书并发送到期提醒(供 systemd timer 调用) 用法:python scripts/check_site_certs.py [threshold_days] 依赖 .env 中的通知渠道配置(TELEGRAM_*/SMTP_*)。 """ import sys from datetime import date from pathlib import Path # 确保能导入 app 包(脚本位于 scripts/ 子目录) sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from sqlmodel import Session, select from app.database import assets_engine, init_db from app.models.ssl import SiteCert from app.services import notification_service, ssl_service def main() -> int: threshold = int(sys.argv[1]) if len(sys.argv) > 1 else ssl_service.EXPIRING_THRESHOLD init_db() with Session(assets_engine) as session: result = ssl_service.check_all_site_certs(session) stats = result["stats"] # 即将到期 / 已过期 / 探测失败的证书 → 发提醒 certs = session.exec(select(SiteCert)).all() urgent = [ c for c in certs if c.status in ("expiring", "expired", "error") ] notifications = [] if urgent: lines = [ssl_service.build_cert_message(c) for c in urgent] message = "【SSL 证书到期提醒】\n" + "\n".join(lines) notifications = notification_service.notify( message, subject="SSL 证书到期提醒" ) print( f"[SSL检查] 共 {stats['total']} 个站点 | 正常 {stats['ok']} | " f"即将到期 {stats['expiring']} | 已过期 {stats['expired']} | 探测失败 {stats['error']}" ) for p in result["problems"]: print(f" - {p['hostname']}: {p['detail']}") for n in notifications: status = "成功" if n["ok"] else f"失败:{n['message']}" print(f" 通知[{n['channel']}]: {status}") return 0 if __name__ == "__main__": sys.exit(main())