Profeto/src/llm/slices/common.py

"""切片共享基础:结果类型 / 比赛头信息 / 赛果与统计可用性判定。

从 context_builder.py 按领域拆出(单文件 → slices 包),仅做搬迁无逻辑修改。
各领域切片见同包 form/h2h/stats/home_away/standings 模块;
聚合入口 build_context 见 aggregate.py;对外统一经 context_builder 再导出。
"""
from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import TYPE_CHECKING

from sqlalchemy import select
from sqlalchemy.orm import selectinload

from src.db.base import AsyncSessionLocal
from src.db.models import Match

if TYPE_CHECKING:
    from sqlalchemy.ext.asyncio import AsyncSession

logger = logging.getLogger(__name__)


def _outcome(home_goals: int, away_goals: int, side: str) -> str:
    """从某队视角看赛果: W/D/L。"""
    if home_goals is None or away_goals is None:
        return "?"
    if side == "home":
        return "W" if home_goals > away_goals else ("D" if home_goals == away_goals else "L")
    return "W" if away_goals > home_goals else ("D" if away_goals == home_goals else "L")


def _is_stats_available(stats, before) -> bool:
    """检查统计数据在 cutoff 时间是否已可用。

    available_at 语义:该条统计「对外可被使用」的最早时间,
    至少不得早于比赛结束。用于回测防泄漏。

    规则:
      - before is None(实盘):available_at 为 None 时允许(兼容旧数据)
      - before is not None(回测):available_at 为 None 视为不可用(保守)
      - available_at > cutoff:不可用(数据在 cutoff 之后才生成)
    """
    if before is None:
        # 实盘模式:无时间信息时允许(兼容旧数据)
        return True
    # 回测模式(cutoff 不为 None):
    # available_at 为 None → 无法确认是否在 cutoff 前可用,保守视为不可用
    if stats.available_at is None:
        return False
    return stats.available_at <= before


@dataclass
class SliceResult:
    """数据切片的显式结果(替代「靠文案子串猜有无数据」)。

    旧实现用 `"无数据" in slice_text` 判断,依赖具体文案 —— 一旦某个切片
    写成「无比分数据」「无伤停数据」这类变体,判断就会静默失配
    (见审查报告 P2-1)。这里让切片函数直接声明 `has_data`,不再猜。
    """
    text: str
    has_data: bool
    n_records: int = 0

    def __str__(self) -> str:  # 让老调用点可直接当 str 用
        return self.text


@dataclass
class MatchContext:
    match_id: int
    text: str
    has_stats: bool
    has_standings: bool
    match_dt: object | None = None  # 比赛时间(回测防泄漏 + 快照用)
    cutoff: object | None = None  # 实际使用的数据截止时间(用于落库记录)


@dataclass
class MatchHeader:
    """比赛基础信息(所有 agent 共享)。"""
    match_id: int
    home_name: str
    away_name: str
    league_name: str
    season: str | None
    match_date: str
    match_dt: object  # 原始 datetime,回测防泄漏用
    stage: str | None
    home_team_id: int
    away_team_id: int
    league_id: int


async def _load_match(db, match_id: int) -> Match:
    stmt = (
        select(Match)
        .where(Match.id == match_id)
        .options(
            selectinload(Match.league),
            selectinload(Match.home_team),
            selectinload(Match.away_team),
            selectinload(Match.stats),
        )
    )
    m = (await db.execute(stmt)).scalar_one_or_none()
    if m is None:
        raise ValueError(f"match {match_id} not found")
    return m


async def load_match_header(match_id: int, db: AsyncSession | None = None) -> MatchHeader:
    """加载比赛头信息(各 agent 共用)。

    Args:
        match_id: 比赛 ID
        db: 可选的共享 session。不传则自建(向后兼容)。
    """
    if db is not None:
        m = await _load_match(db, match_id)
        return _to_header(m)
    async with AsyncSessionLocal() as new_db:
        m = await _load_match(new_db, match_id)
        return _to_header(m)


def _to_header(m: Match) -> MatchHeader:
    return MatchHeader(
        match_id=m.id,
        home_name=m.home_team.name_zh or m.home_team.name,
        away_name=m.away_team.name_zh or m.away_team.name,
        league_name=m.league.name if m.league else "?",
        season=m.season,
        match_date=m.match_date.strftime("%Y-%m-%d %H:%M UTC") if m.match_date else "?",
        match_dt=m.match_date,
        stage=m.match_stage,
        home_team_id=m.home_team_id,
        away_team_id=m.away_team_id,
        league_id=m.league_id,
    )


def header_text(h: MatchHeader) -> str:
    stage = f" {h.stage}" if h.stage else ""
    return (
        f"对阵: {h.home_name} vs {h.away_name} | {h.league_name} {h.season or '?'}{stage} | {h.match_date}"
    )