How to Map Vendor Fields to Canonical Names

Every traffic vendor names the same thing differently: Avion writes HouseNo, Avstar writes house_number, a legacy WideOrbit dump writes CART, and a network SCTE-35 descriptor buries the creative key inside a segmentation_upid. This guide solves one exact task — building a vendor-to-canonical field-name map, applying it to raw export rows to produce records that obey the canonical vocabulary, and flagging any unknown field instead of silently dropping it. It is the ingestion step that sits under The Canonical Traffic Field Data Dictionary, itself part of Broadcast Traffic Architecture & Taxonomy. Get the map wrong and the damage is quiet but expensive: a mis-mapped house_number double-books inventory, a dropped sponsorship_id airs sponsored content without its FCC identifier, and an unnoticed new vendor column means a rate change never reaches billing.

The discipline here is that renaming is lossless and auditable. A field is either mapped to a known canonical name, explicitly ignored by policy, or flagged as unknown for a human to reconcile — never dropped on the floor. That three-way outcome is what keeps an adapter honest when a vendor adds a column mid-quarter, and it is the same fail-closed posture the spot schema applies to identity and the Pydantic validator applies to types.

Prerequisites #

• Python 3.11+ — required for the X | None union syntax, frozenset typing, and dict ordering guarantees used throughout.
• Pydantic v2 — pin it exactly (pydantic==2.7.1); the canonical model in the parent dictionary is a Pydantic v2 BaseModel and this mapper feeds it.
• The canonical field list — the authoritative set of snake_case field names from the data dictionary; the mapper validates every target against it so a typo in the map fails fast.
• A sample export per vendor — at least one real Avion and one real Avstar row (header + values) so the map is built from observed columns, not guessed ones; the Avion vs Avstar export format comparison catalogues the known differences.
• Read-only access to the source export — a mounted file or SELECT-scoped role; the mapper must never mutate the system it reads from.
• A writable review location — a path or object-store prefix where rows with unknown fields are parked for traffic-desk reconciliation.

Step-by-Step Implementation #

The mapper is a pure boundary function over each raw row: look up each source key in the vendor map, rename it to its canonical name, coerce obvious types, and split the outcome into mapped fields plus a list of unknowns. Nothing is dropped silently, and the canonical target of every mapping is checked against the dictionary so the map itself cannot drift.

Figure — Vendor field mapping: known source keys are renamed to canonical names and validated, while any unmapped key is parked for review rather than dropped, surfacing new vendor columns.

Step 1 — Define the canonical field set and the vendor maps #

Goal: fix the authoritative set of canonical names so the mapper can reject a bad target, and declare one rename map per vendor plus an explicit ignore list. Structuring the maps as data — not if/elif branches — means adding a vendor is a config change, not a code change.

from __future__ import annotations

import logging
from dataclasses import dataclass, field

# 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.canonical.vendor_mapper")

# The authoritative canonical field names (subset shown; full set lives in the
# data dictionary). Any map target outside this set is a configuration error.
CANONICAL_FIELDS: frozenset[str] = frozenset({
    "spot_id", "house_number", "isci", "ad_id", "avail_id", "order_id",
    "campaign_id", "sponsor_id", "sponsorship_id", "scheduled_at", "aired_at",
    "duration_ms", "rate_cents", "currency", "billing_code",
    "advertiser_category", "spot_type", "is_sponsored", "political_flag",
    "source_system",
})

# vendor -> {source_column: canonical_name}. Built from observed export headers.
VENDOR_MAPS: dict[str, dict[str, str]] = {
    "avion": {
        "HouseNo": "house_number",
        "ISCI": "isci",
        "AdID": "ad_id",
        "AvailID": "avail_id",
        "OrderNo": "order_id",
        "Campaign": "campaign_id",
        "Rate": "rate_cents",          # dollars string -> cents in Step 3
        "BillCode": "billing_code",
        "Category": "advertiser_category",
        "AirTime": "scheduled_at",
    },
    "avstar": {
        "house_number": "house_number",
        "isci_code": "isci",
        "ad_id": "ad_id",
        "avail_id": "avail_id",
        "order_id": "order_id",
        "campaign_id": "campaign_id",
        "unit_rate": "rate_cents",
        "billing_code": "billing_code",
        "category": "advertiser_category",
        "scheduled_utc": "scheduled_at",
    },
}

# Columns known to be irrelevant to the canonical layer, ignored by policy.
IGNORED_COLUMNS: dict[str, frozenset[str]] = {
    "avion": frozenset({"OperatorInit", "ExportBatch"}),
    "avstar": frozenset({"row_checksum", "export_ts"}),
}

Step 2 — Validate the map itself before using it #

Goal: fail fast if a rename map points at a canonical name that does not exist — a typo like house_numbr must be caught at load time, not silently produce a record missing a field. This check runs once per process start, not per row.

class MapConfigError(ValueError):
    """Raised when a vendor map targets a name outside the canonical set."""


def validate_vendor_map(vendor: str, mapping: dict[str, str]) -> None:
    # Every target must be a real canonical field; catch typos at startup.
    bad = {src: tgt for src, tgt in mapping.items() if tgt not in CANONICAL_FIELDS}
    if bad:
        raise MapConfigError(f"{vendor} map has non-canonical targets: {bad}")
    logger.info("SYSTEM | vendor=%s map validated (%d fields)", vendor, len(mapping))


for _vendor, _map in VENDOR_MAPS.items():
    validate_vendor_map(_vendor, _map)

Expected log line on a clean load: 2026-07-17T09:00:00 | INFO | traffic.canonical.vendor_mapper | SYSTEM | vendor=avion map validated (10 fields). If a target is wrong, the process refuses to start — far cheaper than discovering the gap in a billing reconciliation weeks later.

Step 3 — Apply the map and coerce obvious types #

Goal: rename each source key to its canonical name and perform the small, deterministic coercions the dictionary requires — dollars to integer cents, blank strings to None. Anything more complex than a rename-plus-primitive-coercion is left to the canonical model’s validators, keeping this step single-purpose.

def _to_cents(raw: str) -> int:
    # Vendors emit rate as a dollars string ("2850.00"); the dictionary stores
    # integer cents to avoid float drift across thousands of airings.
    cleaned = raw.replace("$", "").replace(",", "").strip()
    return round(float(cleaned) * 100)


# Per-field coercers keyed by canonical name; absent = pass value through.
COERCERS = {"rate_cents": _to_cents}


@dataclass(frozen=True)
class MappedRow:
    canonical: dict[str, object]
    unknown: dict[str, str] = field(default_factory=dict)


def map_row(vendor: str, raw: dict[str, str]) -> MappedRow:
    mapping = VENDOR_MAPS[vendor]
    ignored = IGNORED_COLUMNS.get(vendor, frozenset())
    canonical: dict[str, object] = {"source_system": vendor}
    unknown: dict[str, str] = {}
    for src_key, value in raw.items():
        if src_key in ignored:
            continue                              # dropped by explicit policy
        target = mapping.get(src_key)
        if target is None:
            unknown[src_key] = value              # surfaced, never silently lost
            continue
        if value == "" or value is None:
            canonical[target] = None              # blank -> explicit null
            continue
        coerce = COERCERS.get(target)
        canonical[target] = coerce(value) if coerce else value
    return MappedRow(canonical=canonical, unknown=unknown)

Step 4 — Flag unknown fields and route the outcome #

Goal: turn the presence of unknown fields into an actionable signal. The mapped portion still proceeds so a single new column does not block an otherwise-good row, but the unknowns are logged and parked so a traffic engineer reconciles them into the map or the ignore list.

from pathlib import Path
import json
from datetime import datetime, timezone


