Automating Weekly CSV Menu Updates

This page walks a Python automation engineer through building a scheduled job that ingests a weekly CSV menu export, diffs it against the current master state, and commits only the changed cost lines — the recurring task that keeps a multi-unit food-cost model current without manual reconciliation. It sits under CSV Bulk Import Automation, narrowing that module’s general ingestion contract to the specific cadence, idempotency, and delta-detection concerns of a once-a-week refresh.

Culinary teams publish a fresh price sheet or recipe revision every week; the job must be safe to re-run, must never double-apply a file, and must surface only the deltas that matter to procurement and POS pricing. The steps below produce a self-contained pipeline you can drop behind cron, Airflow, or Dagster.

Prerequisites and Data Contract

Pin these versions and provision the two tables before the steps apply. The pipeline is deterministic only if the schema is fixed on both sides of the diff.

• Runtime: Python 3.11+, pandas==2.2.*, pandera==0.20.*, SQLAlchemy==2.0.*, psycopg[binary]==3.2.*.
• Monetary rule: every cost value is carried as decimal.Decimal in Python and NUMERIC(12,4) in PostgreSQL — never float. Binary floats silently drift on values like 0.1, and that drift compounds across tens of thousands of SKUs.
• Environment: a DATABASE_URL env var, a writable landing directory for the weekly drop, and read access to the master state table.

The incoming file must satisfy this column contract (extra columns are ignored, missing required columns fail the run):

The master state lives in menu_master and the job writes changes to menu_cost_deltas:

CREATE TABLE menu_master (
    menu_item_id   TEXT        NOT NULL,
    location_code  TEXT        NOT NULL,
    base_cost      NUMERIC(12,4) NOT NULL,
    portion_yield  NUMERIC(6,4)  NOT NULL,
    effective_date DATE        NOT NULL,
    PRIMARY KEY (menu_item_id, location_code)
);

CREATE TABLE menu_cost_deltas (
    menu_item_id   TEXT        NOT NULL,
    location_code  TEXT        NOT NULL,
    effective_date DATE        NOT NULL,
    base_cost_new  NUMERIC(12,4),
    portion_yield_new NUMERIC(6,4),
    delta_cost     NUMERIC(12,4) NOT NULL,
    status         TEXT        NOT NULL,   -- NEW | MODIFIED | DISCONTINUED
    requires_review BOOLEAN    NOT NULL DEFAULT FALSE,
    source_sha256  TEXT        NOT NULL,
    updated_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (menu_item_id, location_code, effective_date)
);

