o2o-triple-pick/backend/app/services/baseball_data.py

"""야구 예측 프롬프트용 데이터 블록 — 자체 DB(정산 결과) + DataCache(프리뷰·순위).

결과 동기화가 쌓은 종료 경기에서 팀 폼·시즌 성적·상대전적을 계산하고,
캐시된 선발투수·공식 상대전적·순위를 덧붙인다. 모두 없으면 None(이름만 예측).
"""
from __future__ import annotations

from sqlalchemy import or_, select

from ..models import DataCache, Match


def _wdl(gf: int, ga: int) -> str:
    return "W" if gf > ga else "L" if gf < ga else "D"


async def _team_games(db, league: str, code: str, before) -> list[dict]:
    rows = (await db.execute(
        select(Match)
        .where(
            Match.league == league,
            or_(Match.team_a_code == code, Match.team_b_code == code),
            Match.result_outcome.isnot(None),
            Match.kickoff_at < before,
        )
        .order_by(Match.kickoff_at)
    )).scalars().all()
    out = []
    for m in rows:
        is_a = m.team_a_code == code  # team_a = 원정
        gf = m.result_score_a if is_a else m.result_score_b
        ga = m.result_score_b if is_a else m.result_score_a
        if gf is None or ga is None:
            continue
        out.append({
            "r": _wdl(gf, ga), "gf": gf, "ga": ga,
            "opp": m.team_b_short if is_a else m.team_a_short,
        })
    return out


def _team_lines(name: str, games: list[dict]) -> str:
    head = f"[{name}]"
    if not games:
        return f"{head}\n  (no season data yet — use general knowledge)"
    n = len(games)
    w = sum(1 for g in games if g["r"] == "W")
    d = sum(1 for g in games if g["r"] == "D")
    l = sum(1 for g in games if g["r"] == "L")
    gf_avg = round(sum(g["gf"] for g in games) / n, 2)
    ga_avg = round(sum(g["ga"] for g in games) / n, 2)
    recent = games[-8:]
    detail = "; ".join("{r} {gf}-{ga} vs {opp}".format(**g) for g in recent[-4:])
    return (
        f"{head}\n"
        f"  Season(in our data): {w}-{d}-{l} (W-D-L), "
        f"runs avg {gf_avg} scored / {ga_avg} allowed over {n} games\n"
        f"  Form(last{len(recent)}): {' '.join(g['r'] for g in recent)} ({detail})"
    )


async def _h2h_line(db, league: str, code_a: str, code_b: str, before) -> str:
    rows = (await db.execute(
        select(Match)
        .where(
            Match.league == league,
            or_(
                (Match.team_a_code == code_a) & (Match.team_b_code == code_b),
                (Match.team_a_code == code_b) & (Match.team_b_code == code_a),
            ),
            Match.result_outcome.isnot(None),
            Match.kickoff_at < before,
        )
        .order_by(Match.kickoff_at)
    )).scalars().all()
    parts = [
        f"{m.team_a_short} {m.result_score_a}-{m.result_score_b} {m.team_b_short}"
        for m in rows[-5:]
        if m.result_score_a is not None
    ]
    return "; ".join(parts) if parts else "no meetings in our data yet"


def _starter_line(side: str, s: dict | None) -> str | None:
    if not s or not s.get("name"):
        return None
    era = f", season ERA {s['era']}" if s.get("era") else ""
    rec = (
        f" ({s['w']}W-{s['l']}L)"
        if s.get("w") is not None and s.get("l") is not None else ""
    )
    # 상대 ERA 0.00 은 대부분 '대전 기록 없음' — 무실점으로 오해하지 않게 생략
    vs_val = s.get("vsEra")
    vs = (
        f", ERA vs this opponent {vs_val}"
        if vs_val not in (None, "", 0, "0", "0.00", "-")
        else ""
    )
    return f"[{side} starting pitcher] {s['name']}{era}{rec}{vs}"


def _standing_line(name: str, st: dict | None) -> str | None:
    if not st:
        return None
    gb = f", GB {st['gb']}" if st.get("gb") not in (None, "", 0, "0.0") else ""
    ba = f", team AVG {st['avg']}" if st.get("avg") else ""
    era = f", team ERA {st['era']}" if st.get("era") else ""
    extra = f", last5 {st['last5']}" if st.get("last5") else ""
    return (
        f"[{name} standings] rank {st.get('rank')}, {st.get('w')}W-"
        f"{st.get('l')}L (pct {st.get('wra')}){gb}{ba}{era}{extra}"
    )


async def _cached_extras(db, match: Match) -> list[str]:
    lines: list[str] = []
    prev = await db.get(DataCache, f"preview:{match.match_id}")
    if prev:
        p = prev.payload
        for line in (
            _starter_line("Away", p.get("starterA")),
            _starter_line("Home", p.get("starterB")),
        ):
            if line:
                lines.append(line)
        vs = p.get("seasonVs")
        if vs and vs.get("aWin") is not None:
            lines.append(
                f"[Season head-to-head (official)] away {vs['aWin']}W - "
                f"{vs.get('draw', 0)}D - home {vs.get('bWin')}W"
            )
    st_row = await db.get(DataCache, f"standings:{match.league}")
    if st_row:
        table = st_row.payload
        for line in (
            _standing_line(match.team_a_short, table.get(match.team_a_code)),
            _standing_line(match.team_b_short, table.get(match.team_b_code)),
        ):
            if line:
                lines.append(line)
    return lines


async def _context_extras(db, match: Match) -> list[str]:
    """전날 결과·활약 + 확정 라인업(타자 시즌 타율) — 응원가 파이프라인 헬퍼 재사용.

    라인업은 발표 전이면 생략(예측이 킥오프 22h 전부터 생성되므로 보통 미포함).
    """
    from .songs import _lineup_line, _yesterday_info, fetch_lineups

    lines: list[str] = []
    avg_maps: dict[str, dict] = {}
    for side, code, name in (
        ("a", match.team_a_code, match.team_a_short),
        ("b", match.team_b_code, match.team_b_short),
    ):
        label = "Away" if side == "a" else "Home"
        try:
            y, avg_map = await _yesterday_info(db, match, code, name)
        except Exception:  # noqa: BLE001 — 부가 데이터 실패는 생략
            y, avg_map = None, {}
        avg_maps[side] = avg_map
        if y:
            lines.append(f"[{label} yesterday] {y}")
    try:
        lu = await fetch_lineups(match)
    except Exception:  # noqa: BLE001
        lu = None
    if lu and lu.get("announced"):
        for side in ("a", "b"):
            label = "Away" if side == "a" else "Home"
            line = _lineup_line(lu, side, avg_maps.get(side))
            if line:
                lines.append(f"[{label}] {line}")
    return lines


async def build_baseball_data_block(db, match: Match) -> str | None:
    ga = await _team_games(db, match.league, match.team_a_code, match.kickoff_at)
    gb = await _team_games(db, match.league, match.team_b_code, match.kickoff_at)
    extras = await _cached_extras(db, match)
    if not ga and not gb and not extras:
        return None
    h2h = await _h2h_line(
        db, match.league, match.team_a_code, match.team_b_code, match.kickoff_at
    )
    context = await _context_extras(db, match)
    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}",
        *extras,
        *context,
        "===",
    ])