Commit 52ab7077 authored by Pierre Dittgen's avatar Pierre Dittgen

No need for extra code to manage header errors

parent 89b557db
Pipeline #2396 passed with stage
in 1 minute and 39 seconds
from .cohesive_columns_value import CohesiveColumnsValue
from .compare_columns_value import CompareColumnsValue
from .extra_or_missing_header import ExtraOrMissingHeader
from .french_siren_value import FrenchSirenValue
from .french_siret_value import FrenchSiretValue
from .nomenclature_actes_value import NomenclatureActesValue
......
"""Reimplementation of NonMatchingHeader check,
taking into account missing header, extra header and wrong header order."""
from frictionless import errors, Check
class ExtraHeadersError(errors.HeaderError):
"""Custom error."""
code = "extra-headers"
name = "colonne(s) surnuméraire(s)"
tags = ["#head", "#structure"]
template = "{note}"
description = ""
class MissingHeadersError(errors.HeaderError):
"""Custom error."""
code = "missing-headers"
name = "colonne(s) manquante(s)"
tags = ["#head", "#structure"]
template = "{note}"
description = ""
class WrongHeadersOrderError(errors.HeaderError):
"""Custom error."""
code = "wrong-headers-order"
name = "en-têtes non ordonnés"
tags = ["#head", "#structure"]
template = "les colonnes du tableau ne sont pas dans l\'ordre défini par le schéma"
description = ""
class ExtraOrMissingHeader(Check):
"""Custom check."""
possible_Errors = [ExtraHeadersError, MissingHeadersError, WrongHeadersOrderError]
def validate_header(self, header):
field_names = self.table.schema.field_names
field_names_set = set(field_names)
header_set = set(header)
default_error_params = {
"cell": "",
"field_name": "",
"field_number":0,
"field_position":0
}
missing_headers = field_names_set - header_set
if missing_headers:
if len(missing_headers) == 1:
note = f"la colonne '{list(missing_headers)[0]}' n'a pas été trouvée dans le fichier"
else:
cols = ", ".join(f"'{col}'" for col in missing_headers)
note = f"les colonnes {cols} n'ont pas été trouvées dans le fichier"
yield MissingHeadersError(note=note, cells=list(missing_headers), **default_error_params)
extra_headers = header_set - field_names_set
if extra_headers:
if len(extra_headers) == 1:
note = f"la colonne '{list(extra_headers)[0]}' n'est pas déclarée dans le schéma"
else:
cols = ", ".join(f"'{col}'" for col in extra_headers)
note = f"les colonnes {cols} ne sont pas déclarées dans le schéma"
yield ExtraHeadersError(note=note, cells=list(extra_headers), **default_error_params)
if field_names_set == header_set and field_names != header:
yield WrongHeadersOrderError(note="", cells=[], **default_error_params)
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment