From 00494e9e93793cfe38654ecd2c56ba0c7f3797f1 Mon Sep 17 00:00:00 2001 From: jwkim Date: Fri, 14 Aug 2026 14:38:48 +0900 Subject: [PATCH] =?UTF-8?q?feat(mlb):=20=EB=AC=B8=EC=9E=90=EC=A4=91?= =?UTF-8?q?=EA=B3=84=20=EC=B6=94=EA=B0=80=20=E2=80=94=20feed/live=20allPla?= =?UTF-8?q?ys=20=EA=B7=9C=EC=B9=99=20=EA=B8=B0=EB=B0=98=20=ED=95=9C?= =?UTF-8?q?=EA=B8=80=20=EB=B3=80=ED=99=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 투구 호칭·eventType 한글 매핑, 타구 방향은 desc 의 야수 표현에서 추출 - 주자 이동(홈인·진루·주루사)은 정규식 파싱 — 선수명은 이니셜("J.P.")만 마침표 허용해 문장 경계를 넘지 않게 함 - 교체는 "A replaces B" 구문 분해 (투수 교체·대타·대주자·수비 교체) - 득점 강조는 KBO 와 동일하게 플레이 종료 점수 합 변화로 판정 + 홈런 결과 라인 - 미지원 문장·이벤트는 영문 원문 폴백. 노이즈(타자 타임·수비 위치이동 등) 제외 - KBO relay 와 동일 스키마라 프론트(LiveField RelayFeed) 변경 없음 Co-Authored-By: Claude Opus 5 --- backend/app/services/baseball_details.py | 200 ++++++++++++++++++++++- 1 file changed, 199 insertions(+), 1 deletion(-) diff --git a/backend/app/services/baseball_details.py b/backend/app/services/baseball_details.py index bc96156..3e26d9a 100644 --- a/backend/app/services/baseball_details.py +++ b/backend/app/services/baseball_details.py @@ -10,6 +10,7 @@ from __future__ import annotations import asyncio import logging +import re import time from datetime import datetime, timedelta, timezone @@ -541,6 +542,195 @@ async def _kbo_relay_groups(client, m: Match, gid: str, t: dict) -> list[dict]: return [g for i in sorted(cache) for g in cache[i]] +# ── MLB 문자중계 (feed/live allPlays 규칙 기반 한글 변환) ────── +# MLB 중계문은 정형 영문 템플릿 — eventType·투구 호칭을 한글로 매핑하고 +# 타구 방향·주자 진루는 정규식으로 뽑는다. 미지원 문장은 영문 원문 폴백. + +_MLB_PITCH_KO = { + "Ball": "볼", "Ball In Dirt": "볼 (원바운드)", "Intent Ball": "고의사구 볼", + "Called Strike": "스트라이크", "Swinging Strike": "헛스윙", + "Swinging Strike (Blocked)": "헛스윙", "Foul": "파울", "Foul Tip": "파울팁", + "Foul Bunt": "번트 파울", "Missed Bunt": "번트 헛스윙", + "Hit By Pitch": "몸에 맞는 볼", "Pitchout": "피치아웃", + "Automatic Ball": "자동 볼 (피치클록)", "Automatic Strike": "자동 스트라이크 (피치클록)", + "In play, out(s)": "타격", "In play, no out": "타격", "In play, run(s)": "타격", +} +_MLB_EVENT_KO = { + "Single": "1루타", "Double": "2루타", "Triple": "3루타", "Home Run": "홈런", + "Walk": "볼넷", "Intent Walk": "고의4구", "Strikeout": "삼진 아웃", + "Strikeout Double Play": "삼진 (병살)", "Groundout": "땅볼 아웃", + "Flyout": "플라이 아웃", "Lineout": "직선타 아웃", "Pop Out": "팝플라이 아웃", + "Bunt Groundout": "번트 아웃", "Bunt Pop Out": "번트 팝플라이 아웃", + "Forceout": "포스 아웃", "Grounded Into DP": "병살타", "Double Play": "병살", + "Triple Play": "삼중살", "Sac Fly": "희생플라이", "Sac Bunt": "희생번트", + "Field Error": "실책 출루", "Fielders Choice": "야수선택", + "Fielders Choice Out": "야수선택 아웃", "Hit By Pitch": "몸에 맞는 볼", + "Catcher Interference": "포수 타격방해", "Runner Out": "주자 아웃", + "Caught Stealing 2B": "도루자 (2루)", "Caught Stealing 3B": "도루자 (3루)", + "Caught Stealing Home": "도루자 (홈)", "Pickoff 1B": "견제사 (1루)", + "Pickoff 2B": "견제사 (2루)", "Pickoff 3B": "견제사 (3루)", +} +# 타구 방향/처리 야수 — desc 에서 가장 먼저 나오는 표현 (긴 표현 우선 매칭) +_MLB_DIR_KO = [ + ("left fielder", "좌익수"), ("left field", "좌익수"), + ("center fielder", "중견수"), ("center field", "중견수"), + ("right fielder", "우익수"), ("right field", "우익수"), + ("shortstop", "유격수"), ("first baseman", "1루수"), + ("second baseman", "2루수"), ("third baseman", "3루수"), + ("first base", "1루수"), ("second base", "2루수"), ("third base", "3루수"), + ("pitcher", "투수"), ("catcher", "포수"), +] +# 대문자 시작 단어 연속 = 선수명 ("J.P. Crawford", "Wenceel Pérez"). +# 마침표는 이니셜("J.P.")에만 허용 — 일반 단어 끝 마침표를 막아 문장 경계를 넘지 않게 한다. +_MLB_NAME_W = r"(?:(?:[A-ZÀ-Ý]\.)+|[A-ZÀ-Ý][\w'\-]*)" +_MLB_NAME = rf"{_MLB_NAME_W}(?: {_MLB_NAME_W})*" +_BASE_KO = {"2nd": "2루", "3rd": "3루", "home": "홈"} + + +def _mlb_direction(desc: str) -> str: + hits = [(desc.find(en), ko) for en, ko in _MLB_DIR_KO if en in desc] + return min(hits)[1] if hits else "" + + +def _mlb_runs(desc: str) -> list[dict]: + """desc 의 주자 이동 문장 → run 이벤트 (홈인·진루·주루사).""" + out: list[dict] = [] + for nm in re.findall(rf"({_MLB_NAME}) scores\b", desc): + out.append({"kind": "run", "text": f"{nm} : 홈인"}) + for nm, base in re.findall(rf"({_MLB_NAME})(?: advances)? to (2nd|3rd)\b", desc): + out.append({"kind": "run", "text": f"{nm} : {_BASE_KO[base]}까지 진루"}) + for nm, base in re.findall(rf"({_MLB_NAME}) out at (2nd|3rd|home)\b", desc): + out.append({"kind": "run", "text": f"{nm} : {_BASE_KO[base]}에서 아웃"}) + seen: set[str] = set() # "to 3rd"+"advances to 3rd" 같은 중복 문장 제거 + return [e for e in out if not (e["text"] in seen or seen.add(e["text"]))] + + +def _tr_mlb_result(res: dict, batter: str) -> str: + event, desc = res.get("event") or "", res.get("description") or "" + ko = _MLB_EVENT_KO.get(event) + if not ko: # 미지원 이벤트 — 영문 원문 폴백 + return f"{batter} : {desc}" if desc else f"{batter} : {event}" + if event == "Strikeout": + if "called out on strikes" in desc: + ko += " (루킹)" + elif "strikes out swinging" in desc: + ko += " (헛스윙)" + direction = _mlb_direction(desc) + return f"{batter} : {direction} {ko}".replace(" ", " ").replace(" : ", " : ") + + +def _tr_mlb_action(det: dict) -> tuple[str, str] | None: + """playEvents 액션 → (kind, text). None 이면 표시하지 않음 (노이즈).""" + event, desc = det.get("event") or "", det.get("description") or "" + if event in ("Game Advisory", "Batter Timeout") or not (event or desc): + return None + if event.startswith("Stolen Base"): + base = "2루" if event.endswith("2B") else "3루" if event.endswith("3B") else "홈" + nm = re.match(rf"({_MLB_NAME}) steals", desc) + who = nm.group(1) if nm else "" + return "run", f"{who} : 도루로 {base}까지 진루".replace(" ", " ") + if event.startswith("Caught Stealing"): + return "run", _MLB_EVENT_KO.get(event, "도루자") + if event == "Wild Pitch": + return "note", "폭투" + if event == "Passed Ball": + return "note", "포일" + if event == "Mound Visit": + return "note", "마운드 방문" + if event in ("Pitching Substitution", "Offensive Substitution", + "Defensive Sub", "Defensive Substitution"): + label = { + "Pitching Substitution": "투수 교체", + "Offensive Substitution": "대타" if "hitter" in desc else "대주자", + }.get(event, "수비 교체") + # "Pitching Change: A replaces B." / "...: Pinch-hitter A replaces + # right fielder B, batting 2nd..." → "{label} — B → A" + body = desc.split(": ", 1)[-1].rstrip(".") + if " replaces " in body: + new, old = body.split(" replaces ", 1) + new = re.sub(r"^Pinch-(?:hitter|runner) ", "", new) + old = re.sub(r"^[a-z][a-z ]* ", "", old) # 포지션 수식어 제거 + old = old.split(", batting")[0].split(" because")[0] + return "sub", f"{label} — {old} → {new}" + return "sub", desc + if event == "Defensive Switch": + return None # 포지션 이동 — 중계 피드에선 노이즈 + if desc.startswith("Pickoff Attempt"): + return "note", "견제 시도" + if desc.startswith("Pitcher Step Off"): + return None + return "note", desc or event # 미지원 액션 — 영문 폴백 + + +def _transform_mlb_relay(feed: dict, m: Match) -> list[dict]: + """allPlays → 문자중계 그룹 (KBO relay 와 동일 스키마, 최신 타석순).""" + plays = ((feed.get("liveData") or {}).get("plays") or {}).get("allPlays") or [] + away_s = m.team_a_short or m.team_a_code + home_s = m.team_b_short or m.team_b_code + groups: list[dict] = [] + prev_half: tuple | None = None + prev_total = 0 + no = 0 + for p in plays: + about = p.get("about") or {} + inn = about.get("inning") + top = about.get("halfInning") == "top" + if (inn, top) != prev_half: # 이닝 구분 헤더 (KBO 와 동일한 빈 그룹) + prev_half = (inn, top) + no += 1 + groups.append({ + "inn": inn, "half": "T" if top else "B", "no": no, + "title": f"{inn}회{'초' if top else '말'} {away_s if top else home_s} 공격", + "events": [], + }) + events: list[dict] = [] + for ev in p.get("playEvents") or []: + det = ev.get("details") or {} + if ev.get("isPitch"): + call = det.get("description") or "" + n = ev.get("pitchNumber") + ko = _MLB_PITCH_KO.get(call, call) + events.append({"kind": "pitch", "text": f"{n}구 {ko}" if n else ko}) + else: + ke = _tr_mlb_action(det) + if ke: + events.append({"kind": ke[0], "text": ke[1]}) + events += _mlb_runs(det.get("description") or "") # 폭투 득점 등 + res = p.get("result") or {} + batter = ((p.get("matchup") or {}).get("batter") or {}).get("fullName") or "" + if res.get("event"): + events.append({"kind": "result", "text": _tr_mlb_result(res, batter)}) + events += _mlb_runs(res.get("description") or "") + # 득점 판정 — 플레이 종료 시점 점수 합 변화 (KBO 와 동일 원리) + if about.get("isComplete"): + total = int(res.get("awayScore") or 0) + int(res.get("homeScore") or 0) + if total > prev_total: + marked = False + for e in events: + if e["text"].endswith("홈인"): + e["score"] = True + marked = True + if not marked and events: # 홈인 문장 미파싱 (솔로 홈런 등) — 결과 라인 강조 + events[-1]["score"] = True + # 홈런은 타자 본인 득점이 홈인 문장에 없음 — 결과 라인도 강조 + if res.get("event") == "Home Run": + for e in events: + if e["kind"] == "result": + e["score"] = True + prev_total = total + if not events: + continue + no += 1 + groups.append({ + "inn": inn, "half": "T" if top else "B", "no": no, + "title": batter, "events": events, + }) + groups.reverse() + ended = ((feed.get("gameData") or {}).get("status") or {}).get( + "abstractGameState") == "Final" + return groups if ended else groups[:_RELAY_MAX_GROUPS] + + def _transform_mlb_feed(feed: dict) -> dict: ld = feed.get("liveData") or {} ls = ld.get("linescore") or {} @@ -655,7 +845,15 @@ async def fetch_live(m: Match) -> dict: if pk: r2 = await c.get(f"{settings.mlb_api_base}/v1.1/game/{pk}/feed/live") r2.raise_for_status() - payload = _transform_mlb_feed(r2.json()) + feed = r2.json() + payload = _transform_mlb_feed(feed) + try: # 문자중계는 부가 정보 — 실패해도 필드 뷰는 그대로 서빙 + relay = _transform_mlb_relay(feed, m) + if relay: + payload["relay"] = relay + payload["available"] = True # 종료 경기 다시보기 지원 + except Exception as e: # noqa: BLE001 + log.warning("mlb 문자중계 실패 %s: %s", m.match_id, e) except Exception as e: # noqa: BLE001 log.warning("live 실패 %s: %s", m.match_id, e) payload = {"available": False}