datacheck/core/validation.py

'''


    @author: olivier.massot, sept. 2018
'''
from datetime import datetime
import zipfile

from cerberus.validator import Validator
from path import Path, TempDir

from core.validation_errors import MissingFile, FormatError


class BaseModel():
    index = {}
    pk = ""
    schema = {}
    def __init__(self, **kwargs):
        self.__dict__.update(kwargs)
        
    @classmethod
    def indexer(cls, instance):
        if instance.getitem(cls.pk) in cls.index:
            raise ValueError("Duplicate PK")
        cls.index[instance.getitem(cls.pk)]
        
class BaseGeoModel(BaseModel):
    def __init__(self, geom, **kwargs):
        super(BaseGeoModel, self).__init__(**kwargs)
        self.geom = geom



class ValidationReport():
    def __init__(self, title = ""):
        self.title = title
        self.errors = {}
    
    @property
    def valid(self):
        return len(self.error) == 0




def is_french_date(field, value, error):
    try:
        datetime.strptime(value, '%d/%m/%Y')
    except:
        error(field, 'Doit être une date au format jj/mm/aaaa')

class BaseValidator():
    
    FILES = {}
    
    @classmethod
    def submit(cls, subject):    
        """ prends un dossier ou une archive en entrée et vérifie son contenu  """
        subject = Path(subject)
        
        if subject.isfile():
            with TempDir() as dirname:
                zip_ref = zipfile.ZipFile(subject, 'r')
                zip_ref.extractall(dirname)
                zip_ref.close()
                if Path(dirname / subject.stem).isdir(): # cas où l'archive contient un dossier qui lui-même contient les fichiers
                    dirname /= subject.stem
                return cls._submit_folder(dirname)
            
        elif subject.isdir():
            return cls._submit_folder(subject)
        else:
            raise FileNotFoundError(f"Impossible de trouver le fichier ou répertoire: {subject}")

    @classmethod
    def _submit_folder(cls, folder):
        
        dataset = {}
        report = ValidationReport("Contrôle des données de {} au format {}".format(folder.name, cls.name))
        
        # Charge les données en mémoire
        for filename, model in cls.files.items():
            path_ = Path(folder) / filename
            
            if not path_.isfile():
                report.errors[MissingFile] = MissingFile("Le fichier '{}' est manquant".format(filename))
                continue
            
            dataset[model] = cls._load_file(model, path_)
        
        # Controle la structure des données (champs, formats et types)
        cls._structure_validation(dataset)
        
        # Contrôle la géométrie (optionnel)
        cls._geometry_validation(dataset)
        
        # Validation technique
        cls._technical_validation(dataset)
    
        return report
    
    @classmethod
    def _load_file(cls, model, filename):
        """ Charge les données du fichier et les associe à un modèle.
        Attention: pas de contrôle de format o de validité à ce niveau!  """
        raise NotImplementedError()
    
    @classmethod        
    def _structure_validation(cls, dataset):
        
        errors = {}
        errors[FormatError] = []
        
        for model in dataset:
            v = Validator(model.schema)
            
            for item in dataset[model]:
                v.validate(item.__dict__)
                
                for fieldname, verrors in v.errors.items():
                    for err in verrors:
                        errors[FormatError].append(FormatError("{}: {}".format(fieldname, err)))
        
        return errors
    
    @classmethod
    def _geometry_validation(cls, dataset):
        pass
    
    @classmethod
    def _organize_dataset(cls, dataset):
        raise NotImplementedError()
    
    @classmethod
    def _technical_validation(cls, dataset):
        raise NotImplementedError()