""" Suppression engine: applies validation results to actual forecast data. Modifies rm_forecast_cache.forecast_data to suppress invalid player props. Handles rollback (unpublish) when previously-valid props become invalid. """ import json import logging from datetime import datetime, timezone from typing import List, Dict, Any, Tuple import psycopg2.extras from .enums import ValidationState, AuditAction from .models import PropEligibility, AuditLogEntry from .repositories import PropEligibilityRepo, AuditLogRepo from .normalization import normalize_player_name log = logging.getLogger("pub-integrity") def suppress_ineligible_props( conn, eligibilities: List[PropEligibility], run_id: str, dry_run: bool = False, ) -> Dict[str, Any]: """ Apply suppression to forecast data based on eligibility results. For each event with suppressed props: 1. Load forecast_data from rm_forecast_cache 2. Filter prop_highlights to only include eligible players 3. Store suppressed props in a separate field for audit 4. Update the forecast_data JSONB Returns stats about what was suppressed. """ stats = { "total_props": 0, "verified": 0, "suppressed": 0, "pending_lineup": 0, "events_modified": 0, "suppressions": [], } # Group eligibilities by event by_event: Dict[str, List[PropEligibility]] = {} for elig in eligibilities: by_event.setdefault(elig.game_id, []).append(elig) cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) for event_id, event_eligs in by_event.items(): # Load current forecast data cur.execute(""" SELECT id, forecast_data FROM rm_forecast_cache WHERE event_id = %s """, (event_id,)) row = cur.fetchone() if not row: continue fd = row["forecast_data"] or {} prop_highlights = fd.get("prop_highlights", []) if not prop_highlights: continue # Build suppression lookup suppress_names = set() for elig in event_eligs: stats["total_props"] += 1 if elig.validation_state.is_suppressed: suppress_names.add(normalize_player_name(elig.player_name)) stats["suppressed"] += 1 stats["suppressions"].append({ "player": elig.player_name, "event_id": event_id, "reason": elig.suppress_reason, "detail": elig.suppress_reason_human, }) elif elig.validation_state == ValidationState.VERIFIED_PENDING_LINEUP: stats["pending_lineup"] += 1 if not elig.publish_allowed: suppress_names.add(normalize_player_name(elig.player_name)) stats["suppressed"] += 1 else: stats["verified"] += 1 if not suppress_names: continue # Filter prop_highlights verified_props = [] suppressed_props = [] for prop in prop_highlights: pname = normalize_player_name(prop.get("player", "")) if pname in suppress_names: suppressed_props.append(prop) else: verified_props.append(prop) if not suppressed_props: continue if dry_run: log.info(f"[DRY RUN] Would suppress {len(suppressed_props)} props in {event_id}") continue # Update forecast_data: replace prop_highlights, store suppressed for audit fd["prop_highlights"] = verified_props fd["_suppressed_props"] = suppressed_props fd["_suppression_run_id"] = run_id fd["_suppression_at"] = datetime.now(timezone.utc).isoformat() cur.execute(""" UPDATE rm_forecast_cache SET forecast_data = %s WHERE id = %s """, (json.dumps(fd), row["id"])) stats["events_modified"] += 1 log.info( f"[suppress] {event_id}: kept {len(verified_props)} props, " f"suppressed {len(suppressed_props)}" ) # Audit log for each suppression for elig in event_eligs: if elig.validation_state.is_suppressed or ( elig.validation_state == ValidationState.VERIFIED_PENDING_LINEUP and not elig.publish_allowed ): prior = PropEligibilityRepo.get_prior_state(conn, event_id, elig.player_id) AuditLogRepo.insert(conn, AuditLogEntry( run_id=run_id, game_id=event_id, player_id=elig.player_id, player_name=elig.player_name, team_id=elig.team_id, league=elig.league, action=AuditAction.SUPPRESSED, validation_state=elig.validation_state, suppress_reason=elig.suppress_reason, suppress_reason_human=elig.suppress_reason_human, winning_source=elig.winning_source, conflict_payload=elig.conflict_payload, prior_state=prior, )) cur.close() return stats def restore_props_if_valid( conn, eligibilities: List[PropEligibility], run_id: str, ) -> int: """ Restore previously-suppressed props that are now valid. Returns count of restored props. """ restored = 0 cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) by_event: Dict[str, List[PropEligibility]] = {} for elig in eligibilities: if elig.validation_state == ValidationState.VERIFIED: by_event.setdefault(elig.game_id, []).append(elig) for event_id, event_eligs in by_event.items(): cur.execute(""" SELECT id, forecast_data FROM rm_forecast_cache WHERE event_id = %s """, (event_id,)) row = cur.fetchone() if not row: continue fd = row["forecast_data"] or {} suppressed = fd.get("_suppressed_props", []) if not suppressed: continue # Check which suppressed props are now verified verified_names = {normalize_player_name(e.player_name) for e in event_eligs} to_restore = [] still_suppressed = [] for prop in suppressed: pname = normalize_player_name(prop.get("player", "")) if pname in verified_names: to_restore.append(prop) else: still_suppressed.append(prop) if not to_restore: continue # Restore props current_props = fd.get("prop_highlights", []) fd["prop_highlights"] = current_props + to_restore fd["_suppressed_props"] = still_suppressed fd["_last_restore_at"] = datetime.now(timezone.utc).isoformat() cur.execute(""" UPDATE rm_forecast_cache SET forecast_data = %s WHERE id = %s """, (json.dumps(fd), row["id"])) restored += len(to_restore) log.info(f"[restore] {event_id}: restored {len(to_restore)} previously suppressed props") # Audit for prop in to_restore: AuditLogRepo.insert(conn, AuditLogEntry( run_id=run_id, game_id=event_id, player_id=normalize_player_name(prop.get("player", "")), player_name=prop.get("player", ""), team_id="", league=fd.get("league", ""), action=AuditAction.PUBLISHED, validation_state=ValidationState.VERIFIED, prior_state="SUPPRESSED", )) cur.close() return restored