365 lines
14 KiB
Python
365 lines
14 KiB
Python
from flask import (
|
|
Blueprint,
|
|
current_app,
|
|
flash,
|
|
redirect,
|
|
render_template,
|
|
request,
|
|
session,
|
|
url_for,
|
|
)
|
|
from sqlalchemy import text
|
|
|
|
from app import db
|
|
from app.decorators import login_required
|
|
|
|
comptes_bp = Blueprint("comptes", __name__)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gestion des comptes
|
|
# ---------------------------------------------------------------------------
|
|
@comptes_bp.route("/gestion_comptes", methods=["GET", "POST"])
|
|
@login_required
|
|
def gestion_comptes():
|
|
from app.models import Societe, Compte
|
|
|
|
societes = Societe.query.order_by(Societe.nom.asc()).all()
|
|
|
|
# La société sélectionnée vient du formulaire (changement manuel),
|
|
# ou d'un paramètre d'URL (redirection après création/modification),
|
|
# ou de la session (mémorisation entre navigations).
|
|
selected_societe_id = (
|
|
request.form.get("societe_id")
|
|
or request.args.get("societe_id")
|
|
or session.get("selected_compte_societe_id")
|
|
)
|
|
if not selected_societe_id and societes:
|
|
selected_societe_id = societes[0].id
|
|
else:
|
|
try:
|
|
selected_societe_id = int(selected_societe_id) if selected_societe_id else None
|
|
except ValueError:
|
|
selected_societe_id = societes[0].id if societes else None
|
|
|
|
# Mémorisation en session : la société active ne change que manuellement
|
|
if selected_societe_id is not None:
|
|
session["selected_compte_societe_id"] = selected_societe_id
|
|
|
|
selected_societe_nom = ""
|
|
ecritures = []
|
|
total_credit = 0.0
|
|
total_debit = 0.0
|
|
# Agrégats financiers (voir règles de calcul dans la documentation)
|
|
tcredit = 0.0 # Depot : somme des crédits des écritures manuelles
|
|
tdebit = 0.0 # Retrait : somme des débits des écritures manuelles
|
|
engagement = 0.0 # ENGA. : positions actives (achat +, vente -)
|
|
pv = 0.0 # PV : positions soldées (achat -, vente +)
|
|
dispo = 0.0 # DISPO = TOTAL CREDIT - TOTAL DEBIT (toutes écritures)
|
|
solde = 0.0 # SOLDE = DISPO + ENGA.
|
|
|
|
if selected_societe_id:
|
|
societe_obj = Societe.query.get(selected_societe_id)
|
|
if societe_obj:
|
|
selected_societe_nom = societe_obj.nom
|
|
|
|
# Tri par date décroissante (du plus récent au plus ancien)
|
|
ecritures = (
|
|
Compte.query.filter_by(id_societe=selected_societe_id)
|
|
.order_by(Compte.date_operation.desc(), Compte.id.desc())
|
|
.all()
|
|
)
|
|
# Totaux bruts du tableau (toutes écritures confondues)
|
|
total_credit = sum(float(e.credit or 0) for e in ecritures)
|
|
total_debit = sum(float(e.debit or 0) for e in ecritures)
|
|
|
|
# --- Dépôts / Retraits : uniquement les écritures manuelles ---
|
|
for e in ecritures:
|
|
if e.source == "manuel":
|
|
tcredit += float(e.credit or 0)
|
|
tdebit += float(e.debit or 0)
|
|
|
|
# --- DISPO = TOTAL CREDIT - TOTAL DEBIT (toutes écritures) ---
|
|
dispo = total_credit - total_debit
|
|
|
|
# --- Engagement & PV : depuis les lignes d'ordres (ordre_pied) ---
|
|
# Règles :
|
|
# etat='actif' : achat -> engagement += qté*(prix_brut+frais)
|
|
# vente -> engagement -= qté*(prix_brut-frais)
|
|
# etat='soldé' : achat -> PV -= qté*(prix_brut+frais)
|
|
# vente -> PV += qté*(prix_brut-frais)
|
|
# etat='annulé' : ignoré (filtré en SQL)
|
|
# Filtres : société en cours + lignes non annulées
|
|
query_ordres = text(
|
|
"SELECT op.sens, op.etat, op.quantite, op.prix_brut, "
|
|
"op.frais_broker, op.frais_etat, op.frais_autre "
|
|
"FROM ordre o "
|
|
"JOIN ordre_pied op ON op.id_entete = o.id "
|
|
"WHERE o.id_societe = :soc_id "
|
|
"AND op.etat <> 'annulé'"
|
|
)
|
|
for row in db.session.execute(query_ordres, {"soc_id": selected_societe_id}):
|
|
quantite = float(row.quantite or 0)
|
|
prix_brut = float(row.prix_brut or 0)
|
|
frais_broker = float(row.frais_broker or 0)
|
|
frais_etat = float(row.frais_etat or 0)
|
|
frais_autre = float(row.frais_autre or 0)
|
|
|
|
# Achat : on paie le prix brut + les frais
|
|
montant_achat = quantite * (prix_brut + frais_broker + frais_etat + frais_autre)
|
|
# Vente : on encaisse le prix brut - les frais
|
|
montant_vente = quantite * (prix_brut - frais_broker - frais_etat - frais_autre)
|
|
|
|
sens = (row.sens or "").strip().lower()
|
|
etat = (row.etat or "").strip().lower()
|
|
|
|
if etat == "actif":
|
|
if sens == "achat":
|
|
engagement += montant_achat
|
|
elif sens == "vente":
|
|
engagement -= montant_vente
|
|
elif etat == "soldé" or etat == "solde":
|
|
if sens == "achat":
|
|
pv -= montant_achat
|
|
elif sens == "vente":
|
|
pv += montant_vente
|
|
# etat == 'annulé' : ne rien faire (déjà filtré en SQL)
|
|
|
|
solde = dispo + engagement
|
|
|
|
return render_template(
|
|
"gestion_comptes.html",
|
|
societes=societes,
|
|
selected_societe_id=selected_societe_id,
|
|
selected_societe=selected_societe_nom,
|
|
ecritures=ecritures,
|
|
total_credit=total_credit,
|
|
total_debit=total_debit,
|
|
tcredit=tcredit,
|
|
tdebit=tdebit,
|
|
engagement=engagement,
|
|
pv=pv,
|
|
dispo=dispo,
|
|
solde=solde,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Génération automatique des écritures de compte depuis les ordres
|
|
# ---------------------------------------------------------------------------
|
|
@comptes_bp.route("/generer_ecritures_ordres", methods=["POST"])
|
|
@login_required
|
|
def generer_ecritures_ordres():
|
|
"""Génère les écritures comptables (table `compte`) à partir des ordres
|
|
et de leurs lignes (`ordre_pied`) pour la société actuellement sélectionnée.
|
|
|
|
Règles :
|
|
- achat : debit = quantite * (prix_brut + frais_broker + frais_etat + frais_autre), credit = 0
|
|
- vente : credit = quantite * (prix_brut + frais_broker + frais_etat + frais_autre), debit = 0
|
|
- libelle= "Achat/Vente actions {isin} {company_name}"
|
|
- source = 'ordre' (non modifiable depuis l'IHM)
|
|
Idempotent : supprime d'abord les écritures source='ordre' de la société.
|
|
"""
|
|
id_societe = request.form.get("id_societe") or session.get(
|
|
"selected_compte_societe_id"
|
|
)
|
|
|
|
if not id_societe:
|
|
flash("Aucune société sélectionnée pour la génération.", "danger")
|
|
return redirect(url_for("comptes.gestion_comptes"))
|
|
|
|
try:
|
|
id_societe_int = int(id_societe)
|
|
except (TypeError, ValueError):
|
|
flash("Identifiant de société invalide.", "danger")
|
|
return redirect(url_for("comptes.gestion_comptes"))
|
|
|
|
try:
|
|
# 1. Suppression des anciennes écritures générées pour cette société
|
|
db.session.execute(
|
|
text(
|
|
"DELETE FROM compte WHERE id_societe = :soc_id AND source = 'ordre'"
|
|
),
|
|
{"soc_id": id_societe_int},
|
|
)
|
|
|
|
# 2. Parcours des ordres de la société, puis de leurs lignes
|
|
# Filtres : société en cours + lignes non annulées
|
|
query = text(
|
|
"SELECT o.id, o.isin, a.company_name, "
|
|
"op.date_op, op.quantite, op.prix_brut, "
|
|
"op.frais_broker, op.frais_etat, op.frais_autre, op.sens "
|
|
"FROM ordre o "
|
|
"JOIN ordre_pied op ON op.id_entete = o.id "
|
|
"JOIN actions a ON o.isin = a.isin "
|
|
"WHERE o.id_societe = :soc_id "
|
|
"AND op.etat <> 'annulé' "
|
|
"ORDER BY o.id, op.date_op"
|
|
)
|
|
lignes = db.session.execute(query, {"soc_id": id_societe_int}).fetchall()
|
|
|
|
insert = text(
|
|
"INSERT INTO compte (id_societe, date_operation, debit, credit, "
|
|
"libelle, source) "
|
|
"VALUES (:id_societe, :date_op, :debit, :credit, :libelle, 'ordre')"
|
|
)
|
|
|
|
count = 0
|
|
from markupsafe import escape
|
|
|
|
for row in lignes:
|
|
quantite = float(row.quantite or 0)
|
|
prix_brut = float(row.prix_brut or 0)
|
|
frais_broker = float(row.frais_broker or 0)
|
|
frais_etat = float(row.frais_etat or 0)
|
|
frais_autre = float(row.frais_autre or 0)
|
|
|
|
sens = (row.sens or "").strip().lower()
|
|
isin = row.isin or ""
|
|
# Nom de l'action échappé puis mis en gras (affichage |safe côté template)
|
|
company = f"<strong>{escape(row.company_name or '')}</strong>"
|
|
|
|
if sens == "achat":
|
|
# Achat : on paie le prix brut + les frais
|
|
montant = quantite * (prix_brut + frais_broker + frais_etat + frais_autre)
|
|
debit = montant
|
|
credit = 0.0
|
|
libelle = f"Achat actions {isin} {company}".strip()
|
|
elif sens == "vente":
|
|
# Vente : on encaisse le prix brut - les frais
|
|
montant = quantite * (prix_brut - frais_broker - frais_etat - frais_autre)
|
|
debit = 0.0
|
|
credit = montant
|
|
libelle = f"Vente actions {isin} {company}".strip()
|
|
else:
|
|
# Sens inattendu : on ignore la ligne
|
|
continue
|
|
|
|
db.session.execute(
|
|
insert,
|
|
{
|
|
"id_societe": id_societe_int,
|
|
"date_op": row.date_op,
|
|
"debit": debit,
|
|
"credit": credit,
|
|
"libelle": libelle,
|
|
},
|
|
)
|
|
count += 1
|
|
|
|
db.session.commit()
|
|
flash(
|
|
f"Génération terminée : {count} écriture(s) créée(s) pour "
|
|
f"la société.",
|
|
"success",
|
|
)
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
current_app.logger.error("Erreur génération écritures : %s", e, exc_info=True)
|
|
flash("Erreur lors de la génération des écritures.", "danger")
|
|
|
|
return redirect(url_for("comptes.gestion_comptes", societe_id=id_societe_int))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Création d'une écriture comptable
|
|
# ---------------------------------------------------------------------------
|
|
@comptes_bp.route("/creer_compte_traitement", methods=["POST"])
|
|
@login_required
|
|
def creer_compte_traitement():
|
|
id_societe = request.form.get("id_societe")
|
|
date_operation = request.form.get("date_operation")
|
|
libelle = request.form.get("libelle", "").strip()
|
|
debit = request.form.get("debit", "0")
|
|
credit = request.form.get("credit", "0")
|
|
|
|
if not id_societe or not date_operation:
|
|
flash("Veuillez remplir tous les champs obligatoires.", "danger")
|
|
return redirect(url_for("comptes.gestion_comptes", societe_id=id_societe or ""))
|
|
|
|
try:
|
|
debit_val = float(debit) if debit else 0.0
|
|
credit_val = float(credit) if credit else 0.0
|
|
id_societe_int = int(id_societe)
|
|
except ValueError:
|
|
flash("Veuillez saisir des valeurs numériques valides.", "danger")
|
|
return redirect(url_for("comptes.gestion_comptes", societe_id=id_societe or ""))
|
|
|
|
try:
|
|
db.session.execute(
|
|
text(
|
|
"INSERT INTO compte (id_societe, date_operation, libelle, "
|
|
"debit, credit) VALUES (:id_societe, :date_op, :libelle, :debit, :credit)"
|
|
),
|
|
{
|
|
"id_societe": id_societe_int,
|
|
"date_op": date_operation,
|
|
"libelle": libelle or None,
|
|
"debit": debit_val,
|
|
"credit": credit_val,
|
|
},
|
|
)
|
|
db.session.commit()
|
|
flash("Écriture créée avec succès !", "success")
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
current_app.logger.error("Erreur création écriture : %s", e, exc_info=True)
|
|
flash("Erreur lors de la création de l'écriture.", "danger")
|
|
|
|
# On reste sur la même société après validation
|
|
return redirect(url_for("comptes.gestion_comptes", societe_id=id_societe_int))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Modification d'une écriture comptable
|
|
# ---------------------------------------------------------------------------
|
|
@comptes_bp.route("/modifier_compte/<int:id>", methods=["POST"])
|
|
@login_required
|
|
def modifier_compte(id):
|
|
date_operation = request.form.get("date_operation")
|
|
libelle = request.form.get("libelle", "").strip()
|
|
debit = request.form.get("debit", "0")
|
|
credit = request.form.get("credit", "0")
|
|
|
|
# Récupération de la société de l'écriture avant modification (pour la redirection)
|
|
id_societe = request.form.get("id_societe")
|
|
if not id_societe:
|
|
row = db.session.execute(
|
|
text("SELECT id_societe FROM compte WHERE id = :id"), {"id": id}
|
|
).fetchone()
|
|
id_societe = str(row.id_societe) if row else ""
|
|
|
|
if not date_operation:
|
|
flash("Veuillez remplir tous les champs obligatoires.", "danger")
|
|
return redirect(url_for("comptes.gestion_comptes", societe_id=id_societe))
|
|
|
|
try:
|
|
debit_val = float(debit) if debit else 0.0
|
|
credit_val = float(credit) if credit else 0.0
|
|
except ValueError:
|
|
flash("Veuillez saisir des valeurs numériques valides.", "danger")
|
|
return redirect(url_for("comptes.gestion_comptes", societe_id=id_societe))
|
|
|
|
try:
|
|
db.session.execute(
|
|
text(
|
|
"UPDATE compte SET date_operation = :date_op, "
|
|
"libelle = :libelle, debit = :debit, credit = :credit WHERE id = :id"
|
|
),
|
|
{
|
|
"date_op": date_operation,
|
|
"libelle": libelle or None,
|
|
"debit": debit_val,
|
|
"credit": credit_val,
|
|
"id": id,
|
|
},
|
|
)
|
|
db.session.commit()
|
|
flash("Écriture modifiée avec succès !", "success")
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
current_app.logger.error("Erreur modification écriture : %s", e, exc_info=True)
|
|
flash("Erreur lors de la modification de l'écriture.", "danger")
|
|
|
|
# On reste sur la même société après validation
|
|
return redirect(url_for("comptes.gestion_comptes", societe_id=id_societe)) |