"""热加载 Worker — watchfiles 监听代码变化,自动重启全部 worker 进程。

worker 本身无热重载(`python -m worker` 启动时加载模块代码,改代码必须重启)。
本脚本与 hot_reload_backend.py 同模式:监听 app/worker/.env 文件变化 →
干净杀掉全部 worker → 重新启动,避免"改了代码 worker 还在跑旧逻辑"。

中断安全:worker 执行中被 kill 的任务,后端靠租约(默认 90s)到期后重新
放回队列,由重启后的 worker 重新领取 —— 任务不丢,最多从头重跑一遍。

用法:
  python scripts/hot_reload_worker.py
  (API Key 未设置时自动从项目 .env 读 WORKER_API_KEY,不会打印)

可选环境变量:
  EGODATA_WORKER_COUNT=3        worker 数量(默认 3,与后端进程布局一致)
  EGODATA_WORKER_API_KEY=...    显式指定(优先于 .env)
  EGODATA_SERVER_URL=...        默认 http://127.0.0.1:8000
  EGODATA_DEVICE=auto           上报的设备类型
"""

import os
import socket
import subprocess
import sys
import time
import uuid
from pathlib import Path

import watchfiles

ROOT = Path(__file__).resolve().parent.parent
# .env 也在监视内:改配置(如 WORKER_API_KEY / 临时目录)后自动重启生效
WATCH_DIRS = [ROOT / "app", ROOT / "worker", ROOT / ".env"]
LOG_PATH = ROOT / "data" / "tmp" / "worker" / "worker-hot.log"
WORKER_COUNT = int(os.getenv("EGODATA_WORKER_COUNT", "3"))


def _read_env_file(path: Path) -> dict[str, str]:
    out: dict[str, str] = {}
    if not path.is_file():
        return out
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, _, v = line.partition("=")
        out[k.strip()] = v.strip().strip('"').strip("'")
    return out


def _log(msg: str) -> None:
    line = f"[{time.strftime('%H:%M:%S')}] {msg}"
    print(line, flush=True)
    try:
        LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
        with open(LOG_PATH, "a", encoding="utf-8") as f:
            f.write(line + "\n")
    except Exception:
        pass


def _resolve_api_key() -> str:
    env = _read_env_file(ROOT / ".env")
    return (os.getenv("EGODATA_WORKER_API_KEY")
            or env.get("EGODATA_WORKER_API_KEY")
            or env.get("WORKER_API_KEY")
            or "")


API_KEY = _resolve_api_key()


def start_worker(index: int, generation: str) -> subprocess.Popen:
    env = dict(os.environ)
    env.setdefault("PYTHONPATH", str(ROOT))
    env["EGODATA_WORKER_API_KEY"] = API_KEY
    env.setdefault("EGODATA_SERVER_URL", "http://127.0.0.1:8000")
    env.setdefault("EGODATA_DEVICE", "auto")
    host = os.getenv("EGODATA_WORKER_ID_PREFIX", f"linux-{socket.gethostname()}")
    env["EGODATA_WORKER_ID"] = f"{host}-g{generation}-{index + 1}"
    # 每个 worker 独立临时子目录,避免共目录残留冲突
    env.setdefault("EGODATA_WORK_DIR",
                   str(ROOT / "data" / "tmp" / "worker" / f"w{index + 1}"))
    log_path = LOG_PATH.parent / f"worker-hot-{index + 1}.log"
    log_file = open(log_path, "a", encoding="utf-8")
    proc = subprocess.Popen([sys.executable, "-m", "worker"],
                            cwd=str(ROOT), env=env,
                            stdout=log_file, stderr=log_file)
    _log(f"started worker #{index + 1} pid={proc.pid} id={env['EGODATA_WORKER_ID']}")
    return proc


def stop_worker(proc: subprocess.Popen) -> None:
    if proc is None or proc.poll() is not None:
        return
    proc.terminate()
    try:
        proc.wait(timeout=8)
    except subprocess.TimeoutExpired:
        proc.kill()
        try:
            proc.wait(timeout=5)
        except Exception:
            pass


def main() -> None:
    if not API_KEY:
        print("[error] EGODATA_WORKER_API_KEY 未设置,且 .env 里没有 WORKER_API_KEY",
              file=sys.stderr)
        sys.exit(1)
    generation = uuid.uuid4().hex[:10]
    procs = [start_worker(i, generation) for i in range(WORKER_COUNT)]
    last_restart = time.monotonic()
    try:
        for changes in watchfiles.watch(*WATCH_DIRS, step=500):
            if time.monotonic() - last_restart < 1.5:
                continue  # 防抖:合并连续变更
            files = sorted({str(p) for _, p in changes})
            _log("detected changes: " + ", ".join(files[:5]))
            for p in procs:
                stop_worker(p)
            generation = uuid.uuid4().hex[:10]
            procs = [start_worker(i, generation) for i in range(WORKER_COUNT)]
            last_restart = time.monotonic()
    except KeyboardInterrupt:
        for p in procs:
            stop_worker(p)
        _log("hot reload stopped")


if __name__ == "__main__":
    main()