예제 소스

import_bot_logs.py

"""bot.py play 출력 로그에서 대국 수순을 복원해 기보 DB에 넣음.

  python import_bot_logs.py LOG [LOG ...]

시작 국면부터 모든 수가 엔진 규칙으로 합법인 대국만 넣는다
(봇이 중간 국면에서 시작한 판은 시작 배치를 알 수 없어서 제외).
"""

from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
from pathlib import Path

import bot
import recorder

FORMATION_KEY = {v: k for k, v in bot.FORMATIONS.items()}
GAME_RE = re.compile(r"===== \d+번째 대국 =====")
INIT_RE = re.compile(r"인식 완료: 화면 위쪽=(\S+), 초 상차림=(\w+), 한 상차림=(\w+)")
REINIT_RE = re.compile(r"상차림 다시 인식: 초=(\w+), 한=(\w+)")
MOVE_RE = re.compile(r"^\s*(\d+)\. \[(AI|상대)\] (?:(초|한) \S+ (\d)(\d+)→(\d)(\d+)|한수 쉼)")


def parse(path: Path) -> list[dict]:
    games: list[dict] = []
    cur: dict | None = None
    for line in path.read_text(errors="replace").splitlines():
        if GAME_RE.search(line):
            cur = {"cho": None, "han": None, "moves": [], "ok": True, "src": path.name}
            games.append(cur)
            continue
        if cur is None:
            continue
        if m := INIT_RE.search(line):
            cur["flipped"] = m.group(1) == "초"
            cur["cho"], cur["han"] = m.group(2), m.group(3)
        elif m := REINIT_RE.search(line):
            cur["cho"], cur["han"] = m.group(1), m.group(2)
        elif "(이미 둔 수)" in line:
            cur["ok"] = False  # 시작 직후 복원 수순은 로그에 좌표가 다른 형식
        elif m := MOVE_RE.match(line):
            ply = int(m.group(1))
            if ply != len(cur["moves"]) + 1:
                cur["ok"] = False
            if m.group(3) is None:
                cur["moves"].append("pass")
            else:
                ff, fr, tf, tr = (int(m.group(i)) - 1 for i in range(4, 8))
                cur["moves"].append(f"{ff}{fr}{tf}{tr}")
    return games


def legal(game: dict) -> bool:
    if not game["ok"] or not game["moves"] or not game["cho"] or not game["han"]:
        return False
    cmd = [
        str(bot.ENGINE), "--cho", game["cho"], "--han", game["han"], "--depth", "0",
        "--no-repetition", "--moves", ",".join(game["moves"]),
    ]
    out = json.loads(subprocess.run(cmd, capture_output=True, text=True, check=False).stdout)
    return "error" not in out


def insert(conn, game: dict) -> int:
    board = bot.start_board(FORMATION_KEY[game["cho"]], FORMATION_KEY[game["han"]])
    cur = conn.execute(
        "INSERT INTO games (source, started_at, ended_at, from_start, start_ply, start_board, flipped, my_side,"
        " move_count, end_reason, note) VALUES (?,?,?,?,?,?,?,?,?,?,?)",
        ("bot_log", recorder.now(), recorder.now(), 1, 0, json.dumps(board), int(game.get("flipped", False)),
         bot.CHO, len(game["moves"]), "imported", f"bot.py 로그 {game['src']} (AI=초, 사람=한)"),
    )
    gid = cur.lastrowid
    side = bot.CHO
    for ply, mv in enumerate(game["moves"], 1):
        piece = captured = None
        if mv != "pass":
            f, t = bot.parse_move(mv)  # type: ignore[misc]
            piece, captured = board[f], board[t] or None
            board[t], board[f] = board[f], ""
        conn.execute(
            "INSERT INTO moves (game_id, ply, side, move, piece, captured, board_after, created_at) VALUES (?,?,?,?,?,?,?,?)",
            (gid, ply, side, mv, piece, captured, json.dumps(board), recorder.now()),
        )
        side = bot.opposite_side(side)
    conn.commit()
    return gid


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("logs", nargs="+", type=Path)
    args = ap.parse_args()
    conn = recorder.db()
    seen = {row[0] for row in conn.execute("SELECT note FROM games WHERE source='bot_log'")}
    total = 0
    for path in args.logs:
        for i, game in enumerate(parse(path), 1):
            tag = f"{path.name} #{i} ({len(game['moves'])}수, 초={game['cho']}, 한={game['han']})"
            if not legal(game):
                print(f"  제외 {tag}")
                continue
            game["src"] = f"{path.name}#{i}"
            note = f"bot.py 로그 {game['src']} (AI=초, 사람=한)"
            if note in seen:
                print(f"  이미 있음 {tag}")
                continue
            gid = insert(conn, game)
            total += len(game["moves"])
            print(f"  추가 #{gid} {tag}")
    print(f"총 {total}수 추가")


if __name__ == "__main__":
    sys.exit(main())