diff --git a/.$functionnal diagrams.drawio.bkp b/.$functionnal diagrams.drawio.bkp deleted file mode 100644 index a455408..0000000 --- a/.$functionnal diagrams.drawio.bkp +++ /dev/null @@ -1,10 +0,0 @@ - - - - - - - - - - diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..21e0ca1 --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,11 @@ +{ + "python.testing.unittestArgs": [ + "-v", + "-s", + "./tests", + "-p", + "*test*.py" + ], + "python.testing.pytestEnabled": false, + "python.testing.unittestEnabled": true +} \ No newline at end of file diff --git a/__pycache__/main.cpython-311.pyc b/__pycache__/main.cpython-311.pyc new file mode 100644 index 0000000..42da899 Binary files /dev/null and b/__pycache__/main.cpython-311.pyc differ diff --git a/config.ori b/config.ori index a13aef1..a8a3cda 100644 --- a/config.ori +++ b/config.ori @@ -1,4 +1,10 @@ [path] -wp-config=/var/www/bricolesocialclub.org/wp-config.php -budget=budget.csv +budget=files/budget.csv save_directory=/var/www/bsc.pbblanc.fr/wp-content/uploads/compta + +[db] +name=doli_bsc +user= +passwd= +host=localhost +port=3306 \ No newline at end of file diff --git a/budget.csv b/files/budget.csv similarity index 100% rename from budget.csv rename to files/budget.csv diff --git a/budget.svg b/files/budget.svg similarity index 100% rename from budget.svg rename to files/budget.svg diff --git a/compta.svg b/files/compta.svg similarity index 100% rename from compta.svg rename to files/compta.svg diff --git a/ratios.svg b/files/ratios.svg similarity index 100% rename from ratios.svg rename to files/ratios.svg diff --git a/total.svg b/files/total.svg similarity index 100% rename from total.svg rename to files/total.svg diff --git a/libs/DBConnection.py b/libs/DBConnection.py new file mode 100644 index 0000000..55cfca8 --- /dev/null +++ b/libs/DBConnection.py @@ -0,0 +1,41 @@ +import mariadb +import sys +from configparser import ConfigParser as CP +class DBConnection(): + + def __init__(self, conf:CP) : + self.name:str = conf['db']['name'] + self.user:str = conf['db']['user'] + self.password:str = conf['db']['password'] + self.host:str = conf['db']['host'] + try : + self.port:int = int(conf['db']['port']) + except: + msg = f"Le numéro de port indiqué dans le fichier de configuration ne semble pas être un nombre entier: {conf['db']['port']}. Merci de corriger et de relancer le programme." + print(msg) + sys.exit(1) + + self.connect_mariadb() + + + def connect_mariadb(self): + try: + self.conn = mariadb.connect( + user=self.user, + password=self.password, + host=self.host, + port=self.port, + database=self.name + + ) + except mariadb.Error as e: + print(f"Error connecting to MariaDB Platform: {e}") + sys.exit(1) + + + def get_conn(self): + return self.conn + + + def close(self): + self.conn.close() \ No newline at end of file diff --git a/libs/__pycache__/DBConnection.cpython-311.pyc b/libs/__pycache__/DBConnection.cpython-311.pyc new file mode 100644 index 0000000..5721b86 Binary files /dev/null and b/libs/__pycache__/DBConnection.cpython-311.pyc differ diff --git a/libs/__pycache__/reporting_finance.cpython-311.pyc b/libs/__pycache__/reporting_finance.cpython-311.pyc new file mode 100644 index 0000000..3a67458 Binary files /dev/null and b/libs/__pycache__/reporting_finance.cpython-311.pyc differ diff --git a/libs/lib_csv.py b/libs/lib_csv.py deleted file mode 100644 index 7a0e46c..0000000 --- a/libs/lib_csv.py +++ /dev/null @@ -1,2 +0,0 @@ -from typing import TypedDict - diff --git a/libs/lib_mariadb.py b/libs/lib_mariadb.py deleted file mode 100644 index 50ee4e8..0000000 --- a/libs/lib_mariadb.py +++ /dev/null @@ -1,8 +0,0 @@ -def connect_db(): - - return test - - -def request_db(): - - return array \ No newline at end of file diff --git a/libs/lib_matplotlib.py b/libs/lib_matplotlib.py deleted file mode 100644 index 139597f..0000000 --- a/libs/lib_matplotlib.py +++ /dev/null @@ -1,2 +0,0 @@ - - diff --git a/libs/reporting_finance.py b/libs/reporting_finance.py new file mode 100644 index 0000000..122a2b0 --- /dev/null +++ b/libs/reporting_finance.py @@ -0,0 +1,431 @@ +from .DBConnection import DBConnection +import os +from sankeyflow import Sankey +import matplotlib.pyplot as plt +import matplotlib.patches as mpatches +import datetime +import numpy as np +from colorsys import hls_to_rgb +import re + + + +class BudgetLine(): + def __init__(self, desc:str, charge:int, revenue:int): + self.desc = desc + self.charge = charge + self.revenue = revenue + + +class Account(): + def __init__(self, label:str, number:int, tmp:int): + self.label = label + self.number = number + self.type = tmp + + +class FinanceDBConnection(DBConnection): + def get_sum_of_operations(self, date:str, account:Account): + VERBOSE = False + tot_cred = tot_deb = 0 + self.conn.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) + ) + last_entrie = date + for piece_num, numero_compte, label_compte, doc_date, code_journal, debit, credit in self.conn : + tot_cred += credit + tot_deb += debit + if str(doc_date) > last_entrie : + last_entrie = str(doc_date) + + 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), last_entrie + + def get_income_and_expense_accounts(self): + VERBOSE = False + if VERBOSE : print("get_income_and_expense_accounts") + self.conn.execute( + "SELECT account_number, label FROM llx_accounting_account WHERE fk_pcg_version = 'PCG-BSC' AND active = 1;" + ) + accounts = [] + for account_number, label in self.conn: + 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(self, date:str): + VERBOSE = False + if VERBOSE : print('get_bank_reserves') + total = 0 + self.conn.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 self.conn: + #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 + + + +class FinanceTable(): + def __init__(self): + self.ft:list[BudgetLine] + + def sort_des_table(self, li:int): + if li < len(self.ft)-1 : + ch = self.ft[li].charge + rev = self.ft[li].revenue + if (ch > 0 and ch >= self.ft[li+1].charge) or (ch == 0 and self.ft[li+1].charge ==0 and rev >=self.ft[li+1].revenue ): + self.sort_des_table(li+1) + else : + if li == 0 : + self.ft = [self.ft[li+1], self.ft[li]] + self.ft[li+2:] + self.sort_des_table( li) + elif li < len(self.ft) - 2 : + self.ft = self.ft[:li] + [self.ft[li+1], self.ft[li]] + self.ft[li+2:] + self.sort_des_table( li-1) + else : + self.ft = self.ft[:li] + [self.ft[li+1], self.ft[li]] + self.sort_des_table(li-1) + + + def get_revenue_ratios(self, tot:int, budget:list[BudgetLine]): + values = {} + oth = 0 + coef = 4 + i = 0 + for line in budget: + if line.revenue >0 : + if i < coef: + values.update( + {line.desc:line.revenue} + ) + i+=1 + else: + oth += line.revenue + if oth > 0 : + values.update( + {'Autres':oth} + ) + return values + + + def get_charge_ratios(self, tot:int, budget:list[BudgetLine]): + values = {} + oth = 0 + coef = 4 + i = 0 + for line in budget: + if line.charge >0 : + if i < coef: + values.update( + {line.desc:line.charge} + ) + i+=1 + else: + oth += line.charge + if oth > 0 : + values.update( + {'Autres':oth} + ) + return values + + + def get_total_charges(self, budget:list[BudgetLine]): + tot = 0 + for line in budget: + tot += line.charge + return int(tot) + + + def get_total_revenues(self, budget:list[BudgetLine]): + tot = 0 + for line in budget: + tot += line.revenue + return int(tot) + + + + +class Budget(FinanceTable): + def __init__(self, csv:str): + with open(csv, "r") as file: + content = file.read() + self.ft = [] + for line in content.split('\n'): + tmp = line.split(';') + self.ft.append( + BudgetLine( + desc=tmp[0], + charge=int(tmp[1]), + revenue=int(tmp[2]) + ) + ) + +class Finance(FinanceTable): + def __init__(self, cursor): + # 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' + accounts = get_income_and_expense_accounts(cursor) + self.table = [] + self.last_update = first_day + for a in accounts: + tot_deb, tot_cred, last_entrie = get_sum_of_operations( + cursor, + first_day, + a + ) + if a.type == 6 : + #charge + self.table.append(BudgetLine(a.label, tot_deb - tot_cred, 0)) + elif a.type == 7: + self.table.append(BudgetLine(a.label, 0, tot_cred - tot_deb)) + if last_entrie >= self.last_update : + self.last_update = last_entrie + + + + + + +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_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() + +def get_totals_lists_for_charts(tot_ch_bud, tot_ch_real, tot_re_bud, tot_re_real): + rest_ch = abs(tot_ch_bud - tot_ch_real) + rest_re = abs(tot_re_bud - tot_re_real) + if tot_ch_bud >= tot_ch_real and tot_re_bud >= tot_re_real: + + values = { + f'actuel' : [tot_re_real, tot_ch_real], + f'budget' : [rest_re, rest_ch] + } + else : + values = { + f'bot_budget' : [int(0.98 * tot_re_bud), int(0.98 * tot_ch_bud)], + f'budget' : [int(0.02 *tot_re_bud), int(0.02 *tot_ch_bud)], + f'actuel' : [rest_re, rest_ch] + } + return values + + +def hls_to_rgbs(s_hsl): + rgb = hls_to_rgb(s_hsl["Hue"], s_hsl["Luminosité"], s_hsl["Saturation"] ) + return "#" + "".join("%02X" % round(i*255) for i in rgb ) + + +def get_color_list(nb, s_hsl, sens = "Blanc") : + temp = s_hsl + luminosite = temp["Luminosité"] + + if sens == "Blanc": + ecart = (1-luminosite ) /(nb +1) + else : + ecart = luminosite / (nb + 1) + + l_couleur = [] + for x in range(nb) : + l_couleur.append(hls_to_rgbs(temp )) + if sens == "Blanc" : + luminosite += ecart + else : + print(luminosite) + luminosite -= ecart + temp["Luminosité"] = luminosite + return l_couleur + + +def create_pie_diagrams(path:str, + revenue_ratios:dict[str, int], + charge_ratios:dict[str, int], + last_update:str): + + bleu_hls = {"Hue":190 / 360, + "Saturation":100/100, + "Luminosité":30/100} + + marron_hls = {"Hue":23 / 360, + "Saturation":72/100, + "Luminosité":30/100} + + fig, axs = plt.subplots(ncols=2,figsize=(10, 4), layout='constrained') + labels, values = transform_dict_for_diag(revenue_ratios) + colors = get_color_list(len(values),bleu_hls) + p = axs[0].pie(x=values, labels=labels, colors=colors) + + labels, values = transform_dict_for_diag(charge_ratios) + colors = get_color_list(len(values),marron_hls) + p = axs[1].pie(x=values, labels=labels, colors=colors) + + axs[1].set_title('Répartition des dépenses réalisées') + axs[0].set_title('Répartition des recettes réalisées') + path = os.path.join(path, 'ratios.svg') + plt.savefig(path) + plt.close() + + +def transform_dict_for_diag(dicts:dict): + labels = [] + values = [] + for label, value in dicts.items(): + labels.append(f'{label}\n{value}€') + values.append(value) + + return labels, values + + +def create_total_diagram(path:str, + totals: dict[str, list[int]], + last_upadate:str): + + fig, ax = plt.subplots() + plt.grid(True, 'major', 'y', color="#555555") + width = 0.6 + types = ('dépenses', 'revenus') + leg = [mpatches.Patch(color='#007e97', label=f'au {last_upadate}'), + mpatches.Patch(color='#999999', label='budgeté') + ] + bottom = np.zeros(2) + if len(totals) ==2 : + col = ['#007e97', '#999999'] + else: + col = ['#007e97', '#999999', '#007e97'] + + x = 0 + for position, values in totals.items(): + p = ax.bar(types, values, width, label=position, bottom=bottom, color=col[x] ) + bottom += values + x+=1 + + #ax.invert_yaxis() # labels read top-to-bottom + + ax.set_title('Total des dépenses et des revenus de 2025') + box = ax.get_position() + ax.set_position( + [box.x0, box.y0 + box.height * 0.1, + box.width, box.height * 0.9]# type: ignore + ) + ax.legend(handles=leg, loc='upper center', bbox_to_anchor=(0.5, -0.07), + fancybox=True, shadow=True, ncol=2) + path = os.path.join(path, 'total.svg') + plt.savefig(path) + plt.close() + + +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, last_entrie = 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() + diff --git a/main.py b/main.py index 42f78c3..8c78ac2 100644 --- a/main.py +++ b/main.py @@ -1,72 +1,45 @@ # 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 +from libs import reporting_finance as report +from libs.reporting_finance import Budget, FinanceDBConnection +from configparser import ConfigParser as cp import os from sankeyflow import Sankey import matplotlib.pyplot as plt import matplotlib.patches as mpatches import re import datetime -import mariadb -import sys -import math import numpy as np from colorsys import hls_to_rgb #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, + client:FinanceDBConnection, save_dir:str): - budget=import_csv(path_b) - budget = sort_des_budget(budget, 0) + budget = Budget(path_b) + budget.sort_des_table(0) #create_budget_sankey(budget,save_dir) - db_infos = get_db_infos(path_wp) - cursor = connect_mariadb(db_infos) - create_accounting_sankey(cursor, save_dir) - finances, last_update = import_finances(cursor) - #finances = import_csv(path_b) - #finances = test_finance(finances) - #last_update = rf'2025-01-01' - finances = sort_des_budget(finances, 0) - tot_rev = get_total_revenues(finances) - tot_ch = get_total_charges(finances) - totals = get_totals_lists_for_charts( - tot_ch_bud = get_total_charges(budget), + + #create_accounting_sankey(client.get_conn(), save_dir) + + finances, last_update = report.import_finances(client.get_conn()) + + finances = report.sort_des_budget(finances, 0) + tot_rev = report.get_total_revenues(finances) + tot_ch = report.get_total_charges(finances) + totals = report.get_totals_lists_for_charts( + tot_ch_bud = report.get_total_charges(budget), tot_ch_real = tot_ch, - tot_re_bud = get_total_revenues(budget), + tot_re_bud = report.get_total_revenues(budget), tot_re_real = tot_rev ) - revenue_ratios = get_revenue_ratios(tot_rev, finances) - charge_ratios = get_charge_ratios(tot_ch, finances) - create_total_diagram(save_dir, totals, last_update) - create_pie_diagrams(save_dir, revenue_ratios, charge_ratios, last_update) + revenue_ratios = report.get_revenue_ratios(tot_rev, finances) + charge_ratios = report.get_charge_ratios(tot_ch, finances) + report.create_total_diagram(save_dir, totals, last_update) + report.create_pie_diagrams(save_dir, revenue_ratios, charge_ratios, last_update) print('oucou') #connection à la db @@ -77,473 +50,29 @@ def main(path_b:str, # - ajouter valeur dans flow et nodes -def test_finance(budget:list[BudgetLine]): - coef = -500 - for line in budget: - if coef > 0 : - if line.revenue == 0 : - line.charge += coef - else : - line.revenue += coef - if coef < 0 : - if line.revenue == 0 and line.charge > abs(coef) : - line.charge += coef - elif line.charge == 0 and line.revenue > abs(coef) : - line.revenue += coef - return budget -def sort_des_budget(budget:list[BudgetLine], li:int): - if li < len(budget)-1 : - ch = budget[li].charge - rev = budget[li].revenue - if (ch > 0 and ch >= budget[li+1].charge) or (ch == 0 and budget[li+1].charge ==0 and rev >=budget[li+1].revenue ): - budget = sort_des_budget(budget, li+1) - else : - if li == 0 : - budget = [budget[li+1], budget[li]] + budget[li+2:] - budget = sort_des_budget(budget, li) - elif li < len(budget) - 2 : - budget = budget[:li] + [budget[li+1], budget[li]] + budget[li+2:] - budget = sort_des_budget(budget, li-1) - else : - budget = budget[:li] + [budget[li+1], budget[li]] - budget = sort_des_budget(budget, li-1) - - return budget - - -def get_revenue_ratios(tot:int, budget:list[BudgetLine]): - values = {} - oth = 0 - coef = 4 - i = 0 - for line in budget: - if line.revenue >0 : - if i < coef: - values.update( - {line.desc:line.revenue} - ) - i+=1 - else: - oth += line.revenue - if oth > 0 : - values.update( - {'Autres':oth} - ) - return values - - -def get_charge_ratios(tot:int, budget:list[BudgetLine]): - values = {} - oth = 0 - coef = 4 - i = 0 - for line in budget: - if line.charge >0 : - if i < coef: - values.update( - {line.desc:line.charge} - ) - i+=1 - else: - oth += line.charge - if oth > 0 : - values.update( - {'Autres':oth} - ) - return values - -def import_finances(cursor): - # 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' - accounts = get_income_and_expense_accounts(cursor) - finances = [] - last_update = first_day - for a in accounts: - tot_deb, tot_cred, last_entrie = get_sum_of_operations( - cursor, - first_day, - a - ) - if a.type == 6 : - #charge - finances.append(BudgetLine(a.label, tot_deb - tot_cred, 0)) - elif a.type == 7: - finances.append(BudgetLine(a.label, 0, tot_cred - tot_deb)) - if last_entrie >= last_update : - last_update = last_entrie - - return finances, last_update - - -def get_total_charges(budget:list[BudgetLine]): - tot = 0 - for line in budget: - tot += line.charge - return int(tot) - - -def get_total_revenues(budget:list[BudgetLine]): - tot = 0 - for line in budget: - tot += line.revenue - return int(tot) - - -def get_totals_lists_for_charts(tot_ch_bud, tot_ch_real, tot_re_bud, tot_re_real): - rest_ch = abs(tot_ch_bud - tot_ch_real) - rest_re = abs(tot_re_bud - tot_re_real) - if tot_ch_bud >= tot_ch_real and tot_re_bud >= tot_re_real: - - values = { - f'actuel' : [tot_re_real, tot_ch_real], - f'budget' : [rest_re, rest_ch] - } - else : - values = { - f'bot_budget' : [int(0.98 * tot_re_bud), int(0.98 * tot_ch_bud)], - f'budget' : [int(0.02 *tot_re_bud), int(0.02 *tot_ch_bud)], - f'actuel' : [rest_re, rest_ch] - } - return values - - -def hls_to_rgbs(s_hsl): - rgb = hls_to_rgb(s_hsl["Hue"], s_hsl["Luminosité"], s_hsl["Saturation"] ) - return "#" + "".join("%02X" % round(i*255) for i in rgb ) - - -def get_color_list(nb, s_hsl, sens = "Blanc") : - temp = s_hsl - luminosite = temp["Luminosité"] - - if sens == "Blanc": - ecart = (1-luminosite ) /(nb +1) - else : - ecart = luminosite / (nb + 1) - - l_couleur = [] - for x in range(nb) : - l_couleur.append(hls_to_rgbs(temp )) - if sens == "Blanc" : - luminosite += ecart - else : - print(luminosite) - luminosite -= ecart - temp["Luminosité"] = luminosite - return l_couleur - - -def create_pie_diagrams(path:str, - revenue_ratios:dict[str, int], - charge_ratios:dict[str, int], - last_update:str): - - bleu_hls = {"Hue":190 / 360, - "Saturation":100/100, - "Luminosité":30/100} - - marron_hls = {"Hue":23 / 360, - "Saturation":72/100, - "Luminosité":30/100} - - fig, axs = plt.subplots(ncols=2,figsize=(10, 4), layout='constrained') - labels, values = transform_dict_for_diag(revenue_ratios) - colors = get_color_list(len(values),bleu_hls) - p = axs[0].pie(x=values, labels=labels, colors=colors) - - labels, values = transform_dict_for_diag(charge_ratios) - colors = get_color_list(len(values),marron_hls) - p = axs[1].pie(x=values, labels=labels, colors=colors) - - axs[1].set_title('Répartition des dépenses réalisées') - axs[0].set_title('Répartition des recettes réalisées') - path = os.path.join(path, 'ratios.svg') - plt.savefig(path) - plt.close() - - -def transform_dict_for_diag(dicts:dict): - labels = [] - values = [] - for label, value in dicts.items(): - labels.append(f'{label}\n{value}€') - values.append(value) - - return labels, values - - -def create_total_diagram(path:str, - totals: dict[str, list[int]], - last_upadate:str): - - fig, ax = plt.subplots() - plt.grid(True, 'major', 'y', color="#555555") - width = 0.6 - types = ('dépenses', 'revenus') - leg = [mpatches.Patch(color='#007e97', label=f'au {last_upadate}'), - mpatches.Patch(color='#999999', label='budgeté') - ] - bottom = np.zeros(2) - if len(totals) ==2 : - col = ['#007e97', '#999999'] - else: - col = ['#007e97', '#999999', '#007e97'] - - x = 0 - for position, values in totals.items(): - p = ax.bar(types, values, width, label=position, bottom=bottom, color=col[x] ) - bottom += values - x+=1 - - #ax.invert_yaxis() # labels read top-to-bottom - - ax.set_title('Total des dépenses et des revenus de 2025') - box = ax.get_position() - ax.set_position( - [box.x0, box.y0 + box.height * 0.1, - box.width, box.height * 0.9]# type: ignore - ) - ax.legend(handles=leg, loc='upper center', bbox_to_anchor=(0.5, -0.07), - fancybox=True, shadow=True, ncol=2) - path = os.path.join(path, 'total.svg') - plt.savefig(path) - plt.close() - - -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, last_entrie = 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) - ) - last_entrie = date - for piece_num, numero_compte, label_compte, doc_date, code_journal, debit, credit in cursor : - tot_cred += credit - tot_deb += debit - if str(doc_date) > last_entrie : - last_entrie = str(doc_date) - - 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), last_entrie - - -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 = cp() conf.read('config') path_b = conf['path']['budget'] - path_wp_config = conf['path']['wp-config'] path_save = conf['path']['save_directory'] - + + db_client = DBConnection(conf) - if os.path.exists(path_b) and os.path.exists(path_wp_config): + if os.path.exists(path_b): print("les 2 fichiers budget et wp-config ont été trouvé") - main(path_b, path_wp_config, path_save) + main(path_b, db_client, 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') diff --git a/requirements.txt b/requirements.txt index fd62dec..b6d9b01 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,16 +1,16 @@ configparser==7.2.0 -contourpy==1.3.2 +contourpy==1.3.3 cycler==0.12.1 -fonttools==4.58.2 -kiwisolver==1.4.8 -mariadb==1.1.12 -matplotlib==3.10.3 -mysql-connector-python==9.3.0 -numpy==2.3.0 +fonttools==4.60.1 +kiwisolver==1.4.9 +mariadb==1.1.14 +matplotlib==3.10.7 +mysql-connector-python==9.5.0 +numpy==2.3.4 packaging==25.0 -pandas==2.3.0 -pillow==11.2.1 -pyparsing==3.2.3 +pillow==12.0.0 +pip-review==1.3.0 +pyparsing==3.2.5 python-dateutil==2.9.0.post0 pytz==2025.2 sankeyflow==0.4.1 diff --git a/tests/__pycache__/compta_test.cpython-311.pyc b/tests/__pycache__/compta_test.cpython-311.pyc new file mode 100644 index 0000000..e2d01ab Binary files /dev/null and b/tests/__pycache__/compta_test.cpython-311.pyc differ diff --git a/tests/compta_test.py b/tests/compta_test.py new file mode 100644 index 0000000..a85744d --- /dev/null +++ b/tests/compta_test.py @@ -0,0 +1,22 @@ +import unittest +from libs.reporting_finance import BudgetLine, import_csv + +class budget_test(unittest.TestCase): + def set_up(self): + self.budget = import_csv('files/budget.csv') + + def test_finance(self): + self.set_up() + coef = -500 + for line in self.budget: + if coef > 0 : + if line.revenue == 0 : + line.charge += coef + else : + line.revenue += coef + if coef < 0 : + if line.revenue == 0 and line.charge > abs(coef) : + line.charge += coef + elif line.charge == 0 and line.revenue > abs(coef) : + line.revenue += coef + \ No newline at end of file