Broadcast Traffic Scheduling Automation

How to Map Legacy Spot IDs to Modern Schemas

Legacy traffic systems mint spot identifiers in shapes that modern, API-driven advertising stacks refuse to trust: arbitrary alphanumeric strings, vendor-specific hashes, zero-padded integers that collide the moment two source systems overlap. This guide solves one exact task — deterministically transforming a legacy spot ID into a canonical modern identifier (a namespace-bound UUIDv5) and a validated metadata record, with explicit quarantine routing and a circuit breaker for records that cannot be resolved. It is the migration procedure that sits under Understanding Broadcast Spot Schemas and Metadata, itself a subsystem of Broadcast Traffic Architecture & Taxonomy. Getting it right is not cosmetic: a spot ID is the primary key that joins scheduling, playout, and billing, so a mapping that duplicates or drifts fractures billing lineage and leaves an audit trail that fails FCC public-inspection and SOC 2 reproducibility scrutiny.

The core requirement is idempotency. The same legacy record re-exported tomorrow must resolve to the same modern ID, or ingestion double-books inventory and posts revenue twice. That rules out auto-increment surrogate keys — which encode arrival order rather than identity — in favour of a deterministic hash over the record’s stable business attributes, validated against current traffic rules and billing code normalization before anything is committed downstream.

Prerequisites

  • Python 3.11+ — required for the X | None union syntax, frozenset typing, and timezone-aware datetime.now(timezone.utc) used below.
  • Standard library only — uuid, hashlib, json, logging, and pathlib cover the mapper; no third-party dependency is needed for the transformation itself. If you persist a processed-ID ledger, pin your driver exactly (psycopg[binary]==3.2.1 or redis==5.0.4).
  • A resolvable namespace — one stable RFC 4122 namespace UUID per traffic environment, held in config so staging and production never generate the same modern ID from the same legacy input.
  • Read-only access to the legacy source — a mounted export or a DB role scoped to SELECT only; the mapper must never mutate the system it is migrating from.
  • A writable quarantine directory — a local path or object-store prefix where unresolved records are isolated for traffic-desk review.
  • Upstream normalization applied — billing codes standardized per billing code normalization and payloads shaped to the canonical spot schema so the mapper validates identity, not field structure.

Step-by-Step Implementation

The mapper runs as a pure boundary function over each record: normalize the legacy ID, derive a deterministic modern ID, validate the billing code, and either emit a canonical mapping or quarantine the record. A circuit breaker halts the batch once consecutive failures cross a threshold, so a corrupted export never floods the scheduler with malformed writes.

Legacy-to-modern spot ID mapping flow with circuit breaker A legacy spot ID arrives in three casing and whitespace variants. It first passes a circuit breaker gate that decides whether to admit the request; if the breaker is open the batch is paused with no partial writes. Admitted records enter the stateless mapper, which normalizes the ID to one canonical identity, hashes it into a deterministic namespace-bound UUIDv5, and validates the billing code, failing closed on anything shorter than four alphanumeric characters. A valid record becomes a Modern Spot Mapping with status SUCCESS and a canonical UUIDv5; an invalid one is written to quarantine with status FAILED. Each failure feeds record_failure back to the breaker, incrementing its counter and tripping it open once consecutive failures cross the threshold. Circuit Breaker OPEN PAUSED · no partial writes Legacy Spot ID case / whitespace / newline CRM-88213 ··crm-88213· CRM-88213↵ Circuit Breaker allow_request() Legacy Spot Mapper · pure fn 1 Normalize ID collapse case · ws · ctrl chars 2 Generate UUIDv5 uuid5(ns, name) · deterministic 3 Validate billing code ≥ 4 alnum · else fail closed valid? Modern Spot Mapping status = SUCCESS 6ba7b811-9dad-…-430c8 Quarantine status = FAILED quarantine/<id>.json valid error OPEN record_failure() → failures++ · trips breaker at threshold

Figure — Legacy-to-modern mapping flow: resolved IDs are hashed into a deterministic UUIDv5 schema, unresolved records are quarantined, and repeated failures trip the circuit breaker to halt processing.

Step 1 — Structured audit logging and record models

Goal: emit machine-parseable audit lines in the traffic-ops timestamp | level | module | spot_id shape, and fix the typed models that carry a record through the pipeline. The audit log is the compliance artifact — every mapping decision must be reconstructable from it alone.

