Spaces:
Sleeping
Sleeping
File size: 3,364 Bytes
a9710f3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | 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]
|