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

126 lines
5.0 KiB
Python
Raw Permalink Normal View History

2021-09-14 10:33:08 +02:00
from db import dbGetMany
from loguru import logger
errorCnt = 0
def perform(dbh, params):
global errorCnt
2021-09-14 10:34:33 +02:00
checkTenant(dbh, params)
2021-09-14 10:33:08 +02:00
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")