Configuring Competitive Separation Windows by Daypart

# separation_policy.yaml — reviewed and diffed like code.
dayparts:
  overnight: { start: "00:00", end: "06:00" }
  daytime:   { start: "06:00", end: "18:00" }
  prime:     { start: "18:00", end: "23:00" }
  late:      { start: "23:00", end: "24:00" }

rules:
  # Autos: tight in prime, relaxed overnight.
  - { category: auto,             daypart: prime,   scope: rolling,   min_gap_seconds: 360 }
  - { category: auto,             daypart: daytime, scope: rolling,   min_gap_seconds: 240 }
  - { category: auto,             daypart: overnight, scope: rolling, min_gap_seconds: 120 }
  # QSR: never two in one break, any daypart.
  - { category: qsr,              daypart: prime,   scope: same_break, min_gap_seconds: 0 }
  - { category: qsr,              daypart: daytime, scope: rolling,   min_gap_seconds: 180 }
  # Telco parent rule cascades to carriers unless overridden below.
  - { category: telco,            daypart: prime,   scope: rolling,   min_gap_seconds: 240 }
  - { category: wireless_carrier, daypart: prime,   scope: rolling,   min_gap_seconds: 360 }
  # Pharma sole-sponsorship in prime: at most one airing in the whole daypart.
  - { category: pharma,           daypart: prime,   scope: daypart,   min_gap_seconds: 0 }

from __future__ import annotations

import logging
from datetime import time
from enum import Enum

from pydantic import BaseModel, Field, field_validator, model_validator

# Traffic-ops structured logging: timestamp | level | module | spot_id
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("traffic.separation.config")


def _log(level: int, spot_id: str, msg: str) -> None:
    logger.log(level, "%s | %s", spot_id, msg)


class WindowScope(str, Enum):
    SAME_BREAK = "same_break"    # No two of the category in one break
    ROLLING = "rolling"          # Minimum elapsed gap between airings
    DAYPART = "daypart"          # At most one airing in the whole daypart


class DaypartBoundary(BaseModel):
    name: str
    start: time                  # Broadcast-local start, inclusive
    end: time                    # Broadcast-local end, exclusive

    @model_validator(mode="after")
    def _ordered(self) -> "DaypartBoundary":
        # A daypart that ends before it starts would swallow the wrong spots.
        # "24:00" is modelled as time.max upstream; equal start/end is empty.
        if self.start >= self.end:
            raise ValueError(f"daypart '{self.name}' start must precede end")
        return self


class SeparationRuleConfig(BaseModel):
    category: str
    daypart: str
    scope: WindowScope
    min_gap_seconds: int = Field(ge=0)

    @field_validator("category")
    @classmethod
    def _lower(cls, v: str) -> str:
        return v.lower()

    @model_validator(mode="after")
    def _coherent_gap(self) -> "SeparationRuleConfig":
        # A rolling window with a zero gap forbids nothing — almost always an
        # authoring slip. same_break and daypart scopes legitimately use 0.
        if self.scope is WindowScope.ROLLING and self.min_gap_seconds == 0:
            raise ValueError(
                f"rolling rule for {self.category}/{self.daypart} has zero gap"
            )
        return self

    @property
    def min_gap_ms(self) -> int:
        return self.min_gap_seconds * 1000


class SeparationPolicy(BaseModel):
    dayparts: list[DaypartBoundary]
    rules: list[SeparationRuleConfig]

def load_policy(
    raw: dict, known_categories: set[str]
) -> SeparationPolicy:
    policy = SeparationPolicy.model_validate(raw)  # Pydantic type-checks rows.

    daypart_names = {d.name for d in policy.dayparts}
    seen: set[tuple[str, str]] = set()
    for rule in policy.rules:
        # 1. Orphan rule: category not in the taxonomy — would never match.
        if rule.category not in known_categories:
            raise ValueError(
                f"ERR_ORPHAN_RULE: '{rule.category}' not in taxonomy"
            )
        # 2. Unknown daypart: rule keyed to a part the calendar never emits.
        if rule.daypart not in daypart_names:
            raise ValueError(
                f"ERR_UNKNOWN_DAYPART: '{rule.daypart}' not in calendar"
            )
        # 3. Duplicate cell: two rules for one (category, daypart) is ambiguous.
        key = (rule.category, rule.daypart)
        if key in seen:
            raise ValueError(f"ERR_RULE_DUPLICATE: {key} defined twice")
        seen.add(key)

    logger.info("separation policy loaded | %d dayparts | %d rules",
                len(policy.dayparts), len(policy.rules))
    return policy

