V2.0: Refonte système bancaire - DB SQLite, RIB FR, TVA auto, multi-banques
- Remplace JSON par SQLite (aiosqlite) pour atomicité et intégrité - Nouveau format RIB français avec checksum modulo 97 - Chaque banque a son propre rôle Discord (isolation par serveur) - TVA automatique Particulier -> Entreprise, redirection au Fisc - Institutions gouvernementales avec droit d'audit optionnel - Commandes: /gouv, /banque, /audit, /mes-ribs - Supprime: 16 anciens fichiers (entreprises, FBM, URSAFF, etc.) - Reset complet des données, repart de zéro
This commit is contained in:
utilisateurqu
parent
2e38d05152
commit
c9ad155bf6
38 changed files with 3689 additions and 5106 deletions
500
utils/bank_manager.py
Normal file
500
utils/bank_manager.py
Normal file
|
|
@ -0,0 +1,500 @@
|
|||
import json
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
import aiosqlite
|
||||
|
||||
from utils.db import get_db
|
||||
from utils.rib import generer_rib, valider_rib, decoder_rib
|
||||
|
||||
GUIICHET_PARTICULIER = 4
|
||||
GUIICHET_ENTREPRISE = 3
|
||||
GUIICHET_BANQUE = 2
|
||||
GUIICHET_INSTITUTION = 1
|
||||
|
||||
|
||||
class BankManager:
|
||||
|
||||
async def creer_banque(self, guild_id: int, name: str, role_id: int) -> dict:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"SELECT COALESCE(MAX(establishment_code), 9999) FROM banks WHERE guild_id = ?",
|
||||
(guild_id,),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
last_code = row[0]
|
||||
new_code = last_code + 1
|
||||
if new_code > 89999:
|
||||
raise ValueError("Plus de codes d'établissement disponibles.")
|
||||
|
||||
cursor = await db.execute(
|
||||
"INSERT INTO banks (guild_id, name, role_id, establishment_code) VALUES (?, ?, ?, ?)",
|
||||
(guild_id, name, role_id, new_code),
|
||||
)
|
||||
await db.commit()
|
||||
bank_id = cursor.lastrowid
|
||||
return {"id": bank_id, "name": name, "role_id": role_id, "establishment_code": new_code}
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def supprimer_banque(self, guild_id: int, bank_id: int) -> bool:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"DELETE FROM banks WHERE id = ? AND guild_id = ?", (bank_id, guild_id)
|
||||
)
|
||||
await db.commit()
|
||||
return cursor.rowcount > 0
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def get_banque(self, guild_id: int, bank_id: int) -> Optional[dict]:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"SELECT * FROM banks WHERE id = ? AND guild_id = ?", (bank_id, guild_id)
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
return dict(row) if row else None
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def get_banque_by_role(self, guild_id: int, role_id: int) -> Optional[dict]:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"SELECT * FROM banks WHERE guild_id = ? AND role_id = ?", (guild_id, role_id)
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
return dict(row) if row else None
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def get_banque_by_name(self, guild_id: int, name: str) -> Optional[dict]:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"SELECT * FROM banks WHERE guild_id = ? AND name = ?", (guild_id, name)
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
return dict(row) if row else None
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def lister_banques(self, guild_id: int) -> list:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"SELECT * FROM banks WHERE guild_id = ? ORDER BY name", (guild_id,)
|
||||
)
|
||||
rows = await cursor.fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def ouvrir_compte(
|
||||
self,
|
||||
guild_id: int,
|
||||
bank_id: int,
|
||||
user_id: int,
|
||||
account_type: str,
|
||||
tva_rate: float = 0.0,
|
||||
password: str = None,
|
||||
) -> dict:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"SELECT * FROM banks WHERE id = ? AND guild_id = ?", (bank_id, guild_id)
|
||||
)
|
||||
bank = await cursor.fetchone()
|
||||
if not bank:
|
||||
raise ValueError("Banque introuvable.")
|
||||
|
||||
guichet = GUIICHET_ENTREPRISE if account_type == "entreprise" else GUIICHET_PARTICULIER
|
||||
|
||||
cursor = await db.execute(
|
||||
"SELECT COUNT(*) FROM accounts WHERE guild_id = ? AND bank_id = ? AND user_id = ? AND account_type = ?",
|
||||
(guild_id, bank_id, user_id, account_type),
|
||||
)
|
||||
count = (await cursor.fetchone())[0]
|
||||
if count > 0:
|
||||
raise ValueError(f"Ce membre a déjà un compte {account_type} dans cette banque.")
|
||||
|
||||
rib = generer_rib(bank["establishment_code"], guichet)
|
||||
|
||||
cursor = await db.execute(
|
||||
"INSERT INTO accounts (guild_id, bank_id, user_id, rib, account_type, tva_rate, password) VALUES (?, ?, ?, ?, ?, ?, ?)",
|
||||
(guild_id, bank_id, user_id, rib, account_type, tva_rate, password),
|
||||
)
|
||||
await db.commit()
|
||||
return {"id": cursor.lastrowid, "rib": rib, "account_type": account_type, "tva_rate": tva_rate}
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def get_compte_by_rib(self, guild_id: int, rib: str) -> Optional[dict]:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"""SELECT a.*, b.name as bank_name, b.establishment_code, b.role_id as bank_role_id
|
||||
FROM accounts a JOIN banks b ON a.bank_id = b.id
|
||||
WHERE a.rib = ? AND a.guild_id = ?""",
|
||||
(rib, guild_id),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
return dict(row) if row else None
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def get_comptes_user(self, guild_id: int, user_id: int) -> list:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"""SELECT a.*, b.name as bank_name, b.establishment_code
|
||||
FROM accounts a JOIN banks b ON a.bank_id = b.id
|
||||
WHERE a.guild_id = ? AND a.user_id = ?
|
||||
ORDER BY b.name, a.account_type""",
|
||||
(guild_id, user_id),
|
||||
)
|
||||
rows = await cursor.fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def get_comptes_banque(self, guild_id: int, bank_id: int) -> list:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"""SELECT a.*, b.name as bank_name
|
||||
FROM accounts a JOIN banks b ON a.bank_id = b.id
|
||||
WHERE a.guild_id = ? AND a.bank_id = ?
|
||||
ORDER BY a.user_id, a.account_type""",
|
||||
(guild_id, bank_id),
|
||||
)
|
||||
rows = await cursor.fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def modifier_tva(self, guild_id: int, rib: str, nouveau_taux: float) -> bool:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"UPDATE accounts SET tva_rate = ? WHERE rib = ? AND guild_id = ? AND account_type = 'entreprise'",
|
||||
(nouveau_taux, rib, guild_id),
|
||||
)
|
||||
await db.commit()
|
||||
return cursor.rowcount > 0
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def virement(
|
||||
self,
|
||||
guild_id: int,
|
||||
rib_source: str,
|
||||
rib_dest: str,
|
||||
montant: float,
|
||||
emetteur_id: int = None,
|
||||
destinataire_id: int = None,
|
||||
) -> dict:
|
||||
if montant <= 0:
|
||||
raise ValueError("Le montant doit être supérieur à 0.")
|
||||
if rib_source == rib_dest:
|
||||
raise ValueError("Impossible de se virer à soi-même.")
|
||||
|
||||
db = await get_db()
|
||||
try:
|
||||
source = await self._get_account(db, guild_id, rib_source)
|
||||
if not source:
|
||||
raise ValueError("Compte source introuvable.")
|
||||
dest = await self._get_account(db, guild_id, rib_dest)
|
||||
if not dest:
|
||||
raise ValueError("Compte destination introuvable.")
|
||||
|
||||
if source["solde"] < montant:
|
||||
raise ValueError(f"Solde insuffisant. Disponible : {source['solde']:.2f}€")
|
||||
|
||||
tva_montant = 0.0
|
||||
montant_net = montant
|
||||
|
||||
if source["account_type"] == "particulier" and dest["account_type"] == "entreprise":
|
||||
tva_rate = dest["tva_rate"] or 0.0
|
||||
if tva_rate > 0:
|
||||
tva_montant = round(montant * tva_rate / 100, 2)
|
||||
montant_net = round(montant - tva_montant, 2)
|
||||
|
||||
cursor = await db.execute(
|
||||
"UPDATE accounts SET solde = solde - ? WHERE rib = ? AND guild_id = ?",
|
||||
(montant, rib_source, guild_id),
|
||||
)
|
||||
cursor = await db.execute(
|
||||
"UPDATE accounts SET solde = solde + ? WHERE rib = ? AND guild_id = ?",
|
||||
(montant_net, rib_dest, guild_id),
|
||||
)
|
||||
|
||||
if tva_montant > 0:
|
||||
fisc_rib = await self._get_fisc_rib(db, guild_id)
|
||||
if fisc_rib:
|
||||
cursor = await db.execute(
|
||||
"UPDATE accounts SET solde = solde + ? WHERE rib = ? AND guild_id = ?",
|
||||
(tva_montant, fisc_rib, guild_id),
|
||||
)
|
||||
|
||||
cursor = await db.execute(
|
||||
"UPDATE banks SET clients_total_balance = clients_total_balance - ? WHERE id = ? AND guild_id = ?",
|
||||
(montant, source["bank_id"], guild_id),
|
||||
)
|
||||
cursor = await db.execute(
|
||||
"UPDATE banks SET clients_total_balance = clients_total_balance + ? WHERE id = ? AND guild_id = ?",
|
||||
(montant_net, dest["bank_id"], guild_id),
|
||||
)
|
||||
|
||||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
await db.execute(
|
||||
"""INSERT INTO transactions (guild_id, emetteur_rib, destinataire_rib, emetteur_id, destinataire_id, montant, type, date)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
||||
(guild_id, rib_source, rib_dest, emetteur_id or source["user_id"], destinataire_id or dest["user_id"], montant, "virement", now),
|
||||
)
|
||||
if tva_montant > 0 and fisc_rib:
|
||||
await db.execute(
|
||||
"""INSERT INTO transactions (guild_id, destinataire_rib, montant, type, date, metadata)
|
||||
VALUES (?, ?, ?, ?, ?, ?)""",
|
||||
(guild_id, fisc_rib, tva_montant, "tva_prelevee", now, json.dumps({"source_rib": rib_source, "dest_rib": rib_dest})),
|
||||
)
|
||||
|
||||
await db.commit()
|
||||
|
||||
result = {
|
||||
"montant_envoye": montant,
|
||||
"montant_recu": montant_net,
|
||||
"tva_prelevee": tva_montant,
|
||||
}
|
||||
if tva_montant > 0:
|
||||
result["tva_dest"] = fisc_rib
|
||||
return result
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def depot_banquier(
|
||||
self, guild_id: int, rib: str, montant: float, banquier_id: int
|
||||
) -> dict:
|
||||
if montant <= 0:
|
||||
raise ValueError("Le montant doit être supérieur à 0.")
|
||||
|
||||
db = await get_db()
|
||||
try:
|
||||
compte = await self._get_account(db, guild_id, rib)
|
||||
if not compte:
|
||||
raise ValueError("Compte introuvable.")
|
||||
|
||||
cursor = await db.execute(
|
||||
"UPDATE accounts SET solde = solde + ? WHERE rib = ? AND guild_id = ?",
|
||||
(montant, rib, guild_id),
|
||||
)
|
||||
cursor = await db.execute(
|
||||
"UPDATE banks SET clients_total_balance = clients_total_balance + ? WHERE id = ? AND guild_id = ?",
|
||||
(montant, compte["bank_id"], guild_id),
|
||||
)
|
||||
|
||||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
await db.execute(
|
||||
"""INSERT INTO transactions (guild_id, destinataire_rib, destinataire_id, montant, type, date, metadata)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
||||
(guild_id, rib, compte["user_id"], montant, "depot_banquier", now, json.dumps({"banquier_id": banquier_id})),
|
||||
)
|
||||
await db.commit()
|
||||
return {"nouveau_solde": compte["solde"] + montant}
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def retrait_banquier(
|
||||
self, guild_id: int, rib: str, montant: float, banquier_id: int
|
||||
) -> dict:
|
||||
if montant <= 0:
|
||||
raise ValueError("Le montant doit être supérieur à 0.")
|
||||
|
||||
db = await get_db()
|
||||
try:
|
||||
compte = await self._get_account(db, guild_id, rib)
|
||||
if not compte:
|
||||
raise ValueError("Compte introuvable.")
|
||||
if compte["solde"] < montant:
|
||||
raise ValueError(f"Solde insuffisant. Disponible : {compte['solde']:.2f}€")
|
||||
|
||||
cursor = await db.execute(
|
||||
"UPDATE accounts SET solde = solde - ? WHERE rib = ? AND guild_id = ?",
|
||||
(montant, rib, guild_id),
|
||||
)
|
||||
cursor = await db.execute(
|
||||
"UPDATE banks SET clients_total_balance = clients_total_balance - ? WHERE id = ? AND guild_id = ?",
|
||||
(montant, compte["bank_id"], guild_id),
|
||||
)
|
||||
|
||||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
await db.execute(
|
||||
"""INSERT INTO transactions (guild_id, emetteur_rib, emetteur_id, montant, type, date, metadata)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
||||
(guild_id, rib, compte["user_id"], montant, "retrait_banquier", now, json.dumps({"banquier_id": banquier_id})),
|
||||
)
|
||||
await db.commit()
|
||||
return {"nouveau_solde": compte["solde"] - montant}
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def get_transactions(self, guild_id: int, rib: str, limit: int = 10) -> list:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"""SELECT * FROM transactions
|
||||
WHERE guild_id = ? AND (emetteur_rib = ? OR destinataire_rib = ?)
|
||||
ORDER BY date DESC LIMIT ?""",
|
||||
(guild_id, rib, rib, limit),
|
||||
)
|
||||
rows = await cursor.fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def creer_institution(
|
||||
self, guild_id: int, name: str, role_id: int = None, can_audit_all: bool = False
|
||||
) -> dict:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"INSERT INTO institutions (guild_id, name, role_id, can_audit_all) VALUES (?, ?, ?, ?)",
|
||||
(guild_id, name, role_id, int(can_audit_all)),
|
||||
)
|
||||
await db.commit()
|
||||
return {"id": cursor.lastrowid, "name": name, "can_audit_all": can_audit_all}
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def get_institution(self, guild_id: int, institution_id: int) -> Optional[dict]:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"SELECT * FROM institutions WHERE id = ? AND guild_id = ?",
|
||||
(institution_id, guild_id),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
return dict(row) if row else None
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def get_institution_by_role(self, guild_id: int, role_id: int) -> Optional[dict]:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"SELECT * FROM institutions WHERE guild_id = ? AND role_id = ?",
|
||||
(guild_id, role_id),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
return dict(row) if row else None
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def lister_institutions(self, guild_id: int) -> list:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"SELECT * FROM institutions WHERE guild_id = ? ORDER BY name", (guild_id,)
|
||||
)
|
||||
rows = await cursor.fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def configurer_fisc(self, guild_id: int, institution_id: int) -> bool:
|
||||
db = await get_db()
|
||||
try:
|
||||
inst = await self.get_institution(guild_id, institution_id)
|
||||
if not inst:
|
||||
return False
|
||||
|
||||
guichet = GUIICHET_INSTITUTION
|
||||
bank = await self._get_gouvernement_bank(db, guild_id)
|
||||
if not bank:
|
||||
cursor = await db.execute(
|
||||
"INSERT INTO banks (guild_id, name, role_id, establishment_code) VALUES (?, ?, ?, ?)",
|
||||
(guild_id, "Gouvernement", 0, 1),
|
||||
)
|
||||
bank_id = cursor.lastrowid
|
||||
bank = {"id": bank_id, "establishment_code": 1}
|
||||
else:
|
||||
bank_id = bank["id"]
|
||||
|
||||
rib = generer_rib(bank["establishment_code"], guichet)
|
||||
|
||||
existing = await db.execute(
|
||||
"SELECT rib FROM accounts WHERE guild_id = ? AND bank_id = ? AND user_id = ? AND account_type = 'entreprise'",
|
||||
(guild_id, bank_id, institution_id),
|
||||
)
|
||||
existing_row = await existing.fetchone()
|
||||
if existing_row:
|
||||
rib = existing_row[0]
|
||||
else:
|
||||
cursor = await db.execute(
|
||||
"INSERT INTO accounts (guild_id, bank_id, user_id, rib, account_type, solde) VALUES (?, ?, ?, ?, 'entreprise', 0.0)",
|
||||
(guild_id, bank_id, institution_id, rib),
|
||||
)
|
||||
|
||||
await db.execute(
|
||||
"INSERT OR REPLACE INTO fisc_config (guild_id, institution_id) VALUES (?, ?)",
|
||||
(guild_id, institution_id),
|
||||
)
|
||||
await db.commit()
|
||||
return True
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def get_fisc_config(self, guild_id: int) -> Optional[dict]:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"""SELECT fc.*, i.name as institution_name, i.role_id as institution_role_id
|
||||
FROM fisc_config fc
|
||||
JOIN institutions i ON fc.institution_id = i.id
|
||||
WHERE fc.guild_id = ?""",
|
||||
(guild_id,),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
return dict(row) if row else None
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def get_next_bank_codes(self, guild_id: int) -> list:
|
||||
db = await get_db()
|
||||
try:
|
||||
cursor = await db.execute(
|
||||
"SELECT establishment_code FROM banks WHERE guild_id = ? ORDER BY establishment_code",
|
||||
(guild_id,),
|
||||
)
|
||||
rows = await cursor.fetchall()
|
||||
return [r[0] for r in rows]
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
async def _get_account(self, db: aiosqlite.Connection, guild_id: int, rib: str) -> Optional[dict]:
|
||||
cursor = await db.execute(
|
||||
"SELECT * FROM accounts WHERE rib = ? AND guild_id = ?", (rib, guild_id)
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
return dict(row) if row else None
|
||||
|
||||
async def _get_fisc_rib(self, db: aiosqlite.Connection, guild_id: int) -> Optional[str]:
|
||||
cursor = await db.execute(
|
||||
"""SELECT a.rib FROM accounts a
|
||||
JOIN fisc_config fc ON a.user_id = fc.institution_id
|
||||
WHERE fc.guild_id = ? AND a.account_type = 'entreprise'""",
|
||||
(guild_id,),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
return row[0] if row else None
|
||||
|
||||
async def _get_gouvernement_bank(self, db: aiosqlite.Connection, guild_id: int) -> Optional[dict]:
|
||||
cursor = await db.execute(
|
||||
"SELECT * FROM banks WHERE guild_id = ? AND name = 'Gouvernement'",
|
||||
(guild_id,),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
return dict(row) if row else None
|
||||
|
|
@ -1,58 +0,0 @@
|
|||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
from .logger import log_error
|
||||
|
||||
class BanqueManager:
|
||||
def __init__(self):
|
||||
|
||||
self.file_path = "comptes_bancaires.json"
|
||||
self.comptes = {}
|
||||
self.load_data()
|
||||
|
||||
def load_data(self):
|
||||
"""Charge les données bancaires depuis un fichier JSON."""
|
||||
if os.path.exists(self.file_path):
|
||||
try:
|
||||
with open(self.file_path, "r", encoding="utf-8") as f:
|
||||
data = f.read().strip()
|
||||
self.comptes = json.loads(data) if data else {}
|
||||
except json.JSONDecodeError:
|
||||
log_error("Erreur de lecture dans comptes_bancaires.json, réinitialisation...")
|
||||
self.comptes = {}
|
||||
self.save_data()
|
||||
|
||||
def save_data(self):
|
||||
"""Sauvegarde les comptes et transactions dans comptes_bancaires.json."""
|
||||
with open(self.file_path, "w", encoding="utf-8") as f:
|
||||
json.dump(self.comptes, f, indent=4, ensure_ascii=False)
|
||||
|
||||
def enregistrer_transaction(self, emetteur_id, destinataire_id, montant, type_transaction):
|
||||
"""
|
||||
Enregistre une transaction dans l'historique et recharge les données en temps réel.
|
||||
"""
|
||||
self.load_data()
|
||||
|
||||
transaction = {
|
||||
"date": datetime.now().strftime("%d/%m/%Y %H:%M:%S"),
|
||||
"type": type_transaction,
|
||||
"montant": montant,
|
||||
"emetteur": emetteur_id,
|
||||
"destinataire": destinataire_id
|
||||
}
|
||||
|
||||
if emetteur_id not in self.comptes:
|
||||
self.comptes[emetteur_id] = {"solde": 0, "prets": [], "transactions": []}
|
||||
elif "transactions" not in self.comptes[emetteur_id]:
|
||||
self.comptes[emetteur_id]["transactions"] = []
|
||||
|
||||
self.comptes[emetteur_id]["transactions"].append(transaction)
|
||||
|
||||
if destinataire_id not in self.comptes:
|
||||
self.comptes[destinataire_id] = {"solde": 0, "prets": [], "transactions": []}
|
||||
elif "transactions" not in self.comptes[destinataire_id]:
|
||||
self.comptes[destinataire_id]["transactions"] = []
|
||||
|
||||
self.comptes[destinataire_id]["transactions"].append(transaction)
|
||||
|
||||
self.save_data()
|
||||
|
|
@ -16,7 +16,8 @@ class CardGenerator:
|
|||
|
||||
def create_template(self):
|
||||
try:
|
||||
CI_path = "/home/discord/Bot/Vicy/ressources/assets/CIITAL.png"
|
||||
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
CI_path = os.path.join(base_dir, "ressources", "assets", "CIITAL.png")
|
||||
|
||||
if not os.path.exists(CI_path):
|
||||
raise FileNotFoundError(f"Image de fond introuvable : {CI_path}")
|
||||
|
|
|
|||
82
utils/db.py
Normal file
82
utils/db.py
Normal file
|
|
@ -0,0 +1,82 @@
|
|||
import aiosqlite
|
||||
import os
|
||||
|
||||
DB_PATH = os.path.join(os.path.dirname(__file__), "..", "data", "bank_v2.db")
|
||||
|
||||
SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS banks (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
guild_id INTEGER NOT NULL,
|
||||
name TEXT NOT NULL,
|
||||
role_id INTEGER NOT NULL,
|
||||
establishment_code INTEGER NOT NULL,
|
||||
vault_balance REAL DEFAULT 0.0,
|
||||
clients_total_balance REAL DEFAULT 0.0,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
UNIQUE(guild_id, name),
|
||||
UNIQUE(guild_id, establishment_code)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS accounts (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
guild_id INTEGER NOT NULL,
|
||||
bank_id INTEGER NOT NULL,
|
||||
user_id INTEGER NOT NULL,
|
||||
rib TEXT NOT NULL UNIQUE,
|
||||
account_type TEXT NOT NULL CHECK(account_type IN ('particulier', 'entreprise')),
|
||||
solde REAL DEFAULT 0.0,
|
||||
tva_rate REAL DEFAULT 0.0,
|
||||
password TEXT,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
FOREIGN KEY (bank_id) REFERENCES banks(id) ON DELETE CASCADE,
|
||||
UNIQUE(guild_id, bank_id, user_id, account_type)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS institutions (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
guild_id INTEGER NOT NULL,
|
||||
name TEXT NOT NULL,
|
||||
role_id INTEGER,
|
||||
can_audit_all INTEGER DEFAULT 0,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
UNIQUE(guild_id, name)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS fisc_config (
|
||||
guild_id INTEGER PRIMARY KEY,
|
||||
institution_id INTEGER,
|
||||
FOREIGN KEY (institution_id) REFERENCES institutions(id) ON DELETE SET NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS transactions (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
guild_id INTEGER NOT NULL,
|
||||
emetteur_rib TEXT,
|
||||
destinataire_rib TEXT,
|
||||
emetteur_id INTEGER,
|
||||
destinataire_id INTEGER,
|
||||
montant REAL NOT NULL,
|
||||
type TEXT NOT NULL,
|
||||
date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
metadata TEXT
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_accounts_guild_user ON accounts(guild_id, user_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_accounts_rib ON accounts(rib);
|
||||
CREATE INDEX IF NOT EXISTS idx_accounts_bank ON accounts(guild_id, bank_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_transactions_guild ON transactions(guild_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_transactions_ribs ON transactions(emetteur_rib, destinataire_rib);
|
||||
CREATE INDEX IF NOT EXISTS idx_banks_guild ON banks(guild_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_institutions_guild ON institutions(guild_id);
|
||||
"""
|
||||
|
||||
|
||||
async def get_db() -> aiosqlite.Connection:
|
||||
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
|
||||
db = await aiosqlite.connect(DB_PATH)
|
||||
db.row_factory = aiosqlite.Row
|
||||
await db.execute("PRAGMA journal_mode=WAL")
|
||||
await db.execute("PRAGMA foreign_keys=ON")
|
||||
await db.executescript(SCHEMA)
|
||||
await db.commit()
|
||||
return db
|
||||
|
|
@ -18,9 +18,6 @@ DEFAULT_CONFIG = {
|
|||
"visite_medicale": None,
|
||||
"permis": None,
|
||||
"auto_ecole": None,
|
||||
"fbm": None,
|
||||
"banque_nationale": None,
|
||||
"ursaff": None,
|
||||
"session_reminders": None,
|
||||
},
|
||||
"channels": {
|
||||
|
|
@ -47,9 +44,6 @@ ROLE_LABELS = {
|
|||
"visite_medicale": "Visite Médicale",
|
||||
"permis": "Permis",
|
||||
"auto_ecole": "Auto-École",
|
||||
"fbm": "FBM",
|
||||
"banque_nationale": "Banque Nationale",
|
||||
"ursaff": "URSAFF",
|
||||
"session_reminders": "Session Reminders",
|
||||
}
|
||||
|
||||
|
|
|
|||
55
utils/rib.py
Normal file
55
utils/rib.py
Normal file
|
|
@ -0,0 +1,55 @@
|
|||
import random
|
||||
import string
|
||||
|
||||
|
||||
def _checksum(digits: str) -> int:
|
||||
return (97 - (int(digits) % 97)) % 97
|
||||
|
||||
|
||||
def generer_rib(establishment_code: int, guichet_code: int) -> str:
|
||||
etab = str(establishment_code).zfill(5)
|
||||
guichet = str(guichet_code).zfill(5)
|
||||
numero = "".join(random.choices(string.digits, k=11))
|
||||
checksum_digits = f"76{etab}{guichet}{numero}"
|
||||
cle = str(_checksum(checksum_digits)).zfill(2)
|
||||
return f"FR76 {etab} {guichet} {numero} {cle}"
|
||||
|
||||
|
||||
def valider_rib(rib: str) -> bool:
|
||||
parts = rib.replace("FR76 ", "").split()
|
||||
if len(parts) != 3:
|
||||
return False
|
||||
etab, guichet, rest = parts
|
||||
if len(etab) != 5 or len(guichet) != 5:
|
||||
return False
|
||||
if len(rest) != 13:
|
||||
return False
|
||||
numero = rest[:11]
|
||||
cle = rest[11:]
|
||||
if not etab.isdigit() or not guichet.isdigit() or not numero.isdigit() or not cle.isdigit():
|
||||
return False
|
||||
checksum_digits = f"76{etab}{guichet}{numero}"
|
||||
return _checksum(checksum_digits) == int(cle)
|
||||
|
||||
|
||||
def decoder_rib(rib: str) -> dict:
|
||||
parts = rib.replace("FR76 ", "").split()
|
||||
if len(parts) != 3:
|
||||
return None
|
||||
etab, guichet, rest = parts
|
||||
numero = rest[:11]
|
||||
cle = rest[11:]
|
||||
etab_int = int(etab)
|
||||
guichet_int = int(guichet)
|
||||
|
||||
TYPE_MAP = {4: "particulier", 3: "entreprise", 2: "banque", 1: "institution"}
|
||||
|
||||
return {
|
||||
"etablissement": etab_int,
|
||||
"guichet": guichet_int,
|
||||
"numero": numero,
|
||||
"cle": int(cle),
|
||||
"type_compte": TYPE_MAP.get(guichet_int, "inconnu"),
|
||||
"est_gouvernement": etab_int < 100,
|
||||
"valide": valider_rib(rib),
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue