diff --git a/edifact_summary.py b/edifact_summary.py index 1ae4faf..4b2fbfb 100644 --- a/edifact_summary.py +++ b/edifact_summary.py @@ -31,6 +31,9 @@ "AAA": "Calculation net", } +PREFERRED_QTY_QUALIFIERS = ("21", "113", "194", "263") +PREFERRED_PRICE_QUALIFIERS = ("AAA",) + def safe_value(segment: dict[str, Any], element_index: int) -> str | None: elements = segment.get("elements", []) @@ -57,6 +60,38 @@ return values +def component_metadata(segment: dict[str, Any], element_index: int, component_index: int) -> dict[str, Any]: + elements = segment.get("elements", []) + element_idx = element_index - 1 + component_idx = component_index - 1 + if element_idx < 0 or element_idx >= len(elements): + return {} + components = elements[element_idx].get("components", []) + if not isinstance(components, list): + return {} + if component_idx < 0 or component_idx >= len(components): + return {} + component = components[component_idx] + return component if isinstance(component, dict) else {} + + +def qualifier_name( + segment: dict[str, Any], + element_index: int, + code: str, + fallback_names: dict[str, str], +) -> str: + name = fallback_names.get(code, "") + if name: + return name + semantic = component_metadata(segment, element_index, 1).get("semantic", {}) + if isinstance(semantic, dict): + semantic_name = semantic.get("qualifier_desc") + if isinstance(semantic_name, str): + return semantic_name + return "" + + def parse_decimal(value: str | None) -> float | None: if value is None or value == "": return None @@ -79,7 +114,7 @@ dates.append( { "qualifier": qualifier, - "qualifier_name": DATE_QUALIFIER_NAMES.get(qualifier, ""), + "qualifier_name": qualifier_name(segment, 1, qualifier, DATE_QUALIFIER_NAMES), "value": value, } ) @@ -97,8 +132,8 @@ refs.append( { "qualifier": parts[0], - "qualifier_name": REFERENCE_QUALIFIER_NAMES.get(parts[0], ""), - "value": parts[1] if len(parts) > 1 else "", + "qualifier_name": qualifier_name(segment, 1, parts[0], REFERENCE_QUALIFIER_NAMES), + "value": ":".join(parts[1:]) if len(parts) > 1 else "", } ) return refs @@ -175,7 +210,7 @@ current["quantities"].append( { "qualifier": parts[0], - "qualifier_name": QTY_QUALIFIER_NAMES.get(parts[0], ""), + "qualifier_name": qualifier_name(segment, 1, parts[0], QTY_QUALIFIER_NAMES), "value": parts[1], } ) @@ -187,7 +222,7 @@ current["prices"].append( { "qualifier": parts[0], - "qualifier_name": PRICE_QUALIFIER_NAMES.get(parts[0], ""), + "qualifier_name": qualifier_name(segment, 1, parts[0], PRICE_QUALIFIER_NAMES), "value": parts[1], } ) @@ -211,6 +246,14 @@ return segments +def pick_measure(entries: list[dict[str, Any]], preferred_qualifiers: tuple[str, ...]) -> dict[str, Any]: + for qualifier in preferred_qualifiers: + found = next((entry for entry in entries if entry.get("qualifier") == qualifier), None) + if found: + return found + return entries[0] if entries else {} + + def summarize_transaction(transaction: dict[str, Any]) -> dict[str, Any]: summary: dict[str, Any] = { "message_ref": transaction.get("id"), @@ -248,8 +291,10 @@ amount_sum = 0.0 has_amount = False for item in line_items: - qty = next((x.get("value") for x in item.get("quantities", []) if x.get("qualifier") == "21"), None) - price = next((x.get("value") for x in item.get("prices", []) if x.get("qualifier") == "AAA"), None) + qty_info = pick_measure(item.get("quantities", []), PREFERRED_QTY_QUALIFIERS) + price_info = pick_measure(item.get("prices", []), PREFERRED_PRICE_QUALIFIERS) + qty = qty_info.get("value") + price = price_info.get("value") qty_num = parse_decimal(qty if isinstance(qty, str) else None) price_num = parse_decimal(price if isinstance(price, str) else None) if qty_num is not None and price_num is not None: @@ -349,8 +394,8 @@ if line_items: line_rows: list[list[Any]] = [] for item in line_items: - qty_info = next((x for x in item.get("quantities", []) if x.get("qualifier") == "21"), {}) - price_info = next((x for x in item.get("prices", []) if x.get("qualifier") == "AAA"), {}) + qty_info = pick_measure(item.get("quantities", []), PREFERRED_QTY_QUALIFIERS) + price_info = pick_measure(item.get("prices", []), PREFERRED_PRICE_QUALIFIERS) qty = qty_info.get("value", "") price = price_info.get("value", "") qty_num = parse_decimal(qty if isinstance(qty, str) else None)