Code source de Consistency

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""Fonctions utilitaires pour assurer la cohérence de la base de données."""


# useful
import os
import sys
import time
import re
import random
# pour hasher les mdp
import hashlib
# module d'accès à la bdd
import psycopg2
import psycopg2.extras

# module qui définit des fonctions essentielles
import BaseFonctions
# module qui définit les erreurs
import ExceptionsNote
# module de configuration
sys.path.append("../config")
import config
# module d'envoi des mails
sys.path.append("../mail")
import mail

[docs]def dereference_historique_pseudo(pseudo, cur): """Enlève les références de l'ancien ``pseudo`` vers le compte, pour pouvoir l'utiliser comme pseudo ou alias. Ne commite pas le curseur si il a été fourni. """ cur.execute("UPDATE historique SET valide = false WHERE avant = %s;", (pseudo,))
[docs]def add_historique_pseudo(idbde, pseudo, cur): """Ajoute un changement de pseudo dans la table historique en faisant les modifications qui s'imposent.""" # D'abord, si quelqu'un avait ce pseudo dans son historique, ça ne le référence plus lui dereference_historique_pseudo(pseudo, cur) # On a ensuite besoin de connaître le pseudo actuel pour le mettre dans le champ "avant" cur.execute("SELECT pseudo FROM comptes WHERE idbde = %s;", (idbde,)) avant = cur.fetchall()[0]["pseudo"] # On insère la nouvelle valeur dans l'historique à la date now() cur.execute("INSERT INTO historique (idbde, avant, apres, date, valide) VALUES (%s, %s, %s, now(), true);", (idbde, avant, pseudo))
[docs]def expire_historique_pseudo(): """Invalide les anciens pseudos qui sont plus vieux que ``configurations.historique_pseudo_timeout`` jours""" con, cur = BaseFonctions.getcursor() cur.execute("""UPDATE historique SET valide = false WHERE date < now() - (SELECT CAST(historique_pseudo_timeout || 'd' AS interval) FROM configurations WHERE used = true);""") cur.execute("COMMIT;")
[docs]def check_consistency(sendmail=False, debug=False): """Vérifie que la base de données est cohérente selon plein de règles. En cas d'incohérence, et si ``sendmail = True``, envoie un mail à :py:data:`config.mails_integrity_problem` """ con, cur = BaseFonctions.getcursor() header = u"Note Kfet 2015 Consistency report : THERE IS A PROBLEM SOMEWHERE\n\n" footer = u"-- \nConsistency.py:check_consistency\nNote Kfet 2015" errmsg = u"" # Une et une seule configuration doit être à used = true if debug: print "Une seule configuration" cur.execute("SELECT count(*) FROM configurations WHERE used = true;") nbconfig = cur.fetchone()[0] if nbconfig != 1: errmsg += u" * La table configurations contient %s entrées à used = true (devrait être 1).\n\n" % nbconfig # La somme des transactions valides d'un utilisateur doit être égale à son solde if debug: print "somme des transactions valides = solde" cur.execute(""" SELECT idbde, pseudo, gagne-perdu AS somme_transactions, effectif AS solde_effectif, effectif-(gagne-perdu) AS en_trop FROM (SELECT idbde, pseudo, COALESCE((SELECT sum(t.montant * t.quantite) FROM transactions AS t WHERE t.destinataire = main.idbde AND t.valide), 0) AS gagne, COALESCE((SELECT sum(t.montant * t.quantite) FROM transactions AS t WHERE t.emetteur = main.idbde AND t.valide), 0) AS perdu, main.solde AS effectif FROM comptes AS main) AS calculs WHERE gagne-perdu != effectif ORDER BY idbde;""") l = cur.fetchall() if l: errmsg += u" * Les soldes des comptes suivants ne correspondent pas à la somme des transactions les concernant :\n" keys = ["idbde", "pseudo", "somme_transactions", "solde_effectif", "en_trop"] errmsg += BaseFonctions.sql_pretty_print(l, keys) + u"\n" # La somme de tous les soldes existants doit être nulle (en incluant bien sûr Espèces, Virements, Chèques et Bde) if debug: print "somme des soldes = 0" cur.execute("SELECT sum(solde) FROM comptes") total = cur.fetchone()[0] if total != 0: errmsg += u" * La somme de tous les soldes est %s (devrait être 0).\n\n" % total # La somme des montants des chèques doit être - le solde de Cheques (Cheques *donne* de l'argent sur un crédit) # Pareil pour les virements et les cartes bancaires if debug: print "somme des chèques/virements/cartes bancaires = montant de la note correspondante" for (table, mot, idbde, pseudo) in [("cheques", u"chèques", -1, u"Cheques"), ("virements", u"virements", -3, u"Virements"), ("carte_bancaires", u"carte bancaires", -4, "Carte bancaire")]: cur.execute("""SELECT COALESCE((SELECT sum(montant * quantite) FROM %s INNER JOIN transactions ON idtransaction=transactions.id WHERE valide AND NOT retrait), 0) - COALESCE((SELECT sum(montant * quantite) FROM %s INNER JOIN transactions ON idtransaction=transactions.id WHERE valide AND retrait), 0);""" % (table, table)) calcul = cur.fetchone()[0] cur.execute("SELECT solde FROM comptes WHERE idbde = %s;", (idbde,)) solde = cur.fetchone()[0] if calcul != - solde: errmsg += u" * La somme des montants des %s est %s alors que le solde de %s est %s.\n" % (mot, calcul, pseudo, solde) # Tout bouton pointe vers un club if debug: print "tout bouton pointe vers un club" cur.execute("""SELECT b.*, c.pseudo FROM boutons AS b, comptes AS c WHERE c.idbde = b.destinataire AND NOT c.type = 'club';""") l = cur.fetchall() if l: errmsg += u" * Les boutons suivants ne pointent pas vers un club :\n" errmsg += BaseFonctions.sql_pretty_print(l, ["id", "label", "montant", "destinataire", "pseudo", "categorie"]) + "\n" # Il ne doit pas y avoir de conflit pseudo/alias/historique valide if debug: print "pas de doublon pseudo/alias/historique valide" cur.execute(""" SELECT terme FROM ( SELECT DISTINCT idbde, terme FROM ((SELECT idbde, avant AS terme FROM historique WHERE valide) UNION ALL (SELECT idbde, alias AS terme FROM aliases) UNION ALL (SELECT idbde, pseudo AS terme FROM comptes) ) AS concat ) AS uniqueconcat GROUP BY terme HAVING count(*) > 1;""") l = cur.fetchall() if l: errmsg += u" * Les pseudos suivants apparaissent plus d'une fois dans les pseudos/alias/historiques :\n" errmsg += u"\n".join([repr(i["terme"]) for i in l]) + "\n\n" # comptes.last_adhesion doit pointer sur une adhésion existante (et d'idbde correspondant) if debug: print "les comptes ont une référence vers leur adhésion" cur.execute("""SELECT c.idbde, c.pseudo, c.nom, c.prenom, c.last_adhesion FROM comptes AS c WHERE idbde > 0 AND c.type = 'personne' AND NOT c.deleted AND NOT EXISTS (SELECT 1 FROM adhesions AS adh WHERE adh.idbde = c.idbde AND c.last_adhesion = adh.id);""") l = cur.fetchall() if l: errmsg += u" * Le champ last_adhesion des comptes suivant ne pointe pas vers une adhésion d'idbde correspondant :\n" errmsg += BaseFonctions.sql_pretty_print(l, ["idbde", "pseudo", "nom", "prenom", "last_adhesion"]) + "\n" # Les notes spéciales (et BDE) ne doivent pas être bloqués if debug: print "notes spéciales non bloquées" cur.execute("SELECT idbde, pseudo FROM comptes WHERE idbde <= 0 AND bloque;") l = cur.fetchall() if l: errmsg += u" * Les notes suivantes sont bloquées (et ne devraient pas l'être) :\n" errmsg += BaseFonctions.sql_pretty_print(l, ["idbde", "pseudo"]) + "\n" if errmsg: if sendmail: mail.queue_mail(config.mails_from, config.mails_integrity_problem, "[Note Kfet] Integrity test : **FAILED**", header + errmsg + footer) if debug: print errmsg.encode("utf-8") return errmsg else: return errmsg
if __name__ == "__main__": sendmail = False debug = False if "-m" in sys.argv or "--mail" in sys.argv: sendmail = True if "-d" in sys.argv or "--debug" in sys.argv: debug = True if "--expire-historique" in sys.argv: expire_historique_pseudo() if "--check-consistency" in sys.argv: errmsg = check_consistency(sendmail, debug) if errmsg and debug: sys.exit(42)