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/_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()