feat(mlb): 문자중계 추가 — feed/live allPlays 규칙 기반 한글 변환
- 투구 호칭·eventType 한글 매핑, 타구 방향은 desc 의 야수 표현에서 추출
- 주자 이동(홈인·진루·주루사)은 정규식 파싱 — 선수명은 이니셜("J.P.")만
마침표 허용해 문장 경계를 넘지 않게 함
- 교체는 "A replaces B" 구문 분해 (투수 교체·대타·대주자·수비 교체)
- 득점 강조는 KBO 와 동일하게 플레이 종료 점수 합 변화로 판정 + 홈런 결과 라인
- 미지원 문장·이벤트는 영문 원문 폴백. 노이즈(타자 타임·수비 위치이동 등) 제외
- KBO relay 와 동일 스키마라 프론트(LiveField RelayFeed) 변경 없음
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
parent
13a4997cc2
commit
00494e9e93
@ -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}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user