from __future__ import annotations from copy import deepcopy from decimal import Decimal, ROUND_HALF_UP from hashlib import sha256 import json from django.db import transaction from django.db.models import Q from core_runtime.models import OutboxEvent from federations.constants import FederationAdminRole from federations.services import FederationService from matches.constants import MatchResultType, MatchStatus from matches.engine.serialization import payload_sha256 from season_ratification.models import SeasonRatificationRecord from season_ratification.services import SeasonRatificationIntegrityService from .constants import ( CompetitionArchiveType, HallOfFameEntityType, HistoricalEntityType, HistoricalRecordMetric, HistoricalRecordRelation, HonorType, ProjectionSourceType, ) from .exceptions import OfficialHistoryIntegrityError, OfficialHistoryPatchError, OfficialHistoryProjectionError from .models import ( OfficialClubSeasonStatistic, OfficialCompetitionArchive, OfficialHallOfFameEntry, OfficialHistoricalRecord, OfficialHistoryProjection, OfficialHonor, OfficialLeagueStandingRow, OfficialMatchArchive, OfficialPlayerSeasonStatistic, ) PROJECTION_SCHEMA_VERSION = 1 SPORTING_ARCHIVE_SCHEMA_VERSION = 1 def stable_hash(payload) -> str: encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str).encode("utf-8") return sha256(encoded).hexdigest() def _decimal_str(value, places=3): value = Decimal(str(value or 0)) quant = Decimal("1").scaleb(-places) return str(value.quantize(quant, rounding=ROUND_HALF_UP)) def _verified_result_certification_for_archive(archive): """Return the final certification sealed by an official-history match row. A superseded original replay row intentionally points at the terminal replay certification, so its certification does not belong to ``archive.match_id``. In that case the immutable replay consolidation is the legal bridge between the archived original and the terminal certified result. """ from match_result_governance.models import MatchResultCertification if not archive.result_certification_id: return None if archive.replay_lineage_role == "SUPERSEDED_ORIGINAL": from replay_execution.models import ReplayResultConsolidation if not archive.replay_consolidation_id or not archive.replay_consolidation_hash: raise OfficialHistoryIntegrityError( "Superseded replay history is missing consolidation evidence." ) try: consolidation = ReplayResultConsolidation.objects.select_related( "execution", "terminal_certification" ).get(pk=archive.replay_consolidation_id) except ReplayResultConsolidation.DoesNotExist as exc: raise OfficialHistoryIntegrityError( "Replay consolidation evidence is no longer available." ) from exc certification = consolidation.terminal_certification if ( consolidation.execution.original_match_id != archive.match_id or consolidation.consolidation_hash != archive.replay_consolidation_hash or stable_hash(consolidation.lineage_payload) != consolidation.consolidation_hash or consolidation.terminal_certification_id != archive.result_certification_id or consolidation.lineage_payload.get("terminal_certification_hash") != certification.certification_hash or consolidation.terminal_match_id != certification.match_id or consolidation.outcome != archive.result_certification_outcome or consolidation.effective_home_score != archive.home_score or consolidation.effective_away_score != archive.away_score ): raise OfficialHistoryIntegrityError( "Replay consolidation evidence failed lineage/hash verification." ) else: try: certification = MatchResultCertification.objects.get( pk=archive.result_certification_id, match_id=archive.match_id ) except MatchResultCertification.DoesNotExist as exc: raise OfficialHistoryIntegrityError( "Ratified result-certification evidence is no longer available." ) from exc if ( certification.certification_hash != archive.result_certification_hash or certification.outcome != archive.result_certification_outcome or stable_hash(certification.certification_payload) != certification.certification_hash or certification.finality_at is None ): raise OfficialHistoryIntegrityError( "Ratified result-certification evidence failed hash/finality verification." ) return certification def _competition_identity(match): fixture = match.fixture if fixture.competition_id: return CompetitionArchiveType.LEAGUE, str(fixture.competition_id) return CompetitionArchiveType.FEDERATION, str(fixture.federation_competition_id) def _competition_key(match): archive_type, source_id = _competition_identity(match) return f"{archive_type}:{source_id}" def _empty_club_stat(*, club_id, name): return { "club_id": str(club_id), "club_name": str(name), "matches": 0, "wins": 0, "draws": 0, "losses": 0, "goals_for": 0, "goals_against": 0, "clean_sheets": 0, "yellow_cards": 0, "red_cards": 0, "competition_breakdown": {}, } def _empty_player_stat(*, player_id, name, nationality, position): return { "player_id": str(player_id), "player_name": str(name), "nationality": str(nationality or ""), "position": str(position or ""), "appearances": 0, "starts": 0, "minutes": 0, "goals": 0, "assists": 0, "clean_sheets": 0, "yellow_cards": 0, "red_cards": 0, "motm_awards": 0, "xg": Decimal("0"), "xa": Decimal("0"), "rating_total": Decimal("0"), "rating_count": 0, "club_breakdown": {}, "competition_breakdown": {}, } def _bump_breakdown(container, key, fields): row = container.setdefault(key, {field: 0 for field in fields}) return row class RatifiedSportingSnapshotBuilder: """Build the exact sporting/statistical manifest bound into season ratification. Match scores are copied into the ratified payload. Immutable simulation and analytics artifacts are referenced by id + hash. Season aggregates are computed at ratification time, so future historical queries do not depend on mutable live tables or current club/player names. """ @classmethod def snapshot(cls, season): from game_statistics.models import ( AdvancedMatchPlayerStatistics, AdvancedMatchTeamStatistics, MatchAnalyticsRun, MatchPlayerStatistics, MatchTeamStatistics, ) from matches.models import Match matches = list( Match.objects.filter( Q(fixture__competition__season=season) | Q(fixture__federation_competition__season=season) ) .filter(status__in=[MatchStatus.COMPLETED, MatchStatus.CANCELLED]) .select_related( "fixture__home_club", "fixture__away_club", "fixture__competition__league", "fixture__federation_competition__federation", ) .order_by("fixture__round_number", "fixture_id", "pk") ) match_ids = [match.pk for match in matches] # Replay fixtures remain first-class historical matches, but only the # terminal replay in a consolidated lineage contributes sporting # aggregates. The superseded original remains archived and points to # the consolidation used by standings/bracket projection. from replay_execution.models import ReplayFixtureExecution replay_executions = list( ReplayFixtureExecution.objects.filter( Q(original_match_id__in=match_ids) | Q(replacement_match_id__in=match_ids) ).select_related("replacement_match", "original_match") ) replay_by_original = {str(row.original_match_id): row for row in replay_executions} replay_by_replacement = {str(row.replacement_match_id): row for row in replay_executions} superseded_original_ids = { row.original_match_id for row in replay_executions if hasattr(row, "consolidation") } countable_match_ids = [mid for mid in match_ids if mid not in superseded_original_ids] advanced_team = { (str(row.match_id), str(row.club_id)): row for row in AdvancedMatchTeamStatistics.objects.filter(match_id__in=match_ids).select_related("club") } basic_team = { (str(row.match_id), str(row.club_id)): row for row in MatchTeamStatistics.objects.filter(match_id__in=match_ids).select_related("club") } analytics_runs = { str(row.match_id): row for row in MatchAnalyticsRun.objects.filter(match_id__in=match_ids).select_related("report") } club_totals = {} match_archive = [] for match in matches: fixture = match.fixture archive_type, source_id = _competition_identity(match) competition_key = f"{archive_type}:{source_id}" report = None try: report = match.simulation_snapshot.simulation_run.report except Exception: report = None report_id = None report_hash = "" if report is not None: if payload_sha256(report.report_payload) != report.report_hash: raise OfficialHistoryIntegrityError( f"Simulation report {report.pk} failed hash verification during season ratification." ) report_id = str(report.pk) report_hash = report.report_hash analytics = analytics_runs.get(str(match.pk)) analytics_id = None analytics_hash = "" if analytics is not None: if payload_sha256(analytics.analytics_payload) != analytics.output_hash: raise OfficialHistoryIntegrityError( f"Analytics run {analytics.pk} failed hash verification during season ratification." ) analytics_id = str(analytics.pk) analytics_hash = analytics.output_hash official_report_id = None official_report_hash = "" official_report_ratification_hash = "" if fixture.federation_competition_id and match.result_type == MatchResultType.PLAYED: from match_officials_governance.exceptions import MatchOfficialsGovernanceError from match_officials_governance.services import OfficialMatchReportService try: official_report = OfficialMatchReportService.assert_official_history_ready(match=match) except MatchOfficialsGovernanceError as exc: raise OfficialHistoryIntegrityError(str(exc)) from exc if official_report is not None: official_report_id = str(official_report.pk) official_report_hash = official_report.report_hash official_report_ratification_hash = official_report.ratification_hash result_certification_id = None result_certification_hash = "" result_certification_outcome = "" effective_home_score = match.home_score effective_away_score = match.away_score effective_result_type = match.result_type if match.status == MatchStatus.COMPLETED: from match_result_governance.exceptions import MatchResultGovernanceError from match_result_governance.services import ResultProjectionService try: certified = ResultProjectionService.effective_result(match=match, require_final=True) except MatchResultGovernanceError as exc: raise OfficialHistoryIntegrityError(str(exc)) from exc if certified is None: raise OfficialHistoryIntegrityError( f"Completed match {match.pk} has no final countable result certification." ) effective_home_score = certified["home_score"] effective_away_score = certified["away_score"] effective_result_type = certified["result_type"] result_certification_id = certified.get("certification_id") result_certification_hash = certified.get("certification_hash") or "" result_certification_outcome = certified.get("outcome") or "" replay_out = replay_by_original.get(str(match.pk)) replay_in = replay_by_replacement.get(str(match.pk)) replay_consolidation = None if replay_out is not None: try: replay_consolidation = replay_out.consolidation except Exception: replay_consolidation = None statistics_countable = match.pk not in superseded_original_ids match_body = { "match_id": str(match.pk), "fixture_id": str(fixture.pk), "archive_type": archive_type, "competition_source_id": source_id, "round_number": fixture.round_number, "leg": fixture.leg, "competition_type": fixture.competition_type, "home_club_id": str(fixture.home_club_id), "home_club_name": fixture.home_club.name, "away_club_id": str(fixture.away_club_id), "away_club_name": fixture.away_club.name, "status": match.status, "result_type": effective_result_type, "home_score": effective_home_score, "away_score": effective_away_score, "raw_result_type": match.result_type, "raw_home_score": match.home_score, "raw_away_score": match.away_score, "scheduled_at": match.scheduled_at.isoformat() if match.scheduled_at else None, "started_at": match.started_at.isoformat() if match.started_at else None, "completed_at": match.completed_at.isoformat() if match.completed_at else None, "cancelled_at": match.cancelled_at.isoformat() if match.cancelled_at else None, "simulation_report_id": report_id, "simulation_report_hash": report_hash, "analytics_run_id": analytics_id, "analytics_output_hash": analytics_hash, "official_match_report_id": official_report_id, "official_match_report_hash": official_report_hash, "official_match_report_ratification_hash": official_report_ratification_hash, "result_certification_id": result_certification_id, "result_certification_hash": result_certification_hash, "result_certification_outcome": result_certification_outcome, "replay_lineage_role": ( "SUPERSEDED_ORIGINAL" if replay_out is not None and replay_consolidation is not None else "REPLAY_REPLACEMENT" if replay_in is not None else "NONE" ), "replay_execution_id": str((replay_out or replay_in).pk) if (replay_out or replay_in) else None, "replay_consolidation_id": str(replay_consolidation.pk) if replay_consolidation else None, "replay_consolidation_hash": replay_consolidation.consolidation_hash if replay_consolidation else "", "statistics_countable": statistics_countable, } match_body["entry_hash"] = stable_hash(match_body) match_archive.append(match_body) if match.status != MatchStatus.COMPLETED or not statistics_countable: continue home = club_totals.setdefault( str(fixture.home_club_id), _empty_club_stat(club_id=fixture.home_club_id, name=fixture.home_club.name) ) away = club_totals.setdefault( str(fixture.away_club_id), _empty_club_stat(club_id=fixture.away_club_id, name=fixture.away_club.name) ) home["matches"] += 1 away["matches"] += 1 home["goals_for"] += int(effective_home_score or 0) home["goals_against"] += int(effective_away_score or 0) away["goals_for"] += int(effective_away_score or 0) away["goals_against"] += int(effective_home_score or 0) if effective_home_score > effective_away_score: home["wins"] += 1 away["losses"] += 1 elif effective_home_score < effective_away_score: away["wins"] += 1 home["losses"] += 1 else: home["draws"] += 1 away["draws"] += 1 if effective_away_score == 0: home["clean_sheets"] += 1 if effective_home_score == 0: away["clean_sheets"] += 1 breakdown_fields = ["matches", "wins", "draws", "losses", "goals_for", "goals_against", "clean_sheets", "yellow_cards", "red_cards"] hb = _bump_breakdown(home["competition_breakdown"], competition_key, breakdown_fields) ab = _bump_breakdown(away["competition_breakdown"], competition_key, breakdown_fields) for row, score_for, score_against in ((hb, effective_home_score, effective_away_score), (ab, effective_away_score, effective_home_score)): row["matches"] += 1 row["goals_for"] += int(score_for or 0) row["goals_against"] += int(score_against or 0) if score_for > score_against: row["wins"] += 1 elif score_for < score_against: row["losses"] += 1 else: row["draws"] += 1 if score_against == 0: row["clean_sheets"] += 1 for club_id, total, breakdown in ( (str(fixture.home_club_id), home, hb), (str(fixture.away_club_id), away, ab), ): stat = advanced_team.get((str(match.pk), club_id)) or basic_team.get((str(match.pk), club_id)) if stat is not None: yellow = int(getattr(stat, "yellow_cards", 0) or 0) red = int(getattr(stat, "red_cards", 0) or 0) total["yellow_cards"] += yellow total["red_cards"] += red breakdown["yellow_cards"] += yellow breakdown["red_cards"] += red advanced_rows = list( AdvancedMatchPlayerStatistics.objects.filter(match_id__in=countable_match_ids) .select_related("player", "club", "match__fixture") .order_by("match_id", "player_id") ) advanced_match_ids = {row.match_id for row in advanced_rows} basic_rows = list( MatchPlayerStatistics.objects.filter(match_id__in=countable_match_ids) .exclude(match_id__in=advanced_match_ids) .select_related("player", "club", "match__fixture") .order_by("match_id", "player_id") ) player_totals = {} for row in [*advanced_rows, *basic_rows]: player = row.player total = player_totals.setdefault( str(player.pk), _empty_player_stat( player_id=player.pk, name=f"{player.first_name} {player.last_name}".strip(), nationality=player.nationality, position=player.position, ), ) total["appearances"] += 1 total["starts"] += int(bool(row.started)) total["minutes"] += int(row.minutes_played or 0) total["goals"] += int(row.goals or 0) total["assists"] += int(getattr(row, "assists", 0) or 0) total["clean_sheets"] += int(bool(row.clean_sheet)) total["yellow_cards"] += int(row.yellow_cards or 0) total["red_cards"] += int(row.red_cards or 0) total["motm_awards"] += int(bool(getattr(row, "is_man_of_the_match", False))) total["xg"] += Decimal(str(getattr(row, "xg", 0) or 0)) total["xa"] += Decimal(str(getattr(row, "xa", 0) or 0)) rating = getattr(row, "rating", None) if rating is not None: total["rating_total"] += Decimal(str(rating)) total["rating_count"] += 1 club_key = str(row.club_id) club_breakdown = total["club_breakdown"].setdefault( club_key, {"club_id": club_key, "club_name": row.club.name, "appearances": 0, "starts": 0, "minutes": 0, "goals": 0, "assists": 0}, ) club_breakdown["appearances"] += 1 club_breakdown["starts"] += int(bool(row.started)) club_breakdown["minutes"] += int(row.minutes_played or 0) club_breakdown["goals"] += int(row.goals or 0) club_breakdown["assists"] += int(getattr(row, "assists", 0) or 0) competition_key = _competition_key(row.match) comp = total["competition_breakdown"].setdefault( competition_key, {"appearances": 0, "starts": 0, "minutes": 0, "goals": 0, "assists": 0, "motm_awards": 0}, ) comp["appearances"] += 1 comp["starts"] += int(bool(row.started)) comp["minutes"] += int(row.minutes_played or 0) comp["goals"] += int(row.goals or 0) comp["assists"] += int(getattr(row, "assists", 0) or 0) comp["motm_awards"] += int(bool(getattr(row, "is_man_of_the_match", False))) club_rows = [] for key in sorted(club_totals): payload = club_totals[key] payload["goal_difference"] = payload["goals_for"] - payload["goals_against"] payload["statistic_hash"] = stable_hash(payload) club_rows.append(payload) player_rows = [] for key in sorted(player_totals): total = player_totals[key] rating_count = total.pop("rating_count") rating_total = total.pop("rating_total") total["xg"] = _decimal_str(total["xg"], 3) total["xa"] = _decimal_str(total["xa"], 3) total["average_rating"] = ( _decimal_str(rating_total / rating_count, 2) if rating_count else None ) total["statistic_hash"] = stable_hash(total) player_rows.append(total) body = { "schema_version": SPORTING_ARCHIVE_SCHEMA_VERSION, "season_id": str(season.pk), "matches": match_archive, "club_season_statistics": club_rows, "player_season_statistics": player_rows, "summary": { "archived_matches": len(match_archive), "completed_matches": sum(1 for row in match_archive if row["status"] == MatchStatus.COMPLETED), "cancelled_matches": sum(1 for row in match_archive if row["status"] == MatchStatus.CANCELLED), "clubs": len(club_rows), "players": len(player_rows), "advanced_analytics_matches": sum(1 for row in match_archive if row["analytics_run_id"]), }, } body["snapshot_hash"] = stable_hash(body) return body @classmethod def verify(cls, snapshot): if not snapshot or int(snapshot.get("schema_version") or 0) != SPORTING_ARCHIVE_SCHEMA_VERSION: return False expected = snapshot.get("snapshot_hash") body = deepcopy(snapshot) body.pop("snapshot_hash", None) if stable_hash(body) != expected: return False matches = snapshot.get("matches") or [] club_rows = snapshot.get("club_season_statistics") or [] player_rows = snapshot.get("player_season_statistics") or [] for row in matches: candidate = dict(row) row_hash = candidate.pop("entry_hash", "") if not row_hash or stable_hash(candidate) != row_hash: return False for key in ("club_season_statistics", "player_season_statistics"): for row in snapshot.get(key) or []: candidate = dict(row) row_hash = candidate.pop("statistic_hash", "") if not row_hash or stable_hash(candidate) != row_hash: return False summary = snapshot.get("summary") or {} if int(summary.get("archived_matches", -1)) != len(matches): return False if int(summary.get("completed_matches", -1)) != sum(1 for row in matches if row.get("status") == MatchStatus.COMPLETED): return False if int(summary.get("cancelled_matches", -1)) != sum(1 for row in matches if row.get("status") == MatchStatus.CANCELLED): return False if int(summary.get("clubs", -1)) != len(club_rows) or int(summary.get("players", -1)) != len(player_rows): return False derived = {} for match in matches: if match.get("status") != MatchStatus.COMPLETED: continue hs = int(match.get("home_score") or 0) aws = int(match.get("away_score") or 0) for club_id, club_name, gf, ga in ( (str(match.get("home_club_id")), match.get("home_club_name"), hs, aws), (str(match.get("away_club_id")), match.get("away_club_name"), aws, hs), ): row = derived.setdefault(club_id, _empty_club_stat(club_id=club_id, name=club_name or club_id)) row["matches"] += 1 row["goals_for"] += gf row["goals_against"] += ga if gf > ga: row["wins"] += 1 elif gf < ga: row["losses"] += 1 else: row["draws"] += 1 if ga == 0: row["clean_sheets"] += 1 official_by_club = {str(row.get("club_id")): row for row in club_rows} if set(derived) != set(official_by_club): return False for club_id, expected_row in derived.items(): actual = official_by_club[club_id] for field in ("matches", "wins", "draws", "losses", "goals_for", "goals_against", "clean_sheets"): if int(actual.get(field) or 0) != int(expected_row.get(field) or 0): return False return True class OfficialHistoryPatchService: """Small deterministic RFC-6902-style subset for official-record addenda.""" ALLOWED_OPS = {"add", "replace", "remove"} @staticmethod def _tokens(path): if not isinstance(path, str) or not path.startswith("/"): raise OfficialHistoryPatchError("Official-history patch paths must be JSON-pointer paths.") return [part.replace("~1", "/").replace("~0", "~") for part in path[1:].split("/") if part != ""] @classmethod def apply(cls, document, operations): result = deepcopy(document) if not isinstance(operations, list): raise OfficialHistoryPatchError("official_history_patch.operations must be a list.") for operation in operations: if not isinstance(operation, dict) or operation.get("op") not in cls.ALLOWED_OPS: raise OfficialHistoryPatchError("Unsupported official-history patch operation.") tokens = cls._tokens(operation.get("path")) if not tokens: raise OfficialHistoryPatchError("Replacing the entire sporting archive is not permitted.") if tokens[0] in {"schema_version", "season_id"}: raise OfficialHistoryPatchError("Sporting archive identity/schema fields cannot be patched.") target = result for token in tokens[:-1]: if isinstance(target, list): target = target[int(token)] else: if token not in target: raise OfficialHistoryPatchError(f"Patch path segment does not exist: {token}") target = target[token] leaf = tokens[-1] op = operation["op"] if isinstance(target, list): if leaf == "-" and op == "add": target.append(deepcopy(operation.get("value"))) continue index = int(leaf) if op == "remove": target.pop(index) elif op == "replace": target[index] = deepcopy(operation.get("value")) elif op == "add": target.insert(index, deepcopy(operation.get("value"))) else: if op == "remove": if leaf not in target: raise OfficialHistoryPatchError(f"Patch remove target does not exist: {leaf}") target.pop(leaf) else: target[leaf] = deepcopy(operation.get("value")) return result class OfficialHistoryProjectionService: @staticmethod def _rehash_snapshot(snapshot): effective = deepcopy(snapshot) effective.pop("snapshot_hash", None) for row in effective.get("matches") or []: row.pop("entry_hash", None) row["entry_hash"] = stable_hash(row) for key in ("club_season_statistics", "player_season_statistics"): for row in effective.get(key) or []: row.pop("statistic_hash", None) row["statistic_hash"] = stable_hash(row) effective_hash = stable_hash(effective) effective["snapshot_hash"] = effective_hash return effective @classmethod def source_chain(cls, ratification, *, chain_head=None): all_addenda = list(ratification.addenda.order_by("sequence")) if not chain_head or chain_head == ratification.official_hash: if chain_head == ratification.official_hash: return [], ratification.official_hash return all_addenda, (all_addenda[-1].addendum_hash if all_addenda else ratification.official_hash) selected = [] found = False for addendum in all_addenda: selected.append(addendum) if addendum.addendum_hash == chain_head: found = True break if not found: raise OfficialHistoryIntegrityError("Requested official-record chain head does not belong to this ratification.") return selected, chain_head @classmethod def effective_sporting_snapshot(cls, ratification, *, chain_head=None): snapshot = deepcopy((ratification.official_payload or {}).get("sporting_archive") or {}) if not RatifiedSportingSnapshotBuilder.verify(snapshot): raise OfficialHistoryIntegrityError("Ratified sporting archive is missing or failed its snapshot hash.") base_hash = snapshot["snapshot_hash"] addenda, head = cls.source_chain(ratification, chain_head=chain_head) applied = [] effective = deepcopy(snapshot) for addendum in addenda: patch = (addendum.correction_payload or {}).get("official_history_patch") if patch: effective = OfficialHistoryPatchService.apply(effective, patch.get("operations") or []) applied.append({ "addendum_id": str(addendum.pk), "sequence": addendum.sequence, "addendum_hash": addendum.addendum_hash, "operations_hash": stable_hash(patch.get("operations") or []), }) effective = cls._rehash_snapshot(effective) if not RatifiedSportingSnapshotBuilder.verify(effective): raise OfficialHistoryIntegrityError( "Effective sporting archive is internally inconsistent after official addendum patches." ) effective_hash = effective["snapshot_hash"] return effective, { "base_snapshot_hash": base_hash, "effective_snapshot_hash": effective_hash, "source_chain_head_hash": head, "addenda": [ {"id": str(row.pk), "sequence": row.sequence, "hash": row.addendum_hash, "type": row.addendum_type} for row in addenda ], "applied_history_patches": applied, } @staticmethod def _competition_payloads(ratification): official = ratification.official_payload or {} payloads = [] standing_rows = [] honors = [] for row in official.get("league_records") or []: source_id = str(row["competition_id"]) table = row.get("table_payload") or {} label = f"{row.get('country') or ''} {row.get('competition_type') or ''} League Level {row.get('level')}".strip() comp = { "archive_type": CompetitionArchiveType.LEAGUE, "source_id": source_id, "label": label, "competition_type": row.get("competition_type") or "", "competition_kind": "LEAGUE", "country": row.get("country") or "", "level": row.get("level"), "status": "COMPLETED", "champion_club_id": row.get("champion_club_id"), "runner_up_club_id": row.get("runner_up_club_id"), "champion_reference": row.get("champion_club_id") or "", "runner_up_reference": row.get("runner_up_club_id") or "", "standings_payload": table, "source_evidence_hash": row.get("standings_hash") or "", } comp["archive_hash"] = stable_hash(comp) payloads.append(comp) rows = table.get("rows") or [] for standing in rows: payload = {"competition_source_id": source_id, "league_id": str(row.get("league_id") or ""), **standing} payload["row_hash"] = stable_hash(payload) standing_rows.append(payload) if rows: for rank, htype in ((1, HonorType.LEAGUE_CHAMPION), (2, HonorType.LEAGUE_RUNNER_UP)): if len(rows) >= rank: standing = rows[rank - 1] hp = { "honor_key": f"{htype}:{source_id}:{standing['club_id']}", "honor_type": htype, "entity_type": HistoricalEntityType.CLUB, "entity_reference": str(standing["club_id"]), "entity_name_snapshot": standing.get("club_name") or str(standing["club_id"]), "club_id": str(standing["club_id"]), "player_id": None, "competition_archive_type": CompetitionArchiveType.LEAGUE, "competition_source_id": source_id, "rank": rank, "metric_key": "LEAGUE_POSITION", "metric_value": _decimal_str(rank, 4), "evidence_payload": {"standings_hash": row.get("standings_hash"), "standing_row": standing}, } hp["honor_hash"] = stable_hash(hp) honors.append(hp) for row in official.get("federation_competitions") or []: source_id = str(row["competition_id"]) comp = { "archive_type": CompetitionArchiveType.FEDERATION, "source_id": source_id, "label": row.get("name") or f"Federation competition {source_id}", "competition_type": row.get("competition_type") or "", "competition_kind": row.get("kind") or "", "country": row.get("country") or "", "level": None, "status": row.get("status") or "", "champion_club_id": row.get("champion_club_id"), "runner_up_club_id": row.get("runner_up_club_id"), "champion_reference": row.get("champion_club_id") or row.get("champion_youth_team_id") or "", "runner_up_reference": row.get("runner_up_club_id") or row.get("runner_up_youth_team_id") or "", "standings_payload": {}, "source_evidence_hash": stable_hash(row), } comp["archive_hash"] = stable_hash(comp) payloads.append(comp) for rank, htype, club_key, youth_key, name_key in ( (1, HonorType.CUP_CHAMPION, "champion_club_id", "champion_youth_team_id", "champion_name"), (2, HonorType.CUP_RUNNER_UP, "runner_up_club_id", "runner_up_youth_team_id", "runner_up_name"), ): reference = row.get(club_key) or row.get(youth_key) if not reference: continue entity_type = HistoricalEntityType.CLUB if row.get(club_key) else HistoricalEntityType.YOUTH_TEAM hp = { "honor_key": f"{htype}:{source_id}:{reference}", "honor_type": htype, "entity_type": entity_type, "entity_reference": str(reference), "entity_name_snapshot": row.get(name_key) or str(reference), "club_id": str(row.get(club_key)) if row.get(club_key) else None, "player_id": None, "competition_archive_type": CompetitionArchiveType.FEDERATION, "competition_source_id": source_id, "rank": rank, "metric_key": "COMPETITION_FINISH", "metric_value": _decimal_str(rank, 4), "evidence_payload": {"competition_record": row}, } hp["honor_hash"] = stable_hash(hp) honors.append(hp) return payloads, standing_rows, honors @staticmethod def _statistical_honors(player_rows): honors = [] metrics = [ (HonorType.TOP_SCORER, "goals", lambda row: Decimal(str(row.get("goals") or 0))), (HonorType.TOP_ASSISTS, "assists", lambda row: Decimal(str(row.get("assists") or 0))), (HonorType.MOST_MOTM, "motm_awards", lambda row: Decimal(str(row.get("motm_awards") or 0))), (HonorType.HIGHEST_RATING, "average_rating", lambda row: Decimal(str(row["average_rating"])) if row.get("average_rating") is not None else None), ] for honor_type, metric_key, extractor in metrics: values = [(row, extractor(row)) for row in player_rows] values = [(row, value) for row, value in values if value is not None] if not values: continue best = max(value for _, value in values) if best <= 0 and metric_key != "average_rating": continue winners = sorted((row for row, value in values if value == best), key=lambda x: str(x["player_id"])) for row in winners: hp = { "honor_key": f"{honor_type}:SEASON:{row['player_id']}", "honor_type": honor_type, "entity_type": HistoricalEntityType.PLAYER, "entity_reference": str(row["player_id"]), "entity_name_snapshot": row.get("player_name") or str(row["player_id"]), "club_id": None, "player_id": str(row["player_id"]), "competition_archive_type": "", "competition_source_id": "", "rank": 1, "metric_key": metric_key, "metric_value": _decimal_str(best, 4), "evidence_payload": {"statistic_hash": row.get("statistic_hash"), "metric": metric_key}, } hp["honor_hash"] = stable_hash(hp) honors.append(hp) return honors @classmethod def _record_candidates(cls, *, ratification, club_rows, player_rows, standing_rows): candidates = [] metric_rows = [ (HistoricalRecordMetric.PLAYER_GOALS_SEASON, HistoricalEntityType.PLAYER, player_rows, "player_id", "player_name", "goals", "goals", "GLOBAL"), (HistoricalRecordMetric.PLAYER_ASSISTS_SEASON, HistoricalEntityType.PLAYER, player_rows, "player_id", "player_name", "assists", "assists", "GLOBAL"), (HistoricalRecordMetric.PLAYER_MOTM_SEASON, HistoricalEntityType.PLAYER, player_rows, "player_id", "player_name", "motm_awards", "awards", "GLOBAL"), (HistoricalRecordMetric.PLAYER_MINUTES_SEASON, HistoricalEntityType.PLAYER, player_rows, "player_id", "player_name", "minutes", "minutes", "GLOBAL"), (HistoricalRecordMetric.CLUB_WINS_SEASON, HistoricalEntityType.CLUB, club_rows, "club_id", "club_name", "wins", "wins", "GLOBAL"), (HistoricalRecordMetric.CLUB_GOALS_SEASON, HistoricalEntityType.CLUB, club_rows, "club_id", "club_name", "goals_for", "goals", "GLOBAL"), (HistoricalRecordMetric.CLUB_GOAL_DIFFERENCE_SEASON, HistoricalEntityType.CLUB, club_rows, "club_id", "club_name", "goal_difference", "goals", "GLOBAL"), ] for metric, entity_type, rows, id_key, name_key, value_key, unit, scope in metric_rows: if not rows: continue best = max(Decimal(str(row.get(value_key) or 0)) for row in rows) winners = [row for row in rows if Decimal(str(row.get(value_key) or 0)) == best] for row in winners: candidates.append({ "metric_key": metric, "scope_key": scope, "entity_type": entity_type, "entity_reference": str(row[id_key]), "entity_name_snapshot": row.get(name_key) or str(row[id_key]), "club_id": str(row[id_key]) if entity_type == HistoricalEntityType.CLUB else None, "player_id": str(row[id_key]) if entity_type == HistoricalEntityType.PLAYER else None, "value": best, "unit": unit, "evidence_payload": {"season_id": str(ratification.season_id), "statistic_hash": row.get("statistic_hash")}, }) by_comp = {} for row in standing_rows: by_comp.setdefault(row["competition_source_id"], []).append(row) for source_id, rows in by_comp.items(): if not rows: continue best = max(Decimal(str(row.get("points") or 0)) for row in rows) for row in rows: if Decimal(str(row.get("points") or 0)) != best: continue candidates.append({ "metric_key": HistoricalRecordMetric.LEAGUE_POINTS, "scope_key": f"LEAGUE:{row.get('league_id') or source_id}", "entity_type": HistoricalEntityType.CLUB, "entity_reference": str(row["club_id"]), "entity_name_snapshot": row.get("club_name") or str(row["club_id"]), "club_id": str(row["club_id"]), "player_id": None, "value": best, "unit": "points", "evidence_payload": {"season_id": str(ratification.season_id), "standing_row_hash": row.get("row_hash")}, }) return candidates @staticmethod def _qualifying_record(candidate, ratification): latest_projection_ids = OfficialHistorySelector.latest_projection_ids() qs = OfficialHistoricalRecord.objects.filter( projection_id__in=latest_projection_ids, metric_key=candidate["metric_key"], scope_key=candidate["scope_key"], ).exclude(projection__ratification=ratification).order_by( "-value", "-projection__season__start_date", "-projection__sequence", "id" ) previous = qs.first() if previous is None: return HistoricalRecordRelation.INITIAL, None if candidate["value"] > previous.value: return HistoricalRecordRelation.BROKE, previous if candidate["value"] == previous.value: return HistoricalRecordRelation.TIED, previous return None, previous @classmethod def history_context(cls, ratification): latest_by_season = {} rows = ( OfficialHistoryProjection.objects.filter(season__start_date__lt=ratification.season.start_date) .select_related("season") .order_by("season__start_date", "season_id", "sequence") ) for row in rows: latest_by_season[row.season_id] = row payload = [ { "season_id": str(row.season_id), "projection_id": str(row.pk), "projection_sequence": row.sequence, "projection_hash": row.projection_hash, "source_chain_head_hash": row.source_chain_head_hash, } for _, row in sorted(latest_by_season.items(), key=lambda item: (item[1].season.start_date, item[0])) ] return payload, stable_hash(payload) @classmethod @transaction.atomic def project(cls, *, ratification_id, source_event_id=None, source_chain_head_hash=None): ratification = ( SeasonRatificationRecord.objects.select_for_update() .select_related("season", "mother_federation") .get(pk=ratification_id) ) integrity = SeasonRatificationIntegrityService.verify(ratification) if not integrity.get("valid"): raise OfficialHistoryIntegrityError( "Season ratification failed integrity verification: " + ", ".join(integrity.get("errors") or []) ) snapshot, source_meta = cls.effective_sporting_snapshot(ratification, chain_head=source_chain_head_hash) chain_head = source_meta["source_chain_head_hash"] history_context, history_context_hash = cls.history_context(ratification) existing = OfficialHistoryProjection.objects.filter( ratification=ratification, source_chain_head_hash=chain_head, history_context_hash=history_context_hash, ).first() if existing is not None: return existing competition_payloads, standing_rows, competition_honors = cls._competition_payloads(ratification) club_rows = list(snapshot.get("club_season_statistics") or []) player_rows = list(snapshot.get("player_season_statistics") or []) match_rows = list(snapshot.get("matches") or []) honors = competition_honors + cls._statistical_honors(player_rows) record_candidates = cls._record_candidates( ratification=ratification, club_rows=club_rows, player_rows=player_rows, standing_rows=standing_rows ) record_rows = [] for candidate in record_candidates: relation, previous = cls._qualifying_record(candidate, ratification) if relation is None: continue payload = { **candidate, "value": _decimal_str(candidate["value"], 4), "relation": relation, "previous_record_id": str(previous.pk) if previous else None, } payload["record_hash"] = stable_hash(payload) payload["previous_record"] = previous record_rows.append(payload) sequence = (OfficialHistoryProjection.objects.filter(ratification=ratification).order_by("-sequence").values_list("sequence", flat=True).first() or 0) + 1 projection_payload = { "schema_version": PROJECTION_SCHEMA_VERSION, "ratification_id": str(ratification.pk), "season_id": str(ratification.season_id), "season_name": ratification.season.name, "ratification_official_hash": ratification.official_hash, "source_chain_head_hash": chain_head, "source_snapshot_hash": source_meta["effective_snapshot_hash"], "history_context_hash": history_context_hash, "history_context": history_context, "source_addenda": source_meta["addenda"], "applied_history_patches": source_meta["applied_history_patches"], "counts": { "competitions": len(competition_payloads), "standing_rows": len(standing_rows), "matches": len(match_rows), "club_statistics": len(club_rows), "player_statistics": len(player_rows), "honors": len(honors), "new_or_tied_records": len(record_rows), }, } projection_hash = stable_hash(projection_payload) projection = OfficialHistoryProjection.objects.create( ratification=ratification, season=ratification.season, sequence=sequence, source_type=ProjectionSourceType.ADDENDUM if source_meta["addenda"] else ProjectionSourceType.RATIFICATION, source_chain_head_hash=chain_head, source_snapshot_hash=source_meta["effective_snapshot_hash"], history_context_hash=history_context_hash, schema_version=PROJECTION_SCHEMA_VERSION, projection_payload=projection_payload, projection_hash=projection_hash, source_event_id=source_event_id, ) comp_map = {} for row in competition_payloads: comp = OfficialCompetitionArchive.objects.create( projection=projection, archive_type=row["archive_type"], source_id=row["source_id"], label=row["label"], competition_type=row["competition_type"], competition_kind=row["competition_kind"], country=row["country"], level=row["level"], status=row["status"], champion_club_id=row["champion_club_id"] or None, runner_up_club_id=row["runner_up_club_id"] or None, champion_reference=row["champion_reference"], runner_up_reference=row["runner_up_reference"], standings_payload=row["standings_payload"], source_evidence_hash=row["source_evidence_hash"], archive_hash=row["archive_hash"], ) comp_map[(row["archive_type"], row["source_id"])] = comp OfficialLeagueStandingRow.objects.bulk_create([ OfficialLeagueStandingRow( projection=projection, competition_archive=comp_map[(CompetitionArchiveType.LEAGUE, row["competition_source_id"])], position=row["position"], club_id=row["club_id"], club_name_snapshot=row.get("club_name") or str(row["club_id"]), played=row.get("played") or 0, won=row.get("won") or 0, drawn=row.get("drawn") or 0, lost=row.get("lost") or 0, goals_for=row.get("goals_for") or 0, goals_against=row.get("goals_against") or 0, goal_difference=row.get("goal_difference") or 0, points=row.get("points") or 0, unresolved_tie=bool(row.get("unresolved_tie")), row_payload={k: v for k, v in row.items() if k not in {"row_hash", "competition_source_id"}}, row_hash=row["row_hash"], ) for row in standing_rows ]) OfficialMatchArchive.objects.bulk_create([ OfficialMatchArchive( projection=projection, match_id=row["match_id"], competition_archive_type=row["archive_type"], competition_source_id=row["competition_source_id"], round_number=row["round_number"], leg=row["leg"], competition_type=row["competition_type"], home_club_id=row["home_club_id"], away_club_id=row["away_club_id"], home_club_name_snapshot=row["home_club_name"], away_club_name_snapshot=row["away_club_name"], status=row["status"], result_type=row["result_type"], home_score=row["home_score"], away_score=row["away_score"], completed_at=row.get("completed_at"), simulation_report_id=row.get("simulation_report_id"), simulation_report_hash=row.get("simulation_report_hash") or "", analytics_run_id=row.get("analytics_run_id"), analytics_output_hash=row.get("analytics_output_hash") or "", official_match_report_id=row.get("official_match_report_id"), official_match_report_hash=row.get("official_match_report_hash") or "", official_match_report_ratification_hash=row.get("official_match_report_ratification_hash") or "", result_certification_id=row.get("result_certification_id"), result_certification_hash=row.get("result_certification_hash") or "", result_certification_outcome=row.get("result_certification_outcome") or "", replay_lineage_role=row.get("replay_lineage_role") or "NONE", replay_execution_id=row.get("replay_execution_id"), replay_consolidation_id=row.get("replay_consolidation_id"), replay_consolidation_hash=row.get("replay_consolidation_hash") or "", statistics_countable=bool(row.get("statistics_countable", True)), evidence_payload={k: v for k, v in row.items() if k != "entry_hash"}, archive_hash=row["entry_hash"], ) for row in match_rows ]) OfficialClubSeasonStatistic.objects.bulk_create([ OfficialClubSeasonStatistic( projection=projection, club_id=row["club_id"], club_name_snapshot=row["club_name"], matches=row.get("matches") or 0, wins=row.get("wins") or 0, draws=row.get("draws") or 0, losses=row.get("losses") or 0, goals_for=row.get("goals_for") or 0, goals_against=row.get("goals_against") or 0, clean_sheets=row.get("clean_sheets") or 0, yellow_cards=row.get("yellow_cards") or 0, red_cards=row.get("red_cards") or 0, competition_breakdown=row.get("competition_breakdown") or {}, statistic_payload={k: v for k, v in row.items() if k != "statistic_hash"}, statistic_hash=row["statistic_hash"], ) for row in club_rows ]) OfficialPlayerSeasonStatistic.objects.bulk_create([ OfficialPlayerSeasonStatistic( projection=projection, player_id=row["player_id"], player_name_snapshot=row["player_name"], nationality_snapshot=row.get("nationality") or "", position_snapshot=row.get("position") or "", appearances=row.get("appearances") or 0, starts=row.get("starts") or 0, minutes=row.get("minutes") or 0, goals=row.get("goals") or 0, assists=row.get("assists") or 0, clean_sheets=row.get("clean_sheets") or 0, yellow_cards=row.get("yellow_cards") or 0, red_cards=row.get("red_cards") or 0, motm_awards=row.get("motm_awards") or 0, xg=Decimal(str(row.get("xg") or 0)), xa=Decimal(str(row.get("xa") or 0)), average_rating=Decimal(str(row["average_rating"])) if row.get("average_rating") is not None else None, club_breakdown=row.get("club_breakdown") or {}, competition_breakdown=row.get("competition_breakdown") or {}, statistic_payload={k: v for k, v in row.items() if k != "statistic_hash"}, statistic_hash=row["statistic_hash"], ) for row in player_rows ]) OfficialHonor.objects.bulk_create([ OfficialHonor( projection=projection, honor_key=row["honor_key"], honor_type=row["honor_type"], entity_type=row["entity_type"], club_id=row.get("club_id"), player_id=row.get("player_id"), entity_reference=row["entity_reference"], entity_name_snapshot=row["entity_name_snapshot"], competition_archive_type=row.get("competition_archive_type") or "", competition_source_id=row.get("competition_source_id") or "", rank=row.get("rank") or 1, metric_key=row.get("metric_key") or "", metric_value=Decimal(str(row["metric_value"])) if row.get("metric_value") is not None else None, evidence_payload=row.get("evidence_payload") or {}, honor_hash=row["honor_hash"], ) for row in honors ]) OfficialHistoricalRecord.objects.bulk_create([ OfficialHistoricalRecord( projection=projection, metric_key=row["metric_key"], scope_key=row["scope_key"], entity_type=row["entity_type"], club_id=row.get("club_id"), player_id=row.get("player_id"), entity_reference=row["entity_reference"], entity_name_snapshot=row["entity_name_snapshot"], value=Decimal(str(row["value"])), unit=row["unit"], relation=row["relation"], previous_record=row.get("previous_record"), evidence_payload=row.get("evidence_payload") or {}, record_hash=row["record_hash"], ) for row in record_rows ]) from regulatory_evidence.services import RegulatoryEvidenceBridgeService RegulatoryEvidenceBridgeService.capture_official_history_projection(projection=projection, actor=ratification.ratified_by) return projection @classmethod def latest_for_season(cls, season_id): return OfficialHistoryProjection.objects.filter(season_id=season_id).order_by("-sequence").first() @classmethod def reproject_downstream(cls, *, after_season): ratifications = ( SeasonRatificationRecord.objects.filter(season__start_date__gt=after_season.start_date) .select_related("season") .order_by("season__start_date", "season_id") ) built = [] for ratification in ratifications: built.append(cls.project(ratification_id=ratification.pk)) return built class OfficialHistoryIntegrityService: @classmethod def verify_projection(cls, projection): errors = [] expected_projection_hash = stable_hash(projection.projection_payload) if expected_projection_hash != projection.projection_hash: errors.append("PROJECTION_HASH_MISMATCH") if (projection.projection_payload or {}).get("history_context_hash") != projection.history_context_hash: errors.append("HISTORY_CONTEXT_HASH_MISMATCH") if stable_hash((projection.projection_payload or {}).get("history_context") or []) != projection.history_context_hash: errors.append("HISTORY_CONTEXT_PAYLOAD_HASH_MISMATCH") try: snapshot, source_meta = OfficialHistoryProjectionService.effective_sporting_snapshot( projection.ratification, chain_head=projection.source_chain_head_hash ) except Exception as exc: errors.append(f"SOURCE_SNAPSHOT_ERROR:{type(exc).__name__}") snapshot = None source_meta = None if source_meta is not None: if source_meta["source_chain_head_hash"] != projection.source_chain_head_hash: errors.append("SOURCE_CHAIN_HEAD_MISMATCH") if source_meta["effective_snapshot_hash"] != projection.source_snapshot_hash: errors.append("SOURCE_SNAPSHOT_HASH_MISMATCH") for row in projection.competitions.all(): payload = { "archive_type": row.archive_type, "source_id": row.source_id, "label": row.label, "competition_type": row.competition_type, "competition_kind": row.competition_kind, "country": row.country, "level": row.level, "status": row.status, "champion_club_id": str(row.champion_club_id) if row.champion_club_id else None, "runner_up_club_id": str(row.runner_up_club_id) if row.runner_up_club_id else None, "champion_reference": row.champion_reference, "runner_up_reference": row.runner_up_reference, "standings_payload": row.standings_payload, "source_evidence_hash": row.source_evidence_hash, } if stable_hash(payload) != row.archive_hash: errors.append(f"COMPETITION_ARCHIVE_HASH_MISMATCH:{row.pk}") for row in projection.standing_rows.all(): payload = {"competition_source_id": row.competition_archive.source_id, **row.row_payload} if stable_hash(payload) != row.row_hash: errors.append(f"STANDING_ROW_HASH_MISMATCH:{row.pk}") for row in projection.matches.all(): if stable_hash(row.evidence_payload) != row.archive_hash: errors.append(f"MATCH_ARCHIVE_HASH_MISMATCH:{row.pk}") if row.result_certification_id: try: _verified_result_certification_for_archive(row) except OfficialHistoryIntegrityError: errors.append(f"RESULT_CERTIFICATION_INVALID:{row.pk}") for row in projection.club_statistics.all(): if stable_hash(row.statistic_payload) != row.statistic_hash: errors.append(f"CLUB_STAT_HASH_MISMATCH:{row.pk}") for row in projection.player_statistics.all(): if stable_hash(row.statistic_payload) != row.statistic_hash: errors.append(f"PLAYER_STAT_HASH_MISMATCH:{row.pk}") for row in projection.honors.all(): payload = { "honor_key": row.honor_key, "honor_type": row.honor_type, "entity_type": row.entity_type, "entity_reference": row.entity_reference, "entity_name_snapshot": row.entity_name_snapshot, "club_id": str(row.club_id) if row.club_id else None, "player_id": str(row.player_id) if row.player_id else None, "competition_archive_type": row.competition_archive_type, "competition_source_id": row.competition_source_id, "rank": row.rank, "metric_key": row.metric_key, "metric_value": _decimal_str(row.metric_value, 4) if row.metric_value is not None else None, "evidence_payload": row.evidence_payload, } if stable_hash(payload) != row.honor_hash: errors.append(f"HONOR_HASH_MISMATCH:{row.pk}") for row in projection.records.all(): payload = { "metric_key": row.metric_key, "scope_key": row.scope_key, "entity_type": row.entity_type, "entity_reference": row.entity_reference, "entity_name_snapshot": row.entity_name_snapshot, "club_id": str(row.club_id) if row.club_id else None, "player_id": str(row.player_id) if row.player_id else None, "value": _decimal_str(row.value, 4), "unit": row.unit, "evidence_payload": row.evidence_payload, "relation": row.relation, "previous_record_id": str(row.previous_record_id) if row.previous_record_id else None, } if stable_hash(payload) != row.record_hash: errors.append(f"HISTORICAL_RECORD_HASH_MISMATCH:{row.pk}") return {"valid": not errors, "errors": errors} @classmethod def verify_hall_of_fame_entry(cls, entry): errors = [] if stable_hash(entry.evidence_payload) != entry.evidence_hash: errors.append("HALL_OF_FAME_EVIDENCE_HASH_MISMATCH") source_projection_hash = (entry.evidence_payload or {}).get("projection_hash") if source_projection_hash != entry.source_projection.projection_hash: errors.append("HALL_OF_FAME_SOURCE_PROJECTION_HASH_MISMATCH") body = { "entity_type": entry.entity_type, "club_id": str(entry.club_id) if entry.club_id else None, "player_id": str(entry.player_id) if entry.player_id else None, "entity_name_snapshot": entry.entity_name_snapshot, "inducted_season_id": str(entry.inducted_season_id), "source_projection_id": str(entry.source_projection_id), "source_projection_hash": entry.source_projection.projection_hash, "induction_reason": entry.induction_reason, "evidence_hash": entry.evidence_hash, "inducted_by_id": str(entry.inducted_by_id), } if stable_hash(body) != entry.entry_hash: errors.append("HALL_OF_FAME_ENTRY_HASH_MISMATCH") return {"valid": not errors, "errors": errors} class HallOfFameService: @classmethod @transaction.atomic def induct(cls, *, projection, actor, entity_type, club=None, player=None, reason, evidence_ids=None): projection = OfficialHistoryProjection.objects.select_related("ratification__mother_federation", "season").get(pk=projection.pk) FederationService.require_role( federation=projection.ratification.mother_federation, user=actor, roles=[FederationAdminRole.GOVERNOR], ) eligible_projection_ids = [] latest_by_season = {} for row in OfficialHistoryProjection.objects.filter( season__start_date__lte=projection.season.start_date ).select_related("season").order_by("season__start_date", "season_id", "sequence"): latest_by_season[row.season_id] = row.pk eligible_projection_ids = list(latest_by_season.values()) if entity_type == HallOfFameEntityType.CLUB: if club is None or player is not None: raise OfficialHistoryProjectionError("Club Hall of Fame induction requires exactly one club.") entity_name = club.name official_honors = list(OfficialHonor.objects.filter( projection_id__in=eligible_projection_ids, club=club ).values("id", "honor_type", "honor_hash", "projection_id")) official_records = list(OfficialHistoricalRecord.objects.filter( projection_id__in=eligible_projection_ids, club=club ).values("id", "metric_key", "record_hash", "projection_id")) elif entity_type == HallOfFameEntityType.PLAYER: if player is None or club is not None: raise OfficialHistoryProjectionError("Player Hall of Fame induction requires exactly one player.") entity_name = f"{player.first_name} {player.last_name}".strip() official_honors = list(OfficialHonor.objects.filter( projection_id__in=eligible_projection_ids, player=player ).values("id", "honor_type", "honor_hash", "projection_id")) official_records = list(OfficialHistoricalRecord.objects.filter( projection_id__in=eligible_projection_ids, player=player ).values("id", "metric_key", "record_hash", "projection_id")) else: raise OfficialHistoryProjectionError("Unsupported Hall of Fame entity type.") if not str(reason or "").strip(): raise OfficialHistoryProjectionError("Hall of Fame induction requires a reason.") if not official_honors and not official_records: raise OfficialHistoryProjectionError("Hall of Fame induction requires ratified honor or record evidence in the source projection.") evidence = { "projection_id": str(projection.pk), "projection_hash": projection.projection_hash, "honors": official_honors, "records": official_records, "requested_evidence_ids": sorted(str(x) for x in (evidence_ids or [])), } evidence_hash = stable_hash(evidence) body = { "entity_type": entity_type, "club_id": str(club.pk) if club else None, "player_id": str(player.pk) if player else None, "entity_name_snapshot": entity_name, "inducted_season_id": str(projection.season_id), "source_projection_id": str(projection.pk), "source_projection_hash": projection.projection_hash, "induction_reason": str(reason).strip(), "evidence_hash": evidence_hash, "inducted_by_id": str(actor.pk), } entry = OfficialHallOfFameEntry.objects.create( entity_type=entity_type, club=club, player=player, entity_name_snapshot=entity_name, inducted_season=projection.season, source_projection=projection, induction_reason=body["induction_reason"], evidence_payload=evidence, evidence_hash=evidence_hash, entry_hash=stable_hash(body), inducted_by=actor, ) from regulatory_evidence.services import RegulatoryEvidenceBridgeService RegulatoryEvidenceBridgeService.capture_hall_of_fame_entry(entry=entry, actor=actor) return entry class OfficialMatchStatisticsService: @classmethod def get(cls, archive): if archive.result_certification_id: _verified_result_certification_for_archive(archive) if archive.official_match_report_id: from match_officials_governance.models import OfficialMatchReport try: official_report = OfficialMatchReport.objects.get( pk=archive.official_match_report_id, match_id=archive.match_id ) except OfficialMatchReport.DoesNotExist as exc: raise OfficialHistoryIntegrityError( "Ratified Official Match Report evidence is no longer available." ) from exc if ( official_report.report_hash != archive.official_match_report_hash or official_report.ratification_hash != archive.official_match_report_ratification_hash or stable_hash(official_report.report_payload) != official_report.report_hash ): raise OfficialHistoryIntegrityError( "Ratified Official Match Report evidence failed hash verification." ) from game_statistics.models import ( AdvancedMatchPlayerStatistics, AdvancedMatchTeamStatistics, MatchAnalyticsRun, MatchPlayerStatistics, MatchTeamStatistics, ) from matches.models import MatchSimulationReport if archive.analytics_run_id: try: run = MatchAnalyticsRun.objects.get(pk=archive.analytics_run_id, match_id=archive.match_id) except MatchAnalyticsRun.DoesNotExist as exc: raise OfficialHistoryIntegrityError("Ratified analytics evidence is no longer available.") from exc if run.output_hash != archive.analytics_output_hash or payload_sha256(run.analytics_payload) != run.output_hash: raise OfficialHistoryIntegrityError("Ratified analytics evidence failed hash verification.") teams = list(AdvancedMatchTeamStatistics.objects.filter(analytics_run=run).order_by("club_id").values( "club_id", "goals_for", "goals_against", "yellow_cards", "red_cards", "substitutions", "possession_pct", "xg", "shots", "shots_on_target", "big_chances", "passes_attempted", "passes_completed", "pass_accuracy_pct", "tackles_attempted", "tackles_won", "interceptions", "saves", "corners", "fouls", )) players = list(AdvancedMatchPlayerStatistics.objects.filter(analytics_run=run).order_by("club_id", "player_id").values( "player_id", "club_id", "started", "minutes_played", "goals", "assists", "clean_sheet", "yellow_cards", "red_cards", "substituted_in", "substituted_out", "injured", "xg", "xa", "shots", "shots_on_target", "big_chances", "passes_attempted", "passes_completed", "key_passes", "tackles_attempted", "tackles_won", "interceptions", "saves", "dribbles_attempted", "dribbles_completed", "rating", "is_man_of_the_match", )) return { "source": "ADVANCED_ANALYTICS", "match_id": str(archive.match_id), "analytics_run_id": str(run.pk), "analytics_output_hash": run.output_hash, "team_statistics": teams, "player_statistics": players, } if archive.simulation_report_id: try: report = MatchSimulationReport.objects.get(pk=archive.simulation_report_id, run__snapshot__match_id=archive.match_id) except MatchSimulationReport.DoesNotExist as exc: raise OfficialHistoryIntegrityError("Ratified match-report evidence is no longer available.") from exc if report.report_hash != archive.simulation_report_hash or payload_sha256(report.report_payload) != report.report_hash: raise OfficialHistoryIntegrityError("Ratified match report failed hash verification.") teams = list(MatchTeamStatistics.objects.filter(match_id=archive.match_id).order_by("club_id").values( "club_id", "goals_for", "goals_against", "clean_sheet", "yellow_cards", "red_cards", "substitutions", )) players = list(MatchPlayerStatistics.objects.filter(match_id=archive.match_id).order_by("club_id", "player_id").values( "player_id", "club_id", "started", "minutes_played", "goals", "clean_sheet", "yellow_cards", "red_cards", "substituted_in", "substituted_out", "injured", )) return { "source": "SIMULATION_REPORT", "match_id": str(archive.match_id), "simulation_report_id": str(report.pk), "simulation_report_hash": report.report_hash, "team_statistics": teams, "player_statistics": players, } return { "source": "RESULT_ONLY", "match_id": str(archive.match_id), "team_statistics": [], "player_statistics": [], } class OfficialHistorySelector: @staticmethod def latest_projection_ids(): latest = {} for row in OfficialHistoryProjection.objects.order_by("season_id", "sequence").values("id", "season_id", "sequence"): latest[row["season_id"]] = row["id"] return list(latest.values()) @classmethod def current_records(cls): return OfficialHistoricalRecord.objects.filter(projection_id__in=cls.latest_projection_ids()) def consume_domain_event(event_id): """Idempotent outbox consumer that projects newly ratified official history.""" event = OutboxEvent.objects.get(pk=event_id) if event.topic != "season.ratification": return {"event_id": str(event.pk), "handled": False} if event.event_type not in {"season.ratification.ratified", "season.ratification.addendum_issued"}: return {"event_id": str(event.pk), "handled": False} ratification_id = (event.payload or {}).get("ratification_id") or event.aggregate_id target_head = None if event.event_type == "season.ratification.ratified": ratification = SeasonRatificationRecord.objects.get(pk=ratification_id) target_head = ratification.official_hash elif event.event_type == "season.ratification.addendum_issued": target_head = (event.payload or {}).get("addendum_hash") projection = OfficialHistoryProjectionService.project( ratification_id=ratification_id, source_event_id=event.pk, source_chain_head_hash=target_head, ) downstream = [] if event.event_type == "season.ratification.addendum_issued": downstream = OfficialHistoryProjectionService.reproject_downstream(after_season=projection.season) return { "event_id": str(event.pk), "handled": True, "projection_id": str(projection.pk), "downstream_projection_ids": [str(row.pk) for row in downstream], }