432 lines
13 KiB
Python
432 lines
13 KiB
Python
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()
|
|
|