From f5d5008838517411416f061b0081fee9dca04fec Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 28 Oct 2026 04:33:07 -0700 Subject: [PATCH 1/2] wip: SEL-549 preflight (paused) Signed-off-by: Christopher Sellers <1@christophersellers.com> Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- .../app/supplier_readiness/__init__.py | 27 + .../app/supplier_readiness/shipping.py | 501 ++++++++++++++++++ .../tests/fixtures/shipping_messy_partner.csv | 8 + .../tests/fixtures/shipping_pass_partner.csv | 2 + 4 files changed, 538 insertions(+) create mode 100644 services/ingestion/app/supplier_readiness/__init__.py create mode 100644 services/ingestion/app/supplier_readiness/shipping.py create mode 100644 services/ingestion/tests/fixtures/shipping_messy_partner.csv create mode 100644 services/ingestion/tests/fixtures/shipping_pass_partner.csv diff --git a/services/ingestion/app/supplier_readiness/__init__.py b/services/ingestion/app/supplier_readiness/__init__.py new file mode 100644 index 000000000..0dde999b3 --- /dev/null +++ b/services/ingestion/app/supplier_readiness/__init__.py @@ -0,0 +1,27 @@ +"""SEL-549 supplier shipping readiness (preflight only). + +Maps a second shipping layout (EDI 856 ASN) into the same +``REQUIRED_KDES_BY_CTE[shipping]`` set the CSV partner path uses, scores +pass / gap / failure, and flags unknown columns or segments instead of +dropping them. + +Synthetic fixtures only. No production ingest. +""" + +from .shipping import ( + ALBERTSONS_BUYER_FIELDS, + FlagRecord, + ShippingReadinessResult, + dry_run_shipping_csv, + map_asn856_to_shipping_kdes, + score_shipping_readiness, +) + +__all__ = [ + "ALBERTSONS_BUYER_FIELDS", + "FlagRecord", + "ShippingReadinessResult", + "dry_run_shipping_csv", + "map_asn856_to_shipping_kdes", + "score_shipping_readiness", +] diff --git a/services/ingestion/app/supplier_readiness/shipping.py b/services/ingestion/app/supplier_readiness/shipping.py new file mode 100644 index 000000000..6ee598cc9 --- /dev/null +++ b/services/ingestion/app/supplier_readiness/shipping.py @@ -0,0 +1,501 @@ +"""Map ASN 856 (and shipping CSV rows) into shipping KDEs and score readiness. + +Preflight only. Uses synthetic fixtures in tests. Does not call production +ingest or partner networks. +""" + +from __future__ import annotations + +import csv +import io +import re +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Iterable, Mapping, Optional, Sequence + +from ..edi_ingestion.utils import _normalize_unit, _resolve_unit_code +from ..webhook_models import REQUIRED_KDES_BY_CTE, VALID_UNITS_OF_MEASURE, WebhookCTEType + +# Segments the 856 extractor / envelope path already understands. Anything +# else is flagged so partner dialects cannot vanish silently (SEL-549). +_KNOWN_856_SEGMENTS: frozenset[str] = frozenset( + { + "ISA", + "GS", + "ST", + "BSN", + "HL", + "N1", + "N2", + "N3", + "N4", + "LIN", + "SN1", + "SLN", + "PID", + "REF", + "TD5", + "TD1", + "TD3", + "MAN", + "DTM", + "CTT", + "PRF", + "CUR", + "PER", + "MEA", + "FOB", + "SE", + "GE", + "IEA", + } +) + +# Albertsons ASN buyer overlay (archived letter). Not in REQUIRED_KDES_BY_CTE. +ALBERTSONS_BUYER_FIELDS: tuple[str, ...] = ( + "gtin", + "po_number", + "sscc", + "product_date", +) + +_SHIPPING_REQUIRED: tuple[str, ...] = tuple(REQUIRED_KDES_BY_CTE[WebhookCTEType.SHIPPING]) + +_GTIN_QUALIFIERS = frozenset({"UK", "EN", "UI", "UA", "EO"}) +_SSCC_MAN_QUALIFIERS = frozenset({"GM", "W"}) # GS1 SSCC / warehouse pallet id +_PO_REF_QUALIFIERS = frozenset({"PO", "VN"}) # purchase order / vendor order + + +@dataclass(frozen=True) +class FlagRecord: + """One unknown or malformed input observation. Never dropped silently.""" + + reason: str + field: Optional[str] = None + detail: Optional[str] = None + row_number: Optional[int] = None + + +@dataclass +class ShippingReadinessResult: + """One shipping lot/row scored against repo KDEs + Albertsons overlay.""" + + status: str # pass | gap | failure + kdes: dict[str, Any] + buyer: dict[str, Any] + missing_repo_kdes: list[str] = field(default_factory=list) + missing_buyer_fields: list[str] = field(default_factory=list) + failures: list[dict[str, str]] = field(default_factory=list) + flags: list[FlagRecord] = field(default_factory=list) + label: str = "SYNTHETIC" + issue: str = "SEL-549" + source: Optional[str] = None + + +def _present(value: Any) -> bool: + if value is None: + return False + if isinstance(value, (int, float)): + return True + return str(value).strip() != "" + + +def _normalize_ship_date(raw: Any) -> Optional[str]: + """Normalize EDI YYYYMMDD (or ISO) into YYYY-MM-DD. None if unparseable.""" + if raw is None: + return None + text = str(raw).strip() + if not text: + return None + if re.fullmatch(r"\d{8}", text): + try: + return datetime.strptime(text, "%Y%m%d").date().isoformat() + except ValueError: + return None + if re.fullmatch(r"\d{4}-\d{2}-\d{2}", text): + try: + datetime.strptime(text, "%Y-%m-%d") + return text + except ValueError: + return None + # Accept ISO datetime prefix + if "T" in text or " " in text: + try: + return datetime.fromisoformat(text.replace("Z", "+00:00")).date().isoformat() + except ValueError: + return None + return None + + +def _scan_buyer_and_unknowns( + segments: Sequence[Sequence[str]], +) -> tuple[dict[str, Any], list[FlagRecord]]: + """Pull Albertsons buyer fields and flag unknown 856 segments / values.""" + buyer: dict[str, Any] = {"gtin": None, "po_number": None, "sscc": None, "product_date": None} + flags: list[FlagRecord] = [] + seen_unknown_seg: set[str] = set() + + for segment in segments: + if not segment: + continue + seg_id = str(segment[0]).strip().upper() + if not seg_id: + continue + if seg_id not in _KNOWN_856_SEGMENTS and seg_id not in seen_unknown_seg: + seen_unknown_seg.add(seg_id) + flags.append( + FlagRecord( + reason="unrecognized_segment", + field=seg_id, + detail="EDI segment id not in the known 856 set; kept for audit", + ) + ) + + if seg_id == "LIN": + # LIN**qual*value pairs start at element 2 + for i in range(2, len(segment) - 1, 2): + qual = str(segment[i]).strip().upper() + val = str(segment[i + 1]).strip() + if not val: + continue + if qual in _GTIN_QUALIFIERS and not buyer["gtin"]: + buyer["gtin"] = val + elif qual and qual not in _GTIN_QUALIFIERS | {"SK", "VN", "UP", "BP", "MG", "LT"}: + flags.append( + FlagRecord( + reason="unrecognized_value", + field=f"LIN.{qual}", + detail="Unrecognized LIN product-id qualifier; value kept", + ) + ) + + elif seg_id == "REF" and len(segment) > 2: + qual = str(segment[1]).strip().upper() + val = str(segment[2]).strip() + if not val: + continue + if qual in _PO_REF_QUALIFIERS and not buyer["po_number"]: + buyer["po_number"] = val + elif qual == "LT": + pass # lot handled via extracted line_items + elif qual in {"BM", "SI", "CN", "PK"}: + pass # BOL / shipment / tracking → reference_document + else: + flags.append( + FlagRecord( + reason="unrecognized_value", + field=f"REF.{qual}", + detail="Unrecognized REF qualifier; value kept", + ) + ) + + elif seg_id == "MAN" and len(segment) > 2: + # MAN01 qualifier; SSCC often in MAN02/MAN03 depending on dialect + qual = str(segment[1]).strip().upper() + for candidate in segment[2:]: + text = str(candidate).strip() + if text and (qual in _SSCC_MAN_QUALIFIERS or len(re.sub(r"\D", "", text)) in (18, 20)): + if not buyer["sscc"]: + buyer["sscc"] = text + break + if not buyer["sscc"] and any(str(c).strip() for c in segment[2:]): + flags.append( + FlagRecord( + reason="unrecognized_value", + field=f"MAN.{qual or '?'}", + detail="MAN present but SSCC not resolved; raw kept in flags only", + ) + ) + + elif seg_id == "DTM" and len(segment) > 2: + # 036 / 405 / 002 style product or ship dates — record product_date when distinct + qual = str(segment[1]).strip().upper() + raw = str(segment[2]).strip() + if qual in {"036", "405", "511", "036"} and raw and not buyer["product_date"]: + buyer["product_date"] = _normalize_ship_date(raw) or raw + + return buyer, flags + + +def map_asn856_to_shipping_kdes( + extracted: Mapping[str, Any], + segments: Optional[Sequence[Sequence[str]]] = None, + *, + source: str = "edi_corpus/asn_856", +) -> ShippingReadinessResult: + """Map one ASN 856 extraction (plus optional segments) into shipping KDEs. + + Line items become one result each; when ``line_items`` is empty a single + header-only row is scored so missing TLC still surfaces as a gap. + """ + header = {k: v for k, v in extracted.items() if k != "line_items"} + lines = list(extracted.get("line_items") or []) + if not lines: + lines = [{"lot_number": None, "product_description": None, "quantity": None, "unit_of_measure": None}] + + buyer_scan: dict[str, Any] = {"gtin": None, "po_number": None, "sscc": None, "product_date": None} + seg_flags: list[FlagRecord] = [] + if segments is not None: + buyer_scan, seg_flags = _scan_buyer_and_unknowns(segments) + + # Score the first line (fixture has one). Multi-line ASNs can call this + # per merged line from the ingest route later; preflight keeps it simple. + line = lines[0] + merged = {**header, **line} + + lot = merged.get("lot_number") + raw_unit = merged.get("unit_of_measure") + resolved_unit = _resolve_unit_code(raw_unit) if raw_unit is not None else None + unit_failure = False + if raw_unit is not None and str(raw_unit).strip() and resolved_unit is None: + unit_failure = True + unit_of_measure = str(raw_unit).strip() + else: + unit_of_measure = resolved_unit or (_normalize_unit(raw_unit) if raw_unit else None) + + ship_date = _normalize_ship_date(merged.get("ship_date_raw") or merged.get("ship_date")) + + kdes: dict[str, Any] = { + "traceability_lot_code": lot if _present(lot) else None, + "product_description": merged.get("product_description"), + "quantity": merged.get("quantity"), + "unit_of_measure": unit_of_measure, + "ship_date": ship_date, + "ship_from_location": merged.get("ship_from_name") or merged.get("ship_from_location"), + "ship_to_location": merged.get("ship_to_name") or merged.get("ship_to_location"), + "reference_document": merged.get("reference_document_number") or merged.get("reference_document"), + "tlc_source_reference": merged.get("tlc_source_reference"), + # Extras kept (never dropped) + "ship_from_gln": merged.get("ship_from_gln"), + "ship_to_gln": merged.get("ship_to_gln"), + "asn_number": merged.get("asn_number"), + "carrier": merged.get("carrier"), + } + + buyer = { + "gtin": buyer_scan.get("gtin") or merged.get("gtin"), + "po_number": buyer_scan.get("po_number") or merged.get("po_number"), + "sscc": buyer_scan.get("sscc") or merged.get("sscc"), + "product_date": buyer_scan.get("product_date") or merged.get("product_date"), + } + + flags = list(seg_flags) + if unit_failure: + flags.append( + FlagRecord( + reason="unrecognized_value", + field="unit_of_measure", + detail=f"Unit {raw_unit!r} is not in VALID_UNITS_OF_MEASURE / X12 DE355 map", + ) + ) + + return score_shipping_readiness(kdes, buyer, flags=flags, source=source) + + +def score_shipping_readiness( + kdes: Mapping[str, Any], + buyer: Optional[Mapping[str, Any]] = None, + *, + flags: Optional[Iterable[FlagRecord]] = None, + source: Optional[str] = None, +) -> ShippingReadinessResult: + """Score one shipping row: failure > gap > pass.""" + buyer = dict(buyer or {}) + flag_list = list(flags or []) + missing_repo: list[str] = [] + failures: list[dict[str, str]] = [] + + for key in _SHIPPING_REQUIRED: + if not _present(kdes.get(key)): + missing_repo.append(key) + + qty = kdes.get("quantity") + if _present(qty): + try: + qf = float(qty) + if not (qf > 0) or qf != qf: # noqa: PLR0124 — NaN check + failures.append({"kind": "malformed_quantity", "detail": f"quantity={qty!r}"}) + except (TypeError, ValueError): + failures.append({"kind": "malformed_quantity", "detail": f"quantity={qty!r}"}) + + uom = kdes.get("unit_of_measure") + if _present(uom) and str(uom).strip().lower() not in VALID_UNITS_OF_MEASURE: + # Also accept already-canonical values; VALID set is lowercase + if str(uom).strip() not in VALID_UNITS_OF_MEASURE: + failures.append({"kind": "invalid_unit_of_measure", "detail": f"unit_of_measure={uom!r}"}) + + # ship_date present but unnormalizable is a failure when a raw was attempted + if kdes.get("ship_date") is None and _present(kdes.get("_ship_date_raw_failed")): + failures.append({"kind": "malformed_ship_date", "detail": "ship_date not parseable"}) + + missing_buyer = [name for name in ALBERTSONS_BUYER_FIELDS if not _present(buyer.get(name))] + + if failures: + status = "failure" + elif missing_repo or missing_buyer: + status = "gap" + else: + status = "pass" + + return ShippingReadinessResult( + status=status, + kdes=dict(kdes), + buyer=dict(buyer), + missing_repo_kdes=missing_repo, + missing_buyer_fields=missing_buyer, + failures=failures, + flags=flag_list, + source=source, + ) + + +def _csv_row_to_kdes(row: Mapping[str, str]) -> tuple[dict[str, Any], dict[str, Any], list[FlagRecord]]: + """Map a shipping CSV row into KDE + buyer dicts; flag unknown columns.""" + known = set(_SHIPPING_REQUIRED) | set(ALBERTSONS_BUYER_FIELDS) | { + "cte_type", + "ship_from_gln", + "ship_to_gln", + "carrier_name", + "carrier", + "event_time", + "asn_number", + "lot_number", + "lot_code", + "tlc", + } + flags: list[FlagRecord] = [] + extras: dict[str, str] = {} + for col, val in row.items(): + key = (col or "").strip() + if not key: + continue + if key not in known: + flags.append( + FlagRecord( + reason="unrecognized_column", + field=key, + detail="Unknown column kept; not mapped to a shipping KDE", + ) + ) + if val is not None and str(val).strip() != "": + extras[key] = str(val) + + def _get(*names: str) -> Optional[str]: + for name in names: + if name in row and str(row[name]).strip() != "": + return str(row[name]).strip() + return None + + raw_unit = _get("unit_of_measure") + resolved = _resolve_unit_code(raw_unit) if raw_unit else None + if raw_unit and resolved is None: + flags.append( + FlagRecord( + reason="unrecognized_value", + field="unit_of_measure", + detail=f"Unit {raw_unit!r} not recognized", + ) + ) + unit = raw_unit + else: + unit = resolved + + ship_raw = _get("ship_date") + ship_date = _normalize_ship_date(ship_raw) if ship_raw else None + if ship_raw and ship_date is None: + flags.append( + FlagRecord( + reason="malformed_date", + field="ship_date", + detail=f"ship_date {ship_raw!r} not parseable", + ) + ) + + qty_raw = _get("quantity") + qty: Any = qty_raw + if qty_raw is not None: + try: + qty = float(qty_raw) + except ValueError: + flags.append( + FlagRecord( + reason="malformed_quantity", + field="quantity", + detail=f"quantity {qty_raw!r} not numeric", + ) + ) + + kdes = { + "traceability_lot_code": _get("traceability_lot_code", "lot_number", "lot_code", "tlc"), + "product_description": _get("product_description"), + "quantity": qty, + "unit_of_measure": unit, + "ship_date": ship_date, + "ship_from_location": _get("ship_from_location"), + "ship_to_location": _get("ship_to_location"), + "reference_document": _get("reference_document"), + "tlc_source_reference": _get("tlc_source_reference"), + "ship_from_gln": _get("ship_from_gln"), + "ship_to_gln": _get("ship_to_gln"), + **extras, + } + buyer = { + "gtin": _get("gtin"), + "po_number": _get("po_number"), + "sscc": _get("sscc"), + "product_date": _get("product_date"), + } + return kdes, buyer, flags + + +def dry_run_shipping_csv( + text: str, + *, + source: str = "shipping_messy_partner.csv", +) -> dict[str, Any]: + """Preflight dry run: score every row and tally flags by reason. + + Does not ingest. Unknown columns are flagged and kept on the KDE dict. + """ + reader = csv.DictReader(io.StringIO(text)) + rows = list(reader) + results: list[ShippingReadinessResult] = [] + flag_counts: dict[str, int] = {} + status_counts = {"pass": 0, "gap": 0, "failure": 0} + + for idx, row in enumerate(rows, start=2): # header is row 1 + cte = (row.get("cte_type") or "").strip().lower() + kdes, buyer, flags = _csv_row_to_kdes(row) + if cte and cte not in {"shipping", "ship", "s", "distribute"}: + flags.append( + FlagRecord( + reason="unknown_cte_type", + field="cte_type", + detail=f"cte_type={cte!r} is not shipping", + row_number=idx, + ) + ) + numbered = [ + FlagRecord(reason=f.reason, field=f.field, detail=f.detail, row_number=f.row_number or idx) + for f in flags + ] + for f in numbered: + flag_counts[f.reason] = flag_counts.get(f.reason, 0) + 1 + scored = score_shipping_readiness(kdes, buyer, flags=numbered, source=f"{source}:row{idx}") + if cte and cte not in {"shipping", "ship", "s", "distribute"}: + # Scope break: never a pass for this readiness case + if scored.status == "pass": + scored.status = "failure" + scored.failures.append({"kind": "wrong_cte_scope", "detail": f"cte_type={cte!r}"}) + results.append(scored) + status_counts[scored.status] = status_counts.get(scored.status, 0) + 1 + + return { + "label": "SYNTHETIC", + "issue": "SEL-549", + "source": source, + "row_count": len(rows), + "status_counts": status_counts, + "flagged_counts_by_reason": flag_counts, + "results": results, + } diff --git a/services/ingestion/tests/fixtures/shipping_messy_partner.csv b/services/ingestion/tests/fixtures/shipping_messy_partner.csv new file mode 100644 index 000000000..4c127ee1f --- /dev/null +++ b/services/ingestion/tests/fixtures/shipping_messy_partner.csv @@ -0,0 +1,8 @@ +cte_type,traceability_lot_code,product_description,quantity,unit_of_measure,ship_date,ship_from_location,ship_to_location,reference_document,tlc_source_reference,partner_bin,mystery_code,extra_pallet_tag +shipping,LOT-ROM-2026-0312A,Organic Romaine Hearts,200,cases,2026-03-10,Valley Fresh Farms,Metro Distribution Center,BOL-9001,PACK-2026-0310-001,BIN-A,XYZ-1,TAG-9 +shipping,,Romaine Hearts,50,cases,2026-03-11,Valley Fresh Farms,Metro Distribution Center,BOL-9002,PACK-2026-0311-001,,, +shipping,LOT-ROM-2026-0312B,Chopped Romaine Mix,not-a-number,cases,2026-03-12,Valley Fresh Farms,Metro Distribution Center,BOL-9003,PACK-2026-0312-001,,, +shipping,LOT-ROM-2026-0311C,Romaine Hearts 3-Pack,100,blargs,2026-03-11,Valley Fresh Farms,Metro Distribution Center,BOL-9004,PACK-2026-0311-002,,, +unknown_cte,LOT-X-1,Something,10,cases,2026-03-10,Valley Fresh Farms,Metro Distribution Center,BOL-9005,PACK-X,,, +shipping,LOT-ROM-2026-0310D,Romaine Mix,75,cases,2026-03-99,Valley Fresh Farms,Metro Distribution Center,BOL-9006,PACK-2026-0310-002,,, +shipping,LOT-ROM-OK,Romaine Carton,40,cases,2026-03-15,Valley Fresh Farms,Metro Distribution Center,BOL-9007,PACK-2026-0315-001,BIN-B,, diff --git a/services/ingestion/tests/fixtures/shipping_pass_partner.csv b/services/ingestion/tests/fixtures/shipping_pass_partner.csv new file mode 100644 index 000000000..daa1a96db --- /dev/null +++ b/services/ingestion/tests/fixtures/shipping_pass_partner.csv @@ -0,0 +1,2 @@ +cte_type,traceability_lot_code,product_description,quantity,unit_of_measure,ship_date,ship_from_location,ship_to_location,reference_document,tlc_source_reference,gtin,po_number,sscc,product_date +shipping,LOT-ROM-2026-0312A,Organic Romaine Hearts,200,cases,2026-03-10,Valley Fresh Farms,Metro Distribution Center,BOL-9001,PACK-2026-0310-001,00614141123456,PO-2026-7788,00161414100000000018,2026-03-17 From 1d7fbfd381d8472fe2a89ba9f8f6a894589cc8cb Mon Sep 17 00:00:00 2001 From: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> Date: Wed, 28 Oct 2026 04:38:35 -0700 Subject: [PATCH 2/2] feat(ingest): SEL-549 ASN 856 supplier-readiness preflight Map EDI 856 ASN into REQUIRED_KDES_BY_CTE[shipping], score pass/gap/failure with Albertsons buyer overlay, and flag unknown columns/segments/values. Synthetic fixtures and dry-run only; no production ingest. Signed-off-by: Christopher Sellers <1@christophersellers.com> Signed-off-by: Christopher <131207452+PetrefiedThunder@users.noreply.github.com> --- .../app/supplier_readiness/shipping.py | 78 +++++-- .../tests/test_supplier_readiness_sel549.py | 191 ++++++++++++++++++ 2 files changed, 247 insertions(+), 22 deletions(-) create mode 100644 services/ingestion/tests/test_supplier_readiness_sel549.py diff --git a/services/ingestion/app/supplier_readiness/shipping.py b/services/ingestion/app/supplier_readiness/shipping.py index 6ee598cc9..aa56969d6 100644 --- a/services/ingestion/app/supplier_readiness/shipping.py +++ b/services/ingestion/app/supplier_readiness/shipping.py @@ -349,9 +349,10 @@ def score_shipping_readiness( ) -def _csv_row_to_kdes(row: Mapping[str, str]) -> tuple[dict[str, Any], dict[str, Any], list[FlagRecord]]: - """Map a shipping CSV row into KDE + buyer dicts; flag unknown columns.""" - known = set(_SHIPPING_REQUIRED) | set(ALBERTSONS_BUYER_FIELDS) | { +_CSV_KNOWN_COLUMNS: frozenset[str] = frozenset( + set(_SHIPPING_REQUIRED) + | set(ALBERTSONS_BUYER_FIELDS) + | { "cte_type", "ship_from_gln", "ship_to_gln", @@ -363,6 +364,16 @@ def _csv_row_to_kdes(row: Mapping[str, str]) -> tuple[dict[str, Any], dict[str, "lot_code", "tlc", } +) + + +def _csv_row_to_kdes( + row: Mapping[str, str], + *, + flag_unknown_columns: bool = True, +) -> tuple[dict[str, Any], dict[str, Any], list[FlagRecord]]: + """Map a shipping CSV row into KDE + buyer dicts; optionally flag unknown columns.""" + known = _CSV_KNOWN_COLUMNS flags: list[FlagRecord] = [] extras: dict[str, str] = {} for col, val in row.items(): @@ -370,13 +381,14 @@ def _csv_row_to_kdes(row: Mapping[str, str]) -> tuple[dict[str, Any], dict[str, if not key: continue if key not in known: - flags.append( - FlagRecord( - reason="unrecognized_column", - field=key, - detail="Unknown column kept; not mapped to a shipping KDE", + if flag_unknown_columns: + flags.append( + FlagRecord( + reason="unrecognized_column", + field=key, + detail="Unknown column kept; not mapped to a shipping KDE", + ) ) - ) if val is not None and str(val).strip() != "": extras[key] = str(val) @@ -455,18 +467,36 @@ def dry_run_shipping_csv( ) -> dict[str, Any]: """Preflight dry run: score every row and tally flags by reason. - Does not ingest. Unknown columns are flagged and kept on the KDE dict. + Does not ingest. Unknown columns are flagged once (header-level, same + spirit as #3682) and their cell values are still kept on each row's KDEs. """ reader = csv.DictReader(io.StringIO(text)) + fieldnames = list(reader.fieldnames or []) rows = list(reader) results: list[ShippingReadinessResult] = [] flag_counts: dict[str, int] = {} status_counts = {"pass": 0, "gap": 0, "failure": 0} + header_flags: list[FlagRecord] = [] + for col in fieldnames: + key = (col or "").strip() + if key and key not in _CSV_KNOWN_COLUMNS: + header_flags.append( + FlagRecord( + reason="unrecognized_column", + field=key, + detail="Unknown column kept; not mapped to a shipping KDE", + row_number=1, + ) + ) + for f in header_flags: + flag_counts[f.reason] = flag_counts.get(f.reason, 0) + 1 + for idx, row in enumerate(rows, start=2): # header is row 1 cte = (row.get("cte_type") or "").strip().lower() - kdes, buyer, flags = _csv_row_to_kdes(row) - if cte and cte not in {"shipping", "ship", "s", "distribute"}: + kdes, buyer, flags = _csv_row_to_kdes(row, flag_unknown_columns=False) + wrong_cte = bool(cte and cte not in {"shipping", "ship", "s", "distribute"}) + if wrong_cte: flags.append( FlagRecord( reason="unknown_cte_type", @@ -475,18 +505,22 @@ def dry_run_shipping_csv( row_number=idx, ) ) - numbered = [ + # Attach header unknown-column flags to row 2 only so they appear on a result + row_flags = list(header_flags) if idx == 2 else [] + row_flags.extend( FlagRecord(reason=f.reason, field=f.field, detail=f.detail, row_number=f.row_number or idx) for f in flags - ] - for f in numbered: - flag_counts[f.reason] = flag_counts.get(f.reason, 0) + 1 - scored = score_shipping_readiness(kdes, buyer, flags=numbered, source=f"{source}:row{idx}") - if cte and cte not in {"shipping", "ship", "s", "distribute"}: - # Scope break: never a pass for this readiness case - if scored.status == "pass": - scored.status = "failure" - scored.failures.append({"kind": "wrong_cte_scope", "detail": f"cte_type={cte!r}"}) + ) + for f in row_flags: + if f.reason == "unrecognized_column" and f.row_number == 1 and idx != 2: + continue + if not (f.reason == "unrecognized_column" and f.row_number == 1): + flag_counts[f.reason] = flag_counts.get(f.reason, 0) + 1 + scored = score_shipping_readiness(kdes, buyer, flags=row_flags, source=f"{source}:row{idx}") + if wrong_cte: + # Fail-case §2.7: wrong CTE is a scope break, never pass/gap-as-ok + scored.status = "failure" + scored.failures.append({"kind": "wrong_cte_scope", "detail": f"cte_type={cte!r}"}) results.append(scored) status_counts[scored.status] = status_counts.get(scored.status, 0) + 1 diff --git a/services/ingestion/tests/test_supplier_readiness_sel549.py b/services/ingestion/tests/test_supplier_readiness_sel549.py new file mode 100644 index 000000000..7cecddb92 --- /dev/null +++ b/services/ingestion/tests/test_supplier_readiness_sel549.py @@ -0,0 +1,191 @@ +"""SEL-549: second shipping format (EDI 856 ASN) maps into shipping KDEs. + +Preflight only. Synthetic fixtures under tests/fixtures/ only. +Modeled on /workspace/regengine-audit/supplier-readiness-{pass,fail}-case.md. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from app.edi_ingestion.extractors import _extract_856_fields +from app.edi_ingestion.parser import _parse_x12_segments +from app.supplier_readiness import ( + ALBERTSONS_BUYER_FIELDS, + dry_run_shipping_csv, + map_asn856_to_shipping_kdes, + score_shipping_readiness, +) +from app.webhook_models import REQUIRED_KDES_BY_CTE, WebhookCTEType + +FIXTURES = Path(__file__).parent / "fixtures" +EDI_DIR = FIXTURES / "edi_corpus" +ASN_EDI = EDI_DIR / "asn_856.edi" +ASN_EXPECTED = EDI_DIR / "asn_856.expected.json" +PASS_CSV = FIXTURES / "shipping_pass_partner.csv" +MESSY_CSV = FIXTURES / "shipping_messy_partner.csv" + +SHIPPING_KDES = list(REQUIRED_KDES_BY_CTE[WebhookCTEType.SHIPPING]) + + +def test_asn_856_fixture_matches_expected_lot_null(): + """Fixture truth: line_items[0].lot_number is null (fail-case §3).""" + expected = json.loads(ASN_EXPECTED.read_text(encoding="utf-8")) + assert expected["line_items"][0]["lot_number"] is None + assert expected["ship_from_name"] == "Valley Fresh Farms" + assert expected["reference_document_number"] == "BOL-9001" + assert "tlc_source_reference" not in expected + assert "gtin" not in expected + assert "sscc" not in expected + assert "po_number" not in expected + + +def test_asn_856_maps_into_shipping_kde_set_and_reports_gap(): + """Second format (856) → same shipping KDE keys; status=gap per fail case.""" + raw = ASN_EDI.read_text(encoding="utf-8") + segments = _parse_x12_segments(raw) + extracted = _extract_856_fields(segments) + expected = json.loads(ASN_EXPECTED.read_text(encoding="utf-8")) + assert extracted == expected + + result = map_asn856_to_shipping_kdes( + extracted, segments, source="edi_corpus/asn_856.edi" + ) + + # Same KDE set as CSV / webhook shipping contract + for key in SHIPPING_KDES: + assert key in result.kdes + + assert result.status == "gap" + assert result.label == "SYNTHETIC" + assert result.issue == "SEL-549" + + # Fail-case §3 / Lead: lot null → TLC missing; TLC source, GTIN, SSCC, PO missing + assert result.kdes["traceability_lot_code"] is None + assert "traceability_lot_code" in result.missing_repo_kdes + assert "tlc_source_reference" in result.missing_repo_kdes + assert "gtin" in result.missing_buyer_fields + assert "sscc" in result.missing_buyer_fields + assert "po_number" in result.missing_buyer_fields + + # Present fields still mapped (not dropped) + assert result.kdes["product_description"] == "ROMAINE-12CT" + assert result.kdes["quantity"] == 200.0 + assert result.kdes["unit_of_measure"] == "cases" # CA → cases + assert result.kdes["ship_date"] == "2026-03-10" + assert result.kdes["ship_from_location"] == "Valley Fresh Farms" + assert result.kdes["ship_to_location"] == "Metro Distribution Center" + assert result.kdes["reference_document"] == "BOL-9001" + + +def test_asn_856_unknown_segment_flagged_not_dropped(): + """Unknown EDI segment ids are flagged; known content still scores.""" + raw = ASN_EDI.read_text(encoding="utf-8") + # Inject a synthetic unknown segment before SE + mutated = raw.replace("~SE*", "~ZZZ*KEEP-ME*1~SE*") + segments = _parse_x12_segments(mutated) + extracted = _extract_856_fields(segments) + result = map_asn856_to_shipping_kdes(extracted, segments, source="mutated-asn") + + assert any(f.reason == "unrecognized_segment" and f.field == "ZZZ" for f in result.flags) + assert result.kdes["product_description"] == "ROMAINE-12CT" + assert result.status == "gap" + + +def test_pass_case_shipping_csv_scores_pass(): + """Pass-case synthetic shipping CSV: every repo KDE + Albertsons buyer field.""" + text = PASS_CSV.read_text(encoding="utf-8") + dry = dry_run_shipping_csv(text, source="shipping_pass_partner.csv") + assert dry["row_count"] == 1 + assert dry["status_counts"]["pass"] == 1 + result = dry["results"][0] + assert result.status == "pass" + assert result.missing_repo_kdes == [] + assert result.missing_buyer_fields == [] + for key in SHIPPING_KDES: + assert result.kdes.get(key) not in (None, "") + for key in ALBERTSONS_BUYER_FIELDS: + assert result.buyer.get(key) not in (None, "") + + +def test_fail_case_missing_repo_kde_is_gap(): + """Fail-case §2.1: empty TLC → gap (missing_kde), not pass.""" + scored = score_shipping_readiness( + { + "traceability_lot_code": None, + "product_description": "Organic Romaine Hearts", + "quantity": 200, + "unit_of_measure": "cases", + "ship_date": "2026-03-10", + "ship_from_location": "Valley Fresh Farms", + "ship_to_location": "Metro Distribution Center", + "reference_document": "BOL-9001", + "tlc_source_reference": "PACK-1", + }, + {"gtin": "1", "po_number": "PO-1", "sscc": "SSCC-1", "product_date": "2026-03-17"}, + ) + assert scored.status == "gap" + assert scored.missing_repo_kdes == ["traceability_lot_code"] + + +def test_fail_case_malformed_quantity_is_failure(): + """Fail-case §2.3: non-numeric quantity → failure.""" + scored = score_shipping_readiness( + { + "traceability_lot_code": "LOT-A", + "product_description": "Romaine", + "quantity": "not-a-number", + "unit_of_measure": "cases", + "ship_date": "2026-03-10", + "ship_from_location": "A", + "ship_to_location": "B", + "reference_document": "BOL-1", + "tlc_source_reference": "PACK-1", + }, + {"gtin": "1", "po_number": "PO-1", "sscc": "SSCC-1", "product_date": "2026-03-17"}, + ) + assert scored.status == "failure" + assert any(f["kind"] == "malformed_quantity" for f in scored.failures) + + +def test_unknown_csv_columns_flagged_and_kept(): + """Fail-case §2.8: unknown columns flagged, values kept on kdes.""" + text = PASS_CSV.read_text(encoding="utf-8") + # Add an unknown column to the pass fixture + lines = text.strip().splitlines() + header = lines[0] + ",partner_bin" + body = lines[1] + ",BIN-99" + dry = dry_run_shipping_csv(header + "\n" + body + "\n", source="pass-with-unknown") + result = dry["results"][0] + assert any(f.reason == "unrecognized_column" and f.field == "partner_bin" for f in result.flags) + assert result.kdes.get("partner_bin") == "BIN-99" + assert result.status == "pass" # extras do not block pass when KDEs complete + + +def test_messy_csv_dry_run_flag_counts_by_reason(): + """Dry run over synthetic messy shipping CSV; counts by reason for PR body.""" + text = MESSY_CSV.read_text(encoding="utf-8") + dry = dry_run_shipping_csv(text, source="shipping_messy_partner.csv") + + assert dry["label"] == "SYNTHETIC" + assert dry["issue"] == "SEL-549" + assert dry["row_count"] == 7 + + counts = dry["flagged_counts_by_reason"] + # Header-level unknown columns (same spirit as #3682): once each + assert counts.get("unrecognized_column") == 3 # partner_bin, mystery_code, extra_pallet_tag + assert counts.get("malformed_quantity") == 1 + assert counts.get("unrecognized_value") == 1 # bad UOM "blargs" + assert counts.get("malformed_date") == 1 # 2026-03-99 + assert counts.get("unknown_cte_type") == 1 + + assert dry["status_counts"]["gap"] >= 1 + assert dry["status_counts"]["failure"] >= 2 # bad qty + wrong CTE (+ invalid UOM) + statuses = [r.status for r in dry["results"]] + assert "gap" in statuses + assert "failure" in statuses + # Wrong CTE row is a failure (fail-case §2.7), not a silent gap + wrong = [r for r in dry["results"] if any(f.reason == "unknown_cte_type" for f in r.flags)] + assert len(wrong) == 1 and wrong[0].status == "failure"