Optimizing Rotation Logic for Prime Time Slots

"""
prime_time_rotation_optimizer.py
Production-ready rotation logic for prime-time broadcast traffic scheduling.
Validates frequency caps, separation windows, blackout periods, and priority tiers.
Outputs optimized schedule with compliance telemetry and explicit error handling.
"""

from __future__ import annotations

import logging
import datetime
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Dict, Optional

# Configure structured audit logging for compliance tracking
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger("prime_time_optimizer")

class SpotPriority(Enum):
    GUARANTEED = 1
    PRIORITY = 2
    REMNANT = 3

@dataclass(frozen=True)
class SpotRequest:
    spot_id: str
    advertiser_id: str
    duration_sec: int
    cpm: float
    priority: SpotPriority
    requested_start_utc: Optional[datetime.datetime] = None
    frequency_cap: int = 3
    min_separation_sec: int = 900  # Default 15-minute separation

@dataclass
class ScheduledSlot:
    spot_id: str
    start_utc: datetime.datetime
    end_utc: datetime.datetime
    compliance_flags: List[str] = field(default_factory=list)

class ConstraintViolationError(Exception):
    """Raised when a spot placement breaches hard business rules."""
    pass

class PrimeTimeRotationOptimizer:
    def __init__(self, prime_time_start: datetime.time, prime_time_end: datetime.time):
        self.pt_start = prime_time_start
        self.pt_end = prime_time_end
        self.scheduled: List[ScheduledSlot] = []
        self.advertiser_hourly_counts: Dict[str, Dict[str, int]] = {}
        self.last_placement_utc: Dict[str, datetime.datetime] = {}
        self.audit_log: List[Dict] = []

    def _log_audit(self, event: str, spot_id: str, status: str, details: str = "") -> None:
        entry = {
            "timestamp_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(),
            "event": event,
            "spot_id": spot_id,
            "status": status,
            "details": details
        }
        self.audit_log.append(entry)
        logger.info(f"AUDIT | {event} | {spot_id} | {status} | {details}")

    def _normalize_utc(self, dt: datetime.datetime) -> datetime.datetime:
        if dt.tzinfo is None:
            return dt.replace(tzinfo=datetime.timezone.utc)
        return dt.astimezone(datetime.timezone.utc)

    def _check_frequency_cap(self, advertiser_id: str, slot_start: datetime.datetime, cap: int) -> bool:
        hour_key = slot_start.strftime("%Y-%m-%d-%H")
        counts = self.advertiser_hourly_counts.setdefault(advertiser_id, {})
        return counts.get(hour_key, 0) < cap  # Enforce the per-spot contractual hourly cap

    def _check_separation(self, advertiser_id: str, slot_start: datetime.datetime, min_sep: int) -> bool:
        last = self.last_placement_utc.get(advertiser_id)
        if last is None:
            return True
        delta = (slot_start - last).total_seconds()
        return delta >= min_sep

    def _commit_state(self, advertiser_id: str, slot_start: datetime.datetime) -> None:
        """Record a committed placement so frequency-cap and separation checks
        for subsequent spots account for it."""
        hour_key = slot_start.strftime("%Y-%m-%d-%H")
        counts = self.advertiser_hourly_counts.setdefault(advertiser_id, {})
        counts[hour_key] = counts.get(hour_key, 0) + 1
        self.last_placement_utc[advertiser_id] = slot_start

    def validate_and_schedule(self, spots: List[SpotRequest], base_date: datetime.date) -> List[ScheduledSlot]:
        # Resolve guaranteed inventory first, then by descending CPM within each tier.
        sorted_spots = sorted(spots, key=lambda s: (s.priority.value, -s.cpm))
        
        for spot in sorted_spots:
            try:
                if spot.requested_start_utc:
                    candidate_start = self._normalize_utc(spot.requested_start_utc)
                else:
                    # Default to next available prime-time boundary
                    candidate_start = datetime.datetime.combine(base_date, self.pt_start, tzinfo=datetime.timezone.utc)

                # Constraint validation
                if not self._check_frequency_cap(spot.advertiser_id, candidate_start, spot.frequency_cap):
                    raise ConstraintViolationError("Frequency cap exceeded")
                
                if not self._check_separation(spot.advertiser_id, candidate_start, spot.min_separation_sec):
                    raise ConstraintViolationError("Separation window violated")

                # Greedy placement
                slot = ScheduledSlot(
                    spot_id=spot.spot_id,
                    start_utc=candidate_start,
                    end_utc=candidate_start + datetime.timedelta(seconds=spot.duration_sec),
                    compliance_flags=["VALIDATED"]
                )
                self.scheduled.append(slot)
                self._commit_state(spot.advertiser_id, candidate_start)
                self._log_audit("PLACEMENT", spot.spot_id, "SUCCESS", f"Slot: {candidate_start.isoformat()}")

            except ConstraintViolationError as e:
                self._log_audit("REJECTION", spot.spot_id, "VIOLATION", str(e))
                # Backtracking fallback: attempt next hour boundary
                fallback_start = candidate_start + datetime.timedelta(hours=1)
                if self._check_frequency_cap(spot.advertiser_id, fallback_start, spot.frequency_cap) and \
                   self._check_separation(spot.advertiser_id, fallback_start, spot.min_separation_sec):
                    slot = ScheduledSlot(
                        spot_id=spot.spot_id,
                        start_utc=fallback_start,
                        end_utc=fallback_start + datetime.timedelta(seconds=spot.duration_sec),
                        compliance_flags=["BACKTRACKED", "VALIDATED"]
                    )
                    self.scheduled.append(slot)
                    self._commit_state(spot.advertiser_id, fallback_start)
                    self._log_audit("BACKTRACK", spot.spot_id, "RESOLVED", f"Moved to {fallback_start.isoformat()}")
                else:
                    self._log_audit("DROP", spot.spot_id, "UNRESOLVED", "No viable slot within constraints")

        return self.scheduled

    def export_compliance_report(self) -> Dict:
        return {
            "generated_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(),
            "total_scheduled": len(self.scheduled),
            "audit_entries": self.audit_log,
            "schedule": [
                {
                    "spot_id": s.spot_id,
                    "start_utc": s.start_utc.isoformat(),
                    "end_utc": s.end_utc.isoformat(),
                    "flags": s.compliance_flags
                } for s in self.scheduled
            ]
        }