# from libs import lib_mariadb as mdb # from libs import lib_csv as csv # from libs import lib_matplotlib as mtp import configparser as cp import os from sankeyflow import Sankey import matplotlib.pyplot as plt import re import datetime import mariadb import sys #node_central = 'total' class BudgetLine(): def __init__(self, desc:str, charge:int, revenue:int): self.desc = desc self.charge = charge self.revenue = revenue class DBInfos(): name:str user:str password:str host:str port:int class Account(): def __init__(self, label:str, number:int, tmp:int): self.label = label self.number = number self.type = tmp def main(path_b:str, path_wp:str, save_dir:str): budget=import_csv(path_b) create_budget_sankey(budget,save_dir) db_infos = get_db_infos(path_wp) cursor = connect_mariadb(db_infos) create_accounting_sankey(cursor, save_dir) #connection à la db # Lister les comptes qui m'intéressent # Pour chaque compte 7 puis 6 : # - calculer somme entre le 1 er janvier et ajd # - ajouter valeur dans flow et nodes def create_accounting_sankey(cursor, save_dir): NR = [] NC = [] flows = [] # extraire le compte et la caisse au 1 er janvier de l'année year = datetime.datetime.now().year first_day = rf'{year}-01-01' reserves = int(get_bank_reserves(cursor, first_day)) ro_name = 'reserves au début de l\'année' ra_name = 'reserves aujourd\'hui' t_name = 'Total' NR.append((ro_name, reserves, {'label_pos':'left', 'color':'#007e97'})) flows.append((ro_name, t_name, reserves)) # lister les comptes de charges et produit accounts = get_income_and_expense_accounts(cursor) tmp_res = reserves tot = reserves for a in accounts: tot_deb, tot_cred = get_sum_of_operations( cursor, first_day, a ) tmp_res = tmp_res + tot_cred - tot_deb if a.type == 6 : tmp_tot = tot_deb-tot_cred if tmp_tot > 0 : NC.append((a.label, tmp_tot, {'label_pos':'right', 'color':'#007e97' })) flows.append((t_name, a.label, tmp_tot)) else : tmp_tot = tot_cred-tot_deb if tmp_tot > 0 : tot += tmp_tot NR.append((a.label, tmp_tot, {'label_pos':'left', 'color':'#007e97' })) flows.append((a.label, t_name, tmp_tot)) NC.insert(0, (ra_name, tmp_res, {'label_pos':'right', 'color':'#007e97'})) flows.insert(0, (t_name, ra_name, tmp_res)) nodes = [ NR, [('Total', tot, {'label_pos':'top', 'color':'#007e97'})], NC ] plt.figure(figsize=(30, 10), dpi=144) s = Sankey( flows=flows, nodes=nodes, node_opts=dict(label_format='{label}:{value}') ) s.draw() path = os.path.join(save_dir, 'compta.svg') plt.savefig(path) plt.close() def get_income_and_expense_accounts(cursor): VERBOSE = False if VERBOSE : print("get_income_and_expense_accounts") cursor.execute( "SELECT account_number, label FROM llx_accounting_account WHERE fk_pcg_version = 'PCG-BSC' AND active = 1;" ) accounts = [] for account_number, label in cursor: tmp = account_number[0] if (tmp == '6' or tmp == '7') and int(account_number) > 10: accounts.append(Account(label, account_number, int(tmp))) if VERBOSE : [print(f'account {a.number} / {a.label}') for a in accounts] return accounts def get_bank_reserves(cursor, date:str): VERBOSE = False if VERBOSE : print('get_bank_reserves') total = 0 cursor.execute( "SELECT piece_num, numero_compte, label_compte, doc_date, code_journal, debit, credit FROM llx_accounting_bookkeeping WHERE (numero_compte = ? OR numero_compte = ?) AND doc_date = ? AND code_journal = ?;", (5121, 5311, date, 'AN',) ) for piece_num, numero_compte, label_compte, doc_date, code_journal, debit, credit in cursor: #print(piece_num, numero_compte, label_compte, doc_date, code_journal, debit, credit, sep=" | ") if VERBOSE : print(f'ajout de : {debit} provenant du compte {label_compte} au total') total += debit if VERBOSE : print(f'total = {total}') return total def connect_mariadb(db_infos:DBInfos): try: conn = mariadb.connect( user=db_infos.user, password=db_infos.password, host=db_infos.host, port=db_infos.port, database=db_infos.name ) except mariadb.Error as e: print(f"Error connecting to MariaDB Platform: {e}") sys.exit(1) # Get Cursor return conn.cursor() def import_csv(csv:str): with open(csv, "r") as file: content = file.read() values = [] for line in content.split('\n'): tmp = line.split(';') values.append( BudgetLine( desc=tmp[0], charge=int(tmp[1]), revenue=int(tmp[2]) ) ) return values def extract_value(content:str, param:str): tmp = re.search(rf'define...{param}.+', content) if tmp is not None: return tmp.group().split(',')[1].split('\'')[1] else: return "" def get_sum_of_operations(cursor, date:str, account:Account): VERBOSE = False tot_cred = tot_deb = 0 cursor.execute( "SELECT piece_num, numero_compte, label_compte, doc_date, code_journal, debit, credit FROM llx_accounting_bookkeeping WHERE numero_compte = ? AND doc_date >= ?", (account.number, date) ) for piece_num, numero_compte, label_compte, doc_date, code_journal, debit, credit in cursor : tot_cred += credit tot_deb += debit if VERBOSE : print(f'pour le compte {account.number} : credit = {tot_cred} / debit = {tot_deb} à partir du {date}') return int(tot_deb), int(tot_cred) def get_db_infos(path_wp:str): with open(path_wp, "r") as file: content = file.read() db_infos = DBInfos() db_infos.name = "doli_bsc" db_infos.user = extract_value(content, 'DB_USER') db_infos.password = extract_value(content, 'DB_PASSWORD') db_infos.host = extract_value(content, 'DB_HOST') db_infos.port = 3306 return db_infos def get_budget_elements(budget:list[BudgetLine]): nodes = [] flows=[] total = 0 CL = [] RL = [] for line in budget: charge = line.charge desc = line.desc revenue = line.revenue if charge != 0 and revenue == 0: total += charge CL.append((desc, charge, {'label_pos':'right', 'color':'#007e97' })) flows.append(('Total', desc, charge)) elif charge == 0 and revenue != 0: RL.append((desc, revenue, {'label_pos':'left', 'color':'#007e97' })) flows.append((desc, 'Total', revenue)) nodes = [ RL, [('Total', total, {'label_pos':'top', 'color':'#007e97' })], CL ] return nodes, flows def create_budget_sankey( budget:list[BudgetLine], save:str): nodes, flows = get_budget_elements(budget) plt.figure(figsize=(25, 10), dpi=144) s = Sankey( flows=flows, nodes=nodes, node_opts=dict(label_format='{label}:{value}') ) s.draw() path = os.path.join(save, 'budget.svg') plt.savefig(path) plt.close() if __name__ == "__main__": VERBOSE = False conf = cp.ConfigParser() conf.read('config') path_b = conf['path']['budget'] path_wp_config = conf['path']['wp-config'] path_save = conf['path']['save_directory'] if os.path.exists(path_b) and os.path.exists(path_wp_config): print("les 2 fichiers budget et wp-config ont été trouvé") main(path_b, path_wp_config, path_save) else : if not os.path.exists(path_b) : msg = "le chemin indiqué pour le paramètre budget dans le fichier config est incorrect. Le document a été supprimé ou déplacer. Merci de préciser un autre chemin" print(msg) print(f'budget = {path_b}') if not os.path.exists(path_wp_config) : msg = "le chemin indiqué pour le paramètre wp-config dans le fichier config est incorrect. Le document a été supprimé ou déplacer. Merci de préciser un autre chemin" print(msg) print(f'wp-config = {path_wp_config}') if VERBOSE : respo = input('Appuyer sur entrée pour terminer')