3003 lines
125 KiB
Python
3003 lines
125 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import math
|
|
import re
|
|
from pathlib import Path
|
|
from time import time, sleep
|
|
from typing import Any
|
|
import os
|
|
|
|
from playwright.sync_api import Page, Playwright, sync_playwright
|
|
|
|
from browser_launch import launch_browser_context
|
|
from register_game_playwright import (
|
|
DEFAULT_BASE_URL,
|
|
DEFAULT_GAME_ID,
|
|
DEFAULT_REPORT_DIR,
|
|
load_report,
|
|
normalize_game_type,
|
|
normalize_stadium_name,
|
|
normalize_team_name,
|
|
)
|
|
|
|
|
|
PITCH_TYPE_LABEL_MAP = {
|
|
"직구": "패스트볼",
|
|
"패스트볼": "패스트볼",
|
|
"커브": "커브",
|
|
"체인지업": "체인지업",
|
|
"슬라이더": "슬라이더",
|
|
"커터": "커터",
|
|
"스플리터": "스플리터",
|
|
"포크": "포크볼",
|
|
"포크볼": "포크볼",
|
|
"투심": "투심",
|
|
"싱커": "싱커",
|
|
"너클": "너클",
|
|
}
|
|
|
|
PITCH_RESULT_LABEL_MAP = {
|
|
"볼": "볼",
|
|
"스트라이크": "스트라이크(루킹)",
|
|
"헛스윙": "헛스윙(스트라이크)",
|
|
"헛스윙 번트": "번트시도-스트라이크",
|
|
"번트 헛스윙": "번트시도-스트라이크",
|
|
"파울": "파울",
|
|
"번트파울": "번트-파울",
|
|
"몸에 맞는 볼": "몸에 맞는 볼",
|
|
"몸에 맞는 공": "몸에 맞는 볼",
|
|
"사구": "몸에 맞는 볼",
|
|
"고의사구": "고의사구",
|
|
"자동 고의사구": "고의사구",
|
|
"폭투-볼": "폭투-볼",
|
|
"포일-볼": "포일-볼",
|
|
}
|
|
|
|
BATTER_RESULT_LABEL_MAP = {
|
|
"walk": "포볼",
|
|
"intentional_walk": "고의사구",
|
|
"strikeout": "루킹스트라이크-아웃",
|
|
"bunt_strikeout": "번트-삼진",
|
|
"hit_by_pitch": "몸에 맞는 볼",
|
|
"single": "1루타",
|
|
"double": "2루타",
|
|
"triple": "3루타",
|
|
"home_run": "홈런",
|
|
"single_runner_out": "1루타 후 주루아웃",
|
|
"double_runner_out": "2루타 후 주루아웃",
|
|
"triple_runner_out": "3루타 후 주루아웃",
|
|
"single_error_advance": "1루타 후 수비실책진루",
|
|
"double_error_advance": "2루타 후 수비실책진루",
|
|
"triple_error_advance": "3루타 후 수비실책진루",
|
|
"sacrifice_fly": "희생 플라이",
|
|
"sacrifice_bunt": "희생 번트",
|
|
"reach_on_error": "수비실책",
|
|
"reach_on_fielder_choice": "야수선택",
|
|
"reach_on_grounder": "땅볼출루(무안타)",
|
|
"bunt_hit": "번트안타",
|
|
"out": "아웃",
|
|
}
|
|
|
|
RUNNER_EVENT_LABEL_MAP = {
|
|
"advance": "일반 진루",
|
|
"score": "일반 진루",
|
|
"steal": "도루성공",
|
|
"steal_fail": "도루시도 아웃",
|
|
"force_out": "포스아웃",
|
|
"pickoff_out": "견제 아웃",
|
|
"error_advance": "수비 실책",
|
|
"wild_pitch_advance": "폭투-진루성공",
|
|
"passed_ball_advance": "포일-진루성공",
|
|
}
|
|
|
|
FOUL_FLY_LEFT = (2, 70)
|
|
FOUL_FLY_RIGHT = (98, 70)
|
|
|
|
REVIEW_ITEM_RESULT_GROUP_MAP = {
|
|
"홈런타구 페어 파울": ("type1", "페어", "파울"),
|
|
"외야타구 페어 파울": ("type1", "페어", "파울"),
|
|
"포수/태그플레이 아웃/세이프": ("type2", "아웃", "세이프"),
|
|
"야수의 포구": ("type2", "아웃", "세이프"),
|
|
"몸에 맞는 공": ("type3", "인정", "불인정"),
|
|
"파울": ("type3", "인정", "불인정"),
|
|
"헛스윙": ("type3", "인정", "불인정"),
|
|
"기타": ("type3", "인정", "불인정"),
|
|
}
|
|
|
|
POSITION_NUMBER_MAP = {
|
|
"투수": "1",
|
|
"포수": "2",
|
|
"1루수": "3",
|
|
"2루수": "4",
|
|
"3루수": "5",
|
|
"유격수": "6",
|
|
"좌익수": "7",
|
|
"중견수": "8",
|
|
"우익수": "9",
|
|
}
|
|
POSITION_LABEL_MAP = {v: k for k, v in POSITION_NUMBER_MAP.items()}
|
|
|
|
HIT_BALL_TYPE_MAP = {
|
|
"땅볼": "0",
|
|
"일반바운드": "1",
|
|
"플라이": "2",
|
|
"라인드라이브": "3",
|
|
"펜스타구": "4",
|
|
"홈런성타구": "5",
|
|
"번트타구": "6",
|
|
}
|
|
|
|
FIELD_COORDINATES = {
|
|
"투수": (50, 80),
|
|
"포수": (50, 93),
|
|
"1루수": (63, 77),
|
|
"2루수": (60, 65),
|
|
"3루수": (37, 77),
|
|
"유격수": (40, 65),
|
|
"좌익수": (22, 42),
|
|
"중견수": (50, 24),
|
|
"우익수": (78, 42),
|
|
"좌전": (30, 50),
|
|
"중전": (50, 35),
|
|
"우전": (70, 50),
|
|
"좌중간": (34, 34),
|
|
"우중간": (66, 34),
|
|
"좌월": (20, 30),
|
|
"중월": (50, 14),
|
|
"우월": (80, 30),
|
|
}
|
|
|
|
DEFENSE_BUTTON_ID_MAP = {
|
|
"투수": "input[name='defenseNumberBtn']#picher",
|
|
"포수": "input[name='defenseNumberBtn']#catcher",
|
|
"1루수": "input[name='defenseNumberBtn']#runner_1",
|
|
"2루수": "input[name='defenseNumberBtn']#runner_2",
|
|
"3루수": "input[name='defenseNumberBtn']#runner_3",
|
|
"유격수": "input[name='defenseNumberBtn']#shortStop",
|
|
"중견수": "input[name='defenseNumberBtn']#centerFielder",
|
|
"우익수": "input[name='defenseNumberBtn']#rightFielder",
|
|
"좌익수": "input[name='defenseNumberBtn']#leftFielder",
|
|
}
|
|
|
|
POSITION_TO_DEFENSE_NO = {
|
|
"투수": "1",
|
|
"포수": "2",
|
|
"1루수": "3",
|
|
"2루수": "4",
|
|
"3루수": "5",
|
|
"유격수": "6",
|
|
"좌익수": "7",
|
|
"중견수": "8",
|
|
"우익수": "9",
|
|
"지명타자": "10",
|
|
}
|
|
|
|
def click_defense_button_robustly(page: Page, position: str, click_count: int = 1) -> bool:
|
|
# 1) 기존 ID 맵 기반 검색
|
|
button_selector = DEFENSE_BUTTON_ID_MAP.get(position)
|
|
# 2) 값(value) 기반 검색 (1~9번)
|
|
pos_no = POSITION_TO_DEFENSE_NO.get(position)
|
|
value_selector = f"input[name='defenseNumberBtn'][value='{pos_no}']" if pos_no else None
|
|
|
|
selectors = [s for s in [button_selector, value_selector] if s]
|
|
|
|
for _ in range(click_count):
|
|
clicked_this_round = False
|
|
for selector in selectors:
|
|
try:
|
|
defense_button = wait_for_visible_locator(page, selector, timeout_ms=1500)
|
|
if defense_button:
|
|
defense_button.click(force=True)
|
|
clicked_this_round = True
|
|
page.wait_for_timeout(100)
|
|
break
|
|
except Exception:
|
|
continue
|
|
|
|
# 3) 폴백: 자바스크립트로 직접 클릭 (텍스트 기반)
|
|
if not clicked_this_round:
|
|
try:
|
|
page.evaluate(f"""(pos) => {{
|
|
const buttons = [...document.querySelectorAll("input[name='defenseNumberBtn'], button, a")];
|
|
const target = buttons.find(b => b.value === pos || b.id === pos || b.innerText.includes(pos));
|
|
if (target) {{
|
|
target.click();
|
|
target.dispatchEvent(new Event('change', {{ bubbles: true }}));
|
|
}}
|
|
}}""", position)
|
|
page.wait_for_timeout(100)
|
|
clicked_this_round = True
|
|
except: pass
|
|
|
|
if not clicked_this_round:
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
# 초기화 대상 히든 필드 ID 목록 (name 속성이 없어 ID로만 접근 가능)
|
|
_DEFENSE_HIDDEN_FIELD_IDS = [
|
|
"putout", "assist", "error", "upstruction",
|
|
"dat_putout_hitter", "dat_assist_hitter", "dat_error_hitter", "dat_upstruction_hitter",
|
|
"dat_putout_runner1", "dat_assist_runner1", "dat_error_runner1", "dat_upstruction_runner1",
|
|
"dat_putout_runner2", "dat_assist_runner2", "dat_error_runner2", "dat_upstruction_runner2",
|
|
"dat_putout_runner3", "dat_assist_runner3", "dat_error_runner3", "dat_upstruction_runner3",
|
|
"dat_error_type", "dat_error_type1", "dat_error_type2", "dat_error_type3",
|
|
"dat_hitball_speed", "dat_hitball_type", "dat_hitball_xy",
|
|
"dat_multiplay_type",
|
|
"dat_hit_x", "dat_hit_y", "hitPoints",
|
|
]
|
|
|
|
_DEFENSE_CLEAR_JS = """
|
|
() => {
|
|
// 1) 수비수 버튼(라디오) 체크 해제 - 실제 name 속성이 있는 요소
|
|
document.querySelectorAll("input[name='defenseNumberBtn']").forEach(el => { el.checked = false; });
|
|
// 2) 타구 종류 라디오 해제 - 실제 name 속성이 있는 요소
|
|
document.querySelectorAll("input[name='hitBallType']").forEach(el => { el.checked = false; });
|
|
|
|
// 3) 자살/보살/실책/방해 dat_ 계열 히든 필드 초기화 (실제 기록에 영향을 주는 필드들)
|
|
const datIds = [
|
|
"putout", "assist", "error", "upstruction",
|
|
"dat_putout_hitter", "dat_assist_hitter", "dat_error_hitter", "dat_upstruction_hitter",
|
|
"dat_putout_runner1", "dat_assist_runner1", "dat_error_runner1", "dat_upstruction_runner1",
|
|
"dat_putout_runner2", "dat_assist_runner2", "dat_error_runner2", "dat_upstruction_runner2",
|
|
"dat_putout_runner3", "dat_assist_runner3", "dat_error_runner3", "dat_upstruction_runner3",
|
|
"dat_error_type", "dat_error_type1", "dat_error_type2", "dat_error_type3",
|
|
"dat_multiplay_type", "multiplay_type",
|
|
"hitBallXY", "hitBallDistance",
|
|
"dat_hitball_speed", "dat_hitball_type", "dat_hitball_x", "dat_hitball_y", "dat_hitball_xy", "dat_hitball_distance",
|
|
"hitball_speed", "hitball_type", "hitball_xy", "hitball_distance"
|
|
];
|
|
datIds.forEach(id => {
|
|
const el = document.getElementById(id);
|
|
if (el) el.value = "";
|
|
});
|
|
|
|
// 4) 타구 종류 라디오 버튼 및 좌표 초기화 (id^= 스타일)
|
|
document.querySelectorAll("input[name='hitBallType'], [id^='hitBallSpeed'], [id^='hitBallType']").forEach(el => {
|
|
if (el.type === 'radio' || el.type === 'checkbox') {
|
|
el.checked = false;
|
|
} else {
|
|
el.value = "";
|
|
}
|
|
});
|
|
}
|
|
"""
|
|
|
|
|
|
def clear_defense_selections(page: Page) -> None:
|
|
"""수비 가담 및 타구 관련 필드(체크박스/라디오/히든 필드) 선택 상태를 초기화합니다.
|
|
|
|
주의: 해당 히든 필드들은 name 속성이 없고 id로만 식별 가능합니다.
|
|
"""
|
|
page.evaluate(_DEFENSE_CLEAR_JS)
|
|
|
|
|
|
def click_defense_sequence_in_popup(page: Page, sequence: list[str], complete_button_selector: str | None = None) -> None:
|
|
popup_field = None
|
|
for _ in range(15):
|
|
popup_field = get_last_visible_locator(page, "#defenseDiv")
|
|
if popup_field is not None and popup_field.bounding_box() is not None:
|
|
break
|
|
page.wait_for_timeout(150)
|
|
|
|
if popup_field is None:
|
|
return
|
|
|
|
for position in sequence:
|
|
click_defense_button_robustly(page, position)
|
|
|
|
complete_button = None
|
|
if complete_button_selector:
|
|
complete_button = get_last_visible_locator(page, complete_button_selector)
|
|
if complete_button is None:
|
|
complete_button = get_last_visible_locator(page, "#btnNext")
|
|
if complete_button is None:
|
|
complete_button = get_last_visible_locator(page, "#btnAdd")
|
|
|
|
if complete_button:
|
|
try:
|
|
complete_button.click(force=True)
|
|
page.wait_for_timeout(200)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def fill_runner_out_defense(page: Page, text: str, sequence_override: list[str] | None = None) -> None:
|
|
# 아웃 관련 팝업이 뜰 때까지 대기 (도루자, 태그아웃 등)
|
|
page.wait_for_timeout(300)
|
|
sequence = sequence_override or extract_defense_sequence(text)
|
|
if sequence:
|
|
click_defense_sequence_in_popup(page, sequence)
|
|
|
|
|
|
def normalize_lineup_text(text: str) -> str:
|
|
text = (text or "").strip()
|
|
text = text.replace("*", "")
|
|
# [10] 문보경 or 문보경 [10번] 등 다양하게 나올 수 있음
|
|
text = re.sub(r"\[\d+(?:번)?\]", "", text)
|
|
text = re.sub(r"\s*\(.*?\)\s*", "", text)
|
|
# 한글/영문 이름만 남김
|
|
text = "".join(re.findall(r"[가-힣A-Za-z]+", text))
|
|
return text.strip()
|
|
|
|
|
|
def extract_change_actor(text: str) -> tuple[str | None, int | None, str]:
|
|
lhs = (text or "").split(" : ", 1)[0].strip()
|
|
batter_match = re.search(r"(\d+)번타자\s+(.+)$", lhs)
|
|
if batter_match:
|
|
return "batter", int(batter_match.group(1)), batter_match.group(2).strip()
|
|
for role in ("대타", "대주자", "1루주자", "2루주자", "3루주자", "주자", "투수", "포수", "1루수", "2루수", "3루수", "유격수", "좌익수", "중견수", "우익수"):
|
|
if lhs.startswith(role + " "):
|
|
return role, None, lhs[len(role):].strip()
|
|
return None, None, lhs
|
|
|
|
|
|
def is_merged_pitcher_substitution(actor_role: str | None, in_role: str | None) -> bool:
|
|
field_roles = {"포수", "1루수", "2루수", "3루수", "유격수", "좌익수", "중견수", "우익수"}
|
|
return actor_role in field_roles and in_role == "투수"
|
|
|
|
|
|
def normalize_change_event(change_event: dict[str, Any]) -> dict[str, Any]:
|
|
if change_event.get("actor_name") or change_event.get("player_name"):
|
|
return change_event
|
|
|
|
text = change_event.get("text") or ""
|
|
normalized = dict(change_event)
|
|
normalized["change_type"] = "position_change" if "수비위치 변경" in text else "substitution"
|
|
actor_role, bat_order, actor_name = extract_change_actor(text)
|
|
normalized["actor_role"] = actor_role
|
|
normalized["actor_name"] = actor_name
|
|
if bat_order is not None:
|
|
normalized["bat_order"] = bat_order
|
|
|
|
if normalized["change_type"] == "position_change":
|
|
rhs = text.split(" : ", 1)[1] if " : " in text else ""
|
|
normalized["player_name"] = actor_name
|
|
normalized["to_position"] = rhs.split("(으)로", 1)[0].strip()
|
|
return normalized
|
|
|
|
rhs = text.split(" : ", 1)[1] if " : " in text else ""
|
|
rhs = rhs.split("(으)로 교체", 1)[0].strip()
|
|
in_role, _, in_name = extract_change_actor(rhs)
|
|
normalized["out_player"] = actor_name
|
|
normalized["in_player"] = in_name
|
|
normalized["in_role"] = in_role
|
|
if is_merged_pitcher_substitution(actor_role, in_role):
|
|
normalized["change_type"] = "merged_pitcher_substitution"
|
|
normalized["player_name"] = actor_name
|
|
normalized["to_position"] = "지명타자"
|
|
normalized["pitcher_in_player"] = in_name
|
|
return normalized
|
|
if in_role in POSITION_TO_DEFENSE_NO:
|
|
normalized["to_position"] = in_role
|
|
return normalized
|
|
|
|
|
|
def get_lineup_state(page: Page) -> dict[str, Any]:
|
|
return page.evaluate(
|
|
"""() => {
|
|
const buildSide = (side) => {
|
|
const rows = [];
|
|
for (let idx = 0; idx <= 9; idx += 1) {
|
|
const player = document.querySelector(`#${side}_player_id_${idx}`);
|
|
const defense = document.querySelector(`#${side}_defense_no_${idx}`);
|
|
const orgPlayer = document.querySelector(`#org_${side}_player_id_${idx}`);
|
|
const orgDefense = document.querySelector(`#org_${side}_defense_no_${idx}`);
|
|
if (!player && !defense) continue;
|
|
rows.push({
|
|
idx,
|
|
playerText: player ? player.options[player.selectedIndex]?.text || '' : '',
|
|
playerValue: player ? player.value : '',
|
|
defenseValue: defense ? defense.value : '',
|
|
orgPlayerValue: orgPlayer ? orgPlayer.value : '',
|
|
orgDefenseValue: orgDefense ? orgDefense.value : '',
|
|
});
|
|
}
|
|
return rows;
|
|
};
|
|
return { home: buildSide('home'), away: buildSide('away') };
|
|
}"""
|
|
)
|
|
|
|
|
|
def detect_change_side(half_inning: dict[str, Any], change_event: dict[str, Any], lineup_state: dict[str, Any]) -> str:
|
|
actor_role = change_event.get("actor_role")
|
|
actor_name = normalize_lineup_text(change_event.get("actor_name") or change_event.get("player_name") or "")
|
|
offense_side = "away" if half_inning.get("half") == "top" else "home"
|
|
defense_side = "home" if offense_side == "away" else "away"
|
|
|
|
matched_sides: list[str] = []
|
|
for side in ("home", "away"):
|
|
for row in lineup_state.get(side, []):
|
|
if normalize_lineup_text(row.get("playerText") or "") == actor_name:
|
|
matched_sides.append(side)
|
|
break
|
|
if len(matched_sides) == 1:
|
|
return matched_sides[0]
|
|
|
|
if actor_role in {"batter", "대타", "대주자", "1루주자", "2루주자", "3루주자", "주자"}:
|
|
return offense_side
|
|
|
|
return defense_side
|
|
|
|
|
|
def find_change_row(side_rows: list[dict[str, Any]], change_event: dict[str, Any]) -> int | None:
|
|
if change_event.get("bat_order") is not None:
|
|
return int(change_event["bat_order"])
|
|
|
|
actor_name_raw = change_event.get("actor_name") or change_event.get("player_name") or change_event.get("out_player") or ""
|
|
actor_name = normalize_lineup_text(actor_name_raw)
|
|
|
|
# 1단계: 정규화된 이름 완전 일치
|
|
for row in side_rows:
|
|
if normalize_lineup_text(row.get("playerText") or "") == actor_name:
|
|
return int(row["idx"])
|
|
|
|
# 2단계: 부분 일치 (포함 관계)
|
|
for row in side_rows:
|
|
player_text = normalize_lineup_text(row.get("playerText") or "")
|
|
if actor_name and (actor_name in player_text or player_text in actor_name):
|
|
return int(row["idx"])
|
|
|
|
# 3단계: 역할(defenseValue) 실치
|
|
actor_role = change_event.get("actor_role")
|
|
if actor_role in POSITION_TO_DEFENSE_NO:
|
|
defense_no = POSITION_TO_DEFENSE_NO[actor_role]
|
|
for row in side_rows:
|
|
if str(row.get("defenseValue") or "") == defense_no:
|
|
return int(row["idx"])
|
|
|
|
# 4단계: '1루주자 문보경' 처럼 actor_name_raw에 역할이 섞여있을 경우 이름만 다시 추출 시도
|
|
if " " in actor_name_raw:
|
|
potential_name = normalize_lineup_text(actor_name_raw.split()[-1])
|
|
for row in side_rows:
|
|
if normalize_lineup_text(row.get("playerText") or "") == potential_name:
|
|
return int(row["idx"])
|
|
|
|
return None
|
|
|
|
|
|
def find_pitcher_row(side_rows: list[dict[str, Any]]) -> int | None:
|
|
for row in side_rows:
|
|
if str(row.get("defenseValue") or "") == POSITION_TO_DEFENSE_NO["투수"]:
|
|
return int(row["idx"])
|
|
return None
|
|
|
|
|
|
def select_lineup_player(page: Page, side: str, row_idx: int, player_name: str) -> None:
|
|
select_id = f"#{side}_player_id_{row_idx}"
|
|
options = page.locator(f"{select_id} option").evaluate_all(
|
|
"""(nodes) => nodes.map((node) => ({ value: node.value, text: node.textContent.trim() }))"""
|
|
)
|
|
target_value = None
|
|
normalized_target = normalize_lineup_text(player_name)
|
|
for option in options:
|
|
if normalize_lineup_text(option["text"]) == normalized_target:
|
|
target_value = option["value"]
|
|
break
|
|
if not target_value:
|
|
raise ValueError(f"{side} {row_idx}번 행에서 선수 '{player_name}' 옵션을 찾지 못했습니다.")
|
|
page.locator(select_id).select_option(value=target_value)
|
|
|
|
|
|
def set_lineup_defense(page: Page, side: str, row_idx: int, position: str | None) -> None:
|
|
if not position:
|
|
return
|
|
defense_no = POSITION_TO_DEFENSE_NO.get(position)
|
|
if not defense_no:
|
|
return
|
|
page.locator(f"#{side}_defense_no_{row_idx}").select_option(value=defense_no)
|
|
|
|
|
|
def get_current_lineup_selection(page: Page, side: str, row_idx: int) -> tuple[str, str]:
|
|
player_value = page.locator(f"#{side}_player_id_{row_idx}").input_value()
|
|
defense_value = page.locator(f"#{side}_defense_no_{row_idx}").input_value()
|
|
return player_value, defense_value
|
|
|
|
|
|
def get_target_player_value(page: Page, side: str, row_idx: int, player_name: str | None) -> str | None:
|
|
if not player_name:
|
|
return None
|
|
select_id = f"#{side}_player_id_{row_idx}"
|
|
options = page.locator(f"{select_id} option").evaluate_all(
|
|
"""(nodes) => nodes.map((node) => ({ value: node.value, text: node.textContent.trim() }))"""
|
|
)
|
|
normalized_target = normalize_lineup_text(player_name)
|
|
for option in options:
|
|
if normalize_lineup_text(option["text"]) == normalized_target:
|
|
return option["value"]
|
|
return None
|
|
|
|
|
|
def get_target_defense_value(position: str | None) -> str | None:
|
|
if not position:
|
|
return None
|
|
return POSITION_TO_DEFENSE_NO.get(position)
|
|
|
|
|
|
def apply_change_event(page: Page, half_inning: dict[str, Any], change_event: dict[str, Any], change_cache: dict[str, tuple[str, int]]) -> None:
|
|
change_event = normalize_change_event(change_event)
|
|
cache_key = normalize_lineup_text(change_event.get("actor_name") or change_event.get("player_name") or change_event.get("out_player") or "")
|
|
actor_name = normalize_lineup_text(change_event.get("actor_name") or change_event.get("player_name") or change_event.get("out_player") or "")
|
|
lineup_state = get_lineup_state(page)
|
|
cached = change_cache.get(cache_key)
|
|
side = cached[0] if cached else detect_change_side(half_inning, change_event, lineup_state)
|
|
side_rows = lineup_state.get(side, [])
|
|
row_idx = find_change_row(side_rows, change_event)
|
|
|
|
if row_idx is None and cached and cached[0] == side:
|
|
row_idx = cached[1]
|
|
|
|
if row_idx is None:
|
|
all_players = [normalize_lineup_text(r.get("playerText", "")) for r in side_rows]
|
|
raise ValueError(f"{side} 교체 행을 찾지 못했습니다 (Target: {actor_name}, Candidates: {all_players}): {change_event.get('text')}")
|
|
|
|
page.evaluate("""() => { window.alert = () => {}; window.confirm = () => true; }""")
|
|
|
|
current_player_value, current_defense_value = get_current_lineup_selection(page, side, row_idx)
|
|
target_player_value = current_player_value
|
|
target_defense_value = current_defense_value
|
|
|
|
def trigger_lineup_save(idx: int):
|
|
home_away_gb = 2 if side == "home" else 1
|
|
# 1) f_lineup 호출 (기본)
|
|
page.evaluate(
|
|
"""({ batterNo, homeAwayGb }) => {
|
|
if (typeof window.f_lineup === 'function') {
|
|
window.f_lineup(batterNo, homeAwayGb);
|
|
}
|
|
}""",
|
|
{"batterNo": idx, "homeAwayGb": home_away_gb},
|
|
)
|
|
page.wait_for_timeout(200)
|
|
|
|
# 2) 행 옆의 'V' 또는 '저장' 버튼 클릭 시도 (row_idx 기반)
|
|
page.evaluate(
|
|
"""({ side, idx }) => {
|
|
const row = document.querySelector(`#${side}_player_id_${idx}`)?.parentElement?.parentElement;
|
|
if (row) {
|
|
const saveBtn = [...row.querySelectorAll("input[type=button], button, a")].find(el =>
|
|
el.value === 'V' || el.innerText.includes('V') || el.innerText.includes('저장') || el.id.includes('save')
|
|
);
|
|
if (saveBtn) {
|
|
saveBtn.click();
|
|
return true;
|
|
}
|
|
}
|
|
return false;
|
|
}""",
|
|
{"side": side, "idx": idx}
|
|
)
|
|
page.wait_for_timeout(200)
|
|
|
|
if change_event.get("change_type") == "merged_pitcher_substitution":
|
|
actor_player_name = change_event.get("player_name") or change_event.get("actor_name")
|
|
pitcher_in_player = change_event.get("pitcher_in_player") or change_event.get("in_player")
|
|
pitcher_row_idx = find_pitcher_row(side_rows)
|
|
if pitcher_row_idx is None or not pitcher_in_player:
|
|
raise ValueError(f"투수 교체 행을 찾지 못했습니다: {change_event.get('text')}")
|
|
|
|
set_lineup_defense(page, side, row_idx, "지명타자")
|
|
trigger_lineup_save(row_idx)
|
|
trigger_lineup_save(row_idx + 1) # 인덱스 보정 처리
|
|
|
|
select_lineup_player(page, side, pitcher_row_idx, pitcher_in_player)
|
|
set_lineup_defense(page, side, pitcher_row_idx, "투수")
|
|
trigger_lineup_save(pitcher_row_idx)
|
|
trigger_lineup_save(pitcher_row_idx + 1)
|
|
|
|
if actor_player_name: change_cache[normalize_lineup_text(actor_player_name)] = (side, int(row_idx))
|
|
change_cache[normalize_lineup_text(pitcher_in_player)] = (side, int(pitcher_row_idx))
|
|
return
|
|
|
|
if change_event.get("change_type") == "substitution":
|
|
in_player = change_event.get("in_player")
|
|
if not in_player: return
|
|
select_lineup_player(page, side, row_idx, in_player)
|
|
set_lineup_defense(page, side, row_idx, change_event.get("to_position"))
|
|
else:
|
|
set_lineup_defense(page, side, row_idx, change_event.get("to_position"))
|
|
|
|
# 최종 저장 트리거
|
|
trigger_lineup_save(row_idx)
|
|
trigger_lineup_save(row_idx + 1) # 사이트 특성(1-based)인 경우 대비
|
|
|
|
# 검증 로직 추가: 실제로 바뀌었는지 확인
|
|
page.wait_for_timeout(300)
|
|
final_state = get_lineup_state(page)
|
|
final_rows = final_state.get(side, [])
|
|
final_row = next((r for r in final_rows if r["idx"] == row_idx), None)
|
|
|
|
if final_row:
|
|
in_name = normalize_lineup_text(change_event.get("in_player") or change_event.get("player_name") or "")
|
|
current_name_on_site = normalize_lineup_text(final_row.get("playerText") or "")
|
|
if in_name and in_name not in current_name_on_site:
|
|
# 실패 시 최후의 수단: 전역 '라인업저장' 버튼이라도 찾아 누름
|
|
page.evaluate("""() => {
|
|
const btn = [...document.querySelectorAll("input[type=button], button, a")].find(el =>
|
|
el.value?.includes('라인업') || el.innerText.includes('라인업저장') || el.innerText.includes('엔트리저장')
|
|
);
|
|
if (btn) btn.click();
|
|
}""")
|
|
page.wait_for_timeout(300)
|
|
|
|
# 캐시 업데이트
|
|
if change_event.get("change_type") == "substitution":
|
|
in_player_name = normalize_lineup_text(change_event.get("in_player") or "")
|
|
if in_player_name: change_cache[in_player_name] = (side, int(row_idx))
|
|
else:
|
|
player_name = normalize_lineup_text(change_event.get("player_name") or change_event.get("actor_name") or "")
|
|
if player_name: change_cache[player_name] = (side, int(row_idx))
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="게임기록 페이지로 이동하고 투구/타석 입력 자동화 준비를 합니다."
|
|
)
|
|
parser.add_argument("--game-id", default=DEFAULT_GAME_ID, help="예: 20250425NCSS02025")
|
|
parser.add_argument("--report-path", help="기본값: output/<game_id>_report.json")
|
|
parser.add_argument("--base-url", default=DEFAULT_BASE_URL, help="관리자 사이트 기본 URL")
|
|
parser.add_argument("--manager-game-no", help="관리자 게임번호. 없으면 날짜/구장/홈/어웨이로 자동 탐색")
|
|
parser.add_argument("--user-data-dir", default="playwright-user-data", help="Chromium 사용자 데이터 폴더")
|
|
parser.add_argument("--channel", default="chrome", help="브라우저 채널. 예: chrome, msedge")
|
|
parser.add_argument("--headless", action="store_true", help="헤드리스 모드")
|
|
parser.add_argument("--write-events", dest="write_events", action="store_true", help="실제 입력완료 버튼까지 누름")
|
|
parser.add_argument("--no-write-events", dest="write_events", action="store_false", help="페이지 진입 후 입력은 하지 않음")
|
|
parser.add_argument("--close", action="store_true", help="작업 후 브라우저를 닫음")
|
|
parser.add_argument("--job-id", help="DB 로깅용 작업 ID (UUID)")
|
|
parser.add_argument("--review-only", action="store_true", help="경기기록 대신 합의판정만 일괄 등록")
|
|
parser.set_defaults(write_events=True)
|
|
return parser.parse_args()
|
|
|
|
|
|
def get_report_path(args: argparse.Namespace) -> Path:
|
|
if args.report_path:
|
|
return Path(args.report_path)
|
|
return DEFAULT_REPORT_DIR / f"{args.game_id}_report.json"
|
|
|
|
|
|
def find_status_href(page: Page, report: dict[str, Any], manager_game_no: str | None) -> str:
|
|
game_info = report["game_info"]
|
|
target_date = game_info["date"]
|
|
target_stadium = normalize_stadium_name(game_info["stadium"])
|
|
target_home_team = normalize_team_name(game_info["home_team"])
|
|
target_away_team = normalize_team_name(game_info["away_team"])
|
|
target_game_type = normalize_game_type(game_info["game_type"])
|
|
|
|
rows: list[dict[str, Any]] = page.locator("table.gclist tr").evaluate_all(
|
|
"""(rows) => rows.slice(1).map((row) => {
|
|
const cells = [...row.cells].map((cell) => cell.innerText.trim());
|
|
const statusLink = [...row.querySelectorAll('a')].find((anchor) => anchor.textContent.trim() === '게임기록');
|
|
return {
|
|
gameNo: cells[0] || '',
|
|
date: cells[1] || '',
|
|
gameType: cells[2] || '',
|
|
stadium: cells[3] || '',
|
|
homeTeam: cells[4] || '',
|
|
awayTeam: cells[5] || '',
|
|
href: statusLink ? statusLink.getAttribute('href') : '',
|
|
};
|
|
})"""
|
|
)
|
|
|
|
if manager_game_no:
|
|
matched = next((row for row in rows if row["gameNo"] == str(manager_game_no)), None)
|
|
if not matched or not matched["href"] or matched["href"].startswith("javascript:"):
|
|
raise ValueError(f"관리자 게임번호 {manager_game_no} 행의 게임기록 링크를 찾지 못했습니다.")
|
|
return matched["href"]
|
|
|
|
candidates = [
|
|
row
|
|
for row in rows
|
|
if row["href"]
|
|
and not row["href"].startswith("javascript:")
|
|
and row["date"] == target_date
|
|
and (not row["gameType"] or normalize_game_type(row["gameType"]) == target_game_type)
|
|
and normalize_stadium_name(row["stadium"]) == target_stadium
|
|
and normalize_team_name(row["homeTeam"]) == target_home_team
|
|
and normalize_team_name(row["awayTeam"]) == target_away_team
|
|
]
|
|
if not candidates:
|
|
raise ValueError("목록에서 일치하는 게임기록 링크를 찾지 못했습니다.")
|
|
return candidates[0]["href"]
|
|
|
|
|
|
def open_game_status_page(page: Page, base_url: str, report: dict[str, Any], manager_game_no: str | None) -> None:
|
|
if manager_game_no:
|
|
page.goto(f"{base_url}/manager/game/status?game_no={manager_game_no}", wait_until="domcontentloaded")
|
|
page.wait_for_selector("#eventWriteBtn", timeout=10000)
|
|
return
|
|
page.goto(f"{base_url}/manager/game/list", wait_until="domcontentloaded")
|
|
page.wait_for_selector("table.gclist", timeout=10000)
|
|
status_href = find_status_href(page, report, manager_game_no)
|
|
with page.expect_navigation(wait_until="domcontentloaded"):
|
|
page.locator(f"a[href='{status_href}']").first.click()
|
|
page.wait_for_selector("#eventWriteBtn", timeout=10000)
|
|
|
|
|
|
def get_radio_map(page: Page, name: str) -> dict[str, str]:
|
|
rows = page.locator(f"input[type=radio][name='{name}']").evaluate_all(
|
|
"""(nodes) => nodes.map((node) => {
|
|
let label = '';
|
|
let current = node.nextSibling;
|
|
while (current) {
|
|
if (current.nodeType === Node.TEXT_NODE && current.textContent.trim()) {
|
|
label = current.textContent.trim();
|
|
break;
|
|
}
|
|
if (current.nodeType === Node.ELEMENT_NODE && current.textContent.trim()) {
|
|
label = current.textContent.trim();
|
|
break;
|
|
}
|
|
current = current.nextSibling;
|
|
}
|
|
return { label, id: node.id, value: node.value };
|
|
})"""
|
|
)
|
|
return {row["label"]: row["id"] for row in rows if row["label"] and row["id"]}
|
|
|
|
|
|
def set_radio_by_label(page: Page, radio_name: str, label: str) -> None:
|
|
locator = find_visible_radio_by_label(page, radio_name, label)
|
|
if locator is None:
|
|
radio_map = get_radio_map(page, radio_name)
|
|
radio_id = radio_map.get(label)
|
|
if not radio_id:
|
|
raise ValueError(f"{radio_name}에서 '{label}' 라디오를 찾지 못했습니다.")
|
|
locator = page.locator(f"#{radio_id}")
|
|
for _ in range(3):
|
|
try:
|
|
locator.click(force=True)
|
|
page.wait_for_timeout(50)
|
|
if locator.is_checked():
|
|
return
|
|
except Exception:
|
|
locator.evaluate(
|
|
"""(node) => {
|
|
node.disabled = false;
|
|
node.checked = true;
|
|
node.dispatchEvent(new Event('click', { bubbles: true }));
|
|
node.dispatchEvent(new Event('change', { bubbles: true }));
|
|
}"""
|
|
)
|
|
page.wait_for_timeout(50)
|
|
if locator.is_checked():
|
|
return
|
|
|
|
# 최종적으로 선택되었는지 확인 (강제성)
|
|
if not locator.is_checked():
|
|
locator.evaluate("node => { node.checked = true; node.dispatchEvent(new Event('change', {bubbles:true})); }")
|
|
|
|
|
|
def click_radio_by_label(page: Page, radio_name: str, label: str) -> None:
|
|
locator = find_visible_radio_by_label(page, radio_name, label)
|
|
if locator is None:
|
|
radio_map = get_radio_map(page, radio_name)
|
|
radio_id = radio_map.get(label)
|
|
if not radio_id:
|
|
raise ValueError(f"{radio_name}에서 '{label}' 라디오를 찾지 못했습니다.")
|
|
locator = page.locator(f"#{radio_id}")
|
|
|
|
for _ in range(3):
|
|
try:
|
|
locator.click(force=True)
|
|
page.wait_for_timeout(50)
|
|
if locator.is_checked():
|
|
return
|
|
except Exception:
|
|
locator.evaluate(
|
|
"""(node) => {
|
|
node.disabled = false;
|
|
node.checked = true;
|
|
node.dispatchEvent(new Event('click', { bubbles: true }));
|
|
node.dispatchEvent(new Event('change', { bubbles: true }));
|
|
}"""
|
|
)
|
|
page.wait_for_timeout(50)
|
|
if locator.is_checked():
|
|
return
|
|
|
|
# 최종적으로 선택되었는지 확인 (강제성)
|
|
if not locator.is_checked():
|
|
locator.evaluate("node => { node.checked = true; node.dispatchEvent(new Event('change', {bubbles:true})); }")
|
|
|
|
|
|
def get_checked_batter_defense_type(page: Page) -> str:
|
|
return page.evaluate(
|
|
"""() => {
|
|
const nodes = [...document.querySelectorAll("input[type=radio][name='evt_batter']:checked")];
|
|
for (const node of nodes) {
|
|
const rect = node.getBoundingClientRect();
|
|
const style = window.getComputedStyle(node);
|
|
if (rect.width > 0 && rect.height > 0 && style.display !== 'none' && style.visibility !== 'hidden' && !node.disabled) {
|
|
return node.getAttribute('defenseType') || '';
|
|
}
|
|
}
|
|
return nodes.length > 0 ? (nodes[0].getAttribute('defenseType') || '') : '';
|
|
}"""
|
|
)
|
|
|
|
|
|
def infer_review_item(detail_text: str) -> str:
|
|
"""리포트 텍스트 분석을 통해 사이트 표준 판독항목 추출"""
|
|
dt = detail_text.replace(" ", "")
|
|
if "홈런" in dt:
|
|
return "홈런타구 페어 파울"
|
|
if "아웃" in dt or "세이프" in dt or "포스" in dt or "태그" in dt or "견제" in dt or "도루" in dt:
|
|
return "포수/태그플레이 아웃/세이프"
|
|
if "페어" in dt or "파울" in dt:
|
|
return "외야타구 페어 파울"
|
|
if "포구" in dt or "노바운드" in dt or "바운드" in dt:
|
|
return "야수의 포구"
|
|
if "몸에맞" in dt or "데드볼" in dt:
|
|
return "몸에 맞는 볼"
|
|
if "헛스윙" in dt or "스윙" in dt:
|
|
return "헛스윙"
|
|
return "기타"
|
|
|
|
|
|
def set_select_by_partial_text(page_or_popup: Page, selector: str, partial_text: str) -> None:
|
|
"""텍스트 일부분만 맞아도 셀렉트 박스에서 선택 (정규식 지원)"""
|
|
if not partial_text:
|
|
return
|
|
page_or_popup.wait_for_selector(selector, timeout=3000)
|
|
|
|
# 사이트 내의 모든 옵션 텍스트를 가져옴
|
|
options = page_or_popup.locator(f"{selector} option").all_text_contents()
|
|
|
|
# 1단계: 정확히 일치하는지 확인
|
|
target = partial_text.strip()
|
|
for opt in options:
|
|
if opt.strip() == target:
|
|
page_or_popup.select_option(selector, label=opt)
|
|
return
|
|
|
|
# 2단계: 부분 일치 확인 (공백 제거 후 비교)
|
|
target_clean = target.replace(" ", "").replace("/", ",").replace("-", ",")
|
|
for opt in options:
|
|
opt_clean = opt.strip().replace(" ", "").replace("/", ",").replace("-", ",")
|
|
if target_clean in opt_clean or opt_clean in target_clean:
|
|
page_or_popup.select_option(selector, label=opt)
|
|
return
|
|
|
|
# 실패 시 로그만 남기고 에러는 내지 않음
|
|
print(f"DEBUG: '{selector}'에서 '{partial_text}'와 일치하는 옵션을 찾지 못함. (기존 선택 유지)")
|
|
|
|
|
|
def normalize_review_result_token(token: str, review_item: str) -> str | None:
|
|
token = (token or "").strip()
|
|
if not token:
|
|
return None
|
|
if review_item in {"홈런타구 페어 파울", "외야타구 페어 파울"}:
|
|
if "페어" in token:
|
|
return "페어"
|
|
if "파울" in token:
|
|
return "파울"
|
|
elif review_item in {"포수/태그플레이 아웃/세이프", "야수의 포구"}:
|
|
if "아웃" in token:
|
|
return "아웃"
|
|
if "세이프" in token:
|
|
return "세이프"
|
|
elif review_item == "헛스윙":
|
|
# 반드시 "노스윙"을 먼저 체크 ("노스윙"에도 "스윙"이 포함되어 있으므로)
|
|
if "불인정" in token or "노스윙" in token or "공포" in token or "노 스윙" in token:
|
|
return "불인정"
|
|
if "스윙" in token or "인정" in token:
|
|
return "인정"
|
|
else:
|
|
if "불인정" in token or "실패" in token:
|
|
return "불인정"
|
|
if "인정" in token:
|
|
return "인정"
|
|
return token # 모르는 키워드는 원문 그대로 반환
|
|
return None
|
|
|
|
|
|
def parse_review_event_text(text: str) -> dict[str, Any]:
|
|
inning_match = re.search(r"(\d+)회(초|말)", text)
|
|
request_team_match = re.search(r"([가-힣A-Za-z]+)요청\s*(?:비디오 판독|합의 판정)", text)
|
|
# "→노 스윙"처럼 결과 토큰에 공백이 낀 경우를 정규화 (예: "노 스윙" → "노스윙")
|
|
normalized = re.sub(r"→([가-힣]+)\s+([가-힣]+)", r"→\1\2", text)
|
|
detail_match = re.search(r"(?:비디오 판독|합의 판정):\s*(.+?)\s*([가-힣]+)→([가-힣]+)\s*$", normalized)
|
|
detail_text = detail_match.group(1).strip() if detail_match else text
|
|
review_item = infer_review_item(detail_text)
|
|
before_result = normalize_review_result_token(detail_match.group(2), review_item) if detail_match else None
|
|
after_result = normalize_review_result_token(detail_match.group(3), review_item) if detail_match else None
|
|
return {
|
|
"type": "video_review",
|
|
"text": text,
|
|
"requestInningLabel": f"{inning_match.group(1)}{'초' if inning_match.group(2) == '초' else '말'}" if inning_match else None,
|
|
"requestTeam": request_team_match.group(1) if request_team_match else None,
|
|
"reviewItem": review_item,
|
|
"beforeResult": before_result,
|
|
"finalResult": after_result,
|
|
"isSuccess": "성공" if before_result and after_result and before_result != after_result else "실패",
|
|
"timing": "before_pitch" if "초구 전" in text else "after_pitch",
|
|
}
|
|
|
|
|
|
def normalize_review_event(review_event: dict[str, Any]) -> dict[str, Any]:
|
|
# beforeResult/finalResult가 누락되어 있으면 다시 파싱
|
|
has_results = review_event.get("beforeResult") is not None and review_event.get("finalResult") is not None
|
|
if review_event.get("requestInningLabel") and review_event.get("reviewItem") and has_results:
|
|
return review_event
|
|
text = review_event.get("text") or ""
|
|
parsed = parse_review_event_text(text)
|
|
parsed.update({k: v for k, v in review_event.items() if k not in parsed})
|
|
return parsed
|
|
|
|
|
|
def open_challenge_popup(page: Page):
|
|
with page.expect_popup(timeout=5000) as popup_info:
|
|
page.locator("#challengeBtn").click()
|
|
popup = popup_info.value
|
|
popup.wait_for_load_state("domcontentloaded")
|
|
popup.wait_for_selector("#requestInning_0", timeout=5000)
|
|
return popup
|
|
|
|
|
|
def set_select_by_text_or_value(page: Page, selector: str, desired: str) -> None:
|
|
locator = page.locator(selector)
|
|
try:
|
|
locator.select_option(label=desired)
|
|
return
|
|
except Exception:
|
|
pass
|
|
try:
|
|
locator.select_option(value=desired)
|
|
return
|
|
except Exception:
|
|
pass
|
|
locator.select_option(index=0)
|
|
|
|
|
|
def select_review_final_result(popup: Page, row_index: int, review_item: str, final_result: str | None) -> None:
|
|
group_key, default_a, _ = REVIEW_ITEM_RESULT_GROUP_MAP.get(review_item, REVIEW_ITEM_RESULT_GROUP_MAP["기타"])
|
|
result_value = final_result or default_a
|
|
select_selector = f"#finalResult_{group_key}_{row_index}"
|
|
if not popup.locator(select_selector).count():
|
|
select_selector = f"#finalResult_{group_key[-1]}_{row_index}"
|
|
if not popup.locator(select_selector).count():
|
|
select_selector = f"#finalResult_type{group_key[-1]}_{row_index}"
|
|
set_select_by_text_or_value(popup, select_selector, result_value)
|
|
try:
|
|
popup.locator(f"#finalResult_{row_index}").evaluate(
|
|
"""(node, value) => {
|
|
node.value = value;
|
|
node.dispatchEvent(new Event('change', { bubbles: true }));
|
|
}""",
|
|
result_value,
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def fill_review_row(popup: Page, row_index: int, review_event: dict[str, Any]) -> None:
|
|
request_inning = review_event.get("requestInningLabel") or "1초"
|
|
request_team = review_event.get("requestTeam") or popup.locator(f"#requestTeamId_{row_index} option").first.text_content() or ""
|
|
review_item = review_event.get("reviewItem") or "기타"
|
|
final_result = review_event.get("finalResult")
|
|
is_success_val = "Y" if (review_event.get("isSuccess") == "성공") else "N"
|
|
|
|
popup.wait_for_selector(f"#requestInning_{row_index}", timeout=3000)
|
|
set_select_by_text_or_value(popup, f"#requestInning_{row_index}", request_inning)
|
|
|
|
# 팀 선택: 부분 일치 지원
|
|
try:
|
|
set_select_by_partial_text(popup, f"#requestTeamId_{row_index}", request_team)
|
|
except Exception:
|
|
pass
|
|
|
|
# 판독 항목 선택: 부분 일치 지원 (예: '아웃/세이프' -> '아웃,세이프')
|
|
if popup.is_closed(): return
|
|
try:
|
|
set_select_by_partial_text(popup, f"#forWhat_{row_index}", review_item)
|
|
except Exception:
|
|
set_select_by_text_or_value(popup, f"#forWhat_{row_index}", "기타")
|
|
popup.wait_for_timeout(100)
|
|
|
|
if popup.is_closed(): return
|
|
select_review_final_result(popup, row_index, review_item, final_result)
|
|
|
|
if popup.is_closed(): return
|
|
set_select_by_text_or_value(popup, f"#isSuccess_{row_index}", is_success_val)
|
|
|
|
|
|
def append_review_row(popup: Page) -> int:
|
|
before_count = popup.locator("select[id^='requestInning_']").count()
|
|
popup.get_by_role("button", name="신규추가").click()
|
|
popup.wait_for_function(
|
|
"""(expectedCount) => {
|
|
return document.querySelectorAll("select[id^='requestInning_']").length > expectedCount;
|
|
}""",
|
|
arg=before_count,
|
|
timeout=3000,
|
|
)
|
|
row_index = popup.evaluate(
|
|
"""() => {
|
|
const ids = [...document.querySelectorAll("select[id^='requestInning_']")]
|
|
.map((el) => el.id)
|
|
.map((id) => Number(id.split("_").pop()))
|
|
.filter((num) => Number.isFinite(num));
|
|
return ids.length ? Math.max(...ids) : 0;
|
|
}"""
|
|
)
|
|
popup.wait_for_selector(f"#requestInning_{row_index}", timeout=5000)
|
|
return int(row_index)
|
|
|
|
|
|
def can_reuse_initial_review_row(popup: Page) -> bool:
|
|
try:
|
|
row_count = popup.locator("select[id^='requestInning_']").count()
|
|
if row_count != 1:
|
|
return False
|
|
hidden_id = (popup.locator("#id_0").input_value() or "").strip()
|
|
if hidden_id:
|
|
return False
|
|
request_inning = popup.locator("#requestInning_0").input_value()
|
|
request_team = popup.locator("#requestTeamId_0").input_value()
|
|
review_item = popup.locator("#forWhat_0").input_value()
|
|
final_result = (popup.locator("#finalResult_0").input_value() or "").strip()
|
|
is_success = popup.locator("#isSuccess_0").input_value()
|
|
return (
|
|
request_inning == "1"
|
|
and review_item == "홈런타구 페어 파울"
|
|
and final_result == "페어"
|
|
and is_success == "Y"
|
|
and bool(request_team)
|
|
)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def save_review_popup(popup: Page) -> None:
|
|
if popup.is_closed():
|
|
return
|
|
|
|
# 알림/확인 자동 수락
|
|
popup.evaluate("""() => {
|
|
window.confirm = () => true;
|
|
window.alert = () => {};
|
|
}""")
|
|
|
|
# #saveLog 버튼 클릭 + AJAX 응답 대기
|
|
saved = False
|
|
try:
|
|
with popup.expect_response(
|
|
re.compile(r"/manager/game/status/challenge/ajax"),
|
|
timeout=3000,
|
|
) as response_info:
|
|
# saveLog 버튼 클릭
|
|
save_btn = popup.locator("#saveLog")
|
|
if save_btn.count() > 0:
|
|
save_btn.click(force=True)
|
|
else:
|
|
# fallback: 다른 저장 버튼 시도
|
|
popup.evaluate("""() => {
|
|
const btn = document.querySelector('#btnAdd')
|
|
|| document.querySelector('#btnSave')
|
|
|| [...document.querySelectorAll('button, a')].find(
|
|
el => el.innerText.includes('입력완료') || el.innerText.includes('저장')
|
|
);
|
|
if (btn) btn.click();
|
|
}""")
|
|
response = response_info.value
|
|
try:
|
|
body = response.text().strip()
|
|
if body in {"1", '"1"'}:
|
|
saved = True
|
|
except Exception:
|
|
saved = True # 응답 읽기 실패해도 AJAX 자체는 완료됨
|
|
except Exception:
|
|
# AJAX 응답 대기 실패 시 JS로 직접 저장 시도
|
|
try:
|
|
popup.evaluate("""() => {
|
|
const btn = document.querySelector('#saveLog')
|
|
|| document.querySelector('#btnAdd');
|
|
if (btn) btn.click();
|
|
}""")
|
|
popup.wait_for_timeout(1000)
|
|
except Exception:
|
|
pass
|
|
|
|
# 저장 후 팝업 닫기
|
|
try:
|
|
if not popup.is_closed():
|
|
popup.close()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
|
|
def record_review_events(page: Page, review_events: list[dict[str, Any]]) -> None:
|
|
normalized_events = [normalize_review_event(event) for event in (review_events or [])]
|
|
if not normalized_events:
|
|
return
|
|
popup = open_challenge_popup(page)
|
|
reuse_initial_row = can_reuse_initial_review_row(popup)
|
|
for index, review_event in enumerate(normalized_events):
|
|
if index == 0 and reuse_initial_row:
|
|
row_index = 0
|
|
else:
|
|
row_index = append_review_row(popup)
|
|
fill_review_row(popup, row_index, review_event)
|
|
save_review_popup(popup)
|
|
page.wait_for_timeout(300)
|
|
try:
|
|
page.bring_to_front()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def open_game_end_popup(page: Page) -> None:
|
|
page.evaluate("""() => { window.confirm = () => true; window.alert = () => {}; }""")
|
|
page.locator("#gameEndBtn").click(force=True)
|
|
page.wait_for_selector("#btnGameEnd", timeout=5000)
|
|
page.wait_for_selector("input[name^='homeTeamPitcher_'], input[name^='awayTeamPitcher_']", timeout=5000)
|
|
|
|
|
|
def get_game_end_pitcher_rows(page: Page) -> dict[str, list[dict[str, Any]]]:
|
|
return page.evaluate(
|
|
"""() => {
|
|
const rowsFor = (nameAttr) => {
|
|
return [...document.querySelectorAll(`input[name='${nameAttr}']`)].map((input, idx) => {
|
|
const tr = input.closest('tr');
|
|
const firstTd = tr ? tr.querySelector('td') : null;
|
|
return {
|
|
idx,
|
|
name: firstTd ? firstTd.textContent.trim() : '',
|
|
};
|
|
});
|
|
};
|
|
return {
|
|
home: rowsFor('home_player_id'),
|
|
away: rowsFor('away_player_id'),
|
|
};
|
|
}"""
|
|
)
|
|
|
|
|
|
def select_game_end_role(page: Page, side: str, idx: int, role_value: str) -> None:
|
|
selector = f"input[name='{side}TeamPitcher_{idx}'][value='{role_value}']"
|
|
ok = page.evaluate(
|
|
"""(selector) => {
|
|
const node = document.querySelector(selector);
|
|
if (!node) return false;
|
|
node.disabled = false;
|
|
node.checked = true;
|
|
node.dispatchEvent(new MouseEvent('click', { bubbles: true, cancelable: true }));
|
|
node.dispatchEvent(new Event('change', { bubbles: true }));
|
|
return node.checked === true;
|
|
}""",
|
|
selector,
|
|
)
|
|
if not ok:
|
|
page.locator(selector).click(force=True)
|
|
|
|
|
|
def check_game_end_blown_save(page: Page, side: str, idx: int) -> None:
|
|
selector = f"input[name='{side}BlownSave_{idx}']"
|
|
ok = page.evaluate(
|
|
"""(selector) => {
|
|
const node = document.querySelector(selector);
|
|
if (!node) return false;
|
|
node.disabled = false;
|
|
node.checked = true;
|
|
node.dispatchEvent(new MouseEvent('click', { bubbles: true, cancelable: true }));
|
|
node.dispatchEvent(new Event('change', { bubbles: true }));
|
|
return node.checked === true;
|
|
}""",
|
|
selector,
|
|
)
|
|
if not ok:
|
|
page.locator(selector).check(force=True)
|
|
|
|
|
|
def fill_game_end_pitching(page: Page, report: dict[str, Any]) -> None:
|
|
open_game_end_popup(page)
|
|
rows = get_game_end_pitcher_rows(page)
|
|
summary = report.get("pitching_summary") or {}
|
|
home_starter = normalize_lineup_text((((report.get("lineups") or {}).get("home_team") or {}).get("starter_pitcher") or {}).get("name") or "")
|
|
away_starter = normalize_lineup_text((((report.get("lineups") or {}).get("away_team") or {}).get("starter_pitcher") or {}).get("name") or "")
|
|
|
|
winners = {normalize_lineup_text(name) for name in (summary.get("승리투수") or [])}
|
|
losers = {normalize_lineup_text(name) for name in (summary.get("패전투수") or [])}
|
|
holds = {normalize_lineup_text(name) for name in (summary.get("홀드") or [])}
|
|
saves = {normalize_lineup_text(name) for name in (summary.get("세이브") or [])}
|
|
blown_saves = {normalize_lineup_text(name) for name in (summary.get("블론세이브") or [])}
|
|
fixed_roles = winners | losers | holds | saves
|
|
|
|
for side, side_rows in rows.items():
|
|
starter_name = home_starter if side == "home" else away_starter
|
|
for row in side_rows:
|
|
name = normalize_lineup_text(row.get("name") or "")
|
|
idx = int(row["idx"])
|
|
if name in winners:
|
|
select_game_end_role(page, side, idx, "wins")
|
|
elif name in losers:
|
|
select_game_end_role(page, side, idx, "loses")
|
|
elif name in saves:
|
|
select_game_end_role(page, side, idx, "save")
|
|
elif name in holds:
|
|
select_game_end_role(page, side, idx, "holds")
|
|
elif name and name != starter_name and name not in fixed_roles:
|
|
select_game_end_role(page, side, idx, "re")
|
|
|
|
if name in blown_saves:
|
|
check_game_end_blown_save(page, side, idx)
|
|
page.wait_for_timeout(300)
|
|
show_debug_overlay(
|
|
page,
|
|
[
|
|
"게임종료 팝업 입력 완료",
|
|
f"승리: {', '.join(summary.get('승리투수') or []) or '-'}",
|
|
f"패전: {', '.join(summary.get('패전투수') or []) or '-'}",
|
|
f"홀드: {', '.join(summary.get('홀드') or []) or '-'}",
|
|
f"세이브: {', '.join(summary.get('세이브') or []) or '-'}",
|
|
f"블론세이브: {', '.join(summary.get('블론세이브') or []) or '-'}",
|
|
],
|
|
)
|
|
|
|
|
|
def submit_game_end(page: Page) -> None:
|
|
page.evaluate("""() => { window.confirm = () => true; window.alert = () => {}; }""")
|
|
page.locator("#btnGameEnd").click(force=True)
|
|
page.wait_for_timeout(1500)
|
|
|
|
|
|
def set_ball_count(page: Page, balls: int, strikes: int, outs: int) -> None:
|
|
# 볼카운트는 시스템에서 자동 관리되므로 수동 클릭 제거
|
|
# page.locator(f"#evt_ballscore{balls}").check(force=True)
|
|
# page.locator(f"#evt_strikescore{strikes}").check(force=True)
|
|
# page.locator(f"#evt_outscore{outs}").check(force=True)
|
|
pass
|
|
|
|
|
|
def show_debug_overlay(page: Page, lines: list[str]) -> None:
|
|
page.evaluate(
|
|
"""(lines) => {
|
|
let box = document.querySelector('#codex-debug-overlay');
|
|
if (!box) {
|
|
window.codexControl = { paused: false, proceed: 0 };
|
|
box = document.createElement('div');
|
|
box.id = 'codex-debug-overlay';
|
|
box.style.position = 'fixed';
|
|
box.style.top = '12px';
|
|
box.style.right = '12px';
|
|
box.style.zIndex = '999999';
|
|
box.style.background = 'rgba(0, 0, 0, 0.82)';
|
|
box.style.color = '#fff';
|
|
box.style.padding = '10px 12px';
|
|
box.style.borderRadius = '8px';
|
|
box.style.fontSize = '14px';
|
|
box.style.lineHeight = '1.5';
|
|
box.style.maxWidth = '360px';
|
|
box.style.whiteSpace = 'pre-wrap';
|
|
box.style.boxShadow = '0 4px 16px rgba(0,0,0,0.35)';
|
|
const controls = document.createElement('div');
|
|
controls.style.marginBottom = '8px';
|
|
controls.style.display = 'flex';
|
|
controls.style.gap = '8px';
|
|
|
|
const pauseBtn = document.createElement('button');
|
|
pauseBtn.id = 'codex-pause-btn';
|
|
pauseBtn.textContent = '일시정지';
|
|
pauseBtn.onclick = () => {
|
|
window.codexControl.paused = !window.codexControl.paused;
|
|
pauseBtn.textContent = window.codexControl.paused ? '재개' : '일시정지';
|
|
};
|
|
|
|
const nextBtn = document.createElement('button');
|
|
nextBtn.id = 'codex-next-btn';
|
|
nextBtn.textContent = '다음';
|
|
nextBtn.onclick = () => {
|
|
window.codexControl.proceed += 1;
|
|
};
|
|
|
|
controls.appendChild(pauseBtn);
|
|
controls.appendChild(nextBtn);
|
|
|
|
const body = document.createElement('div');
|
|
body.id = 'codex-debug-body';
|
|
|
|
box.appendChild(controls);
|
|
box.appendChild(body);
|
|
document.body.appendChild(box);
|
|
}
|
|
const body = document.querySelector('#codex-debug-body');
|
|
if (body) {
|
|
body.textContent = lines.join('\\n');
|
|
}
|
|
}""",
|
|
lines,
|
|
)
|
|
|
|
|
|
def wait_for_operator_control(page: Page) -> None:
|
|
state = page.evaluate("() => window.codexControl || { paused: false, proceed: 0 }")
|
|
last_proceed = state.get("proceed", 0)
|
|
while True:
|
|
state = page.evaluate("() => window.codexControl || { paused: false, proceed: 0 }")
|
|
if not state.get("paused"):
|
|
return
|
|
if state.get("proceed", 0) > last_proceed:
|
|
page.evaluate("(prev) => { if (window.codexControl && window.codexControl.proceed > prev) { window.codexControl.proceed = prev; } }", last_proceed)
|
|
return
|
|
sleep(0.1)
|
|
|
|
|
|
def get_checked_event_name(page: Page, radio_name: str) -> str:
|
|
return page.evaluate(
|
|
"""(radioName) => {
|
|
const nodes = [...document.querySelectorAll(`input[type=radio][name='${radioName}']:checked`)];
|
|
for (const node of nodes) {
|
|
const rect = node.getBoundingClientRect();
|
|
const style = window.getComputedStyle(node);
|
|
if (rect.width > 0 && rect.height > 0 && style.display !== 'none' && style.visibility !== 'hidden' && !node.disabled) {
|
|
return node.getAttribute('eventName') || '';
|
|
}
|
|
}
|
|
return nodes.length > 0 ? (nodes[0].getAttribute('eventName') || '') : '';
|
|
}""",
|
|
radio_name,
|
|
)
|
|
|
|
|
|
def set_radio_by_label(page: Page, radio_name: str, label: str) -> None:
|
|
radios = page.locator(f"input[type=radio][name='{radio_name}']").all()
|
|
target_radio = None
|
|
|
|
# 1. 정확히 일치하는 라벨 우선 탐색
|
|
for rb in radios:
|
|
if rb.get_attribute("eventname") == label:
|
|
target_radio = rb
|
|
break
|
|
|
|
# 2. 일치하는 게 없으면 포함 관계로 탐색
|
|
if not target_radio:
|
|
for rb in radios:
|
|
if label in (rb.get_attribute("eventname") or ""):
|
|
target_radio = rb
|
|
break
|
|
|
|
if target_radio:
|
|
target_radio.click(force=True)
|
|
|
|
|
|
def get_last_visible_locator(page: Page, selector: str):
|
|
marker = page.evaluate(
|
|
"""(selector) => {
|
|
const nodes = [...document.querySelectorAll(selector)];
|
|
for (let i = nodes.length - 1; i >= 0; i -= 1) {
|
|
const node = nodes[i];
|
|
const rect = node.getBoundingClientRect();
|
|
const style = window.getComputedStyle(node);
|
|
if (rect.width > 0 && rect.height > 0 && style.display !== 'none' && style.visibility !== 'hidden') {
|
|
const marker = `codex-visible-${Math.random().toString(36).slice(2)}`;
|
|
node.setAttribute('data-codex-marker', marker);
|
|
return marker;
|
|
}
|
|
}
|
|
return null;
|
|
}""",
|
|
selector,
|
|
)
|
|
if marker:
|
|
return page.locator(f"[data-codex-marker='{marker}']")
|
|
return None
|
|
|
|
|
|
def get_last_visible_enabled_locator(page: Page, selector: str):
|
|
marker = page.evaluate(
|
|
"""(selector) => {
|
|
const nodes = [...document.querySelectorAll(selector)];
|
|
for (let i = nodes.length - 1; i >= 0; i -= 1) {
|
|
const node = nodes[i];
|
|
const rect = node.getBoundingClientRect();
|
|
const style = window.getComputedStyle(node);
|
|
if (rect.width > 0 && rect.height > 0 && style.display !== 'none' && style.visibility !== 'hidden' && !node.disabled) {
|
|
const marker = `codex-enabled-${Math.random().toString(36).slice(2)}`;
|
|
node.setAttribute('data-codex-marker', marker);
|
|
return marker;
|
|
}
|
|
}
|
|
return null;
|
|
}""",
|
|
selector,
|
|
)
|
|
if marker:
|
|
return page.locator(f"[data-codex-marker='{marker}']")
|
|
return None
|
|
|
|
|
|
def find_visible_radio_by_label(page: Page, radio_name: str, label: str):
|
|
marker = page.evaluate(
|
|
"""({ radioName, expectedLabel }) => {
|
|
const nodes = [...document.querySelectorAll(`input[type=radio][name='${radioName}']`)];
|
|
for (let i = nodes.length - 1; i >= 0; i -= 1) {
|
|
const node = nodes[i];
|
|
const rect = node.getBoundingClientRect();
|
|
const style = window.getComputedStyle(node);
|
|
if (!rect.width || !rect.height || style.display === 'none' || style.visibility === 'hidden' || node.disabled) {
|
|
continue;
|
|
}
|
|
let current = node.nextSibling;
|
|
let text = '';
|
|
while (current) {
|
|
if (current.textContent && current.textContent.trim()) {
|
|
text = current.textContent.trim();
|
|
break;
|
|
}
|
|
current = current.nextSibling;
|
|
}
|
|
if (text === expectedLabel) {
|
|
const marker = `codex-radio-${Math.random().toString(36).slice(2)}`;
|
|
node.setAttribute('data-codex-marker', marker);
|
|
return marker;
|
|
}
|
|
}
|
|
return null;
|
|
}""",
|
|
{"radioName": radio_name, "expectedLabel": label},
|
|
)
|
|
if marker:
|
|
return page.locator(f"[data-codex-marker='{marker}']")
|
|
return None
|
|
|
|
|
|
def wait_for_visible_locator(page: Page, selector: str, timeout_ms: int = 5000):
|
|
deadline = time() + timeout_ms / 1000
|
|
while time() < deadline:
|
|
candidate = get_last_visible_locator(page, selector)
|
|
if candidate is not None:
|
|
return candidate
|
|
sleep(0.1)
|
|
raise TimeoutError(f"{selector} visible locator not found within {timeout_ms}ms")
|
|
|
|
|
|
def get_history_count(page: Page) -> int:
|
|
try:
|
|
return int(
|
|
page.evaluate(
|
|
"""() => document.querySelectorAll("div[name='historyView']").length"""
|
|
)
|
|
)
|
|
except Exception:
|
|
return 0
|
|
|
|
|
|
def get_last_history_text(page: Page) -> str:
|
|
try:
|
|
return (
|
|
page.evaluate(
|
|
"""() => {
|
|
const nodes = document.querySelectorAll("div[name='historyView']");
|
|
const lastNode = nodes[nodes.length - 1];
|
|
return lastNode ? lastNode.textContent.trim() : '';
|
|
}"""
|
|
)
|
|
or ""
|
|
).strip()
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def wait_for_history_increment(page: Page, before_count: int, timeout_ms: int = 2500) -> bool:
|
|
deadline = time() + timeout_ms / 1000
|
|
while time() < deadline:
|
|
if get_history_count(page) > before_count:
|
|
return True
|
|
sleep(0.1)
|
|
return False
|
|
|
|
|
|
def submit_input_complete(page: Page, debug_label: str = "", clear_defense: bool = False, log_info: dict[str, Any] | None = None) -> None:
|
|
from time import time
|
|
t0 = time()
|
|
try:
|
|
# 팝업이나 다이얼로그가 떠 있으면 닫음
|
|
page.evaluate("""() => {
|
|
window.confirm = () => true;
|
|
window.alert = () => {};
|
|
const defenseDiv = document.querySelector('#defenseDiv');
|
|
if (defenseDiv && defenseDiv.style.display !== 'none') {
|
|
const btnAdd = document.querySelector('#btnAdd');
|
|
if (btnAdd) btnAdd.click();
|
|
}
|
|
}""")
|
|
|
|
# clear_defense=True 일 때만 수비/타구 필드 초기화
|
|
# (볼/스트라이크 등 타격 없는 일반 투구만 해당)
|
|
# 타격 결과(안타, 아웃 등)는 팝업이 이미 히든 필드를 채운 상태이므로 지우면 안 됨!
|
|
if clear_defense:
|
|
page.evaluate(_DEFENSE_CLEAR_JS)
|
|
|
|
prev_history = get_history_count(page)
|
|
# 최대 40번 시도 (약 6~8초) - 간격 세분화로 빠른 응답 포착
|
|
for i in range(40):
|
|
curr_history = get_history_count(page)
|
|
if curr_history > prev_history:
|
|
# 입력 성공 후 안정화 대기
|
|
page.wait_for_timeout(30)
|
|
_try_log_pitch(log_info, True, "", "", time() - t0)
|
|
return
|
|
|
|
# 매 루프마다 클릭하면 오히려 부하가 될 수 있으므로 주기적으로 시도
|
|
if i % 8 == 0:
|
|
submit_btn = get_last_visible_locator(page, "#eventWriteBtn")
|
|
if not submit_btn:
|
|
submit_btn = page.get_by_role("button", name="입력완료").last
|
|
|
|
if submit_btn:
|
|
try:
|
|
submit_btn.click(force=True, timeout=500)
|
|
except Exception:
|
|
page.evaluate("document.querySelector('#eventWriteBtn')?.click() || [...document.querySelectorAll('a, button')].find(el => el.innerText.includes('입력완료'))?.click()")
|
|
|
|
page.wait_for_timeout(50)
|
|
page.evaluate("() => { window.confirm = () => true; window.alert = () => {}; }")
|
|
|
|
raise TimeoutError(f"입력완료가 반영되지 않았습니다 (버튼 클릭 후 이력 불일치): {debug_label}")
|
|
except Exception as e:
|
|
_try_log_pitch(log_info, False, type(e).__name__, str(e), time() - t0)
|
|
raise e
|
|
|
|
def _try_log_pitch(log_info: dict[str, Any] | None, is_success: bool, error_code: str, error_detail: str, duration: float) -> None:
|
|
if log_info and log_info.get("job_id"):
|
|
try:
|
|
from db_logging import log_pitch
|
|
log_pitch(
|
|
log_info["job_id"], log_info.get("inning", ""), log_info.get("batter", ""),
|
|
log_info.get("pitch_no", 0), log_info.get("target_value", ""), log_info.get("selected_value", ""),
|
|
is_success, error_code, error_detail, duration
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _try_log_event(log_info: dict[str, Any] | None, is_success: bool, error_msg: str = "") -> None:
|
|
if log_info and log_info.get("job_id"):
|
|
try:
|
|
from db_logging import log_event
|
|
log_event(
|
|
log_info["job_id"], log_info.get("inning", ""), log_info.get("event_type", ""),
|
|
log_info.get("target_player", ""), log_info.get("actual_player", ""),
|
|
is_success, error_msg
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def set_runner_advance(page: Page, from_base: int, to_base: int | None) -> None:
|
|
if to_base is None:
|
|
return
|
|
selector = f"input[name='dat_evt_runner_{from_base}_advance'][value='{to_base}']"
|
|
deadline = time() + 3
|
|
locator = None
|
|
while time() < deadline:
|
|
locator = get_last_visible_enabled_locator(page, selector)
|
|
if locator is not None:
|
|
break
|
|
sleep(0.1)
|
|
if locator is None:
|
|
fallback = page.locator(selector)
|
|
if fallback.count():
|
|
locator = fallback.last
|
|
else:
|
|
raise TimeoutError(f"{selector} enabled locator not found within 3000ms")
|
|
try:
|
|
locator.click(force=True)
|
|
except Exception:
|
|
locator.evaluate(
|
|
"""(node) => {
|
|
node.disabled = false;
|
|
node.checked = true;
|
|
node.dispatchEvent(new Event('click', { bubbles: true }));
|
|
node.dispatchEvent(new Event('change', { bubbles: true }));
|
|
}"""
|
|
)
|
|
|
|
|
|
def set_runner_action(page: Page, from_base: int, label: str) -> None:
|
|
radio_name = f"evt_runner_{from_base}"
|
|
locator = page.evaluate(
|
|
"""({ radioName, eventName }) => {
|
|
const nodes = [...document.querySelectorAll(`input[type=radio][name='${radioName}']`)];
|
|
// 1단계: eventName 속성으로 정확히 매칭 시도
|
|
for (const node of nodes) {
|
|
const rect = node.getBoundingClientRect();
|
|
const style = window.getComputedStyle(node);
|
|
if (!rect.width || !rect.height || style.display === 'none' || style.visibility === 'hidden' || node.disabled) {
|
|
continue;
|
|
}
|
|
if ((node.getAttribute('eventName') || '') === eventName) {
|
|
const marker = `codex-runner-${Math.random().toString(36).slice(2)}`;
|
|
node.setAttribute('data-codex-marker', marker);
|
|
return marker;
|
|
}
|
|
}
|
|
// 2단계: 라벨 텍스트로 부분 매칭 시도 (태그아웃 vs 태그 아웃 등 대응)
|
|
for (const node of nodes) {
|
|
let text = '';
|
|
let p = node.parentElement;
|
|
if (p) text = p.textContent.trim();
|
|
if (!text && node.nextSibling) text = node.nextSibling.textContent || '';
|
|
|
|
if (eventName && text.replace(/\s/g, '').includes(eventName.replace(/\s/g, ''))) {
|
|
const marker = `codex-runner-${Math.random().toString(36).slice(2)}`;
|
|
node.setAttribute('data-codex-marker', marker);
|
|
return marker;
|
|
}
|
|
}
|
|
return null;
|
|
}""",
|
|
{"radioName": radio_name, "eventName": label},
|
|
)
|
|
if locator:
|
|
candidate = page.locator(f"[data-codex-marker='{locator}']")
|
|
for _ in range(5):
|
|
try:
|
|
# 가시성 이슈가 있을 수 있으므로 force=True와 evaluate(JS click) 병행
|
|
candidate.click(force=True, timeout=500)
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
candidate.evaluate(
|
|
"""(node) => {
|
|
node.disabled = false;
|
|
node.checked = true;
|
|
node.dispatchEvent(new Event('click', { bubbles: true }));
|
|
node.dispatchEvent(new Event('change', { bubbles: true }));
|
|
}"""
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
page.wait_for_timeout(100)
|
|
if get_checked_event_name(page, radio_name) == label:
|
|
return
|
|
else:
|
|
for _ in range(5):
|
|
set_radio_by_label(page, radio_name, label)
|
|
page.wait_for_timeout(50)
|
|
if get_checked_event_name(page, radio_name) == label:
|
|
return
|
|
|
|
|
|
def infer_runner_action_label(event: dict[str, Any], runner_event: dict[str, Any]) -> str | None:
|
|
# 0. 리포트에 명시된 라벨이 있으면 최우선으로 사용 (가장 정확함)
|
|
if "action_label" in runner_event:
|
|
return runner_event["action_label"]
|
|
|
|
event_type = runner_event.get("type") or ""
|
|
event_text = runner_event.get("text") or ""
|
|
result_type = ((event.get("result") or {}).get("type") or "")
|
|
result_text = ((event.get("result") or {}).get("text") or "")
|
|
|
|
if "이중도루 실패" in event_text and "진루" in event_text:
|
|
return "기타 진루"
|
|
if "도루" in event_text and "실패" in event_text and "진루" in event_text:
|
|
return "기타 진루"
|
|
|
|
if event_type == "pickoff_out" or "견제사" in event_text:
|
|
return "견제 아웃"
|
|
|
|
if event_type == "steal_fail":
|
|
return "도루시도 아웃"
|
|
if "이중도루 실패" in event_text and "아웃" in event_text:
|
|
return "도루시도 아웃"
|
|
if "도루" in event_text and "실책" in event_text and ("진루" in event_text or event_type == "error_advance"):
|
|
return "도루성공&실책"
|
|
if "도루" in event_text:
|
|
if "실패" in event_text:
|
|
return "도루시도 아웃"
|
|
return "도루성공"
|
|
if "낫아웃" in result_text and event_type == "wild_pitch_advance":
|
|
return "폭투 낫아웃 진루"
|
|
if "낫아웃" in result_text and event_type == "passed_ball_advance":
|
|
return "포일 낫아웃 진루"
|
|
if "포일" in event_text and ("진루" in event_text or event_type == "passed_ball_advance"):
|
|
return "포일-진루성공"
|
|
|
|
if "포일" in event_text and ("진루" in event_text or event_type == "passed_ball_advance"):
|
|
return "포일-진루성공"
|
|
|
|
# 1. 수비 실책: 텍스트에 '실책으로'가 명시된 경우 최우선
|
|
if "실책으로" in event_text:
|
|
return "수비 실책"
|
|
|
|
# 2. 안타/아웃/타격 관련 상황이면 무조건 일반 진루 (볼넷 진루 방지)
|
|
play_types = {"single", "double", "triple", "home_run", "out", "strikeout", "play", "sacrifice_fly", "sacrifice_bunt", "ground_out", "fly_out"}
|
|
if result_type in play_types and event_type in {"advance", "score"}:
|
|
return "일반 진루"
|
|
|
|
# 3. 볼넷 진루: 오직 포볼, 고의사구, 몸에 맞는 볼 상황서만 허용
|
|
walk_types = {"walk", "intentional_walk", "hit_by_pitch"}
|
|
if result_type in walk_types and event_type in {"advance", "score"}:
|
|
return "볼넷 진루"
|
|
|
|
# 4. 나머지는 일반 진루
|
|
if event_type in {"advance", "score"}:
|
|
return "일반 진루"
|
|
|
|
return RUNNER_EVENT_LABEL_MAP.get(event_type)
|
|
|
|
|
|
def get_runner_area_type(event: dict[str, Any], runner_event: dict[str, Any]) -> int:
|
|
event_text = runner_event.get("text") or ""
|
|
|
|
# 도루, 견제, 폭투, 포일, 아웃 등 버튼 조작이 필요한 특수 상황은 액션 영역(2)
|
|
action_keywords = ["도루", "견제", "폭투", "포일", "태그아웃", "포스아웃"]
|
|
if any(k in event_text for k in action_keywords):
|
|
return 2
|
|
|
|
return 1
|
|
|
|
|
|
def open_runner_area(page: Page, from_base: int, area_type: int) -> None:
|
|
function_name = f"changRunnerArea{from_base}"
|
|
page.evaluate(
|
|
"""({ functionName, areaType }) => {
|
|
const fn = window[functionName];
|
|
if (typeof fn === 'function') {
|
|
fn(areaType);
|
|
}
|
|
}""",
|
|
{"functionName": function_name, "areaType": area_type},
|
|
)
|
|
deadline = time() + 2
|
|
radio_name = f"evt_runner_{from_base}"
|
|
advance_name = f"dat_evt_runner_{from_base}_advance"
|
|
while time() < deadline:
|
|
ready = page.evaluate(
|
|
"""({ radioName, advanceName }) => {
|
|
const hasEnabledVisibleRadio = [...document.querySelectorAll(`input[type=radio][name='${radioName}']`)].some((node) => {
|
|
const rect = node.getBoundingClientRect();
|
|
const style = window.getComputedStyle(node);
|
|
return !!(rect.width && rect.height && style.display !== 'none' && style.visibility !== 'hidden' && !node.disabled);
|
|
});
|
|
const hasEnabledVisibleAdvance = [...document.querySelectorAll(`input[type=radio][name='${advanceName}']`)].some((node) => {
|
|
const rect = node.getBoundingClientRect();
|
|
const style = window.getComputedStyle(node);
|
|
return !!(rect.width && rect.height && style.display !== 'none' && style.visibility !== 'hidden' && !node.disabled);
|
|
});
|
|
return hasEnabledVisibleRadio || hasEnabledVisibleAdvance;
|
|
}""",
|
|
{"radioName": radio_name, "advanceName": advance_name},
|
|
)
|
|
if ready:
|
|
return
|
|
sleep(0.1)
|
|
|
|
|
|
def split_complex_runner_event(runner_event: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any] | None]:
|
|
text = runner_event.get("text") or ""
|
|
|
|
# [중요] '실책' 혹은 '/'가 포함된 복합 상황만 분할함.
|
|
if "/" not in text or ("실책" not in text and "아웃" not in text):
|
|
return runner_event, None
|
|
|
|
parts = [p.strip() for p in text.split("/") if p.strip()]
|
|
if len(parts) < 2:
|
|
return runner_event, None
|
|
|
|
# 텍스트에서 중간 베이스 정보를 추출하여 toBase 보정
|
|
# 예: "1루주자 ... 2루까지 진루" -> intermediate = 2
|
|
def extract_base(t: str) -> int | None:
|
|
m = re.search(r"([123])루", t)
|
|
return int(m.group(1)) if m else None
|
|
|
|
# 1차 이벤트: 첫 번째 분절의 행동
|
|
primary = dict(runner_event)
|
|
primary["text"] = parts[0]
|
|
|
|
intermediate_to = extract_base(parts[0])
|
|
if intermediate_to:
|
|
primary["toBase"] = intermediate_to
|
|
|
|
# 2차 이벤트: 두 번째 분절의 행동 (지연 처리용)
|
|
secondary = dict(runner_event)
|
|
secondary["fromBase"] = primary["toBase"] # 1차의 목적지가 2차의 시작점
|
|
secondary["text"] = parts[1]
|
|
|
|
if "실책" in parts[1]:
|
|
secondary["type"] = "error_advance"
|
|
elif "태그" in parts[1]:
|
|
secondary["type"] = "tag_out"
|
|
elif "포스" in parts[1]:
|
|
secondary["type"] = "force_out"
|
|
else:
|
|
secondary["type"] = "out"
|
|
|
|
return primary, secondary
|
|
|
|
def set_runner_events(page: Page, event: dict[str, Any], runner_events: list[dict[str, Any]] | None = None) -> list[dict[str, Any]]:
|
|
is_double_play = is_double_play_result(((event.get("result") or {}).get("text") or "").strip())
|
|
if runner_events is None:
|
|
runner_events = (event.get("runnerEvents") or []).copy()
|
|
|
|
late_events = []
|
|
filtered_events = []
|
|
for re_item in runner_events:
|
|
primary, secondary = split_complex_runner_event(re_item)
|
|
filtered_events.append(primary)
|
|
if secondary:
|
|
late_events.append(secondary)
|
|
|
|
for runner_event in filtered_events:
|
|
from_base = runner_event.get("fromBase")
|
|
if from_base not in {1, 2, 3}:
|
|
continue
|
|
label = infer_runner_action_label(event, runner_event)
|
|
if not label:
|
|
continue
|
|
|
|
# 라벨에 따른 영역 결정 (도루, 견제 등 특수 액션만 2번 영역)
|
|
if any(k in label for k in ["도루", "견제", "폭투", "포일", "아웃"]):
|
|
area_type = 2
|
|
else:
|
|
area_type = 1
|
|
|
|
# [강제] 일반/볼넷 진루 및 실책은 무조건 1번(진루) 영역
|
|
if any(k in label for k in ["일반 진루", "볼넷 진루", "수비 실책"]):
|
|
area_type = 1
|
|
|
|
open_runner_area(page, from_base, area_type)
|
|
set_runner_action(page, from_base, label)
|
|
|
|
runner_text = runner_event.get("text") or ""
|
|
is_error_related_label = (
|
|
(label and "실책" in label)
|
|
or label in {"견제 에러", "수비 실책", "도루성공&실책"}
|
|
or "실책" in runner_text
|
|
)
|
|
if is_error_related_label:
|
|
# 주자 텍스트에 구체적인 수비 위치(예: 유격수)가 명시된 경우만 별도 실책 팝업 처리
|
|
if extract_error_position(runner_text):
|
|
fill_error_defense_popup(page, runner_text)
|
|
|
|
if label in {"태그아웃", "도루시도 아웃", "포스아웃"}:
|
|
fill_runner_out_defense(page, runner_event.get("text") or "")
|
|
|
|
page.wait_for_timeout(150)
|
|
set_runner_advance(page, from_base, runner_event.get("toBase"))
|
|
|
|
try:
|
|
extra_advance = runner_event.get("extra_advance")
|
|
if extra_advance and extra_advance > 0:
|
|
locator = get_last_visible_enabled_locator(page, f"select[name='runner{from_base}_running_add']")
|
|
if locator is not None:
|
|
locator.select_option(value=str(extra_advance))
|
|
except Exception:
|
|
pass
|
|
|
|
page.wait_for_timeout(150)
|
|
|
|
return late_events
|
|
|
|
|
|
def handle_late_runner_events(page: Page, event: dict[str, Any], late_events: list[dict[str, Any]], write_events: bool, job_id: str | None = None) -> None:
|
|
if not late_events or not write_events:
|
|
return
|
|
|
|
# 1차 입력완료 후 사이트 갱신을 위해 대기 (이전보다 단축)
|
|
page.wait_for_timeout(800)
|
|
|
|
# 이미 반영되어 있는지 확인 (중복 입력 방지)
|
|
current_history = get_last_history_text(page)
|
|
all_matched = True
|
|
for le in late_events:
|
|
le_text = le.get("text", "")
|
|
if le_text and le_text not in current_history:
|
|
all_matched = False
|
|
break
|
|
if all_matched:
|
|
return
|
|
|
|
# 2차(지연된) 주루 이벤트 입력
|
|
new_late = set_runner_events(page, event, late_events)
|
|
|
|
# 2차 입력완료 클릭
|
|
submit_input_complete(
|
|
page,
|
|
f"지연 주루 처리: {', '.join(e.get('text', '') for e in late_events)}",
|
|
clear_defense=True,
|
|
log_info={"job_id": job_id} if job_id else None
|
|
)
|
|
|
|
if new_late:
|
|
handle_late_runner_events(page, event, new_late, write_events, job_id)
|
|
|
|
|
|
def build_runner_event_lines(event: dict[str, Any]) -> list[str]:
|
|
lines: list[str] = []
|
|
for runner_event in (event.get("runnerEvents") or []):
|
|
r_text = runner_event.get("text", "")
|
|
from_b = runner_event.get("fromBase", "?")
|
|
to_b = runner_event.get("toBase", "?")
|
|
label = infer_runner_action_label(event, runner_event)
|
|
line = f"🏃 {from_b}루주자 -> {to_b}루 : {r_text}"
|
|
if label:
|
|
line += f" | 라벨: {label}"
|
|
lines.append(line)
|
|
return lines
|
|
|
|
|
|
def get_pitch_runner_events(pitch: dict[str, Any], event: dict[str, Any] | None = None) -> list[dict[str, Any]]:
|
|
pitch_runner_events = list(pitch.get("runnerEvents") or [])
|
|
if pitch_runner_events:
|
|
return pitch_runner_events
|
|
if event and event.get("runnerEvents"):
|
|
return list(event.get("runnerEvents") or [])
|
|
return []
|
|
|
|
|
|
def set_pitch(page: Page, pitch: dict[str, Any], event: dict[str, Any] | None = None) -> None:
|
|
pitch_type = PITCH_TYPE_LABEL_MAP.get(pitch.get("pitchType") or "")
|
|
pitch_result_text = (pitch.get("pitchResultText") or "").strip()
|
|
normalized_pitch_result_text = pitch_result_text.replace(" ", "")
|
|
if "피치클락" in pitch_result_text and "투수위반" in pitch_result_text:
|
|
pitch_result_text = "볼"
|
|
|
|
# 폭투/포일 체크 (투구 단위)
|
|
runner_events = get_pitch_runner_events(pitch, event)
|
|
is_wild_pitch = any(
|
|
re.get("type") == "wild_pitch_advance" or "폭투" in (re.get("text") or "")
|
|
for re in runner_events
|
|
)
|
|
is_passed_ball = any(
|
|
re.get("type") == "passed_ball_advance" or "포일" in (re.get("text") or "")
|
|
for re in runner_events
|
|
)
|
|
|
|
if is_wild_pitch:
|
|
pitch_result = "폭투-볼"
|
|
elif is_passed_ball:
|
|
pitch_result = "포일-볼"
|
|
else:
|
|
if "번트" in normalized_pitch_result_text and "헛스윙" in normalized_pitch_result_text:
|
|
pitch_result = "번트시도-스트라이크"
|
|
elif "번트" in normalized_pitch_result_text and "파울" in normalized_pitch_result_text:
|
|
pitch_result = "번트-파울"
|
|
else:
|
|
pitch_result = PITCH_RESULT_LABEL_MAP.get(pitch_result_text)
|
|
if not pitch_result and pitch.get("pitchResult") in {"BS", "V"}:
|
|
pitch_result = "번트시도-스트라이크"
|
|
if not pitch_result and pitch.get("pitchResult") == "BF":
|
|
pitch_result = "번트-파울"
|
|
if not pitch_result and "고의사구" in pitch_result_text:
|
|
pitch_result = "고의사구"
|
|
if not pitch_result and "파울플라이" in pitch_result_text and "실책" in pitch_result_text:
|
|
pitch_result = "파울플라이-실책"
|
|
|
|
if pitch_type:
|
|
set_radio_by_label(page, "evt_ballType", pitch_type)
|
|
|
|
if pitch_result:
|
|
set_radio_by_label(page, "evt_batter", pitch_result)
|
|
|
|
speed_input = page.locator("#ballspeed")
|
|
speed_input.fill(str(pitch.get("speedKmh") or 0))
|
|
speed_input.evaluate("node => node.dispatchEvent(new Event('change', {bubbles:true}))")
|
|
page.wait_for_timeout(50)
|
|
|
|
|
|
def set_pitch_meta_only(page: Page, pitch: dict[str, Any]) -> None:
|
|
"""구종/구속만 세팅. 인플레이(H) 마지막 구에서 팝업이 일찍 열리지 않도록 evt_batter는 건드리지 않음."""
|
|
pitch_type = PITCH_TYPE_LABEL_MAP.get(pitch.get("pitchType") or "")
|
|
if pitch_type:
|
|
set_radio_by_label(page, "evt_ballType", pitch_type)
|
|
page.locator("#ballspeed").fill(str(pitch.get("speedKmh") or 0))
|
|
|
|
|
|
def normalize_pitch_result_code(pitch: dict[str, Any], event: dict[str, Any] | None = None) -> str:
|
|
pitch_result = (pitch.get("pitchResult") or "").strip()
|
|
pitch_result_text = (pitch.get("pitchResultText") or "").strip()
|
|
normalized_text = pitch_result_text.replace(" ", "")
|
|
if "피치클락" in pitch_result_text and "투수위반" in pitch_result_text:
|
|
return "B"
|
|
if "번트" in normalized_text and "헛스윙" in normalized_text:
|
|
return "BS"
|
|
runner_events = get_pitch_runner_events(pitch, event)
|
|
if any(re.get("type") == "wild_pitch_advance" or "폭투" in (re.get("text") or "") for re in runner_events):
|
|
return "B"
|
|
if any(re.get("type") == "passed_ball_advance" or "포일" in (re.get("text") or "") for re in runner_events):
|
|
return "B"
|
|
return pitch_result
|
|
|
|
|
|
def set_batter_result_type(page: Page, result: dict[str, Any] | None, event: dict[str, Any] | None = None) -> None:
|
|
"""타격 결과 종류(1루타, 수비실책 등)만 세팅. 이 작업은 보통 구장 이미지 팝업을 엶."""
|
|
if not result:
|
|
return
|
|
label = infer_batter_result_label(result, event)
|
|
if not label:
|
|
return
|
|
|
|
if label == "병살-아웃":
|
|
forced = page.evaluate(
|
|
"""(eventName) => {
|
|
const nodes = [...document.querySelectorAll("input[type=radio][name='evt_batter']")];
|
|
for (const node of nodes) {
|
|
const name = (node.getAttribute('eventName') || '').trim();
|
|
if (name === eventName) {
|
|
node.disabled = false;
|
|
node.checked = true;
|
|
node.dispatchEvent(new MouseEvent('click', { bubbles: true, cancelable: true }));
|
|
node.dispatchEvent(new Event('change', { bubbles: true }));
|
|
return true;
|
|
}
|
|
}
|
|
return false;
|
|
}""",
|
|
label,
|
|
)
|
|
if forced:
|
|
page.wait_for_timeout(120)
|
|
if get_checked_event_name(page, "evt_batter") == label:
|
|
return
|
|
|
|
marker = page.evaluate(
|
|
"""(eventName) => {
|
|
const nodes = [...document.querySelectorAll("input[type=radio][name='evt_batter']")];
|
|
for (let i = nodes.length - 1; i >= 0; i -= 1) {
|
|
const node = nodes[i];
|
|
const rect = node.getBoundingClientRect();
|
|
const style = window.getComputedStyle(node);
|
|
if (!rect.width || !rect.height || style.display === 'none' || style.visibility === 'hidden' || node.disabled) {
|
|
continue;
|
|
}
|
|
if ((node.getAttribute('eventName') || '') === eventName) {
|
|
const marker = `codex-batter-${Math.random().toString(36).slice(2)}`;
|
|
node.setAttribute('data-codex-marker', marker);
|
|
return marker;
|
|
}
|
|
}
|
|
return null;
|
|
}""",
|
|
label,
|
|
)
|
|
if marker:
|
|
candidate = page.locator(f"[data-codex-marker='{marker}']")
|
|
for _ in range(3):
|
|
try:
|
|
candidate.click(force=True)
|
|
except Exception:
|
|
candidate.evaluate(
|
|
"""(node) => {
|
|
node.disabled = false;
|
|
node.checked = true;
|
|
node.dispatchEvent(new Event('click', { bubbles: true }));
|
|
node.dispatchEvent(new Event('change', { bubbles: true }));
|
|
}"""
|
|
)
|
|
checked_name = get_checked_event_name(page, "evt_batter")
|
|
if checked_name == label:
|
|
return
|
|
page.wait_for_timeout(100)
|
|
|
|
for _ in range(3):
|
|
click_radio_by_label(page, "evt_batter", label)
|
|
if get_checked_event_name(page, "evt_batter") == label:
|
|
break
|
|
page.wait_for_timeout(100)
|
|
|
|
|
|
def set_batter_advancement(page: Page, result: dict[str, Any] | None) -> None:
|
|
"""타자의 최종 루(1루, 2루 등)와 주루가산 세팅. 구장 팝업이 닫힌 후 메인 폼에서 처리해야 함."""
|
|
if not result:
|
|
return
|
|
|
|
# 타자 최종 루 (1루, 2루, 3루, 홈)
|
|
to_base = result.get("toBase")
|
|
|
|
# 1루타, 수비실책, 야수선택, 몸에 맞는 볼, 볼넷 등은 명시적인 toBase가 없더라도 기본적으로 1루로 진루함
|
|
if to_base is None:
|
|
r_type = result.get("type")
|
|
if r_type in {"single", "bunt_hit", "reach_on_error", "reach_on_fielder_choice", "reach_on_grounder", "hit_by_pitch", "walk", "intentional_walk"}:
|
|
to_base = 1
|
|
elif r_type == "double":
|
|
to_base = 2
|
|
elif r_type == "triple":
|
|
to_base = 3
|
|
elif r_type == "home_run":
|
|
to_base = 4
|
|
|
|
if to_base is not None:
|
|
try:
|
|
selector = f"input[name='dat_evt_batter_advance'][value='{to_base}']"
|
|
locator = get_last_visible_enabled_locator(page, selector)
|
|
if locator is not None:
|
|
locator.check(force=True)
|
|
else:
|
|
fallback = page.locator(selector)
|
|
if fallback.count() > 0:
|
|
fallback.first.check(force=True)
|
|
except Exception:
|
|
pass
|
|
|
|
# 주루가산 (Extra Advance)
|
|
try:
|
|
extra_advance = result.get("extra_advance")
|
|
if extra_advance is not None and extra_advance > 0:
|
|
locator = get_last_visible_enabled_locator(page, "#batterRunningAdd")
|
|
if locator is not None:
|
|
locator.select_option(value=str(extra_advance))
|
|
else:
|
|
fallback = page.locator("#batterRunningAdd")
|
|
if fallback.count() > 0:
|
|
fallback.first.select_option(value=str(extra_advance))
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def set_batter_result(page: Page, result: dict[str, Any] | None, event: dict[str, Any] | None = None) -> None:
|
|
"""하위 호환성을 위해 유지. 결과 종류와 진루를 모두 처리 시도."""
|
|
set_batter_result_type(page, result, event)
|
|
set_batter_advancement(page, result)
|
|
|
|
|
|
def get_last_pitch_result_text(event: dict[str, Any] | None) -> str:
|
|
pitches = (event or {}).get("pitches") or []
|
|
if not pitches:
|
|
return ""
|
|
return ((pitches[-1] or {}).get("pitchResultText") or "").strip()
|
|
|
|
|
|
def is_simple_terminal_result_type(result_type: str) -> bool:
|
|
return result_type in {"strikeout", "strikeout_not_out", "walk", "intentional_walk", "hit_by_pitch"}
|
|
|
|
|
|
def infer_batter_result_label(result: dict[str, Any], event: dict[str, Any] | None = None) -> str | None:
|
|
result_type = result.get("type") or ""
|
|
result_text = (result.get("text") or "").strip()
|
|
runner_events = (event or {}).get("runnerEvents") or []
|
|
last_pitch_result_text = get_last_pitch_result_text(event)
|
|
|
|
if result_type == "strikeout_not_out" or "낫아웃" in result_text:
|
|
if "폭투" in result_text:
|
|
return "폭투 낫아웃 진루"
|
|
if "포일" in result_text:
|
|
return "포일 낫아웃 진루"
|
|
if "아웃" in result_text:
|
|
return "스트라이크-낫아웃"
|
|
return "낫아웃-출루"
|
|
|
|
if result_type == "strikeout":
|
|
if "헛스윙" in last_pitch_result_text or "헛스윙" in result_text:
|
|
return "스윙 스트라이크-아웃"
|
|
return "루킹스트라이크-아웃"
|
|
|
|
if "희생 번트" in result_text or "희생번트" in result_text:
|
|
return "희생 번트"
|
|
|
|
if "번트 아웃" in result_text or "번트아웃" in result_text:
|
|
return "번트-아웃"
|
|
|
|
if any("보크" in (runner_event.get("text") or "") and "진루" in (runner_event.get("text") or "") for runner_event in runner_events):
|
|
if "볼" in last_pitch_result_text:
|
|
return "보크-볼"
|
|
return "보크"
|
|
# 폭투-볼 처리 (최우선)
|
|
if any((re.get("type") == "wild_pitch_advance") for re in runner_events):
|
|
return "폭투-볼"
|
|
|
|
# 포볼(Ball 4) 처리
|
|
if result_type == "walk":
|
|
if any(re.get("type") == "wild_pitch_advance" or "폭투" in (re.get("text") or "") for re in runner_events):
|
|
return "폭투-포볼"
|
|
if any(re.get("type") == "passed_ball_advance" or "포일" in (re.get("text") or "") for re in runner_events):
|
|
return "포일-포볼"
|
|
return "포볼"
|
|
if any((runner_event.get("type") or "") == "passed_ball_advance" for runner_event in runner_events):
|
|
if "볼" in last_pitch_result_text:
|
|
return "포일-볼"
|
|
return "포일-스트라이크"
|
|
if result_type == "reach_on_error" or "실책" in result_text:
|
|
return "수비실책"
|
|
|
|
if result_type in {"reach_on_fielder_choice"}:
|
|
return "야수선택"
|
|
|
|
if result_type in {"reach_on_grounder"}:
|
|
return "땅볼출루(무안타)"
|
|
|
|
if result_type == "double_play":
|
|
if "번트" in result_text:
|
|
return "번트-병살"
|
|
return "병살-아웃"
|
|
if result_type == "single_runner_out":
|
|
return "1루타 후 주루아웃"
|
|
if result_type == "double_runner_out":
|
|
return "2루타 후 주루아웃"
|
|
if result_type == "triple_runner_out":
|
|
return "3루타 후 주루아웃"
|
|
if result_type == "single_error_advance":
|
|
return "1루타 후 수비실책진루"
|
|
if result_type == "double_error_advance":
|
|
return "2루타 후 수비실책진루"
|
|
if result_type == "triple_error_advance":
|
|
return "3루타 후 수비실책진루"
|
|
|
|
if "파울희생플라이" in result_text or "파울 희생플라이" in result_text:
|
|
return "희생 플라이"
|
|
|
|
if result_type == "out":
|
|
if "병살" in result_text:
|
|
if "번트" in result_text:
|
|
return "번트-병살"
|
|
return "병살-아웃"
|
|
if "희생 플라이" in result_text:
|
|
return "희생 플라이"
|
|
if "희생플라이" in result_text:
|
|
return "희생 플라이"
|
|
if "인필드플라이" in result_text:
|
|
return "인필드플라이"
|
|
if "파울플라이" in result_text:
|
|
return "파울플라이-아웃"
|
|
return "아웃"
|
|
|
|
if result_type == "bunt_hit":
|
|
return "번트안타"
|
|
if result_type == "single":
|
|
if "번트안타" in result_text:
|
|
return "번트안타"
|
|
if "내야안타" in result_text:
|
|
return "내야안타"
|
|
if result_type == "hit_by_pitch" or "헤드샷" in result_text:
|
|
return "몸에 맞는 볼"
|
|
return BATTER_RESULT_LABEL_MAP.get(result_type)
|
|
|
|
|
|
def is_ball_in_play_event(event: dict[str, Any]) -> bool:
|
|
pitches = event.get("pitches") or []
|
|
result = event.get("result") or {}
|
|
if not pitches or not result:
|
|
return False
|
|
return pitches[-1].get("pitchResult") == "H"
|
|
|
|
|
|
def infer_hit_ball_type(result_text: str) -> str:
|
|
if "번트" in result_text:
|
|
return "번트타구"
|
|
if "몸에 맞는 타구" in result_text:
|
|
return "땅볼"
|
|
if "파울희생플라이" in result_text or "파울 희생플라이" in result_text:
|
|
return "플라이"
|
|
if "파울플라이" in result_text:
|
|
return "플라이"
|
|
if "라인드라이브" in result_text or "직선타" in result_text:
|
|
return "라인드라이브"
|
|
if "플라이" in result_text:
|
|
return "플라이"
|
|
if "땅볼" in result_text:
|
|
return "땅볼"
|
|
if "홈런" in result_text:
|
|
return "홈런성타구"
|
|
return "일반바운드"
|
|
|
|
|
|
def infer_field_zone(result_text: str) -> str:
|
|
if "몸에 맞는 타구" in result_text:
|
|
return "1루수"
|
|
ordered_zones = (
|
|
"좌중간",
|
|
"우중간",
|
|
"좌전",
|
|
"중전",
|
|
"우전",
|
|
"좌월",
|
|
"중월",
|
|
"우월",
|
|
"좌익수",
|
|
"중견수",
|
|
"우익수",
|
|
"유격수",
|
|
"3루수",
|
|
"2루수",
|
|
"1루수",
|
|
"투수",
|
|
"포수",
|
|
)
|
|
for zone in ordered_zones:
|
|
if zone in result_text:
|
|
return zone
|
|
return "중견수"
|
|
|
|
|
|
def extract_direction_offsets(result_text: str) -> tuple[int, int]:
|
|
x_delta = 0
|
|
y_delta = 0
|
|
|
|
if "왼쪽" in result_text:
|
|
x_delta -= 1
|
|
if "오른쪽" in result_text:
|
|
x_delta += 1
|
|
if "앞" in result_text:
|
|
y_delta += 1
|
|
if "뒤" in result_text:
|
|
y_delta -= 1
|
|
|
|
return x_delta, y_delta
|
|
|
|
|
|
def is_infield_zone(zone: str) -> bool:
|
|
return zone in {"투수", "포수", "1루수", "2루수", "3루수", "유격수"}
|
|
|
|
|
|
def extract_defense_sequence(result_text: str) -> list[str]:
|
|
# 1) '2-6', '2-5-3' 같은 숫자 패턴 처리
|
|
num_seq_match = re.search(r"(\d+(?:-\d+)+)", result_text)
|
|
if num_seq_match:
|
|
nums = num_seq_match.group(1).split("-")
|
|
pos_names = []
|
|
for n in nums:
|
|
# 1:투수, 2:포수, ..., 9:우익수 매핑
|
|
name = POSITION_LABEL_MAP.get(n)
|
|
if name: pos_names.append(name)
|
|
if pos_names: return pos_names
|
|
|
|
parenthetical_match = re.search(r"\(([^)]*)\)", result_text)
|
|
if parenthetical_match:
|
|
sequence = re.findall(r"(투수|포수|1루수|2루수|3루수|유격수|좌익수|중견수|우익수)", parenthetical_match.group(1))
|
|
if sequence:
|
|
return sequence
|
|
leading_text = result_text.split("(", 1)[0]
|
|
sequence = re.findall(r"(투수|포수|1루수|2루수|3루수|유격수|좌익수|중견수|우익수)", leading_text)
|
|
if sequence:
|
|
return sequence
|
|
zone = infer_field_zone(result_text)
|
|
if zone in POSITION_NUMBER_MAP:
|
|
return [zone]
|
|
return []
|
|
|
|
|
|
def is_double_play_result(result_text: str) -> bool:
|
|
return "병살" in result_text
|
|
|
|
|
|
def build_double_play_first_sequence(event: dict[str, Any]) -> list[str]:
|
|
result_text = ((event.get("result") or {}).get("text") or "").strip()
|
|
return extract_defense_sequence(result_text)
|
|
|
|
|
|
def is_error_result(result_text: str) -> bool:
|
|
return "실책" in result_text
|
|
|
|
|
|
def is_throwing_error(result_text: str) -> bool:
|
|
keywords = ("송구실책", "송구 실책", "악송구", "throwing error", "송구에러")
|
|
return any(keyword in result_text for keyword in keywords)
|
|
|
|
|
|
def extract_error_position(result_text: str) -> str | None:
|
|
parenthetical_match = re.search(r"\(([^)]*실책[^)]*)\)", result_text)
|
|
search_texts = [parenthetical_match.group(1)] if parenthetical_match else []
|
|
search_texts.append(result_text)
|
|
for text in search_texts:
|
|
positions = re.findall(r"(투수|포수|1루수|2루수|3루수|유격수|좌익수|중견수|우익수)", text)
|
|
if positions:
|
|
return positions[0]
|
|
return None
|
|
|
|
|
|
def infer_error_position_fallback(text: str) -> str:
|
|
if "야수선택" in text:
|
|
return "야수선택"
|
|
if "도루" in text:
|
|
return "포수"
|
|
if "포구" in text:
|
|
return "포수"
|
|
if "송구" in text:
|
|
return "투수"
|
|
return "포수"
|
|
|
|
|
|
def fill_error_defense_popup(page: Page, text: str) -> None:
|
|
defense_sequence = extract_defense_sequence(text)
|
|
if len(defense_sequence) >= 2:
|
|
click_defense_sequence_in_popup(page, defense_sequence)
|
|
else:
|
|
error_position = extract_error_position(text)
|
|
if not error_position:
|
|
error_position = infer_error_position_fallback(text)
|
|
click_count = 2 if is_throwing_error(text) else 1
|
|
click_defense_button_robustly(page, error_position, click_count)
|
|
|
|
complete_button = get_last_visible_locator(page, "#btnNext")
|
|
if complete_button is None:
|
|
complete_button = get_last_visible_locator(page, "#btnAdd")
|
|
if complete_button:
|
|
complete_button.click()
|
|
page.wait_for_timeout(120)
|
|
|
|
|
|
def build_hit_ball_payload(page: Page, result_text: str) -> dict[str, str]:
|
|
zone = infer_field_zone(result_text)
|
|
x, y = FIELD_COORDINATES.get(zone, FIELD_COORDINATES["중견수"])
|
|
meter_per_px_text = page.locator("#dat_meterPerPx").input_value() or "0"
|
|
try:
|
|
meter_per_px = float(meter_per_px_text)
|
|
except ValueError:
|
|
meter_per_px = 0.0
|
|
|
|
result_type = classify_result_text_type(result_text)
|
|
hit_ball_type_label = infer_hit_ball_type(result_text)
|
|
x, y = apply_hit_ball_variation(result_text, result_type, zone, x, y)
|
|
px_x = math.floor(650 * x / 100)
|
|
px_y = math.floor(621 * y / 100)
|
|
distance = math.floor(math.sqrt((50 - x) ** 2 + (95 - y) ** 2) * math.hypot(6.5, 6.21) * meter_per_px / 100) if meter_per_px else 0
|
|
return {
|
|
"type": HIT_BALL_TYPE_MAP.get(hit_ball_type_label, "1"),
|
|
"label": hit_ball_type_label,
|
|
"x": str(px_x),
|
|
"y": str(px_y),
|
|
"xy": f"{x},{y}",
|
|
"distance": str(distance),
|
|
}
|
|
|
|
|
|
def classify_result_text_type(result_text: str) -> str:
|
|
if "홈런" in result_text:
|
|
return "home_run"
|
|
if "아웃" in result_text or "희생" in result_text:
|
|
return "out"
|
|
return "safe"
|
|
|
|
|
|
def deterministic_offset(seed_text: str, radius: int) -> tuple[int, int]:
|
|
digest = hashlib.md5(seed_text.encode("utf-8")).digest()
|
|
x_offset = (digest[0] % (radius * 2 + 1)) - radius
|
|
y_offset = (digest[1] % (radius * 2 + 1)) - radius
|
|
return x_offset, y_offset
|
|
|
|
|
|
def apply_hit_ball_variation(result_text: str, result_type: str, zone: str, x: int, y: int) -> tuple[int, int]:
|
|
dir_x, dir_y = extract_direction_offsets(result_text)
|
|
|
|
if "파울플라이" in result_text or "파울희생플라이" in result_text or "파울 희생플라이" in result_text:
|
|
is_left = any(token in result_text for token in ("좌", "3루", "유격"))
|
|
foul_x, foul_y = FOUL_FLY_LEFT if is_left else FOUL_FLY_RIGHT
|
|
x_offset, y_offset = deterministic_offset(result_text, 2)
|
|
return (
|
|
max(0, min(100, foul_x + x_offset)),
|
|
max(50, min(100, foul_y + y_offset)),
|
|
)
|
|
|
|
if result_type == "home_run":
|
|
x_offset, y_offset = deterministic_offset(result_text, 2)
|
|
return (
|
|
max(15, min(85, x + x_offset)),
|
|
max(12, min(22, y + y_offset)),
|
|
)
|
|
|
|
if is_infield_zone(zone):
|
|
base_shift = 3
|
|
random_radius = 2 if result_type == "out" else 3
|
|
else:
|
|
base_shift = 12
|
|
random_radius = 5 if result_type == "out" else 7
|
|
|
|
x_offset, y_offset = deterministic_offset(result_text, random_radius)
|
|
return (
|
|
max(10, min(90, x + dir_x * base_shift + x_offset)),
|
|
max(18, min(96, y + dir_y * base_shift + y_offset)),
|
|
)
|
|
|
|
|
|
def wait_for_user_confirmation(event: dict[str, Any], stage: str = "확인") -> None:
|
|
result_text = ((event.get("result") or {}).get("text") or "").strip()
|
|
prompt = f"[{stage} 대기] {result_text or '타구 결과'} 확인 후 Enter를 누르세요: "
|
|
try:
|
|
input(prompt)
|
|
except EOFError:
|
|
pass
|
|
|
|
|
|
def set_hit_ball_and_defense(page: Page, event: dict[str, Any]) -> bool:
|
|
result = event.get("result") or {}
|
|
result_text = (result.get("text") or "").strip()
|
|
if not result_text:
|
|
return False
|
|
|
|
if is_double_play_result(result_text):
|
|
first_popup_sequence = build_double_play_first_sequence(event)
|
|
if first_popup_sequence:
|
|
click_defense_sequence_in_popup(page, first_popup_sequence, "#btnNext")
|
|
page.wait_for_timeout(250)
|
|
|
|
hit_ball = build_hit_ball_payload(page, result_text)
|
|
popup_field = None
|
|
for _ in range(10):
|
|
popup_field = get_last_visible_locator(page, "#defenseDiv")
|
|
if popup_field is not None and popup_field.bounding_box() is not None:
|
|
break
|
|
page.wait_for_timeout(100)
|
|
|
|
if popup_field is None:
|
|
return False
|
|
|
|
defense_box = popup_field.bounding_box()
|
|
if not defense_box:
|
|
return False
|
|
|
|
page.mouse.click(defense_box["x"] + int(hit_ball["x"]), defense_box["y"] + int(hit_ball["y"]))
|
|
page.wait_for_timeout(100)
|
|
|
|
# hitBallType 라디오 버튼 선택: 값(value)으로 강제 체크
|
|
# value가 없으면 라벨 텍스트로 찾아서 click
|
|
hit_type_val = hit_ball["type"]
|
|
hit_type_label = hit_ball.get("label", "")
|
|
selected = page.evaluate(
|
|
"""([val, label]) => {
|
|
const nodes = [...document.querySelectorAll("input[name='hitBallType']")];
|
|
for (const node of nodes) {
|
|
if (node.value === val) {
|
|
node.checked = true;
|
|
node.dispatchEvent(new Event('change', {bubbles: true}));
|
|
return true;
|
|
}
|
|
}
|
|
// 라벨 텍스트로 재시도
|
|
for (const node of nodes) {
|
|
let text = '';
|
|
let p = node.parentElement;
|
|
if (p) text = p.textContent.trim();
|
|
if (!text && node.nextSibling) text = node.nextSibling.textContent || '';
|
|
if (label && text.includes(label)) {
|
|
node.checked = true;
|
|
node.dispatchEvent(new Event('change', {bubbles: true}));
|
|
return true;
|
|
}
|
|
}
|
|
return false;
|
|
}""",
|
|
[hit_type_val, hit_type_label],
|
|
)
|
|
page.wait_for_timeout(150)
|
|
|
|
if is_error_result(result_text):
|
|
error_position = extract_error_position(result_text)
|
|
if error_position:
|
|
click_count = 2 if is_throwing_error(result_text) else 1
|
|
click_defense_button_robustly(page, error_position, click_count)
|
|
|
|
next_button = get_last_visible_locator(page, "#btnNext")
|
|
if next_button is not None:
|
|
next_button.click()
|
|
page.wait_for_timeout(400)
|
|
|
|
complete_button = get_last_visible_locator(page, "#btnAdd")
|
|
# 세이프 결과(안타 등) 혹은 실책 처리가 #btnNext 로 종료될 수 있으므로 체크
|
|
if complete_button is None:
|
|
return True
|
|
|
|
if result.get("type") in {"out", "sacrifice_fly", "sacrifice_bunt", "reach_on_fielder_choice", "reach_on_grounder", "double_play", "reach_on_error"}:
|
|
positions = extract_defense_sequence(result_text)
|
|
if "직선타" in result_text or "라인드라이브" in result_text or "플라이" in result_text:
|
|
if positions:
|
|
positions = [positions[0]]
|
|
elif result.get("type") == "double_play":
|
|
# 병살타의 경우: 두 번째 상세 팝업에서는 첫 번째 수비수를 제외한 나머지 구간(예: 6-3)만 입력
|
|
if len(positions) > 1:
|
|
positions = positions[1:]
|
|
elif result.get("type") == "reach_on_error":
|
|
# 실책의 경우: 이미 2285번 라인에서 처리했을 수 있으나, 여기서도 positions가 있다면 보장
|
|
if not positions:
|
|
err_pos = extract_error_position(result_text)
|
|
if err_pos:
|
|
positions = [err_pos]
|
|
|
|
for position in positions:
|
|
click_defense_button_robustly(page, position)
|
|
|
|
if complete_button:
|
|
complete_button.click(force=True)
|
|
page.wait_for_timeout(400) # 팝업이 완전히 닫힐 때까지 대기
|
|
return True
|
|
|
|
|
|
def advance_count(balls: int, strikes: int, pitch_result: str) -> tuple[int, int]:
|
|
if pitch_result == "B":
|
|
return min(balls + 1, 3), strikes
|
|
if pitch_result in {"T", "S"}:
|
|
return balls, min(strikes + 1, 2)
|
|
if pitch_result == "F":
|
|
return balls, strikes if strikes >= 2 else strikes + 1
|
|
return balls, strikes
|
|
|
|
|
|
def process_only_reviews(page: Page, report: dict[str, Any], write_events: bool, job_id: str = None) -> None:
|
|
"""리포트 전체에서 합의판정 전용 데이터만 추출하여 일괄 등록"""
|
|
all_reviews = []
|
|
for half_inning in report["game_contents"]:
|
|
for event in half_inning.get("events", []):
|
|
if event.get("event_type") == "at_bat":
|
|
reviews = event.get("reviewEvents") or []
|
|
for r in reviews:
|
|
all_reviews.append(normalize_review_event(r))
|
|
|
|
if not all_reviews:
|
|
show_debug_overlay(page, ["입력할 합의판정 기록이 없습니다."])
|
|
page.wait_for_timeout(2000)
|
|
return
|
|
|
|
show_debug_overlay(page, [f"합의판정 {len(all_reviews)}건 일괄 등록 시작"])
|
|
if write_events:
|
|
record_review_events(page, all_reviews)
|
|
show_debug_overlay(page, ["합의판정 일괄 등록 완료"])
|
|
page.wait_for_timeout(1000)
|
|
|
|
|
|
def process_report(page: Page, report: dict[str, Any], write_events: bool, job_id: int = 0) -> None:
|
|
outs = 0
|
|
change_cache: dict[str, tuple[str, int]] = {}
|
|
applied_change_texts: set[str] = set()
|
|
for half_inning in report["game_contents"]:
|
|
inning = half_inning.get("inning", "")
|
|
outs = 0
|
|
for event in half_inning["events"]:
|
|
if event.get("event_type") == "change":
|
|
change_text = (event.get("text") or "").strip()
|
|
show_debug_overlay(
|
|
page,
|
|
[
|
|
f"교체 입력: {change_text or '-'}",
|
|
],
|
|
)
|
|
wait_for_operator_control(page)
|
|
if write_events:
|
|
if change_text and change_text in applied_change_texts:
|
|
show_debug_overlay(
|
|
page,
|
|
[
|
|
"교체 중복 건너뜀",
|
|
change_text,
|
|
],
|
|
)
|
|
page.wait_for_timeout(250)
|
|
continue
|
|
try:
|
|
# job_id는 이미 함수의 인자로 전달받으므로 별도로 다시 구할 필요 없음
|
|
log_id = job_id or os.environ.get("JOB_ID")
|
|
log_info_event = {
|
|
"job_id": log_id,
|
|
"inning": inning,
|
|
"event_type": event.get("change_type", "change"),
|
|
"target_player": event.get("in_player") or event.get("to_position", ""),
|
|
"actual_player": event.get("actor_name") or event.get("player_name", "")
|
|
} if job_id else None
|
|
|
|
apply_change_event(page, half_inning, event, change_cache)
|
|
_try_log_event(log_info_event, True)
|
|
except Exception as e:
|
|
_try_log_event(log_info_event, False, str(e))
|
|
raise e
|
|
if change_text:
|
|
applied_change_texts.add(change_text)
|
|
show_debug_overlay(
|
|
page,
|
|
[
|
|
"교체 완료",
|
|
f"{change_text or '-'}",
|
|
],
|
|
)
|
|
page.wait_for_timeout(120)
|
|
continue
|
|
if event.get("event_type") != "at_bat":
|
|
continue
|
|
|
|
clear_defense_selections(page)
|
|
balls = 0
|
|
strikes = 0
|
|
pitches = event.get("pitches") or []
|
|
result = event.get("result") or {}
|
|
for pitch_index, pitch in enumerate(pitches):
|
|
pitch_result_text = (pitch.get("pitchResultText") or "").strip()
|
|
normalized_pitch_result = normalize_pitch_result_code(pitch, event)
|
|
is_balk_strike = "보크" in pitch_result_text and ("스트라이크" in pitch_result_text or "헛스윙" in pitch_result_text)
|
|
if "피치클락" in pitch_result_text and "투수위반" in pitch_result_text:
|
|
pitch_result_text = "피치클락 투수위반 볼"
|
|
show_debug_overlay(
|
|
page,
|
|
[
|
|
f"다음 카운트: {balls}볼 {strikes}스트 {outs}아웃",
|
|
f"다음 공: {pitch.get('pitchNo')}구 {pitch_result_text}",
|
|
f"구종/구속: {(pitch.get('pitchType') or '-')} / {(pitch.get('speedKmh') or '-')}",
|
|
f"타석: {event.get('batter') or '-'}",
|
|
],
|
|
)
|
|
wait_for_operator_control(page)
|
|
# set_ball_count(page, balls, strikes, outs) # 시스템 자동 관리로 제거
|
|
is_last_pitch = pitch_index == len(pitches) - 1
|
|
# 인플레이 타격(H), 몸에 맞는 볼, 삼진, 볼넷 등 타석을 종료시키는 모든 결과는 '인플레이/액션'으로 간주하여
|
|
# 루프 끝난 뒤 타석 결과 블록에서 일괄 처리합니다. (팝업 및 타석 결과 라디오 선택을 위함)
|
|
is_action_result = is_last_pitch and result.get("type") in {
|
|
"hit_by_pitch", "strikeout", "strikeout_not_out", "walk", "intentional_walk", "out", "single", "double", "triple",
|
|
"home_run", "sacrifice_fly", "sacrifice_bunt", "double_play", "reach_on_error", "reach_on_fielder_choice", "bunt_hit",
|
|
"single_runner_out", "double_runner_out", "triple_runner_out"
|
|
}
|
|
is_in_play = (pitch.get("pitchResult") == "H") or is_action_result
|
|
# 인플레이(H)이거나 볼넷/삼진 등 타석이 종료되는 마지막 구인 경우:
|
|
# 개별 투구 조작(볼/스트라이크 버튼 클릭)을 생략하고
|
|
# 루프 밖의 타석 결과 처리 블록에서 한 번에 처리합니다.
|
|
if is_last_pitch and is_in_play:
|
|
continue
|
|
|
|
if is_balk_strike:
|
|
if write_events:
|
|
current_late = []
|
|
# 1) 투구별 주루 이벤트 (도루 등)
|
|
p_runner_events = pitch.get("runnerEvents")
|
|
if p_runner_events:
|
|
current_late.extend(set_runner_events(page, event, p_runner_events))
|
|
|
|
# 2) 타석 종료 주루 이벤트 (보크로 인한 진루 등)
|
|
if is_last_pitch and event.get("runnerEvents"):
|
|
current_late.extend(set_runner_events(page, event))
|
|
|
|
submit_input_complete(
|
|
page,
|
|
f"{event.get('batter') or '-'} / {pitch.get('pitchNo')}구 보크",
|
|
clear_defense=True,
|
|
log_info={
|
|
"job_id": job_id, "inning": inning, "batter": event.get("batter", ""),
|
|
"pitch_no": pitch.get("pitchNo", 0), "target_value": "보크",
|
|
"selected_value": "보크"
|
|
} if job_id else None
|
|
)
|
|
# 지연된 이벤트가 있다면 사후 처리
|
|
if current_late:
|
|
handle_late_runner_events(page, event, current_late, True, job_id)
|
|
|
|
page.wait_for_timeout(80)
|
|
set_pitch_meta_only(page, pitch)
|
|
if "헛스윙" in pitch_result_text:
|
|
set_radio_by_label(page, "evt_batter", "헛스윙(스트라이크)")
|
|
else:
|
|
set_radio_by_label(page, "evt_batter", "스트라이크(루킹)")
|
|
if write_events:
|
|
submit_input_complete(
|
|
page,
|
|
f"{event.get('batter') or '-'} / {pitch.get('pitchNo')}구 보크 후 {'헛스윙' if '헛스윙' in pitch_result_text else '스트라이크'}",
|
|
clear_defense=True,
|
|
log_info={
|
|
"job_id": job_id, "inning": inning, "batter": event.get("batter", ""),
|
|
"pitch_no": pitch.get("pitchNo", 0), "target_value": pitch_result_text,
|
|
"selected_value": pitch_result_text
|
|
} if job_id else None
|
|
)
|
|
else:
|
|
set_pitch(page, pitch, event)
|
|
if write_events:
|
|
current_late = []
|
|
# 폭투/포일 로그 보강
|
|
p_runner_events = get_pitch_runner_events(pitch, event)
|
|
is_wild_pitch = any(
|
|
re.get("type") == "wild_pitch_advance" or "폭투" in (re.get("text") or "")
|
|
for re in p_runner_events
|
|
)
|
|
is_passed_ball = any(
|
|
re.get("type") == "passed_ball_advance" or "포일" in (re.get("text") or "")
|
|
for re in p_runner_events
|
|
)
|
|
|
|
extra_log = ""
|
|
if is_wild_pitch:
|
|
extra_log = " (폭투)"
|
|
elif is_passed_ball:
|
|
extra_log = " (포일)"
|
|
|
|
if p_runner_events:
|
|
current_late.extend(set_runner_events(page, event, p_runner_events))
|
|
|
|
# 2) 타석 종료 주루 이벤트 (폭투 낫아웃 등)
|
|
if is_last_pitch and event.get("runnerEvents"):
|
|
current_late.extend(set_runner_events(page, event))
|
|
|
|
if "파울플라이" in pitch_result_text and "실책" in pitch_result_text:
|
|
fill_error_defense_popup(page, pitch_result_text)
|
|
|
|
submit_input_complete(
|
|
page,
|
|
f"{event.get('batter') or '-'} / {pitch.get('pitchNo')}구 {pitch_result_text or '-'}{extra_log}",
|
|
clear_defense=True,
|
|
log_info={
|
|
"job_id": job_id, "inning": inning, "batter": event.get("batter", ""),
|
|
"pitch_no": pitch.get("pitchNo", 0), "target_value": f"{pitch_result_text}{extra_log}",
|
|
"selected_value": "폭투-볼" if is_wild_pitch else "포일-볼" if is_passed_ball else pitch_result_text
|
|
} if job_id else None
|
|
)
|
|
# 지연된 이벤트 사후 처리
|
|
if current_late:
|
|
handle_late_runner_events(page, event, current_late, True, job_id)
|
|
balls, strikes = advance_count(balls, strikes, normalized_pitch_result)
|
|
|
|
if result:
|
|
last_pitch = pitches[-1] if pitches else {}
|
|
# 타격(H)이거나 타석 종료 결과(삼진, 볼넷 등)가 있는 마지막 구 처리
|
|
action_result_types = {
|
|
"hit_by_pitch", "strikeout", "strikeout_not_out", "walk", "intentional_walk", "out", "single", "double", "triple",
|
|
"home_run", "sacrifice_fly", "sacrifice_bunt", "double_play", "reach_on_error", "reach_on_fielder_choice", "reach_on_grounder", "bunt_hit",
|
|
"single_runner_out", "double_runner_out", "triple_runner_out", "play"
|
|
}
|
|
if last_pitch.get("pitchResult") == "H" or result.get("type") in action_result_types:
|
|
runner_lines = []
|
|
for runner_event in (event.get("runnerEvents") or []):
|
|
r_text = runner_event.get("text", "")
|
|
from_b = runner_event.get("fromBase", "?")
|
|
to_b = runner_event.get("toBase", "?")
|
|
line = f"🏃 {from_b}루주자 -> {to_b}루 : {r_text}"
|
|
label = infer_runner_action_label(event, runner_event)
|
|
if label in {"태그아웃", "도루시도 아웃", "포스아웃"}:
|
|
r_seq = extract_defense_sequence(r_text)
|
|
if r_seq:
|
|
line += f" | ⚾ 수비 클릭됨: {' -> '.join(r_seq)}"
|
|
runner_lines.append(line)
|
|
|
|
result_text = result.get('text') or ''
|
|
def_seq = []
|
|
if is_error_result(result_text):
|
|
err_pos = extract_error_position(result_text)
|
|
if err_pos:
|
|
click_count = 2 if is_throwing_error(result_text) else 1
|
|
def_seq = [err_pos] * click_count
|
|
elif result.get("type") in {"out", "double_play", "sacrifice_fly", "sacrifice_bunt", "strikeout", "reach_on_fielder_choice", "reach_on_grounder",
|
|
"single_runner_out", "double_runner_out", "triple_runner_out", "play"}:
|
|
def_seq = extract_defense_sequence(result_text)
|
|
if "직선타" in result_text or "라인드라이브" in result_text or "플라이" in result_text:
|
|
if def_seq:
|
|
def_seq = [def_seq[0]]
|
|
|
|
defense_lines = []
|
|
if def_seq:
|
|
seq_str = ", ".join(def_seq)
|
|
defense_lines = [f"⚾ 누를 수비수: {seq_str}"]
|
|
|
|
show_debug_overlay(
|
|
page,
|
|
[
|
|
f"📌 타격 결과: {result_text or '-'}",
|
|
f"🎯 현재 카운트: {balls}B {strikes}S {outs}O",
|
|
*defense_lines,
|
|
*runner_lines,
|
|
],
|
|
)
|
|
wait_for_operator_control(page)
|
|
# set_ball_count(page, balls, strikes, outs) # 시스템 자동 관리로 제거
|
|
# 구종/구속만 세팅 (evt_batter는 여기서 건드리지 않음)
|
|
# → set_batter_result가 실제 결과(1루타 등)를 선택할 때 팝업이 올바르게 열림
|
|
set_pitch_meta_only(page, last_pitch)
|
|
if write_events:
|
|
page.wait_for_timeout(120)
|
|
simple_terminal_result = is_simple_terminal_result_type(result.get("type") or "")
|
|
expected_batter_event = infer_batter_result_label(result, event) or ""
|
|
|
|
simple_terminal_result = is_simple_terminal_result_type(result.get("type") or "")
|
|
expected_batter_event = infer_batter_result_label(result, event) or ""
|
|
|
|
# [필독] 모든 결과에 대해 라디오 버튼(포볼, 삼진 등)은 반드시 선택해야 함
|
|
if expected_batter_event:
|
|
for _ in range(5):
|
|
set_batter_result_type(page, result, event)
|
|
page.wait_for_timeout(50)
|
|
if get_checked_event_name(page, "evt_batter") == expected_batter_event:
|
|
break
|
|
else:
|
|
set_batter_result_type(page, result, event)
|
|
page.wait_for_timeout(50)
|
|
|
|
popup_defense_used = False
|
|
if not simple_terminal_result:
|
|
# ① 팝업 처리 (타격 시에만 팝업이 열림)
|
|
# 실책(reach_on_error)이나 병살(double_play)은 defenseType 속성이 없어도 팝업이 뜨는 경우가 있으므로 조건에 추가
|
|
if result.get("type") in {"reach_on_error", "double_play"} or get_checked_batter_defense_type(page):
|
|
set_hit_ball_and_defense(page, event)
|
|
popup_defense_used = True
|
|
|
|
# ② 타자 진루 및 주루가산 (팝업 닫힌 후 메인폼에서 처리)
|
|
set_batter_advancement(page, result)
|
|
|
|
# ③ 주루 이벤트 (타자 외 주자들 - 공통)
|
|
current_late = []
|
|
all_runner_events = (event.get("runnerEvents") or []).copy()
|
|
if last_pitch.get("runnerEvents"):
|
|
all_runner_events.extend(last_pitch["runnerEvents"])
|
|
|
|
if all_runner_events:
|
|
current_late.extend(set_runner_events(page, event, all_runner_events))
|
|
page.wait_for_timeout(30)
|
|
|
|
# ④ 최종 전송 (모든 결과 공통)
|
|
submit_input_complete(
|
|
page,
|
|
f"{event.get('batter') or '-'} / {result_text or '-'}",
|
|
clear_defense=not popup_defense_used,
|
|
log_info={
|
|
"job_id": job_id, "inning": inning, "batter": event.get("batter", ""),
|
|
"pitch_no": (last_pitch.get("pitchNo", 0) if isinstance(last_pitch, dict) else 0),
|
|
"target_value": result_text,
|
|
"selected_value": expected_batter_event
|
|
} if job_id else None
|
|
)
|
|
# 타석 종료 후 지연된 주루 아웃이 있다면 별도로 처리 (실책 진루 후 아웃 등)
|
|
if current_late:
|
|
handle_late_runner_events(page, event, current_late, True, job_id)
|
|
|
|
|
|
result_type = result.get("type")
|
|
result_text = (result.get("text") or "").strip()
|
|
if result_type == "double_play":
|
|
outs += 2
|
|
if outs >= 3:
|
|
break
|
|
elif "낫아웃" in result_text and not any(token in result_text for token in ("폭투", "포일", "진루", "출루", "세이프")):
|
|
outs += 1
|
|
if outs >= 3:
|
|
break
|
|
elif result_type in {"out", "strikeout", "sacrifice_fly", "sacrifice_bunt", "single_runner_out", "double_runner_out", "triple_runner_out"}:
|
|
outs += 1
|
|
if outs >= 3:
|
|
break
|
|
|
|
|
|
def run(playwright: Playwright, args: argparse.Namespace, report: dict[str, Any]) -> None:
|
|
browser = launch_browser_context(
|
|
playwright=playwright,
|
|
user_data_dir=args.user_data_dir,
|
|
channel=args.channel,
|
|
headless=args.headless,
|
|
)
|
|
page = browser.pages[0] if browser.pages else browser.new_page()
|
|
|
|
import os
|
|
job_id = getattr(args, "job_id", None) or os.environ.get("JOB_ID")
|
|
|
|
if args.write_events and not job_id:
|
|
try:
|
|
import db_logging
|
|
import uuid
|
|
# 독립 실행 시 새 ID 생성 및 등록
|
|
job_id = f"standalone-{uuid.uuid4().hex[:8]}"
|
|
db_logging.start_job(job_id, args.game_id, getattr(args, "start_inning", ""), getattr(args, "end_inning", ""))
|
|
except Exception:
|
|
job_id = None
|
|
|
|
try:
|
|
open_game_status_page(page, args.base_url, report, args.manager_game_no)
|
|
if getattr(args, "review_only", False):
|
|
process_only_reviews(page, report, args.write_events, job_id=job_id)
|
|
else:
|
|
process_report(page, report, args.write_events, job_id=job_id)
|
|
if not args.close:
|
|
page.wait_for_timeout(3600 * 1000)
|
|
finally:
|
|
if args.write_events and job_id:
|
|
try:
|
|
import db_logging
|
|
db_logging.finish_job(job_id)
|
|
except Exception:
|
|
pass
|
|
if args.close:
|
|
try:
|
|
browser.close()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
if args.game_id:
|
|
args.game_id = "".join(args.game_id.split())
|
|
report = load_report(get_report_path(args))
|
|
with sync_playwright() as playwright:
|
|
run(playwright, args, report)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|