python 
from __future__ import annotations

import json
import logging
import time
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path

# Traffic-ops structured logging: timestamp | level | module | message(spot_id=...)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[logging.FileHandler("spot_mapping_audit.log"), logging.StreamHandler()],
)
logger = logging.getLogger("traffic.migration.legacy_spot_mapper")


class MappingStatus(str, Enum):
    SUCCESS = "success"
    QUARANTINED = "quarantined"
    PAUSED = "paused"       # circuit breaker open — batch halted
    FAILED = "failed"


@dataclass(frozen=True)
class LegacySpotRecord:
    legacy_id: str
    billing_code: str | None = None
    placement_type: str | None = None
    raw_metadata: dict[str, str] | None = None


@dataclass(frozen=True)
class ModernSpotMapping:
    modern_id: str
    legacy_id: str
    billing_code: str
    namespace: str
    mapped_at: str
    status: MappingStatus
    compliance_flags: list[str] = field(default_factory=list)
    error_trace: str | None = None

Step 2 — Normalize the legacy ID, then hash it deterministically

Goal: collapse casing, whitespace, and stray control characters into one canonical string, then derive a namespace-bound UUIDv5. Identity is computed from the normalized input, so crm-88213 , CRM-88213, and CRM-88213\n all resolve to a single modern ID.

python 
class IdNormalizationError(ValueError):
    """Raised when a legacy ID cannot be reduced to a usable canonical form."""


def normalize_legacy_id(legacy_id: str) -> str:
    # Strip surrounding whitespace and a UTF-8 BOM, drop interior control chars,
    # and uppercase so vendor casing variants converge on one identity.
    cleaned = legacy_id.strip().lstrip("").upper()
    cleaned = "".join(ch for ch in cleaned if ch.isprintable() and not ch.isspace())
    if not cleaned:
        raise IdNormalizationError("legacy_id is empty after normalization")
    return cleaned


def generate_modern_id(legacy_id: str, namespace: uuid.UUID, ns_label: str) -> str:
    # UUIDv5 = SHA-1 over (namespace, name): same inputs -> same output, forever.
    # A per-environment ns_label prefix isolates staging from production IDs.
    canonical = f"{ns_label}:{normalize_legacy_id(legacy_id)}"
    modern_id = str(uuid.uuid5(namespace, canonical))
    logger.debug("normalized legacy_id=%s -> spot_id=%s", legacy_id, modern_id)
    return modern_id

Because uuid.uuid5 is a pure function of its namespace and name, re-running Step 2 on the same legacy ID is a no-op that yields byte-identical output — the property audit replay depends on. This is the same idempotency discipline the parent spot schema applies when it assigns canonical keys at the ingestion boundary.

Step 3 — Validate the billing code against traffic rules

Goal: fail closed on any record whose billing code cannot guarantee revenue attribution. A missing or malformed code is a hard rejection, not a warning, because it is the join key downstream billing uses to post revenue.

python 
class BillingValidationError(ValueError):
    """Raised when a billing code violates the canonical traffic-system format."""


def validate_billing_code(code: str | None) -> str:
    if not code:
        raise BillingValidationError("missing billing code; revenue attribution unguaranteed")
    sanitized = code.strip().upper()
    # Canonical codes are >= 4 alphanumeric characters after normalization.
    if len(sanitized) < 4 or not sanitized.isalnum():
        raise BillingValidationError(
            f"billing code '{sanitized}' violates alphanumeric/length constraints"
        )
    return sanitized

Step 4 — Guard downstream systems with a circuit breaker

Goal: protect the scheduler and billing engine from a corrupted export. After failure_threshold consecutive failures the breaker opens and the mapper stops issuing writes; after recovery_timeout it admits a single probe and closes on the first success.

