scripts-pde/qgis_sync_etancheite.py

"""

    Met à jour les résultats des contrôles d'étanchéité des regards et tronçons de la base POSTGIS

    Les résultats à jour sont issus de la base Contrôles
    Pour des raisons de performances, seuls sont contrôlés les essais datant de moins de X mois,
    où X vaut 24 par défaut
    Pour être analysés, les noms des tronçons de la base Contrôles doivent être de la forme
    'R1-R2' ou 'R2-R1' (les espaces sont acceptés avant et aprés le tiret séparateur)

@author: olivier.massot, mai 2018
"""
from datetime import datetime
import logging
import re

from dateutil.relativedelta import relativedelta
from path import Path

from core import logconf
from core.model import Sql
from core.pde import ControlesDb, WincanDb, CSigDb


logger = logging.getLogger("qgis_sync_etancheite")
logconf.start("qgis_sync_etancheite", logging.DEBUG)

IMPORT_DEPUIS = 24  # Ne cherche des données à importer que sur les X derniers mois (mettre à 0 pour ignorer)



# # POUR TESTER, décommenter les lignes suivantes
##-----------------------------------------------

ControlesDb._path = Path(r"\\h2o\local\4-transversal\BDD\mdb_test\cg67Parc_data.mdb")
CSigDb.server = "TR-POSTGIS-02"
logger.handlers = [h for h in logger.handlers if (type(h) == logging.StreamHandler)]
logger.warning("<<<<<<<<<<<<<<   Mode TEST   >>>>>>>>>>>>>>>>>")

##-----------------------------------------------

# Connexion à ControlesSig (postgres)
csig_db = CSigDb(autocommit=False)

# Connexion à Controles
controles_db = ControlesDb(autocommit=False)

# Connexion à Wincan
wincan_db = WincanDb(autocommit=False)


# filter results on the last X months, depending on corresponding parameter
date_min = datetime.today() - relativedelta(months=IMPORT_DEPUIS) if IMPORT_DEPUIS >= 0 else datetime(1899, 12, 30, 0, 0, 0)


logger.info("Loading data")

# NB: utiliser une erquête access pré-enregistrée permet d'améliorer les perfs (la requete est pré-compilée)
qessais = controles_db.read(Sql.format("""SELECT lngChantierId, bytIntervId, strTrcRegard, strResSigne, strObjetEssai
                                          FROM csig_etancheite_results WHERE dtmEssai > {:date}""", date_min))

nb_r, nb_t = 0, 0

for essai in qessais:

    # L'essai concerne un regard
    if essai.strObjetEssai == "R":
        # # Cherche le regard correspondant dans qregards

        regard = csig_db.first(Sql.format("""SELECT t_regards.id, t_regards.res_ce
                                    FROM t_regards INNER JOIN t_chantiers ON t_regards.id_chantier = t_chantiers.id
                                    WHERE t_regards.nom={:text} AND t_chantiers.numero={} AND t_regards.archive=False
                                    """, essai.strTrcRegard, essai.lngChantierId))

        if not regard:
            logger.info("Le regard n'existe pas dans CSig: {}, {}".format(essai.lngChantierId, essai.strTrcRegard))
            continue

        if regard.res_ce != essai.strResSigne:
            q = csig_db.execute(Sql.format("""UPDATE t_regards
                                              SET res_ce = {res_ce:text}
                                              WHERE id = {pgid}
                                              """, res_ce=essai.strResSigne, pgid=regard.id))
            nb_r += 1
            logger.info("Résultat mis à jour: %s, %s > %s", essai.lngChantierId, essai.strTrcRegard, essai.strResSigne)


    # L'essai concerne un tronçon
    elif essai.strObjetEssai == "T":

        # Parse le nom de tronçon
        regex = re.search(r"^(\S*)\s?-\s?(\S*)$", essai.strTrcRegard)
        if not regex:
            logger.error(u"Nom de tronçon invalide: '{}' (chantier: {})".format(essai.strTrcRegard, essai.lngChantierId))
            continue
        r1, r2 = regex.group(1), regex.group(2)

        troncon = csig_db.first(Sql.format("""SELECT t_troncons.id, t_chantiers.numero, t_regards_1.nom AS r1,
                                    t_regards.nom AS r2, t_troncons.res_ce, t_troncons.nom as nom
                                    FROM ((t_troncons INNER JOIN t_chantiers ON t_troncons.id_chantier = t_chantiers.id)
                                    INNER JOIN t_regards AS t_regards_1 ON t_troncons.id_regard_depart = t_regards_1.id)
                                    INNER JOIN t_regards ON t_troncons.id_regard_fin = t_regards.id
                                    WHERE t_regards_1.nom={:text} AND t_regards.nom={:text} AND t_chantiers.numero={} AND t_troncons.archive=False
                                    """, r1, r2, essai.lngChantierId))

        if not troncon:
            logger.info("Le tronçon n'existe pas dans CSig: {}, {}".format(essai.lngChantierId, essai.strTrcRegard))
            continue

        if troncon.res_ce != essai.strResSigne:
            q = csig_db.execute(Sql.format("""UPDATE t_troncons
                                              SET res_ce = {res_ce:text}
                                              WHERE id = {pgid};""", res_ce=essai.strResSigne, pgid=troncon.id))
            nb_t += 1
            logger.info("Résultat mis à jour: %s, %s > %s", essai.lngChantierId, essai.strTrcRegard, essai.strResSigne)

#     csig_db.commit()

logger.info("- Opération terminée -")
logger.info("%s regards et %s tronçons mis à jour", nb_r, nb_t)