hv2-all-in-one/cli/ConsistencyCheck.py

from db import dbGetMany
from loguru import logger

errorCnt = 0

def perform(dbh, params):
    global errorCnt
    checkTenant(dbh, params)
    checkFlats(dbh, params)
    checkParkings(dbh, params)
    checkCommercialPremise(dbh, params)

    if (errorCnt > 0):
        logger.error(f"Total error count: {errorCnt}")

def checkTenant(dbh, params):
    global errorCnt
    tenants = dbGetMany(dbh, { "statement": "SELECT * FROM tenant_t", "params": () })
    for tenant in tenants:
        outPre = f"Tenant: {tenant['firstname']} {tenant['lastname']}"
        logger.info(outPre)

        # check tenancies
        tenancyCnt = 0
        flatTenancyCnt = 0
        tenancies = dbGetMany(dbh, { 
                "statement": "SELECT * FROM tenancy_t WHERE tenant = %s AND startdate < now() AND (enddate > now() or enddate is null)", 
                "params": (tenant['id'], )
            }
        )
        for tenancy in tenancies:
            tenancyCnt += 1
            if (tenancy['flat']):
                flatTenancyCnt += 1
                logger.info(f"{outPre}: Flat tenancy: {tenancy['id']}, start: {tenancy['startdate']}, end: {tenancy['enddate']}")
            if (tenancy['parking']):
                logger.info(f"{outPre}: Garage tenancy: {tenancy['id']}, start: {tenancy['startdate']}, end: {tenancy['enddate']}")
            if (tenancy['commercial_premise']):
                logger.info(f"{outPre}: Commercial premise tenancy: {tenancy['id']}, start: {tenancy['startdate']}, end: {tenancy['enddate']}")

        if (flatTenancyCnt == 0):
            logger.warning(f"{outPre}: no flat tenancy")
        if (flatTenancyCnt > 1):
            logger.warning(f"{outPre}: more than one flat tenancy ({flatTenancyCnt})")
        if (tenancyCnt == 0):
            logger.error(f"{outPre}: no tenancy at all")
            errorCnt += 1
            noCurrentTenancies = dbGetMany(dbh, {
                    "statement": "SELECT * FROM tenancy_t WHERE tenant = %s",
                    "params": (tenant['id'], )
                }
            )
            for noCurrentTenancy in noCurrentTenancies:
                logger.error(f"{outPre}: but: flat {noCurrentTenancy['flat']}, parking: {noCurrentTenancy['parking']}, commercial premise: {noCurrentTenancy['commercial_premise']}, start: {noCurrentTenancy['startdate']}, end: {noCurrentTenancy['enddate']}")

def _checkRentals(dbh, params, rentalType):
    global errorCnt
    table = f"{rentalType}_t"
    rentals = dbGetMany(dbh, {
            "statement": f"SELECT * FROM {table}",
            "params": ()
        }
    )
    for rental in rentals:
        outPre = f"{rentalType}: {rental['description']}, premise: {rental['premise']}"
        logger.info(outPre)

        if (rentalType == 'flat'):
            overheadMappingCnt = 0
            overheadMappings = dbGetMany(dbh, {
                    "statement": "SELECT * FROM overhead_advance_flat_mapping_t WHERE flat = %s",
                    "params": (rental['id'], )
                }
            )
            for overheadMapping in overheadMappings:
                overheadMappingCnt += 1
                logger.info(f"{outPre}: overhead mapping: {overheadMapping['id']}")
            if (overheadMappingCnt == 0):
                errorCnt += 1
                logger.error(f"{outPre}: no overhead mapping available")
            if (overheadMappingCnt > 1):
                errorCnt += 1
                logger.error(f"{outPre}: more than one overhead mapping available")

        tenancyCnt = 0
        tenancies = dbGetMany(dbh, {
                "statement": f"SELECT * FROM tenancy_t WHERE {rentalType} = %s AND startdate < now() AND (enddate > now() or enddate is null)",
                "params": (rental['id'], )
            }
        )
        for tenancy in tenancies:
            tenancyCnt += 1
            logger.info(f"{outPre}: tenant: {tenancy['tenant']}, start: {tenancy['startdate']}, end: {tenancy['enddate']}")

            feeMappingCnt = 0
            feeMappings = dbGetMany(dbh, {
                    "statement": "SELECT * FROM tenancy_fee_mapping_t where tenancy = %s",
                    "params": (tenancy['id'], )
                }
            )
            for feeMapping in feeMappings:
                feeMappingCnt += 1
                logger.info(f"{outPre}: fee mapping: {feeMapping['id']}")
            if (feeMappingCnt == 0):
                errorCnt += 1
                logger.error(f"{outPre}: no fee mapping available")
            if (feeMappingCnt > 1):
                errorCnt += 1
                logger.error(f"{outPre}: more than one fee mapping available")

        if (tenancyCnt == 0):
            errorCnt += 1
            logger.error(f"{outPre}: vacant")
        if (tenancyCnt > 1):
            errorCnt += 1
            logger.error(f"{outPre}: overbooked")


def checkFlats(dbh, params):
    _checkRentals(dbh, params, "flat")

def checkParkings(dbh, params):
    _checkRentals(dbh, params, "parking")

def checkCommercialPremise(dbh, params):
    _checkRentals(dbh, params, "commercial_premise")