datacheck/main.py

'''
Python 3.7+

@author: olivier.massot, sept 2018
'''
from dataclasses import dataclass
from datetime import datetime
import logging
import sys

from path import Path
import shapefile
import yaml

from core import logconf
from core.constants import MAIN


logger = logging.getLogger("netgeo_checker")
logconf.start("netgeo_checker", logging.INFO)

ETUDE, IN_DIR = False, MAIN / "work" / "SCOPELEC_CAP_097AP0_REC_180829_OK"
# ETUDE, IN_DIR = True, MAIN / "rsc" / "data_in" / "SCOPELEC_101BP0_APD_180725"


subject = MAIN / "work" / "SCOPELEC_CAP_097AP0_REC_180829_OK"
checker = MAIN / "resources" / "netgeo_v2-2_doe.yaml"

# prendre une archive en entrée
# contrôler la presence des 6 fichiers shp
# Pour chaque couche, vérifier le type de géometrie, la presence de données, la projection, l'inclusion du bounding rect dans le departement
# Charger les modèles
# Vérifier enregistrement par enregistrement la validité des données



def check(subject, checker):    
    """ prends un dossier ou une archive en entier et vérifie son contenu selon les règles données par le fichier de config """
    archive, checker = Path(subject), Path(checker)
    
    if archive.isfile():
        # extraire vers un dossier temp
        # dirname = tempdir
        pass
    elif archive.isdir():
        dirname = subject
    else:
        raise IOError(f"Impossible de trouver le fichier ou répertoire: {subject}")
    
    logging.info("*****   Traitement de '%s'   *****", subject.name)
    
    logging.info("> Controlleur: '%s'", checker.name)
    
    with open(checker, "r") as cf:
        config = yaml.load(cf)
    
    for filename, model in config["files"].items():
        path_ = dirname / filename
        logging.info("* Traitement de %s", path_.name)
        
        try:
            sf = shapefile.Reader(path_)
        except shapefile.ShapefileException:
            logger.error("Fichier SHAPE illisible")
            continue
        
        shape_names = {1:"Point", 3:"Polyligne", 5:"Polygone"}
        if sf.shapeType != model["shape_type"]:
            logger.error("Le fichier shapefile n'est pas de type %s", shape_names[model["shape_type"]])
            del sf
            continue

        records = sf.shapeRecords()
        if not records:
            if not model["can_be_empty"]:
                logger.error("Le fichier shapefile ne contient aucune donnees")
                del sf, records
                continue
            else:
                logger.warning("Le fichier shapefile ne contient aucune donnees")
        
        if not "fields" in model:
            continue
        
        fields = [f[0] for f in sf.fields if f[0] != 'DeletionFlag']
        
        # controle d'éventuels champs inconnus
        for f in fields:
            if f not in model["fields"]:
                logger.warning("Champs inconnu: %s", f)
        
        # parcours et controle des enregistrements
        for i, record in enumerate(records):
            
            logging.info("\n> Enregistrement n°%s\n", i)
            record_data = {field: record.record[i] for i, field in enumerate(fields)}
            
            for fieldname, fieldmodel in model["fields"].items():
                
                try:
                    val = record_data[fieldname]
                except KeyError:
                    if fieldmodel.get("required", True):
                        logger.error("%s - Champs manquant", fieldname)
                    continue
                
                type_ = fieldmodel.get("type", "str")
                if type_ == "float":
                    try:
                        _ = float(val)
                    except (TypeError, ValueError):
                        logger.error("%s - Valeur invalide, un flottant est attendu ('%s')", fieldname, val)
                        continue
                elif type_ == "datetime":
                    try:
                        _ = datetime.strptime(val, fieldmodel.get("date_format", "%d/%m/%Y"))
                    except ValueError:
                        logger.error("%s - Valeur invalide, une date est attendu ('%s')", fieldname, val)
                        continue
                else: 
                    if not fieldmodel.get("allow_empty", False) and not val:
                        logger.error("%s - Champs vide", fieldname)
                        continue
            
                try:
                    if not val in fieldmodel["in_list"]:
                        logger.error("%s - Valeur invalide, pas dans la liste ('%s')", fieldname, val)
                        continue
                except KeyError:
                    pass
        
        del sf, records


if __name__ == "__main__":
    check(subject, checker)
    logger.info("-- Fin --")