ingest/sofascore/compact_stats.py

from __future__ import annotations

import json
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import pandas as pd
import structlog

from config import settings
from ingest.sofascore.paths import MATCH_STATS_PARQUET
from schemas.national_teams import normalize_national_team

logger = structlog.get_logger()

JSON_SUFFIX = "_stats.json"


@dataclass(frozen=True)
class CompactStatsReport:
    json_files: int
    rows_written: int
    rows_from_json: int
    rows_from_existing: int
    parquet_path: Path

    def to_dict(self) -> dict[str, Any]:
        return {
            "json_files": self.json_files,
            "rows_written": self.rows_written,
            "rows_from_json": self.rows_from_json,
            "rows_from_existing": self.rows_from_existing,
            "parquet_path": str(self.parquet_path),
        }


def _parse_fetched_at(value: object) -> datetime:
    if value is None or (isinstance(value, float) and pd.isna(value)):
        return datetime.min.replace(tzinfo=timezone.utc)
    if isinstance(value, datetime):
        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
    parsed = pd.to_datetime(value, utc=True, errors="coerce")
    if pd.isna(parsed):
        return datetime.min.replace(tzinfo=timezone.utc)
    return parsed.to_pydatetime()


def _load_json_rows(stats_dir: Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for path in sorted(stats_dir.glob(f"*{JSON_SUFFIX}")):
        try:
            payload = json.loads(path.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, OSError) as exc:
            logger.warning("sofascore_json_skip", path=str(path), error=str(exc))
            continue
        if not isinstance(payload, dict) or payload.get("event_id") is None:
            logger.warning("sofascore_json_skip", path=str(path), reason="missing_event_id")
            continue
        payload["_source_path"] = str(path)
        payload["_source_mtime"] = path.stat().st_mtime
        rows.append(payload)
    return rows


def _normalize_stats_df(df: pd.DataFrame) -> pd.DataFrame:
    if df.empty:
        return df

    out = df.copy()
    if "event_id" not in out.columns:
        return out

    out["event_id"] = pd.to_numeric(out["event_id"], errors="coerce")
    out = out.dropna(subset=["event_id"])
    out["event_id"] = out["event_id"].astype(int)

    if "home_team" in out.columns:
        out["home_team"] = out["home_team"].map(normalize_national_team)
    if "away_team" in out.columns:
        out["away_team"] = out["away_team"].map(normalize_national_team)

    if "match_date" in out.columns:
        out["match_date"] = pd.to_datetime(out["match_date"], utc=True, errors="coerce")

    if "fetched_at" in out.columns:
        out["_sort_ts"] = out["fetched_at"].map(_parse_fetched_at)
    elif "_source_mtime" in out.columns:
        out["_sort_ts"] = pd.to_datetime(out["_source_mtime"], unit="s", utc=True)
    else:
        out["_sort_ts"] = pd.NaT

    meta_cols = {"_source_path", "_source_mtime", "_sort_ts"}
    out = out.sort_values("_sort_ts", na_position="first")
    out = out.drop_duplicates(subset=["event_id"], keep="last")
    drop_cols = [c for c in meta_cols if c in out.columns]
    return out.drop(columns=drop_cols).reset_index(drop=True)


def compact_match_stats_json(
    *,
    stats_dir: Path | None = None,
    merge_existing: bool = True,
) -> CompactStatsReport:
    """Consolida *_stats.json (e parquet existente) em match_stats.parquet."""
    root = stats_dir or settings.sofascore_stats_dir
    root.mkdir(parents=True, exist_ok=True)
    parquet_path = root / MATCH_STATS_PARQUET

    json_rows = _load_json_rows(root)
    frames: list[pd.DataFrame] = []

    if json_rows:
        frames.append(pd.DataFrame(json_rows))

    existing_rows = 0
    if merge_existing and parquet_path.is_file():
        existing = pd.read_parquet(parquet_path)
        existing_rows = len(existing)
        if not existing.empty:
            frames.append(existing)

    if not frames:
        empty = pd.DataFrame()
        empty.to_parquet(parquet_path, index=False)
        logger.info("sofascore_compact_empty", path=str(parquet_path))
        return CompactStatsReport(
            json_files=len(json_rows),
            rows_written=0,
            rows_from_json=0,
            rows_from_existing=0,
            parquet_path=parquet_path,
        )

    combined = pd.concat(frames, ignore_index=True)
    normalized = _normalize_stats_df(combined)
    normalized.to_parquet(parquet_path, index=False)

    report = CompactStatsReport(
        json_files=len(json_rows),
        rows_written=len(normalized),
        rows_from_json=len(json_rows),
        rows_from_existing=existing_rows,
        parquet_path=parquet_path,
    )
    logger.info("sofascore_compact_done", **report.to_dict())
    return report