# -*- coding: utf-8 -*- """轻量后台任务:企业微信 23:00(北京时间)日报。""" from __future__ import annotations import logging import os import threading from datetime import datetime, timedelta, timezone from pathlib import Path from app.config import Config, DATA_DIR from app import db as database from app import wecom log = logging.getLogger("gold_monitor.scheduler") BJT = timezone(timedelta(hours=8)) _started = False _lock = threading.Lock() def _try_claim_daily(day: str) -> bool: """多 worker 下用独占锁文件抢占当日日报;成功后再写 settings 备查。""" DATA_DIR.mkdir(parents=True, exist_ok=True) lock_path = Path(DATA_DIR) / f"wecom_daily_{day}.sent" try: fd = os.open(str(lock_path), os.O_CREAT | os.O_EXCL | os.O_WRONLY) try: os.write(fd, f"pid={os.getpid()} ts={time_stamp()}\n".encode("utf-8")) finally: os.close(fd) except FileExistsError: return False try: with database.get_db() as conn: database.set_setting(conn, wecom.SETTING_DAILY_LAST, day) except Exception: # noqa: BLE001 log.exception("record daily last failed") return True def time_stamp() -> str: return datetime.now(BJT).isoformat() def _daily_tick() -> None: hour = int(Config.WECOM_DAILY_HOUR) minute = int(Config.WECOM_DAILY_MINUTE) now = datetime.now(BJT) if now.hour != hour or now.minute != minute: return if not wecom.is_enabled() or not wecom.webhook_url(): return day = now.strftime("%Y-%m-%d") if not _try_claim_daily(day): return lock_path = Path(DATA_DIR) / f"wecom_daily_{day}.sent" try: ok, detail = wecom.send_daily_reports(day) log.info("wecom daily %s ok=%s detail=%s", day, ok, detail) if not ok: try: lock_path.unlink(missing_ok=True) except OSError: pass except Exception: # noqa: BLE001 log.exception("wecom daily failed") try: lock_path.unlink(missing_ok=True) except OSError: pass def _loop(stop_event: threading.Event) -> None: # 错开启动,减少多 worker 同时抢 jitter = (Path(DATA_DIR).stat().st_mtime_ns % 17) if DATA_DIR.exists() else 0 stop_event.wait(3 + int(jitter) % 5) while not stop_event.is_set(): try: _daily_tick() except Exception: # noqa: BLE001 log.exception("scheduler tick error") stop_event.wait(20) def start_background_jobs() -> None: """幂等启动;gunicorn 多 worker 各自起线程,靠 settings 去重。""" global _started with _lock: if _started: return _started = True stop = threading.Event() t = threading.Thread(target=_loop, args=(stop,), name="wecom-daily", daemon=True) t.start() log.info( "wecom daily scheduler started (BJT %02d:%02d)", int(Config.WECOM_DAILY_HOUR), int(Config.WECOM_DAILY_MINUTE), )