from __future__ import annotations from datetime import datetime, timezone import json import logging import math import os from pathlib import Path import shutil import tempfile from huggingface_hub import HfApi, hf_hub_download from app.models import FinishedMatch logger = logging.getLogger(__name__) def _reject_non_finite(value: str): raise ValueError(f"constante JSON não finita: {value}") class StorageCorruptionError(RuntimeError): pass class StateStore: def __init__(self, data_dir: Path, hf_token: str = "", hf_dataset_repo: str = ""): self.data_dir = data_dir self.state_path = data_dir / "state.json" self.history_path = data_dir / "history.json" self.matches_path = data_dir / "matches.json" self.hf_token = hf_token self.hf_dataset_repo = hf_dataset_repo def _atomic_json(self, path: Path, payload) -> None: path.parent.mkdir(parents=True, exist_ok=True) fd, tmp = tempfile.mkstemp(prefix=path.name, dir=str(path.parent)) try: with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False, indent=2, allow_nan=False) f.flush() os.fsync(f.fileno()) os.replace(tmp, path) finally: if os.path.exists(tmp): os.unlink(tmp) @staticmethod def _quarantine(path: Path) -> Path | None: if not path.exists(): return None try: marker = path.stat().st_mtime_ns target = path.with_name(f"{path.name}.{marker}.corrupt") if not target.exists(): shutil.copy2(path, target) logger.error("Cópia de segurança do arquivo inválido criada em %s", target) return target except Exception as exc: logger.error("Não foi possível preservar o arquivo inválido %s: %s", path, exc) return None @staticmethod def _empty_state(*, status: str = "waiting", warning: str | None = None) -> dict: warnings = [warning] if warning else ["Faça o primeiro scan depois de configurar os Secrets."] return { "generated_at": None, "status": status, "summary": { "events": 0, "historical_matches": 0, "approved": 0, "rejected": 0, "radar": 0, }, "picks": [], "radar": [], "tickets": {}, "performance": {}, "providers": {}, "warnings": warnings, } def load_state(self) -> dict: if not self.state_path.exists(): return self._empty_state() try: value = json.loads( self.state_path.read_text(encoding="utf-8"), parse_constant=_reject_non_finite, ) if not isinstance(value, dict): raise ValueError("a raiz precisa ser um objeto JSON") return value except Exception as exc: logger.error("state.json inválido: %s", exc) self._quarantine(self.state_path) return self._empty_state( status="error", warning="O estado persistido estava inválido e foi ignorado.", ) def save_state(self, state: dict) -> None: self._atomic_json(self.state_path, state) def load_history(self) -> list[dict]: if not self.history_path.exists(): return [] try: value = json.loads( self.history_path.read_text(encoding="utf-8"), parse_constant=_reject_non_finite, ) if not isinstance(value, list): raise ValueError("a raiz não é uma lista") valid: list[dict] = [] invalid_rows = 0 for item in value: if not isinstance(item, dict): invalid_rows += 1 continue row = dict(item) try: for field in ( "probability", "raw_model_probability", "market_probability", "conservative_probability", ): if row.get(field) is None: continue number = float(row[field]) if not math.isfinite(number) or not 0.0 <= number <= 1.0: raise ValueError(field) row[field] = number if row.get("odd") is not None: odd = float(row["odd"]) if not math.isfinite(odd) or not 1.0 <= odd <= 1000.0: raise ValueError("odd") row["odd"] = odd if row.get("profit_units") is not None: profit = float(row["profit_units"]) if not math.isfinite(profit): raise ValueError("profit_units") row["profit_units"] = profit if row.get("result") not in {None, "win", "loss"}: raise ValueError("result") except (TypeError, ValueError): invalid_rows += 1 continue valid.append(row) if invalid_rows: self._quarantine(self.history_path) logger.warning( "history.json: %d registro(s) inválido(s) ignorado(s)", invalid_rows, ) return valid except Exception as exc: logger.error("history.json inválido: %s", exc) self._quarantine(self.history_path) raise StorageCorruptionError("history.json inválido; original preservado") from exc def save_history(self, history: list[dict]) -> None: self._atomic_json(self.history_path, history) def load_matches(self) -> list[FinishedMatch]: if not self.matches_path.exists(): return [] try: rows = json.loads( self.matches_path.read_text(encoding="utf-8"), parse_constant=_reject_non_finite, ) except Exception as exc: logger.error("matches.json inválido: %s", exc) self._quarantine(self.matches_path) raise StorageCorruptionError("matches.json inválido; original preservado") from exc if not isinstance(rows, list): logger.error("matches.json inválido: a raiz não é uma lista") self._quarantine(self.matches_path) raise StorageCorruptionError("matches.json inválido; original preservado") out: list[FinishedMatch] = [] invalid_rows = 0 for row in rows: try: dt = datetime.fromisoformat(str(row["utc_date"]).replace("Z", "+00:00")) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) out.append(FinishedMatch( match_id=str(row.get("match_id") or ""), competition=str(row["competition"]), utc_date=dt, home=str(row["home"]), away=str(row["away"]), home_goals=int(row["home_goals"]), away_goals=int(row["away_goals"]), home_id=str(row.get("home_id") or ""), away_id=str(row.get("away_id") or ""), home_aliases=tuple(row.get("home_aliases") or ()), away_aliases=tuple(row.get("away_aliases") or ()), )) except Exception: invalid_rows += 1 continue if invalid_rows: self._quarantine(self.matches_path) logger.warning("matches.json: %d registro(s) inválido(s) ignorado(s)", invalid_rows) return out def save_matches(self, matches: list[FinishedMatch]) -> None: rows = [{ "match_id": m.match_id, "competition": m.competition, "utc_date": m.utc_date.isoformat(), "home": m.home, "away": m.away, "home_goals": m.home_goals, "away_goals": m.away_goals, "home_id": m.home_id, "away_id": m.away_id, "home_aliases": list(m.home_aliases), "away_aliases": list(m.away_aliases), } for m in matches] self._atomic_json(self.matches_path, rows) def restore_from_hub_if_needed(self) -> None: if not self.hf_token or not self.hf_dataset_repo: return files = ( ("state/state.json", self.state_path), ("state/history.json", self.history_path), ("state/matches.json", self.matches_path), ) for filename, target in files: if target.exists(): continue try: downloaded = hf_hub_download( repo_id=self.hf_dataset_repo, filename=filename, repo_type="dataset", token=self.hf_token, ) target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(downloaded, target) logger.info("Restaurado %s do Dataset HF", filename) except Exception as exc: logger.warning("Não foi possível restaurar %s: %s", filename, exc) def backup_to_hub(self) -> None: if not self.hf_token or not self.hf_dataset_repo: return api = HfApi(token=self.hf_token) try: api.create_repo(self.hf_dataset_repo, repo_type="dataset", exist_ok=True, private=True) available = [ path.name for path in (self.state_path, self.history_path, self.matches_path) if path.exists() ] if available: api.upload_folder( folder_path=str(self.data_dir), path_in_repo="state", allow_patterns=available, repo_id=self.hf_dataset_repo, repo_type="dataset", commit_message="Update Safe Bet precision state", ) except Exception as exc: logger.warning("Backup HF falhou (não interrompe o bot): %s", exc)