Spaces:
Sleeping
Sleeping
| from typing import Any | |
| from cbh.api.client.dto import InputField | |
| from cbh.api.importer.utils import normalize_header | |
| FIELD_SYNONYMS: dict[InputField, tuple[str, ...]] = { | |
| InputField.CODE: ( | |
| "goederencode", | |
| "goederen code", | |
| "gn code", | |
| "gncode", | |
| "hs code", | |
| "hscode", | |
| "taric", | |
| "taric code", | |
| "commodity code", | |
| "goods code", | |
| "artikelcode", | |
| "code", | |
| ), | |
| InputField.COUNTRY: ( | |
| "land", | |
| "landcode", | |
| "land van oorsprong", | |
| "oorsprong", | |
| "land van bestemming", | |
| "bestemming", | |
| "country", | |
| "country of origin", | |
| "origin", | |
| "destination", | |
| "iso", | |
| ), | |
| InputField.DIRECTION: ( | |
| "richting", | |
| "direction", | |
| "invoer/uitvoer", | |
| "import/export", | |
| "soort aangifte", | |
| "movement", | |
| ), | |
| InputField.DATE: ( | |
| "datum", | |
| "date", | |
| "aangiftedatum", | |
| "declaration date", | |
| "verzenddatum", | |
| ), | |
| InputField.ADDITIONAL_CODE: ( | |
| "aanvullende code", | |
| "additional code", | |
| "taric additional code", | |
| ), | |
| InputField.DESCRIPTION: ( | |
| "omschrijving", | |
| "goederenomschrijving", | |
| "description", | |
| "goods description", | |
| "artikel", | |
| ), | |
| InputField.QUANTITY: ("aantal", "quantity", "qty", "hoeveelheid", "stuks"), | |
| InputField.PACKAGES: ("colli", "packages", "aantal colli", "pakketten", "cartons"), | |
| InputField.GROSS_WEIGHT: ( | |
| "brutogewicht", | |
| "bruto gewicht", | |
| "gross weight", | |
| "brutto", | |
| "gross", | |
| ), | |
| InputField.NET_WEIGHT: ("nettogewicht", "netto gewicht", "net weight", "netto", "net"), | |
| InputField.VALUE: ( | |
| "waarde", | |
| "value", | |
| "factuurwaarde", | |
| "invoice value", | |
| "statistische waarde", | |
| "amount", | |
| ), | |
| } | |
| def suggest_column_mapping(headers: list[Any]) -> dict[str, str]: | |
| normalized = [(normalize_header(header), str(header)) for header in headers if header] | |
| mapping: dict[str, str] = {} | |
| taken: set[str] = set() | |
| for field, synonyms in FIELD_SYNONYMS.items(): | |
| for synonym in synonyms: | |
| target = normalize_header(synonym) | |
| match = next( | |
| ( | |
| original | |
| for key, original in normalized | |
| if key == target and original not in taken | |
| ), | |
| None, | |
| ) | |
| if match: | |
| mapping[field.value] = match | |
| taken.add(match) | |
| break | |
| if field.value in mapping: | |
| continue | |
| for synonym in synonyms: | |
| target = normalize_header(synonym) | |
| match = next( | |
| ( | |
| original | |
| for key, original in normalized | |
| if target and target in key and original not in taken | |
| ), | |
| None, | |
| ) | |
| if match: | |
| mapping[field.value] = match | |
| taken.add(match) | |
| break | |
| return mapping | |
| def unmapped_headers(headers: list[Any], mapping: dict[str, str]) -> list[str]: | |
| claimed = set(mapping.values()) | |
| return [str(header) for header in headers if header and str(header) not in claimed] | |