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) #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): nodes = 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 = get_bank_reserves(cursor, first_day) NR.append(('reserves', reserves, {'label_pos':'left'})) # lister les comptes de charges et produit accounts = get_income_and_expense_accounts(cursor) for a in accounts: tot_deb, tot_cred = get_sum_of_operations( cursor, first_day, a ) def get_income_and_expense_accounts(cursor): print("get_income_and_expense_accounts") cursor.execute( "SELECT account_number, label FROM llx_accounting_account WHERE reconcilable = 'PCG-BSC'" ) accounts = [] for account_number, label in cursor: tmp = str(account_number)[0] if tmp == '6' or tmp == '7': accounts.append(Account(label, account_number, int(tmp))) [print(f'account {a.number} / {a.label}') for a in accounts] return accounts def get_bank_reserves(cursor, date:str): 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 debit, label_compte in cursor: print(f'ajout de : {debit} provenant du compte {label_compte} au total') total += debit print(f'total = {total}') return total def connect_mariadb(db_infos:DBInfos): try: conn = mariadb.connect( user="db_user", password="db_user_passwd", host="192.0.2.1", port=3306, database="employees" ) 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) print('test') if tmp is not None: return tmp.group().split(',')[1].split('\'')[1] else: return "" def get_sum_of_operations(cursor, date:str, account:Account): 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 debit, credit in cursor : tot_cred += credit tot_deb += debit print(f'pour le compte {account.number} : credit = {tot_cred} / debit = {tot_deb} à partir du {date}') return tot_deb, 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 = extract_value(content, 'DB_NAME') 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'})) flows.append(('Total', desc, charge)) elif charge == 0 and revenue != 0: RL.append((desc, revenue, {'label_pos':'left'})) flows.append((desc, 'Total', revenue)) nodes = [ RL, [('Total', total)], 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) if __name__ == "__main__": 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}') respo = input('Appuyer sur entrée pour terminer')