def process_row(vendor: str, raw: dict[str, str], review_dir: Path) -> MappedRow:
    result = map_row(vendor, raw)
    # Use the mapped house_number (or a placeholder) as the audit key.
    key = str(result.canonical.get("house_number", "UNKNOWN"))
    if result.unknown:
        review_dir.mkdir(parents=True, exist_ok=True)
        target = review_dir / f"{vendor}_{key}.json"
        target.write_text(json.dumps(
            {"vendor": vendor, "unknown_fields": result.unknown,
             "flagged_at": datetime.now(timezone.utc).isoformat()},
            indent=2,
        ))
        logger.warning("%s | unknown fields %s parked -> %s",
                       key, sorted(result.unknown), target)
    else:
        logger.info("%s | mapped %d canonical fields, no unknowns",
                    key, len(result.canonical))
    return result

Expected log line when a vendor slips in a new column: 2026-07-17T09:14:22 | WARNING | traffic.canonical.vendor_mapper | RCPRIME30 | unknown fields ['Promo_Tag'] parked -> review/avion_RCPRIME30.json. The mapped fields still flow onward to the Pydantic validator for constraint enforcement.

Verification & Testing #

Two properties must hold: a known column always lands under its canonical name with the right type, and an unknown column is always surfaced rather than dropped. Both are assertable against fixture rows drawn from real exports.

avion_row = {
    "HouseNo": "RCPRIME30", "ISCI": "AUTX0142", "Rate": "2850.00",
    "BillCode": "RCPRIME30", "Category": "auto", "AirTime": "2026-07-17T23:30:00Z",
    "OperatorInit": "jdoe",              # ignored by policy
    "Promo_Tag": "SUMMER",               # unknown -> flagged
}
result = map_row("avion", avion_row)

# 1. Known columns rename and coerce correctly.
assert result.canonical["house_number"] == "RCPRIME30"
assert result.canonical["rate_cents"] == 285000        # dollars -> integer cents
assert result.canonical["source_system"] == "avion"

# 2. Policy-ignored column is gone; genuinely unknown column is surfaced.
assert "OperatorInit" not in result.canonical
assert result.unknown == {"Promo_Tag": "SUMMER"}       # never silently dropped

Run this suite in CI against a golden fixture per vendor so a change to a rename map that would silently re-route a field fails the build instead of shipping. Because _to_cents is a pure function, Rate of "2850.00" always yields 285000 — the determinism a billing audit relies on when it replays an archived export.

Edge Cases & Failure Handling #

• Two source columns map to one canonical field. If a vendor export carries both ISCI and ISCI_CODE and both map to isci, the second write silently wins. Guard against it by asserting no duplicate targets when the map loads, and prefer the more specific column explicitly — never let dictionary order decide which creative ID survives.
• A required canonical field never appears. The mapper produces whatever the row contains; a missing order_id is not the mapper’s error to raise. It flows to the canonical model, where the constraint fails closed with a clear message. Keep the two responsibilities separate so the failure reason is unambiguous — mapping surfaces unknown fields, validation catches missing ones.
• A new vendor column that is actually important. The review file is the safety net: a Promo_Tag parked today is reconciled tomorrow into either VENDOR_MAPS (if it belongs in the dictionary) or IGNORED_COLUMNS (if it does not). Never resolve it by loosening the mapper to accept arbitrary keys, which is how an un-typed value slips into the canonical layer. When a genuinely new concept appears, it belongs in the canonical data dictionary first, then the map.

FAQ #

Related #

• The Canonical Traffic Field Data Dictionary — the authoritative field vocabulary every rename map must target, with types and constraints for each canonical name.
• Avion vs Avstar Export Format Comparison — the catalogue of vendor column and representation differences this mapper reconciles.
• Schema Validation with Pydantic for Traffic Data — the boundary validator that enforces types and constraints on the fields once they are renamed.