예제 소스

register_eye_games.py

#!/usr/bin/env python3
"""무제한 5판 모니터가 끝나면 그 SQLite를 다음 야간 학습 대기열에 올린다.

대국이 3판 미만이면 등록하지 않는다. 이미 끝난 판의 백 차례는
KataGo 정답으로 바꿀 내용이 있으므로 학습 대상으로 둔다.
"""
from __future__ import annotations

import json
import subprocess
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path

ROOT = Path("~/project")
PID = 72486
DB = ROOT / "logs/heuristic_r5_eye_20261008/matches.sqlite"
PENDING = ROOT / "logs/onpolicy/pending_sources.json"
KST = timezone(timedelta(hours=9))
LABEL = "eye5_20261009"


def alive(pid: int) -> bool:
    r = subprocess.run(["ps", "-p", str(pid)], capture_output=True)
    return r.returncode == 0


def finished_games() -> int:
    import sqlite3

    con = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
    n = con.execute(
        "select count(*) from games where finished_at is not null and finished_at != ''"
    ).fetchone()[0]
    con.close()
    return int(n)


def main() -> int:
    while alive(PID):
        time.sleep(30)
    n = finished_games() if DB.exists() else 0
    data = json.loads(PENDING.read_text(encoding="utf-8")) if PENDING.exists() else {"sources": []}
    now = datetime.now(KST).strftime("%Y-%m-%d %H:%M:%S")
    status = "ready" if n >= 3 else "skip"
    row = {
        "label": LABEL,
        "db": str(DB),
        "status": status,
        "games": n,
        "at": now,
        "note": "휴리스틱 대 R5 무제한 대국. 백 수를 KataGo 최선으로 라벨링해 다음 야간 학습.",
    }
    data["sources"] = [s for s in data.get("sources", []) if s.get("label") != LABEL]
    data["sources"].append(row)
    PENDING.parent.mkdir(parents=True, exist_ok=True)
    PENDING.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(row, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())