"""MLS 데이터 — ESPN 비공식 API (일정·결과·순위·프리뷰·라이브). ESPN site API 는 키 불필요·비공식. 응답 표준 레코드는 야구와 동일 형태 ({league, teamA(원정), teamB(홈), dateKst, kickoffKst, venue, cancelled, [scoreA, scoreB]})라 sync_baseball_schedule/settle 파이프라인을 그대로 탄다. - 일정: scoreboard?dates=YYYYMMDD-YYYYMMDD (dates 는 미국 동부 기준 날짜라 KST 매칭은 이벤트의 UTC 시각을 KST 로 환산해 산출) - 순위: 동부/서부 컨퍼런스 → standings:mls 캐시 (승점제) - 프리뷰: scoreboard 의 form(최근5)·records(시즌 W-D-L)·odds(머니라인) + summary 의 headToHeadGames(맞대결 최근 5) → preview:{match_id} 캐시. odds 는 AI 프롬프트 참고용으로만 쓰고 UI 에는 노출하지 않는다. - 라이브: scoreboard 이벤트의 clock·스코어·득점 이벤트(details) — 15초 TTL """ from __future__ import annotations import logging import re import time import unicodedata from datetime import datetime, timedelta, timezone from difflib import SequenceMatcher from ..config import settings from ..domain import ensure_aware, now_utc from ..models import DataCache, Match from ..teams_mls import MLS_ID_TO_CODE, MLS_TEAMS log = logging.getLogger("triplepick.mls") KST = timezone(timedelta(hours=9)) # 주의: ESPN 은 브라우저 위장 UA(풀 Chrome 문자열)를 보내면 WAF 가 403 으로 차단한다 # (TLS 핑거프린트와 UA 불일치 감지로 추정). httpx 기본 UA 로 보내야 통과 — 헤더 없이 호출. # 취소로 취급하는 ESPN 상태 (연기 포함 — 보강 일정이 새 이벤트로 재등장) _CANCEL_STATUS = {"STATUS_POSTPONED", "STATUS_CANCELED", "STATUS_ABANDONED"} def _sb_url(dates: str) -> str: return ( f"{settings.espn_api_base}/site/v2/{settings.espn_mls_path}" f"/scoreboard?dates={dates}&limit=200" ) def _summary_url(event_id: str) -> str: return ( f"{settings.espn_api_base}/site/v2/{settings.espn_mls_path}" f"/summary?event={event_id}" ) def _standings_url() -> str: return f"{settings.espn_api_base}/v2/{settings.espn_mls_path}/standings" def _parse_event(e: dict) -> dict | None: """scoreboard 이벤트 1건 → 표준 레코드 (+espnId·form·records·odds 원본).""" comp = (e.get("competitions") or [{}])[0] sides: dict[str, dict] = {} for t in comp.get("competitors") or []: sides[t.get("homeAway", "")] = t away, home = sides.get("away"), sides.get("home") if not away or not home: return None a = (away.get("team") or {}).get("abbreviation", "") b = (home.get("team") or {}).get("abbreviation", "") if a not in MLS_TEAMS or b not in MLS_TEAMS: return None # 올스타전 등 제외 gd = e.get("date") if not gd: return None kickoff = datetime.fromisoformat(gd.replace("Z", "+00:00")).astimezone(KST) stype = (e.get("status") or {}).get("type") or {} rec: dict = { "league": "mls", "teamA": a, "teamB": b, "dateKst": kickoff.strftime("%Y%m%d"), "kickoffKst": kickoff.isoformat(), "venue": ((comp.get("venue") or {}).get("fullName")) or "", "cancelled": stype.get("name") in _CANCEL_STATUS, "espnId": e.get("id"), "_away": away, "_home": home, "_comp": comp, "_status": e.get("status"), } if stype.get("state") == "post" and stype.get("completed") and not rec["cancelled"]: sa, sb = away.get("score"), home.get("score") if sa is not None and sb is not None: rec["scoreA"], rec["scoreB"] = int(sa), int(sb) return rec async def _fetch_events(client, dates: str) -> list[dict]: r = await client.get(_sb_url(dates)) r.raise_for_status() out = [] for e in r.json().get("events") or []: rec = _parse_event(e) if rec: out.append(rec) return out def _clean(rec: dict) -> dict: """sync/정산에 넘길 때 원본 참조 필드 제거.""" return {k: v for k, v in rec.items() if not k.startswith("_")} async def fetch_mls_schedule(days_back: int | None = None) -> list[dict]: """일정+결과 수집 (취소 포함). 실패 시 빈 리스트 — 기존 일정 유지. days_back 기본값은 mls_days_back. settle_baseball 이 미정산 경기 중 가장 오래된 날짜를 덮도록 더 큰 값을 넘길 수 있다(오래 방치된 미정산 경기 구제). """ import httpx today = datetime.now(KST).date() # 시작 = 백필 윈도우(주말 위주 편성 공백 방지). ESPN dates 는 미국 동부 날짜라 # KST 범위를 놓치지 않게 하루 여유를 둔다. start = today - timedelta(days=(days_back if days_back is not None else settings.mls_days_back) + 1) end = today + timedelta(days=settings.baseball_days_ahead) try: async with httpx.AsyncClient(timeout=20) as c: records = await _fetch_events( c, f"{start.strftime('%Y%m%d')}-{end.strftime('%Y%m%d')}" ) records = [_clean(r) for r in records] from .baseball_sync import assign_seq assign_seq(records) # MLS 는 더블헤더가 없어 전부 seq=1 이지만 키 일관성 유지 log.info("mls schedule: %d경기 수집", len(records)) return records except Exception as e: # noqa: BLE001 log.error("mls schedule 실패: %s — 기존 일정 유지", e) return [] async def fetch_mls_results(days_back: int | None = None) -> list[dict]: """확정 스코어만 → 정산(settle) 입력.""" return [r for r in await fetch_mls_schedule(days_back) if "scoreA" in r] # ── 순위 (동/서부 컨퍼런스, 승점제) ───────────────────────────── _CONF_KEY = {"Eastern Conference": "EAST", "Western Conference": "WEST"} async def _refresh_standings(db) -> bool: import httpx table: dict[str, dict] = {} try: async with httpx.AsyncClient(timeout=15) as c: r = await c.get(_standings_url()) r.raise_for_status() data = r.json() for conf in data.get("children") or []: key = _CONF_KEY.get(conf.get("name", ""), conf.get("abbreviation", "")) for ent in ((conf.get("standings") or {}).get("entries")) or []: code = (ent.get("team") or {}).get("abbreviation", "") if code not in MLS_TEAMS: continue stats = {s.get("name"): s for s in ent.get("stats") or []} def _v(name: str): # noqa: ANN202 s = stats.get(name) or {} return s.get("value") if s.get("value") is not None else s.get("displayValue") table[code] = { "div": key, "rank": int(_v("rank") or 0) or None, "gp": int(_v("gamesPlayed") or 0), "w": int(_v("wins") or 0), "d": int(_v("ties") or 0), "l": int(_v("losses") or 0), "pts": int(_v("points") or 0), "gf": int(_v("pointsFor") or 0), "ga": int(_v("pointsAgainst") or 0), "diff": (stats.get("pointDifferential") or {}).get("displayValue"), } except Exception as e: # noqa: BLE001 log.warning("mls standings 실패: %s", e) return False if not table: return False row = await db.get(DataCache, "standings:mls") if row: row.payload = table row.fetched_at = now_utc() else: db.add(DataCache(key="standings:mls", payload=table, fetched_at=now_utc())) return True # ── 프리뷰 (폼·시즌 성적·맞대결·머니라인) ────────────────────── def _odds_of(comp: dict) -> dict | None: """scoreboard odds → {home, draw, away, provider} (없으면 None).""" o = (comp.get("odds") or [{}])[0] or {} # odds: [null] 인 경기도 있음 ml = o.get("moneyline") or {} def _pick(side: dict | None) -> str | None: if not side: return None for k in ("current", "close", "open"): v = (side.get(k) or {}).get("odds") if v and v != "OFF": return str(v) return None home, away = _pick(ml.get("home")), _pick(ml.get("away")) draw = _pick(ml.get("draw")) or ( str((o.get("drawOdds") or {}).get("moneyLine") or "") or None ) if not (home or away or draw): return None return { "home": home, "draw": draw, "away": away, "provider": ((o.get("provider") or {}).get("displayName")) or "", } def _h2h_of(summary: dict) -> list[dict]: """summary.headToHeadGames → 최근 맞대결 [{date, home, away, scoreH, scoreA}].""" out: list[dict] = [] for grp in summary.get("headToHeadGames") or []: for ev in grp.get("events") or []: h = MLS_ID_TO_CODE.get(int(ev.get("homeTeamId") or 0)) a = MLS_ID_TO_CODE.get(int(ev.get("awayTeamId") or 0)) sh, sa = ev.get("homeTeamScore"), ev.get("awayTeamScore") if not h or not a or sh is None or sa is None: continue out.append({ "date": (ev.get("gameDate") or "")[:10], "home": h, "away": a, "scoreH": int(sh), "scoreA": int(sa), }) break # 첫 그룹(상대팀 기준)만 return out[:5] async def _refresh_previews(db, matches: list[Match]) -> int: import httpx if not matches: return 0 dates = sorted({ensure_aware(m.kickoff_at).astimezone(KST).date() for m in matches}) span = f"{(dates[0] - timedelta(days=1)).strftime('%Y%m%d')}-{dates[-1].strftime('%Y%m%d')}" n = 0 async with httpx.AsyncClient(timeout=20) as c: events = await _fetch_events(c, span) by_key = {(r["dateKst"], r["teamA"], r["teamB"]): r for r in events} for m in matches: d = ensure_aware(m.kickoff_at).astimezone(KST).strftime("%Y%m%d") rec = by_key.get((d, m.team_a_code, m.team_b_code)) if not rec: continue payload: dict = { "formA": (rec["_away"].get("form")) or None, # 최근5 "WWLDW" "formB": (rec["_home"].get("form")) or None, "recordA": next( # 시즌 전적 "8-2-4" (W-L-D) (r.get("summary") for r in rec["_away"].get("records") or [] if r.get("type") == "total"), None), "recordB": next( (r.get("summary") for r in rec["_home"].get("records") or [] if r.get("type") == "total"), None), "odds": _odds_of(rec["_comp"]), } try: # 맞대결(h2h)은 summary 1콜 — 실패해도 나머지 프리뷰는 유지 r2 = await c.get(_summary_url(rec["espnId"])) r2.raise_for_status() payload["h2h"] = _h2h_of(r2.json()) except Exception as e: # noqa: BLE001 log.warning("mls summary 실패 %s: %s", m.match_id, e) if any(v for v in payload.values()): row = await db.get(DataCache, f"preview:{m.match_id}") if row: row.payload = payload row.fetched_at = now_utc() else: db.add(DataCache( key=f"preview:{m.match_id}", payload=payload, fetched_at=now_utc(), )) n += 1 return n async def refresh_mls_details(db, matches: list[Match]) -> None: """임박(48h 내) 미종료 경기 프리뷰 + 컨퍼런스 순위 캐시 갱신.""" horizon = now_utc() + timedelta(hours=48) targets = [ m for m in matches if m.result_outcome is None and m.status != "cancelled" and ensure_aware(m.kickoff_at) <= horizon ] n = await _refresh_previews(db, targets) await _refresh_standings(db) await db.commit() log.info("mls details: 프리뷰 %d경기 캐싱", n) # ── AI 예측 프롬프트 데이터 블록 ──────────────────────────────── async def build_mls_data_block(db, match: Match) -> str | None: """자체 DB 축적 결과(폼·상대전적) + ESPN 프리뷰(폼·시즌·맞대결·배당) 결합.""" from .baseball_data import _h2h_line, _team_games, _team_lines ga = await _team_games(db, "mls", match.team_a_code, match.kickoff_at) gb = await _team_games(db, "mls", match.team_b_code, match.kickoff_at) lines: list[str] = [] prev = await db.get(DataCache, f"preview:{match.match_id}") p = prev.payload if prev else {} if p.get("formA") or p.get("recordA"): lines.append( f"[{match.team_a_short} season] record {p.get('recordA') or '?'} (W-L-D), " f"last5 {p.get('formA') or '?'}" ) if p.get("formB") or p.get("recordB"): lines.append( f"[{match.team_b_short} season] record {p.get('recordB') or '?'} (W-L-D), " f"last5 {p.get('formB') or '?'}" ) if p.get("h2h"): h2h = "; ".join( f"{g['away']} {g['scoreA']}-{g['scoreH']} {g['home']} (away-home, {g['date']})" for g in p["h2h"] ) lines.append(f"[Recent head-to-head] {h2h}") if p.get("odds"): o = p["odds"] lines.append( f"[Bookmaker moneyline ({o.get('provider')})] " f"home {o.get('home') or '?'} / draw {o.get('draw') or '?'} / " f"away {o.get('away') or '?'} (American odds; use as market signal)" ) st_row = await db.get(DataCache, "standings:mls") if st_row: for code, name in ( (match.team_a_code, match.team_a_short), (match.team_b_code, match.team_b_short), ): st = st_row.payload.get(code) if st: lines.append( f"[{name} standings] {st.get('div')} rank {st.get('rank')}, " f"{st.get('w')}W-{st.get('d')}D-{st.get('l')}L, " f"{st.get('pts')}pts, GF {st.get('gf')} GA {st.get('ga')}" ) if not ga and not gb and not lines: return None h2h_db = await _h2h_line(db, "mls", match.team_a_code, match.team_b_code, match.kickoff_at) return "\n".join([ "=== MATCH DATA (factual; weigh heavily over priors) ===", "Away " + _team_lines(match.team_a_name, ga), "Home " + _team_lines(match.team_b_name, gb), f"[Head-to-head in our data] {h2h_db}", *lines, "===", ]) # ── 문자중계 한글 번역 (규칙 기반) ───────────────────────────── # ESPN 중계 문장은 정형 템플릿이라 정규식으로 한국어 요약 변환. 매칭 실패 시 원문. # 선수명은 영문 유지, 팀 풀네임은 한글 축약으로 치환. _TEAM_EN2KO: dict[str, str] = { **{v["en"]: v["short"] for v in MLS_TEAMS.values()}, # ESPN 표기가 우리 en 값과 다른 팀 별칭 "Red Bull New York": "레드불스", "Vancouver Whitecaps": "밴쿠버", "St. Louis City SC": "세인트루이스", } _ATTEMPT_LABEL = {"missed": "슛 빗나감", "blocked": "슛 차단", "saved": "슛 (GK 선방)"} # ESPN 팀 표기(중계 원문) → 팀 코드 — 교체 이벤트 팀 매칭용 _TEAM_EN2CODE: dict[str, str] = { **{v["en"]: k for k, v in MLS_TEAMS.items()}, "Red Bull New York": "RBNY", "Vancouver Whitecaps": "VAN", "St. Louis City SC": "STL", } def _ko_teams(s: str) -> str: for en, ko in _TEAM_EN2KO.items(): s = s.replace(en, ko) return s def _tr_commentary(text: str) -> str: # noqa: PLR0911 t = text.strip() if t in ("First Half begins.", "First Half Kicks Off."): return "전반전 시작" if t.startswith("Second Half begins"): return "후반전 시작" m = re.match(r"^First Half ends, (.+)\.$", t) if m: return f"전반전 종료 — {_ko_teams(m.group(1))}" m = re.match(r"^Match ends, (.+)\.$", t) if m: return f"경기 종료 — {_ko_teams(m.group(1))}" m = re.match(r"^Second Half ends, (.+)\.$", t) if m: return f"후반전 종료 — {_ko_teams(m.group(1))}" m = re.match(r"^Goal!\s+(.+?) (\d+), (.+?) (\d+)\.\s*(.+?) \((.+?)\)(.*)$", t) if m: t1, s1, t2, s2, player, team, rest = m.groups() tag = " (PK)" if "penalty" in rest else " (헤딩)" if "header" in rest else "" return ( f"골! {_ko_teams(t1)} {s1} : {s2} {_ko_teams(t2)} — " f"{player} ({_ko_teams(team)}){tag}" ) m = re.match(r"^Own Goal by (.+?), (.+?) (\d+), (.+?) (\d+)\.", t) if m: og, t1, s1, t2, s2 = m.groups() return f"자책골 — {_ko_teams(og)} · {_ko_teams(t1)} {s1} : {s2} {_ko_teams(t2)}" m = re.match(r"^Attempt (missed|blocked|saved)\.\s*(.+?) \((.+?)\)(.*)$", t) if m: kind, player, team, rest = m.groups() head = "헤딩 " if "header" in rest else "" am = re.search(r"Assisted by ([^.]+?)(?: with| following|\.|$)", rest) assist = f" · 도움 {am.group(1).strip()}" if am else "" return f"{head}{_ATTEMPT_LABEL[kind]} — {player} ({_ko_teams(team)}){assist}" m = re.match(r"^(.+?) \((.+?)\) hits the (left post|right post|bar|crossbar)(.*)$", t) if m: return f"골대! — {m.group(1)} ({_ko_teams(m.group(2))})" m = re.match(r"^Foul by (.+?) \((.+?)\)\.$", t) if m: return f"파울 — {m.group(1)} ({_ko_teams(m.group(2))})" m = re.match(r"^(.+?) \((.+?)\) wins a free kick (.+)\.$", t) if m: return f"프리킥 획득 — {m.group(1)} ({_ko_teams(m.group(2))})" m = re.match(r"^(.+?) \((.+?)\) is shown the (yellow|red) card(.*)\.$", t) if m: card = "옐로카드" if m.group(3) == "yellow" else "레드카드" return f"{card} — {m.group(1)} ({_ko_teams(m.group(2))})" m = re.match( r"^Substitution, (.+?)\.\s*(.+?) replaces (.+?)( because of [^.]*)?\.$", t ) if m: inj = " (부상)" if m.group(4) else "" return f"교체 ({_ko_teams(m.group(1))}) — IN {m.group(2)} · OUT {m.group(3)}{inj}" m = re.match(r"^Corner,\s+(.+?)\. Conceded by (.+?)\.$", t) if m: return f"코너킥 — {_ko_teams(m.group(1))}" m = re.match(r"^Offside, (.+?)\.(.*)$", t) if m: return f"오프사이드 — {_ko_teams(m.group(1))}" m = re.match(r"^Hand ball by (.+?) \((.+?)\)\.$", t) if m: return f"핸드볼 — {m.group(1)} ({_ko_teams(m.group(2))})" m = re.match(r"^Penalty conceded by (.+?) \((.+?)\)(.*)$", t) if m: return f"페널티 유발 — {m.group(1)} ({_ko_teams(m.group(2))})" m = re.match(r"^Penalty (.+?)\. (.+?) draws a foul(.*)$", t) if m: return f"페널티 획득 — {_ko_teams(m.group(1))} ({m.group(2)})" if t.startswith("Delay in match"): return "경기 지연" if t.startswith("Delay over"): return "경기 재개" m = re.match(r"^Fourth official has announced (\d+) minutes? of added time\.$", t) if m: return f"추가시간 {m.group(1)}분" if t.startswith("Lineups are announced"): return "라인업 발표 · 선수 몸풀기" return _ko_teams(t) # 미지원 템플릿 — 팀명만 한글화한 원문 # ── 라이브 (스코어·경기시간·득점 이벤트·라인업) ──────────────── _LIVE_TTL_SEC = 15.0 _live_cache: dict[str, tuple[float, dict]] = {} def _player_out(p: dict) -> dict: ath = p.get("athlete") or {} return { "name": ath.get("displayName", ""), "pos": ((p.get("position") or {}).get("abbreviation")) or "", "jersey": p.get("jersey", ""), "in": bool(p.get("subbedIn")), "out": bool(p.get("subbedOut")), } def _lineups_of(summary: dict) -> dict | None: """summary.rosters → {away, home}: {formation, starters[], bench[]}. starters 는 formationPlace 순(1=GK). subbedIn/Out 플래그로 실시간 교체 표시. 라인업 미발표(경기 한참 전)면 rosters 가 비어 None. """ out: dict = {} for r in summary.get("rosters") or []: side = r.get("homeAway") players = r.get("roster") or [] if side not in ("home", "away") or not players: continue starters = sorted( (p for p in players if p.get("starter")), key=lambda p: int(p.get("formationPlace") or 99), ) bench = [p for p in players if not p.get("starter")] out[side] = { "formation": r.get("formation"), "starters": [_player_out(p) for p in starters], "bench": [_player_out(p) for p in bench], } return out or None def _norm_name(s: str) -> str: """악센트 제거·소문자 — 선수명 비교용.""" s = unicodedata.normalize("NFKD", s) return "".join(ch for ch in s if not unicodedata.combining(ch)).lower().strip() def _make_name_resolver(lineups: dict | None): """중계 원문의 선수명 → 로스터 표기명 스냅. ESPN 중계와 로스터가 같은 선수를 다르게 적는 경우가 흔하다 (미들네임 생략 "Sékou Bangoura"↔"Sékou Tidiany Bangoura", 철자 차이 "Akhundzada"↔"Akhundzade", 애칭 "Máximo"↔"Maxi Carrizo"). 로스터명과 못 맞추면 교체 대체·골/어시 마커가 피치에 못 붙으므로 정규화 일치 → 성(姓) 유일 일치 → 유사도(≥0.75) 순으로 맞춘다. """ roster: list[str] = [] for side in (lineups or {}).values(): for p in (side.get("starters") or []) + (side.get("bench") or []): if p.get("name"): roster.append(p["name"]) by_norm = {_norm_name(n): n for n in roster} def resolve(name: str) -> str: if not name or name in roster: return name n = _norm_name(name) if n in by_norm: return by_norm[n] last = n.rsplit(" ", 1)[-1] cands = [r for r in roster if _norm_name(r).rsplit(" ", 1)[-1] == last] if len(cands) == 1: return cands[0] # 토큰 순서가 뒤바뀐 표기 ("Djé D'Avilla" ↔ "D'Avilla Dje Tah") — 집합 포함 관계로 매칭 toks = set(n.split()) cands = [ r for r in roster if toks <= set(_norm_name(r).split()) or set(_norm_name(r).split()) <= toks ] if len(cands) == 1: return cands[0] best, score = name, 0.0 for r in roster: s = SequenceMatcher(None, n, _norm_name(r)).ratio() if s > score: best, score = r, s return best if score >= 0.75 else name return resolve async def fetch_live_mls(m: Match) -> dict: """라이브 페이로드 {available, soccer, clock, period, score, goals[]}.""" import httpx cached = _live_cache.get(m.match_id) if cached and time.monotonic() - cached[0] < _LIVE_TTL_SEC: return cached[1] payload: dict = {"available": False} try: kst = ensure_aware(m.kickoff_at).astimezone(KST) date_kst = kst.strftime("%Y%m%d") span = ( f"{(kst.date() - timedelta(days=1)).strftime('%Y%m%d')}" f"-{kst.date().strftime('%Y%m%d')}" ) async with httpx.AsyncClient(timeout=15) as c: events = await _fetch_events(c, span) rec = next( (r for r in events if r["dateKst"] == date_kst and r["teamA"] == m.team_a_code and r["teamB"] == m.team_b_code), None, ) if rec: status = rec["_status"] or {} stype = status.get("type") or {} # 라인업+문자중계 — summary 1콜 (경기 전 선발 발표~경기 중 교체·중계 반영) lineups = None subs: list[dict] = [] assist_q: dict[str, list[str]] = {} commentary: list[dict] = [] _rn = _make_name_resolver(None) # summary 실패 시 원문 그대로 try: async with httpx.AsyncClient(timeout=15) as c2: r2 = await c2.get(_summary_url(rec["espnId"])) r2.raise_for_status() sj = r2.json() lineups = _lineups_of(sj) _rn = _make_name_resolver(lineups) # 문자중계 — 규칙 기반 한글 번역, 경기 시작부터 전체(최신순). # 교체·어시스트는 details 에 없어 중계 원문에서 함께 추출한다. for e in sj.get("commentary") or []: raw = e.get("text", "") clk = (e.get("time") or {}).get("displayValue", "") ptype = (((e.get("play") or {}).get("type")) or {}).get("type", "") sm = re.match( r"^Substitution, (.+?)\.\s*(.+?) replaces (.+?)(?: because of [^.]*)?\.$", raw, ) if sm: subs.append({ "clock": clk, "team": _TEAM_EN2CODE.get(sm.group(1), ""), "inName": _rn(sm.group(2).strip()), "outName": _rn(sm.group(3).strip()), }) if raw.startswith("Goal!"): gm = re.match(r"^Goal!\s+.+?\d+, .+?\d+\.\s*(.+?) \(", raw) am = re.search(r"Assisted by ([^.]+?)(?: with| following|\.|$)", raw) if gm and am: assist_q.setdefault(_rn(gm.group(1).strip()), []).append( _rn(am.group(1).strip()) ) commentary.append({ "clock": clk, "text": _tr_commentary(raw), "goal": ( ptype == "goal" or raw.startswith("Goal!") or raw.startswith("Own Goal") ), }) commentary.reverse() except Exception as e: # noqa: BLE001 — 라인업 실패해도 스코어는 서빙 log.warning("mls lineups 실패 %s: %s", m.match_id, e) if stype.get("state") in ("in", "post"): goals = [] cards = [] for det in rec["_comp"].get("details") or []: team_id = int((det.get("team") or {}).get("id") or 0) aths = det.get("athletesInvolved") or [] if det.get("yellowCard") or det.get("redCard"): cards.append({ "clock": (det.get("clock") or {}).get("displayValue", ""), "team": MLS_ID_TO_CODE.get(team_id, ""), "player": _rn((aths[0] if aths else {}).get("displayName", "")), "red": bool(det.get("redCard")), }) continue if not det.get("scoringPlay"): continue player = _rn((aths[0] if aths else {}).get("displayName", "")) # 어시스트: 중계 원문 큐(시간순) 우선, 없으면 details 두 번째 선수 queue = assist_q.get(player) assist = queue.pop(0) if queue else ( _rn(aths[1].get("displayName", "")) if len(aths) > 1 and not det.get("ownGoal") else "" ) goals.append({ "clock": (det.get("clock") or {}).get("displayValue", ""), "team": MLS_ID_TO_CODE.get(team_id, ""), "player": player, "assist": assist, "ownGoal": bool(det.get("ownGoal")), "penalty": bool(det.get("penaltyKick")), }) payload = { "available": True, "soccer": True, "clock": status.get("displayClock", ""), "period": status.get("period"), "state": stype.get("state"), "score": { "away": rec["_away"].get("score"), "home": rec["_home"].get("score"), }, "goals": goals, "cards": cards, "subs": subs, "lineups": lineups, "commentary": commentary, } elif lineups: # 경기 전 선발 라인업 발표됨 — 스코어 없이 라인업만 payload = { "available": True, "soccer": True, "state": "pre", "lineups": lineups, "commentary": commentary, } except Exception as e: # noqa: BLE001 log.warning("mls live 실패 %s: %s", m.match_id, e) payload = {"available": False} _live_cache[m.match_id] = (time.monotonic(), payload) return payload