from datetime import datetime


def resolve_daypart(starts_at: datetime, policy: SeparationPolicy) -> str:
    # Compare the broadcast-local wall-clock time against each boundary.
    # starts_at is timezone-aware; the caller localises to broadcast time first.
    t = starts_at.timetz().replace(tzinfo=None)
    for boundary in policy.dayparts:
        if boundary.start <= t < boundary.end:
            return boundary.name
    # No matching part means the calendar has a hole — fail loudly, never guess.
    raise ValueError(f"ERR_DAYPART_GAP: {t} matches no configured daypart")

class ResolvedWindow(BaseModel):
    rule_id: str
    category_code: str
    daypart: str
    scope: WindowScope
    min_gap_ms: int


def build_index(
    policy: SeparationPolicy,
) -> dict[tuple[str, str], SeparationRuleConfig]:
    # (category, daypart) -> rule, built once and reused across the run.
    return {(r.category, r.daypart): r for r in policy.rules}


def resolve_window(
    category_code: str,
    daypart: str,
    index: dict[tuple[str, str], SeparationRuleConfig],
    parent_of: dict[str, str | None],
) -> ResolvedWindow | None:
    # Walk from the exact category up the taxonomy until a rule matches.
    current: str | None = category_code.lower()
    while current is not None:
        rule = index.get((current, daypart))
        if rule is not None:
            _log(logging.INFO, category_code,
                 f"resolved {rule.scope.value} window {rule.min_gap_seconds}s "
                 f"via {current}/{daypart}")
            return ResolvedWindow(
                rule_id=f"RULE-{current.upper()}-{daypart.upper()}",
                category_code=current,
                daypart=daypart,
                scope=rule.scope,
                min_gap_ms=rule.min_gap_ms,
            )
        current = parent_of.get(current)  # Climb to the parent group.
    # No rule anywhere in the ancestry means this category is unrestricted here.
    _log(logging.INFO, category_code, f"no rule for {daypart} | unrestricted")
    return None

KNOWN = {"auto", "qsr", "telco", "wireless_carrier", "pharma"}
PARENT_OF = {"wireless_carrier": "telco", "telco": None, "auto": None,
             "qsr": None, "pharma": None}

RAW = {
    "dayparts": [
        {"name": "daytime", "start": "06:00", "end": "18:00"},
        {"name": "prime",   "start": "18:00", "end": "23:00"},
    ],
    "rules": [
        {"category": "telco",            "daypart": "prime", "scope": "rolling", "min_gap_seconds": 240},
        {"category": "wireless_carrier", "daypart": "prime", "scope": "rolling", "min_gap_seconds": 360},
        {"category": "pharma",           "daypart": "prime", "scope": "daypart", "min_gap_seconds": 0},
    ],
}

policy = load_policy(RAW, KNOWN)
index = build_index(policy)

# 1. Override wins: a carrier uses its own 360s rule, not the telco 240s parent.
w = resolve_window("wireless_carrier", "prime", index, PARENT_OF)
assert w is not None and w.min_gap_ms == 360_000
assert w.category_code == "wireless_carrier"

# 2. Cascade: a telco with no carrier match still resolves the parent rule.
w2 = resolve_window("telco", "prime", index, PARENT_OF)
assert w2 is not None and w2.min_gap_ms == 240_000

# 3. Fail-closed: an orphan rule is rejected on load, never reaches resolution.
try:
    load_policy({"dayparts": RAW["dayparts"],
                 "rules": [{"category": "crypto", "daypart": "prime",
                            "scope": "rolling", "min_gap_seconds": 300}]}, KNOWN)
    raise AssertionError("orphan rule should have failed to load")
except ValueError as exc:
    assert "ERR_ORPHAN_RULE" in str(exc)