python 
class CircuitBreaker:
    """State machine (closed -> open -> half-open) that halts cascading failures."""

    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 60.0) -> None:
        self.failure_threshold: int = failure_threshold
        self.recovery_timeout: float = recovery_timeout
        self.failure_count: int = 0
        self.last_failure_time: float = 0.0
        self.state: str = "closed"

    def record_failure(self) -> None:
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = "open"
            logger.warning("circuit breaker OPENED after %d consecutive failures", self.failure_count)

    def record_success(self) -> None:
        self.failure_count = 0
        self.state = "closed"

    def allow_request(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.time() - self.last_failure_time >= self.recovery_timeout:
                self.state = "half-open"
                logger.info("circuit breaker transitioning to HALF-OPEN (probe admitted)")
                return True
            return False
        return True  # half-open: admit probes until one succeeds or fails

Step 5 — Map a record and orchestrate the batch

Goal: compose the pieces into a deployable LegacySpotMapper. Each record is checked against the breaker, mapped or quarantined, and appended to the audit log; the batch stops on the first PAUSED result so no partial state is committed.

python 
class LegacySpotMapper:
    def __init__(
        self,
        namespace: uuid.UUID,
        ns_label: str = "production_traffic",
        quarantine_path: Path = Path("quarantine/"),
        failure_threshold: int = 5,
    ) -> None:
        self.namespace: uuid.UUID = namespace
        self.ns_label: str = ns_label
        self.quarantine_path: Path = quarantine_path
        self.quarantine_path.mkdir(parents=True, exist_ok=True)
        self.breaker: CircuitBreaker = CircuitBreaker(failure_threshold=failure_threshold)
        self.audit_log: list[ModernSpotMapping] = []

    def map_record(self, record: LegacySpotRecord) -> ModernSpotMapping:
        now = datetime.now(timezone.utc).isoformat()
        if not self.breaker.allow_request():
            logger.critical("circuit breaker active — halting; legacy_id=%s", record.legacy_id)
            return ModernSpotMapping("", record.legacy_id, "", self.ns_label, now,
                                     MappingStatus.PAUSED, ["CIRCUIT_BREAKER_ACTIVE"])
        try:
            billing = validate_billing_code(record.billing_code)
            modern_id = generate_modern_id(record.legacy_id, self.namespace, self.ns_label)
            mapping = ModernSpotMapping(modern_id, record.legacy_id, billing, self.ns_label,
                                        now, MappingStatus.SUCCESS, ["SCHEMA_COMPLIANT"])
            self.breaker.record_success()
            self.audit_log.append(mapping)
            logger.info("mapped legacy_id=%s -> spot_id=%s billing=%s accepted=True",
                        record.legacy_id, modern_id, billing)
            return mapping
        except (IdNormalizationError, BillingValidationError) as exc:
            self.breaker.record_failure()
            logger.error("mapping FAILED legacy_id=%s reason=%s", record.legacy_id, exc)
            self._quarantine_record(record, str(exc))
            return ModernSpotMapping("", record.legacy_id, record.billing_code or "",
                                     self.ns_label, now, MappingStatus.FAILED,
                                     ["VALIDATION_FAILURE"], str(exc))

    def _quarantine_record(self, record: LegacySpotRecord, error: str) -> None:
        target = self.quarantine_path / f"{record.legacy_id}.json"
        target.write_text(json.dumps(
            {"legacy_record": record.__dict__, "error": error,
             "quarantined_at": datetime.now(timezone.utc).isoformat()},
            indent=2, default=str,
        ))
        logger.warning("quarantined legacy_id=%s -> %s", record.legacy_id, target)

    def process_batch(self, records: list[LegacySpotRecord]) -> list[ModernSpotMapping]:
        results: list[ModernSpotMapping] = []
        for rec in records:
            res = self.map_record(rec)
            results.append(res)
            if res.status is MappingStatus.PAUSED:
                logger.critical("emergency halt — batch suspended at legacy_id=%s", rec.legacy_id)
                break
        return results

A representative accepted-path log line reads 2026-07-03T14:22:07+00:00 | INFO | traffic.migration.legacy_spot_mapper | mapped legacy_id=CRM-88213 -> spot_id=6ba7b811-9dad-11d1-80b4-00c04fd430c8 billing=RCPRIME30 accepted=True.

Verification & Testing

Correct behaviour rests on two properties: determinism (a legacy ID always maps to the same modern ID) and fail-closed validation (a bad record never produces a SUCCESS). Both are assertable against fixture data drawn from a real export.

python 
NAMESPACE = uuid.UUID("6ba7b810-9dad-11d1-80b4-00c04fd430c8")
mapper = LegacySpotMapper(namespace=NAMESPACE, ns_label="production_traffic")

# 1. Determinism: normalized variants of one legacy ID converge on one spot_id.
a = mapper.map_record(LegacySpotRecord("CRM-88213", billing_code="RC-PRIME-30"))
b = mapper.map_record(LegacySpotRecord("  crm-88213 ", billing_code="rc-prime-30"))
assert a.status is MappingStatus.SUCCESS
assert a.modern_id == b.modern_id            # idempotent across casing/whitespace
assert a.billing_code == "RCPRIME30"

# 2. Fail-closed: a short/missing billing code is quarantined, never mapped.
bad = mapper.map_record(LegacySpotRecord("CRM-99001", billing_code="X1"))
assert bad.status is MappingStatus.FAILED
assert "VALIDATION_FAILURE" in bad.compliance_flags
assert Path("quarantine/CRM-99001.json").exists()

Because generate_modern_id is pure, the canonical output for production_traffic:CRM-88213 is stable across machines and runs — replaying an archived export during an audit reproduces byte-identical spot_id values. Run the suite in CI against a golden fixture of legacy → modern pairs so a change to the normalization rules that would silently re-key inventory fails the build instead of shipping.

Edge Cases & Failure Handling

  • Unresolvable legacy ID. A record whose legacy_id is empty, whitespace-only, or all control characters raises IdNormalizationError and is quarantined with an actionable reason rather than hashed into a garbage UUID. Reconcile the quarantine file against the source system and re-run once the upstream export is corrected — the deterministic hash means the recovered record maps to the ID it always would have.
  • Billing code mismatch. A code that survives transport but violates the canonical format (too short, non-alphanumeric, or absent) fails closed in validate_billing_code. This is usually an upstream normalization gap; apply billing code normalization before the mapper rather than loosening the validator, which would let unbillable spots into the schedule.
  • Circuit breaker trip. Once failure_threshold consecutive failures accumulate, the breaker opens, map_record returns PAUSED, and process_batch halts before committing partial state. Inspect spot_mapping_audit.log for the root cause, correct the quarantined records, and restart; after recovery_timeout the breaker admits a probe and closes on the first success. The same breaker discipline governs the downstream engine described in Spot Scheduling Validation & Rule Engines, so an upstream halt and a scheduler halt read identically in the logs.

FAQ

Why UUIDv5 instead of a random UUID or an auto-increment key?

Because identity must be reproducible. A random UUIDv4 or an auto-increment integer changes every time you re-ingest the same record, so a re-exported order double-books inventory and fractures billing lineage. UUIDv5 is a pure hash of (namespace, normalized legacy_id), so the same input always yields the same output — re-ingestion is a no-op. This is the identity model the canonical spot schema assumes for its primary key.

Can I safely re-run a batch after a partial failure?

Yes — that is the point of the deterministic hash. Records that already mapped resolve to the identical spot_id on a re-run, so committing them again is idempotent. To avoid re-doing work, persist processed spot_id values to a state store (PostgreSQL or Redis) and skip any legacy ID whose modern ID is already present. Never gate re-runs on arrival order; gate them on the canonical key.

What happens to quarantined records, and how do I reconcile them?

Each failure is written to quarantine/<legacy_id>.json with the raw record, the error, and a timestamp — nothing is dropped. Cross-reference the payload against your current traffic taxonomy, fix the source data (usually a missing billing code or a malformed ID), and re-run the batch. Retain quarantine files for financial-audit purposes; archiving them to cold storage after 90 days keeps the working directory clean without losing the evidence trail.

How is the circuit breaker here related to the one in the scheduler?

They are the same pattern applied at two boundaries. This breaker protects downstream systems from a corrupted migration batch; the breaker in Spot Scheduling Validation & Rule Engines protects playout from a corrupted schedule. Both open on a consecutive-failure threshold, require operator acknowledgment, and emit the same timestamp | level | module log shape, so a single dashboard can watch migration and scheduling halts side by side.

Should I map every legacy field, or just the ID?

Map the ID deterministically and validate the billing code here; resolve the rest of the record against the full field contract in the parent spot schema using the Pydantic validator pattern. Keeping identity generation and full-record validation as separate passes means each fails for one reason, which makes the quarantine reason unambiguous.

Related content

Up one levelUnderstanding Broadcast Spot Schemas and Metadata