scripts-pde/qgis_sync_videores.py

'''

    Met à jour les résultats des contrôles ITV 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

    @author: olivier.massot, mai 2018
'''
from _datetime import datetime
import logging

from dateutil.relativedelta import relativedelta
from path import Path

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


logger = logging.getLogger("qgis_sync_videores")
logconf.start("qgis_sync_videores", 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"
# CSigDb.pwd = "Am5VOMkdFHU7WwrfVOs9"
# logger.handlers = [h for h in logger.handlers if (type(h) == logging.StreamHandler)]
# logger.warning("<<<<<<<<<<<<<<   Mode TEST   >>>>>>>>>>>>>>>>>")

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

def main():

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

    # Connexion à Controles
    controles_db = ControlesDb(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("Chargement des données")
    qessais = controles_db.read(Sql.format("SELECT lngChantierId, SI_AutoNumber, DG FROM csig_itv_results WHERE SI_Date > {:date}", date_min))

    for essai in qessais:

        troncon = csig_db.first(Sql.format("""SELECT t_chantiers.numero, t_troncons.si_autonumber, t_troncons.res_itv
                                              FROM (t_troncons INNER JOIN t_chantiers ON t_troncons.id_chantier = t_chantiers.id)
                                              WHERE t_troncons.si_autonumber={}
                                              """, essai.SI_AutoNumber))
        if not troncon:
            logger.warning("Le tronçon n'existe pas dans ControlesSIG (si_autonumber=%s)", essai.SI_AutoNumber)
            continue

        if troncon.res_itv != essai.DG:
            logger.info("Mise à jour du tronçon si_autonumber=%s", essai.SI_AutoNumber)
            csig_db.execute(Sql.format("""UPDATE t_troncons
                                          SET res_itv = {dg:text}
                                          WHERE si_autonumber = {si_autonumber}""", dg=essai.DG, si_autonumber=essai.SI_AutoNumber))

    csig_db.commit()


if __name__ == "__main__":
    main()
    logger.info("-- Fin --")