"""Original DESADV detail projection laboratory; never a full EANCOM validator. Run with Python and pydifact==0.1.8. No credentials, network or WMS writes. The deliberately narrow LOCAL profile accepts one identified package per CPS, 201 pallets and CT cartons, single BJ identifiers, SRV product identifiers and integer QTY+12 lines. Other legal EDI shapes are unsupported, not declared invalid. Numeric identifiers below are fictitious, not GS1 allocations or checksum proofs. The input is a DETAIL FRAGMENT: no interchange, complete message or partner certification is asserted. The native tokenizer/parser remains unmodified. """ from collections import Counter from copy import deepcopy from hashlib import sha256 from importlib.metadata import version import json import platform import socket from pydifact.segmentcollection import RawSegmentCollection CHECKS = [] PARSE_OBSERVATIONS = [] CONNECTION_ATTEMPTS = [] def check(condition, name): if not condition: raise AssertionError(name) CHECKS.append(name) def deny_connection(*args, **kwargs): CONNECTION_ATTEMPTS.append("blocked") raise AssertionError("No real connection belongs to this laboratory") socket.create_connection = deny_connection socket.socket.connect = deny_connection socket.socket.connect_ex = deny_connection class UnsupportedProfile(ValueError): pass def reject(reason): raise UnsupportedProfile(reason) def native_parse(text, label): result = RawSegmentCollection.from_str(text) # This native method is a stub in the pinned source; returning None is not # a hierarchy, SSCC, partner profile or physical reception validation. value = result.validate() segments = list(result.segments) PARSE_OBSERVATIONS.append({"case": label, "segments": len(segments), "native_validate_return": value}) return segments def project(segments): nodes, seen_ids, rows = {}, set(), [] current = None pending = None for segment in segments: tag, elements = segment.tag, segment.elements if tag == "CPS": if pending is not None: reject("line_without_despatch_quantity") if not 1 <= len(elements) <= 2: reject("unsupported_CPS_shape") node_id = elements[0] parent = elements[1] if len(elements) == 2 else None if not isinstance(node_id, str) or not node_id or node_id in nodes: reject("missing_or_duplicate_CPS") if parent is not None and parent not in nodes: reject("parent_not_already_defined") if parent is None and nodes: reject("multiple_roots") current = {"cps": node_id, "parent": parent, "package_type": None, "sscc": None, "marking": False} nodes[node_id] = current elif tag == "PAC": if current is None or current["package_type"] is not None: reject("missing_CPS_or_repeated_PAC") if len(elements) != 3 or elements[:2] != ["1", ""]: reject("only_one_explicit_package_per_CPS_supported") kind = elements[2] if kind not in ["201", "CT"]: reject("unsupported_package_type") current["package_type"] = kind elif tag == "PCI": if current is None or current["package_type"] is None: reject("marking_without_package") if elements != ["33E"] or current["marking"]: reject("unsupported_or_repeated_PCI") current["marking"] = True elif tag == "GIN": if current is None or not current["marking"]: reject("identity_without_profile_marking") if len(elements) != 2 or elements[0] != "BJ": reject("only_single_BJ_identifier_supported") identifier = elements[1] if not isinstance(identifier, str) or not identifier: reject("ranges_not_supported") if current["sscc"] is not None or identifier in seen_ids: reject("duplicate_logistic_identifier") current["sscc"] = identifier seen_ids.add(identifier) elif tag == "LIN": if current is None or current["package_type"] != "CT" or not current["sscc"]: reject("line_without_identified_carton") if pending is not None: reject("line_without_despatch_quantity") if (len(elements) != 3 or elements[1] != "" or not isinstance(elements[2], list) or len(elements[2]) != 2 or elements[2][1] != "SRV" or not elements[2][0]): reject("unsupported_LIN_shape") pending = {"cps": current["cps"], "line": elements[0], "gtin": elements[2][0]} if not isinstance(pending["line"], str) or not pending["line"]: reject("missing_line_number") if any(row["line"] == pending["line"] for row in rows): reject("duplicate_line_number") elif tag == "QTY": if pending is None or len(elements) != 1: reject("quantity_without_pending_line") quantity = elements[0] if (not isinstance(quantity, list) or len(quantity) != 2 or quantity[0] != "12" or not quantity[1].isascii() or not quantity[1].isdigit()): reject("only_integer_despatch_quantity_supported") rows.append({**pending, "quantity": int(quantity[1])}) pending = None else: reject("unsupported_segment_" + tag) if not nodes or pending is not None: reject("empty_or_unfinished_detail") for node in nodes.values(): if not node["sscc"] or not node["package_type"]: reject("unidentified_package") parent = nodes.get(node["parent"]) if node["package_type"] == "201" and parent is not None: reject("only_root_pallet_supported") if node["package_type"] == "CT" and (parent is None or parent["package_type"] != "201"): reject("carton_requires_pallet_parent") result = [] for row in rows: carton = nodes[row["cps"]] pallet = nodes[carton["parent"]] result.append({**row, "carton_sscc": carton["sscc"], "pallet_sscc": pallet["sscc"]}) if not result: reject("no_despatch_lines") return result def flat_totals(rows): result = Counter() for row in rows: result[row["gtin"]] += row["quantity"] return dict(result) def require_approved_containment(rows, approved): actual = {(r["carton_sscc"], r["gtin"]): (r["pallet_sscc"], r["quantity"]) for r in rows} # Duplicate semantic keys cannot be silently overwritten by dict creation. if len(actual) != len(rows): reject("repeated_carton_product_requires_another_profile") if actual != approved: reject("containment_differs_from_local_approved_manifest") return deepcopy(rows) # Two independent partial details are combined only at the application level. # Each has ONE root pallet. This is a local projection fixture, not a complete # two-pallet DESADV message. The distinct fragments retain their context. DETAIL_A = """CPS+10' PAC+1++201' PCI+33E' GIN+BJ+000000000000000101' CPS+11+10' PAC+1++CT' PCI+33E' GIN+BJ+000000000000000118' LIN+1++12345678901231:SRV' QTY+12:4'""" DETAIL_B = """CPS+20' PAC+1++201' PCI+33E' GIN+BJ+000000000000000125' CPS+21+20' PAC+1++CT' PCI+33E' GIN+BJ+000000000000000132' LIN+2++12345678901231:SRV' QTY+12:6'""" def main(): check(version("pydifact") == "0.1.8", "pinned native parser") a = native_parse(DETAIL_A, "pallet_A") b = native_parse(DETAIL_B, "pallet_B") check(len(a) == len(b) == 10, "ten actual native segments per detail") check(a[4].elements == ["11", "10"], "native CPS retains own and parent IDs") check(a[3].elements == ["BJ", "000000000000000101"], "native logistic identifier retains leading zeros") check(a[8].elements == ["1", "", ["12345678901231", "SRV"]], "native LIN preserves empty element and product string") check(a[9].elements == [["12", "4"]], "native composite quantity is not a package count") rows = project(a) + project(b) check(rows[0]["pallet_sscc"] == "000000000000000101", "first carton follows its parent pallet") check(rows[1]["pallet_sscc"] == "000000000000000125", "second carton retains different pallet") check([r["quantity"] for r in rows] == [4, 6], "quantities remain at their carton lines") check(flat_totals(rows) == {"12345678901231": 10}, "flat product total is ten") approved = {(r["carton_sscc"], r["gtin"]): (r["pallet_sscc"], r["quantity"]) for r in rows} check(require_approved_containment(rows, approved) == rows, "local reference manifest admits original containment") swapped = deepcopy(rows) swapped[0]["pallet_sscc"], swapped[1]["pallet_sscc"] = swapped[1]["pallet_sscc"], swapped[0]["pallet_sscc"] check(flat_totals(swapped) == flat_totals(rows), "incorrect reassociation invisible in flat total") try: require_approved_containment(swapped, approved) except UnsupportedProfile as error: check(str(error) == "containment_differs_from_local_approved_manifest", "local reference refuses changed carton parents") else: raise AssertionError("changed containment admitted") bad_cases = [ ("missing_parent", DETAIL_A.replace("CPS+11+10", "CPS+11+99"), "parent_not_already_defined"), ("self_parent", DETAIL_A.replace("CPS+11+10", "CPS+11+11"), "parent_not_already_defined"), ("repeated_CPS", DETAIL_A.replace("CPS+11+10", "CPS+10+10"), "missing_or_duplicate_CPS"), ("carton_group", DETAIL_A.replace("PAC+1++CT", "PAC+3++CT"), "only_one_explicit_package_per_CPS_supported"), ("range_identity", DETAIL_A.replace("000000000000000118'", "000000000000000118:000000000000000132'"), "ranges_not_supported"), ("foreign_quantity", DETAIL_A.replace("QTY+12:4", "QTY+21:4"), "only_integer_despatch_quantity_supported"), ("fractional_quantity", DETAIL_A.replace("QTY+12:4", "QTY+12:4.5"), "only_integer_despatch_quantity_supported"), ("repeated_identity", DETAIL_A.replace("000000000000000118", "000000000000000101"), "duplicate_logistic_identifier"), ("missing_quantity", DETAIL_A.replace("QTY+12:4'", ""), "empty_or_unfinished_detail"), ] rejected = [] for label, text, reason in bad_cases: segments = native_parse(text, label) check(isinstance(segments, list), label + ": native raw parsing completed") try: project(segments) except UnsupportedProfile as error: check(str(error) == reason, label + ": narrow local contract refused") rejected.append({"case": label, "local_reason": str(error)}) else: raise AssertionError(label + " was accepted") check(all(x["native_validate_return"] is None for x in PARSE_OBSERVATIONS), "native raw validate stub does not enforce containment") check(not CONNECTION_ATTEMPTS, "no real network attempts") state = {"announcement": "projected_only", "warehouse_receipt": None, "erp_receipt": None} check(state["warehouse_receipt"] is state["erp_receipt"] is None, "announcement does not manufacture physical or ERP receipt") return {"python": platform.python_version(), "pydifact": version("pydifact"), "checks_passed": len(CHECKS), "checks": CHECKS, "native_parse_runs": len(PARSE_OBSERVATIONS), "native_parse_observations": PARSE_OBSERVATIONS, "original_detail_sha256": {"A": sha256(DETAIL_A.encode()).hexdigest(), "B": sha256(DETAIL_B.encode()).hexdigest()}, "projected_rows": rows, "flat_totals": flat_totals(rows), "swapped_parent_rows": swapped, "rejected_local_cases": rejected, "real_connection_attempts": len(CONNECTION_ATTEMPTS), "real_api_calls": 0, "state": state, "limits": ["partial detail fragments only", "no full EANCOM or partner-profile certificate", "no GS1 allocation/checksum verification", "no WMS or ERP write", "local memory only; no durable receipt", "approved manifest itself is fictional, not verified against physical goods"]} if __name__ == "__main__": print(json.dumps(main(), ensure_ascii=False, indent=2))