dekun
75b9bffe96
Desktop can auto-pull stop to open+/-1 after a 3$ favorable move; hub pushes open/close events and a 23:00 BJT daily summary via webhook. Co-authored-by: Cursor <cursoragent@cursor.com>
101 lines
3.0 KiB
Python
101 lines
3.0 KiB
Python
# -*- 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),
|
|
)
|