Unit strings arriving in the CSV are assumed to already be canonical. If your vendor sheets use regional aliases (lbs, #, litre), normalise them upstream with the unit canonicalization frameworks before this job runs, so the diff compares like against like.

Step-by-Step Implementation

Each step is a self-contained block. Run them in order inside one scheduled task.

Step 1 — Declare the schema and Decimal converters

Parse the CSV with explicit converters so monetary columns land as Decimal, not float, and validate the frame against a pandera model. Validation is a hard gate: a malformed weekly drop is rejected here rather than silently corrupting the master state.

from decimal import Decimal
import pandas as pd
import pandera as pa
from pandera.typing import Series


class WeeklyMenuSchema(pa.DataFrameModel):
    menu_item_id: Series[str] = pa.Field(str_matches=r"^[A-Z0-9\-]{4,12}$")
    location_code: Series[str] = pa.Field(str_matches=r"^[A-Z]{2,4}$")
    effective_date: Series[pd.Timestamp] = pa.Field(nullable=False)
    base_cost: Series[object] = pa.Field(nullable=False)      # Decimal objects
    portion_yield: Series[float] = pa.Field(gt=0.0, le=1.0)
    unit_of_measure: Series[str] = pa.Field(isin=["kg", "lb", "oz", "g", "ea"])


def _to_decimal(raw: str) -> Decimal:
    # Canonicalise European decimals and stray thousands separators.
    return Decimal(raw.strip().replace(" ", "").replace(",", "."))


def load_weekly_csv(path: str) -> pd.DataFrame:
    df = pd.read_csv(
        path,
        dtype={"menu_item_id": str, "location_code": str, "unit_of_measure": str},
        converters={"base_cost": _to_decimal},
        parse_dates=["effective_date"],
    )
    return WeeklyMenuSchema.validate(df, lazy=True)

Step 2 — Enforce idempotency with a payload checksum

A weekly job will be retried after transient failures, and an operator may upload the same file twice. Hash the raw bytes and refuse to reprocess a checksum you have already committed. This is what makes the whole run safe to repeat.

import hashlib


def file_sha256(path: str) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as fh:
        for chunk in iter(lambda: fh.read(8192), b""):
            h.update(chunk)
    return h.hexdigest()


def already_processed(conn, digest: str) -> bool:
    row = conn.exec_driver_sql(
        "SELECT 1 FROM menu_cost_deltas WHERE source_sha256 = %s LIMIT 1",
        (digest,),
    ).fetchone()
    return row is not None

Step 3 — Diff against the master state (vectorized)

Load the master state and classify every incoming row with a single merge. An outer merge with indicator=True labels each key as present in both frames (MODIFIED candidate), only in the weekly file (NEW), or only in master (DISCONTINUED) — no row-by-row iteration.

import numpy as np


def classify_rows(master: pd.DataFrame, weekly: pd.DataFrame) -> pd.DataFrame:
    keys = ["menu_item_id", "location_code"]
    merged = weekly.merge(
        master[keys + ["base_cost", "portion_yield"]],
        on=keys, how="outer", suffixes=("_new", "_old"), indicator=True,
    )
    merged["status"] = np.select(
        [merged["_merge"] == "right_only", merged["_merge"] == "left_only"],
        ["DISCONTINUED", "NEW"],
        default="MODIFIED",
    )
    return merged

Step 4 — Compute cost deltas in Decimal

Portion-adjusted cost is base_cost * portion_yield. Because the cost columns are object-dtype Decimal Series, the arithmetic stays element-wise and exact — the delta is new - old, with a missing side treated as Decimal("0").

from decimal import Decimal

ZERO = Decimal("0")


def compute_deltas(merged: pd.DataFrame) -> pd.DataFrame:
    new_yield = merged["portion_yield_new"].fillna(0.0)
    old_yield = merged["portion_yield_old"].fillna(0.0)

    new_cost = merged["base_cost_new"].map(lambda d: d if isinstance(d, Decimal) else ZERO)
    old_cost = merged["base_cost_old"].map(lambda d: d if isinstance(d, Decimal) else ZERO)

    new_portioned = new_cost * new_yield.map(lambda y: Decimal(str(y)))
    old_portioned = old_cost * old_yield.map(lambda y: Decimal(str(y)))
    merged["delta_cost"] = new_portioned - old_portioned
    return merged[merged["delta_cost"] != ZERO].copy()

Step 5 — Gate anomalies to a review queue

A raw delta is not automatically trustworthy: a fat-fingered price or a yield entered as a percentage (85) instead of a fraction (0.85) produces a huge swing. Flag outliers instead of halting, so the good rows still commit while the suspect rows route to a culinary manager. Tune the band with the dynamic variance thresholds approach rather than a hardcoded constant.

def flag_for_review(deltas: pd.DataFrame, abs_cost_limit: Decimal = Decimal("2.50")) -> pd.DataFrame:
    deltas["requires_review"] = deltas["delta_cost"].map(abs) > abs_cost_limit
    return deltas

Step 6 — Persist idempotently with UPSERT

Write every changed row inside one transaction, keyed on (menu_item_id, location_code, effective_date). ON CONFLICT ... DO UPDATE means a re-run overwrites rather than duplicating, and the source_sha256 recorded here is what Step 2 checks on the next run.

from sqlalchemy import create_engine, text


UPSERT = text("""
    INSERT INTO menu_cost_deltas
      (menu_item_id, location_code, effective_date, base_cost_new,
       portion_yield_new, delta_cost, status, requires_review, source_sha256)
    VALUES
      (:menu_item_id, :location_code, :effective_date, :base_cost_new,
       :portion_yield_new, :delta_cost, :status, :requires_review, :source_sha256)
    ON CONFLICT (menu_item_id, location_code, effective_date) DO UPDATE SET
       base_cost_new = EXCLUDED.base_cost_new,
       delta_cost    = EXCLUDED.delta_cost,
       status        = EXCLUDED.status,
       requires_review = EXCLUDED.requires_review,
       source_sha256 = EXCLUDED.source_sha256,
       updated_at    = NOW();
""")


def persist(db_url: str, rows: list[dict]) -> int:
    engine = create_engine(db_url)
    with engine.begin() as conn:
        conn.execute(UPSERT, rows)
    return len(rows)

Step 7 — Schedule the weekly run

Wire the steps behind a scheduler. A cron entry is enough for a single node; for fan-out across many locations, hand the file off to the async batch processing workflow instead of processing inline.

# /etc/cron.d/weekly-menu-sync — Mondays 04:15, off-peak POS window
15 4 * * 1  automation  /opt/venv/bin/python -m menu_sync.weekly \
    --drop /data/menu/incoming/latest.csv >> /var/log/menu_sync.log 2>&1

Verification and Validation

Confirm the run did what you expect before trusting downstream reports.

• Row-count reconciliation. The number of committed rows must equal NEW + MODIFIED + DISCONTINUED. Assert it in-process before commit:

  counts = deltas["status"].value_counts().to_dict()
  assert len(deltas) == sum(counts.values()), "delta count mismatch"
  logging.info("weekly sync classified %s", counts)

• Idempotency proof. Re-run the same file; the second run should log already processed and write zero rows. Query to confirm no duplication:

  SELECT source_sha256, COUNT(*) FROM menu_cost_deltas
  GROUP BY source_sha256 ORDER BY 2 DESC LIMIT 5;

• Review backlog check. Anything the anomaly gate flagged should be visible and non-empty only when a real swing occurred:

  SELECT status, COUNT(*) FILTER (WHERE requires_review) AS flagged
  FROM menu_cost_deltas WHERE effective_date = CURRENT_DATE GROUP BY status;

A healthy run emits one structured log line with the classification counts, a checksum, and a zero (or explained) flagged count.

Gotchas and Edge Cases

• IEEE-754 drift in cost math. If you let base_cost land as float, Decimal(str(...)) later cannot recover the lost precision. Convert at parse time (Step 1), never after arithmetic.
• portion_yield = 0 divide-by-zero. Any downstream cost-per-usable-portion step divides by yield. The gt=0.0 schema rule rejects a zero yield at the gate; do not relax it to ge=0.0.
• Percentage-vs-fraction yields. A yield of 85 slips past a naive >= 0 check but blows up the delta 100x. The le=1.0 bound catches it; route offenders to review, not to master.
• Locale decimal separators. German and French exports use , as the decimal mark and . as thousands. The _to_decimal converter normalises both — never rely on pandas thousands=/decimal= alone when the two collide.
• False DISCONTINUED flags. If a location legitimately omits an item from one weekly drop, an outer-merge diff marks it discontinued and zeroes its cost. Guard high-value items with a grace window before acting on a DISCONTINUED status, and reconcile against the POS taxonomy mapping to confirm the SKU is truly retired.
• Duplicate submissions. Two operators uploading the same sheet under different filenames is common; the Step 2 checksum — not the filename — is the deduplication key.

FAQ

How do I make the weekly job safe to re-run after a crash?

Two mechanisms combine: the SHA-256 payload checksum in Step 2 blocks reprocessing an already-committed file, and the ON CONFLICT DO UPDATE upsert in Step 6 means a partial re-run overwrites rather than duplicating rows. Together they make the pipeline idempotent, so a retry after a mid-run failure converges to the same state.

Why use Decimal instead of float for the cost columns?

Binary floats cannot represent common decimal cents exactly, and the error compounds when you multiply by yield and sum across thousands of SKUs. Carrying Decimal in Python and NUMERIC(12,4) in PostgreSQL keeps every intermediate value exact; rounding is applied only at the reporting boundary.

How do I detect items that were removed from the menu this week?

The outer merge in Step 3 tags keys present in master but absent from the weekly file as DISCONTINUED. Treat that as a signal, not an immediate deletion — a temporary omission looks identical to a genuine retirement, so apply a grace window and cross-check the SKU against your live catalog before zeroing its cost.

Can this run across hundreds of locations without exhausting memory?

Yes. The merge and delta math are vectorized, so a single frame handles hundreds of thousands of rows. Past that, read the CSV in chunks or partition by location_code and dispatch each partition through the async batch workflow, keeping the per-worker footprint bounded.

Where should flagged anomalies go?

Rows the anomaly gate marks requires_review = TRUE stay in menu_cost_deltas but are excluded from the automatic POS push until a culinary manager clears them. Feed the flag threshold from a variance model rather than a constant so it adapts to each category’s volatility.

Related

• Up: CSV Bulk Import Automation — the parent module this weekly job specializes.
• Data Ingestion & Recipe Parsing Workflows — the full ingestion architecture.
• Async Batch Processing Workflows — fan-out execution for high-volume, multi-location drops.
• POS API Polling Strategies — how committed cost baselines reach live pricing.
• Setting Dynamic Variance Thresholds — tuning the anomaly gate in Step 5.