881 lines
34 KiB
Python
881 lines
34 KiB
Python
from flask import Blueprint, render_template, request, redirect, url_for, session, flash, Response, jsonify
|
|
import time
|
|
import requests
|
|
import csv
|
|
import io
|
|
from werkzeug.security import check_password_hash, generate_password_hash
|
|
from sqlalchemy import text
|
|
from app import db
|
|
from datetime import datetime
|
|
|
|
main = Blueprint('main', __name__)
|
|
|
|
# Route pour la page d'accueil
|
|
@main.route('/', methods=['GET', 'POST'])
|
|
def index():
|
|
if 'user_id' in session:
|
|
return redirect(url_for('main.menu'))
|
|
|
|
if 'login_attempts' not in session:
|
|
session['login_attempts'] = 0
|
|
|
|
if request.method == 'POST':
|
|
user_login = request.form.get('login')
|
|
user_pass = request.form.get('pass')
|
|
|
|
query = text("SELECT id, pass AS user_pass, nom, prenom FROM login WHERE login = :login AND actif = 1")
|
|
result = db.session.execute(query, {"login": user_login}).fetchone()
|
|
|
|
if result and check_password_hash(result.user_pass, user_pass):
|
|
session.pop('login_attempts', None)
|
|
session['user_id'] = result.id
|
|
session['user_name'] = f"{result.prenom} {result.nom}"
|
|
return redirect(url_for('main.menu'))
|
|
else:
|
|
session['login_attempts'] += 1
|
|
|
|
if session['login_attempts'] >= 3:
|
|
session.pop('login_attempts', None)
|
|
return redirect('https://giraud-finance.com')
|
|
|
|
essais_restants = 3 - session['login_attempts']
|
|
flash(f"Identifiant ou mot de passe incorrect. Il vous reste {essais_restants} essai(s).", "danger")
|
|
|
|
return render_template('index.html')
|
|
|
|
# Route pour la déconnexion
|
|
@main.route('/logout')
|
|
def logout():
|
|
session.clear()
|
|
return redirect(url_for('main.index'))
|
|
|
|
# Route pour le menu principal
|
|
@main.route('/menu')
|
|
def menu():
|
|
if 'user_id' not in session:
|
|
return redirect(url_for('main.index'))
|
|
return render_template('menu.html')
|
|
|
|
# Route pour la gestion des utilisateurs
|
|
@main.route('/gestion-utilisateurs', methods=['GET', 'POST'])
|
|
def gestion_utilisateurs():
|
|
if 'user_id' not in session:
|
|
return redirect(url_for('main.index'))
|
|
|
|
if request.method == 'POST':
|
|
action = request.form.get('action')
|
|
user_id = request.form.get('id')
|
|
user_actif = request.form.get('actif', '1')
|
|
user_prenom = request.form.get('prenom')
|
|
user_nom = request.form.get('nom')
|
|
user_login = request.form.get('login')
|
|
user_mail = request.form.get('mail')
|
|
user_pass = request.form.get('pass')
|
|
|
|
if action == 'creer':
|
|
if user_pass:
|
|
hashed_pass = generate_password_hash(user_pass)
|
|
query = text("""
|
|
INSERT INTO login (login, nom, prenom, mail, pass, actif)
|
|
VALUES (:login, :nom, :prenom, :mail, :pass, :actif)
|
|
""")
|
|
db.session.execute(query, {
|
|
"login": user_login,
|
|
"nom": user_nom,
|
|
"prenom": user_prenom,
|
|
"mail": user_mail,
|
|
"pass": hashed_pass,
|
|
"actif": user_actif
|
|
})
|
|
db.session.commit()
|
|
flash("Utilisateur créé avec succès.", "success")
|
|
else:
|
|
flash("Le mot de passe est obligatoire pour créer un utilisateur.", "danger")
|
|
|
|
elif action == 'modifier' and user_id:
|
|
if user_pass:
|
|
hashed_pass = generate_password_hash(user_pass)
|
|
query = text("""
|
|
UPDATE login
|
|
SET login = :login, nom = :nom, prenom = :prenom, mail = :mail, pass = :pass, actif = :actif
|
|
WHERE id = :id
|
|
""")
|
|
db.session.execute(query, {
|
|
"id": user_id,
|
|
"login": user_login,
|
|
"nom": user_nom,
|
|
"prenom": user_prenom,
|
|
"mail": user_mail,
|
|
"pass": hashed_pass,
|
|
"actif": user_actif
|
|
})
|
|
else:
|
|
query = text("""
|
|
UPDATE login
|
|
SET login = :login, nom = :nom, prenom = :prenom, mail = :mail, actif = :actif
|
|
WHERE id = :id
|
|
""")
|
|
db.session.execute(query, {
|
|
"id": user_id,
|
|
"login": user_login,
|
|
"nom": user_nom,
|
|
"prenom": user_prenom,
|
|
"mail": user_mail,
|
|
"actif": user_actif
|
|
})
|
|
db.session.commit()
|
|
flash("Utilisateur modifié avec succès.", "success")
|
|
|
|
return redirect(url_for('main.gestion_utilisateurs'))
|
|
|
|
users_query = text("SELECT id, login, nom, prenom, mail, actif FROM login ORDER BY id ASC")
|
|
result = db.session.execute(users_query)
|
|
|
|
users = []
|
|
for row in result:
|
|
users.append({
|
|
'id': row.id,
|
|
'login': row.login if row.login else '',
|
|
'nom': row.nom if row.nom else '',
|
|
'prenom': row.prenom if row.prenom else '',
|
|
'mail': row.mail if row.mail else '',
|
|
'actif': row.actif if row.actif is not None else 1
|
|
})
|
|
|
|
return render_template('gestion_utilisateurs.html', users=users)
|
|
|
|
# Route pour la gestion des actions
|
|
@main.route('/gestion-actions', methods=['GET', 'POST'])
|
|
def gestion_actions():
|
|
if 'user_id' not in session:
|
|
return redirect(url_for('main.index'))
|
|
|
|
if request.method == 'POST':
|
|
action_type = request.form.get('action')
|
|
action_id = request.form.get('id')
|
|
isin = request.form.get('isin') or None
|
|
ticker = request.form.get('ticker')
|
|
company_name = request.form.get('company_name')
|
|
exchange = request.form.get('exchange')
|
|
pays = request.form.get('pays')
|
|
currency = request.form.get('currency')
|
|
|
|
if action_type == 'creer':
|
|
try:
|
|
query = text("""
|
|
INSERT INTO actions (isin, ticker, company_name, exchange, pays, currency, updated_at)
|
|
VALUES (:isin, :ticker, :company_name, :exchange, :pays, :currency, NOW())
|
|
""")
|
|
db.session.execute(query, {
|
|
"isin": isin, "ticker": ticker, "company_name": company_name,
|
|
"exchange": exchange, "pays": pays, "currency": currency
|
|
})
|
|
db.session.commit()
|
|
flash("Action créée avec succès", "success")
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
flash("Erreur lors de la création de l'action.", "danger")
|
|
|
|
elif action_type == 'modifier' and action_id:
|
|
try:
|
|
query = text("""
|
|
UPDATE actions
|
|
SET isin = :isin, ticker = :ticker, company_name = :company_name, exchange = :exchange,
|
|
pays = :pays, currency = :currency, updated_at = NOW()
|
|
WHERE id = :id
|
|
""")
|
|
db.session.execute(query, {
|
|
"id": action_id, "isin": isin, "ticker": ticker, "company_name": company_name,
|
|
"exchange": exchange, "pays": pays, "currency": currency
|
|
})
|
|
db.session.commit()
|
|
flash("Modification effectuée", "success")
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
flash("Erreur lors de la modification.", "danger")
|
|
|
|
return redirect(url_for('main.gestion_actions'))
|
|
|
|
actions_query = text("SELECT id, isin, ticker, company_name, exchange, pays, currency, updated_at FROM actions ORDER BY company_name ASC")
|
|
result = db.session.execute(actions_query)
|
|
|
|
actions_list = []
|
|
for row in result:
|
|
actions_list.append({
|
|
'id': row.id,
|
|
'isin': row.isin if row.isin else '',
|
|
'ticker': row.ticker if row.ticker else '',
|
|
'company_name': row.company_name if row.company_name else 'Sans nom',
|
|
'exchange': row.exchange if row.exchange else '',
|
|
'pays': row.pays if row.pays else '',
|
|
'currency': row.currency if row.currency else '',
|
|
'updated_at': row.updated_at.strftime('%Y-%m-%d %H:%M:%S') if row.updated_at else ''
|
|
})
|
|
|
|
return render_template('gestion_actions.html', actions=actions_list)
|
|
|
|
# Route pour la gestion des sociétés
|
|
@main.route('/gestion-societes', methods=['GET', 'POST'])
|
|
def gestion_societes():
|
|
if 'user_id' not in session:
|
|
return redirect(url_for('main.index'))
|
|
|
|
if request.method == 'POST':
|
|
action = request.form.get('action')
|
|
soc_id = request.form.get('id')
|
|
|
|
nom = request.form.get('Nom')
|
|
forme = request.form.get('Forme')
|
|
capital = request.form.get('Capital')
|
|
siren = request.form.get('siren')
|
|
siret = request.form.get('siret')
|
|
rcs = request.form.get('RCS')
|
|
tva = request.form.get('TVA')
|
|
naf = request.form.get('NAF')
|
|
tel = request.form.get('Tel')
|
|
web = request.form.get('Web')
|
|
mail = request.form.get('Mail')
|
|
|
|
if action == 'creer':
|
|
query = text("""
|
|
INSERT INTO societe (Nom, Forme, Capital, siren, siret, RCS, TVA, NAF, Tel, Web, Mail)
|
|
VALUES (:nom, :forme, :capital, :siren, :siret, :rcs, :tva, :naf, :tel, :web, :mail)
|
|
""")
|
|
db.session.execute(query, {
|
|
"nom": nom, "forme": forme, "capital": capital,
|
|
"siren": siren, "siret": siret, "rcs": rcs,
|
|
"tva": tva, "naf": naf, "tel": tel,
|
|
"web": web, "mail": mail
|
|
})
|
|
db.session.commit()
|
|
flash("Société créée avec succès.", "success")
|
|
|
|
elif action == 'modifier' and soc_id:
|
|
query = text("""
|
|
UPDATE societe
|
|
SET Nom = :nom, Forme = :forme, Capital = :capital, siren = :siren,
|
|
siret = :siret, RCS = :rcs, TVA = :tva, NAF = :naf,
|
|
Tel = :tel, Web = :web, Mail = :mail
|
|
WHERE id = :id
|
|
""")
|
|
db.session.execute(query, {
|
|
"id": soc_id, "nom": nom, "forme": forme, "capital": capital,
|
|
"siren": siren, "siret": siret, "rcs": rcs,
|
|
"tva": tva, "naf": naf, "tel": tel,
|
|
"web": web, "mail": mail
|
|
})
|
|
db.session.commit()
|
|
flash("Société modifiée avec succès.", "success")
|
|
|
|
return redirect(url_for('main.gestion_societes'))
|
|
|
|
societes_query = text("SELECT id, Nom, Forme, Capital, siren, siret, RCS, TVA, NAF, Tel, Web, Mail FROM societe ORDER BY id ASC")
|
|
result = db.session.execute(societes_query)
|
|
|
|
societes = []
|
|
for row in result:
|
|
societes.append({
|
|
'id': row.id,
|
|
'Nom': row.Nom if row.Nom else '',
|
|
'Forme': row.Forme if row.Forme else '',
|
|
'Capital': row.Capital if row.Capital is not None else 0,
|
|
'siren': row.siren if row.siren is not None else '',
|
|
'siret': row.siret if row.siret is not None else '',
|
|
'RCS': row.RCS if row.RCS else '',
|
|
'TVA': row.TVA if row.TVA else '',
|
|
'NAF': row.NAF if row.NAF else '',
|
|
'Tel': row.Tel if row.Tel else '',
|
|
'Web': row.Web if row.Web else '',
|
|
'Mail': row.Mail if row.Mail else ''
|
|
})
|
|
|
|
return render_template('gestion_societes.html', societes=societes)
|
|
|
|
# Routes pour la gestion des imports/exports des actions CSV
|
|
@main.route('/gestion-import-export-actions-csv', methods=['GET'])
|
|
def gestion_import_export_actions_csv():
|
|
if 'user_id' not in session:
|
|
return redirect(url_for('main.index'))
|
|
return render_template('gestion_import_export_actions_csv.html')
|
|
|
|
|
|
# Route pour l'exportation des actions vers un fichier CSV
|
|
@main.route('/export-actions-csv', methods=['GET'])
|
|
def export_actions_csv():
|
|
if 'user_id' not in session:
|
|
return redirect(url_for('main.index'))
|
|
|
|
try:
|
|
query = text("SELECT id, isin, ticker, company_name, exchange, pays, currency, updated_at FROM actions")
|
|
result = db.session.execute(query)
|
|
|
|
output = io.StringIO()
|
|
writer = csv.writer(output, delimiter=';', quoting=csv.QUOTE_MINIMAL)
|
|
|
|
writer.writerow(['id', 'isin', 'ticker', 'company_name', 'exchange', 'pays', 'currency', 'updated_at'])
|
|
|
|
for row in result:
|
|
writer.writerow([
|
|
row.id,
|
|
row.isin or '',
|
|
row.ticker or '',
|
|
row.company_name or '',
|
|
row.exchange or '',
|
|
row.pays or '',
|
|
row.currency or '',
|
|
row.updated_at.strftime('%Y-%m-%d %H:%M:%S') if row.updated_at else ''
|
|
])
|
|
|
|
output.seek(0)
|
|
|
|
return Response(
|
|
output.getvalue(),
|
|
mimetype="text/csv",
|
|
headers={"Content-Disposition": "attachment;filename=actions_export.csv"}
|
|
)
|
|
except Exception as e:
|
|
flash("Erreur lors de l'exportation du fichier CSV.", "danger")
|
|
return redirect(url_for('main.gestion_import_export_actions_csv'))
|
|
|
|
|
|
# Route pour l'importation des actions depuis un fichier CSV
|
|
@main.route('/import-actions-csv', methods=['POST'])
|
|
def import_actions_csv():
|
|
if 'user_id' not in session:
|
|
return redirect(url_for('main.index'))
|
|
|
|
if 'file' not in request.files:
|
|
flash("Aucun fichier sélectionné.", "danger")
|
|
return redirect(url_for('main.gestion_import_export_actions_csv'))
|
|
|
|
file = request.files['file']
|
|
|
|
if file.filename == '':
|
|
flash("Aucun fichier sélectionné.", "danger")
|
|
return redirect(url_for('main.gestion_import_export_actions_csv'))
|
|
|
|
if file and file.filename.endswith('.csv'):
|
|
try:
|
|
stream = io.TextIOWrapper(file.stream, encoding="utf-8")
|
|
csv_reader = csv.reader(stream, delimiter=';')
|
|
|
|
header = next(csv_reader, None)
|
|
if not header:
|
|
flash("Le fichier CSV est vide.", "danger")
|
|
return redirect(url_for('main.gestion_import_export_actions_csv'))
|
|
|
|
count = 0
|
|
for row in csv_reader:
|
|
if len(row) >= 6:
|
|
isin = row[0].strip() if row[0] != '' else None
|
|
ticker = row[1].strip() if row[1] != '' else None
|
|
company_name = row[2].strip() if row[2] != '' else None
|
|
exchange = row[3].strip() if row[3] != '' else ''
|
|
pays = row[4].strip() if row[4] != '' else ''
|
|
currency = row[5].strip() if row[5] != '' else ''
|
|
|
|
if not isin or not ticker:
|
|
continue
|
|
|
|
query = text("""
|
|
INSERT INTO actions (isin, ticker, company_name, exchange, pays, currency, updated_at)
|
|
VALUES (:isin, :ticker, :company_name, :exchange, :pays, :currency, NOW())
|
|
ON DUPLICATE KEY UPDATE
|
|
ticker = :ticker, company_name = :company_name,
|
|
exchange = :exchange, pays = :pays, currency = :currency, updated_at = NOW()
|
|
""")
|
|
db.session.execute(query, {
|
|
"isin": isin, "ticker": ticker, "company_name": company_name,
|
|
"exchange": exchange, "pays": pays, "currency": currency
|
|
})
|
|
count += 1
|
|
|
|
db.session.commit()
|
|
flash(f"Importation réussie : {count} actions traitées.", "success")
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
flash(f"Erreur lors de l'importation du fichier : {str(e)}", "danger")
|
|
|
|
return redirect(url_for('main.gestion_import_export_actions_csv'))
|
|
else:
|
|
flash("Veuillez fournir un fichier au format .csv valide.", "warning")
|
|
return redirect(url_for('main.gestion_import_export_actions_csv'))
|
|
|
|
|
|
# Route pour la gestion des ordres
|
|
@main.route("/gestion_ordres", methods=["GET", "POST"])
|
|
def gestion_ordres():
|
|
if "user_id" not in session:
|
|
return redirect(url_for("main.index"))
|
|
|
|
selected_societe_id = (
|
|
request.form.get("societe_id")
|
|
or request.args.get("societe_id")
|
|
or session.get("selected_societe_id")
|
|
)
|
|
|
|
# Récupération de l'année sélectionnée (par défaut "Tout")
|
|
selected_annee = (
|
|
request.form.get("annee")
|
|
or request.args.get("annee")
|
|
or session.get("selected_annee", "Tout")
|
|
)
|
|
session["selected_annee"] = selected_annee
|
|
|
|
societes = []
|
|
selected_societe = ""
|
|
groupes_dict = {}
|
|
total_plus_value_globale = 0.0
|
|
|
|
with db.engine.connect() as connection:
|
|
result_soc = connection.execute(
|
|
text("SELECT id, Nom FROM societe ORDER BY Nom ASC")
|
|
)
|
|
societes = [dict(row._mapping) for row in result_soc]
|
|
|
|
if societes:
|
|
if not selected_societe_id:
|
|
selected_societe_id = societes[0]["id"]
|
|
else:
|
|
selected_societe_id = int(selected_societe_id)
|
|
|
|
session["selected_societe_id"] = selected_societe_id
|
|
|
|
for soc in societes:
|
|
if soc["id"] == selected_societe_id:
|
|
selected_societe = soc["Nom"]
|
|
break
|
|
|
|
# --- CALCUL DE LA PLUS-VALUE TOTALE ( GLOBALE OU FILTRÉE PAR ANNÉE ) ---
|
|
annee_str = str(selected_annee).strip().lower()
|
|
|
|
if not annee_str or annee_str == "tout":
|
|
query_pv = text("""
|
|
SELECT SUM(PlusValue) as total_pv
|
|
FROM ordre
|
|
WHERE id_societe = :soc_id AND PlusValue IS NOT NULL
|
|
""")
|
|
res_pv = connection.execute(query_pv, {"soc_id": selected_societe_id}).fetchone()
|
|
else:
|
|
query_pv = text("""
|
|
SELECT SUM(PlusValue) as total_pv
|
|
FROM ordre
|
|
WHERE id_societe = :soc_id
|
|
AND PlusValue IS NOT NULL
|
|
AND LEFT(CAST(date_plusvalue AS CHAR), 4) = :annee
|
|
""")
|
|
res_pv = connection.execute(
|
|
query_pv, {"soc_id": selected_societe_id, "annee": annee_str}
|
|
).fetchone()
|
|
|
|
if res_pv and res_pv.total_pv is not None:
|
|
total_plus_value_globale = float(res_pv.total_pv)
|
|
else:
|
|
total_plus_value_globale = 0.0
|
|
|
|
# --- CHARGEMENT DES ORDRES ---
|
|
query = text("""
|
|
SELECT o.ID as ordre_id, o.isin, o.PlusValue, o.date_plusvalue,
|
|
op.ID as pied_id, op.Date_op, op.Quantite, op.Prix_brut, op.Frais_brocker,
|
|
op.Frais_etat, op.Frais_autre, op.Ordre, op.Type, op.Etat,
|
|
a.ticker, a.company_name, a.currency
|
|
FROM ordre o
|
|
JOIN ordre_pied op ON o.ID = op.Id_entete
|
|
JOIN actions a ON o.isin = a.isin
|
|
WHERE o.id_societe = :soc_id
|
|
ORDER BY o.ID DESC, op.Date_op DESC
|
|
""")
|
|
|
|
result_ordres = connection.execute(
|
|
query, {"soc_id": selected_societe_id}
|
|
)
|
|
|
|
for row in result_ordres:
|
|
r = dict(row._mapping)
|
|
ordre_id = r["ordre_id"] # Utilisation de l'ID d'en-tête comme clé unique
|
|
isin = r["isin"]
|
|
|
|
quantite = float(r["Quantite"]) if r["Quantite"] is not None else 0.0
|
|
prix_brut = float(r["Prix_brut"]) if r["Prix_brut"] is not None else 0.0
|
|
frais_brocker = (
|
|
float(r["Frais_brocker"]) if r["Frais_brocker"] is not None else 0.0
|
|
)
|
|
frais_etat = (
|
|
float(r["Frais_etat"]) if r["Frais_etat"] is not None else 0.0
|
|
)
|
|
frais_autre = (
|
|
float(r["Frais_autre"]) if r["Frais_autre"] is not None else 0.0
|
|
)
|
|
|
|
frais_total = frais_brocker + frais_etat + frais_autre
|
|
sens_ordre = str(r["Ordre"]).strip().lower()
|
|
type_ordre = (
|
|
str(r["Type"]).strip().lower()
|
|
if r["Type"]
|
|
else "long"
|
|
)
|
|
|
|
# Calcul Prix Net unitaire
|
|
if sens_ordre == "vente":
|
|
prix_net = prix_brut - frais_total
|
|
else:
|
|
prix_net = prix_brut + frais_total
|
|
|
|
# Calculs Ligne (Pied)
|
|
eng_brut = quantite * prix_brut
|
|
eng_net = quantite * prix_net
|
|
|
|
pied_data = {
|
|
"id": r["pied_id"],
|
|
"date_op": r["Date_op"],
|
|
"quantite": quantite,
|
|
"prix_brut": prix_brut,
|
|
"frais": frais_total,
|
|
"frais_brocker": frais_brocker,
|
|
"frais_etat": frais_etat,
|
|
"frais_autre": frais_autre,
|
|
"prix_net": prix_net,
|
|
"eng_brut": eng_brut,
|
|
"eng_net": eng_net,
|
|
"ordre": r["Ordre"],
|
|
"type_ordre": r["Type"],
|
|
"etat": r["Etat"],
|
|
}
|
|
|
|
# MODIFICATION ICI : On utilise ordre_id comme clé du dictionnaire pour séparer les blocs
|
|
if ordre_id not in groupes_dict:
|
|
groupes_dict[ordre_id] = {
|
|
"id": ordre_id,
|
|
"isin": isin,
|
|
"company_name": r["company_name"],
|
|
"ticker": r["ticker"],
|
|
"pieds": [],
|
|
"total_quantite": 0.0,
|
|
"total_prix_brut": 0.0,
|
|
"total_frais": 0.0,
|
|
"total_prix_net": 0.0,
|
|
"total_eng_brut": 0.0,
|
|
"total_eng_net": 0.0,
|
|
"plus_value": r["PlusValue"],
|
|
"date_plusvalue": r["date_plusvalue"],
|
|
}
|
|
|
|
groupes_dict[ordre_id]["pieds"].append(pied_data)
|
|
groupes_dict[ordre_id]["total_prix_brut"] += prix_brut
|
|
groupes_dict[ordre_id]["total_frais"] += frais_total
|
|
groupes_dict[ordre_id]["total_prix_net"] += prix_net
|
|
|
|
# --- APPLICATION DE VOS RÈGLES POUR LES TOTAUX ---
|
|
|
|
# 1. QUANTITÉ
|
|
if sens_ordre == "achat":
|
|
groupes_dict[ordre_id]["total_quantite"] += quantite
|
|
elif sens_ordre == "vente":
|
|
groupes_dict[ordre_id]["total_quantite"] -= quantite
|
|
|
|
# 2. ENGAGEMENT BRUT
|
|
if sens_ordre == "achat":
|
|
groupes_dict[ordre_id]["total_eng_brut"] -= eng_brut
|
|
elif sens_ordre == "vente":
|
|
groupes_dict[ordre_id]["total_eng_brut"] += eng_brut
|
|
|
|
# 3. ENGAGEMENT NET
|
|
if sens_ordre == "achat":
|
|
groupes_dict[ordre_id]["total_eng_net"] -= eng_net
|
|
elif sens_ordre == "vente":
|
|
groupes_dict[ordre_id]["total_eng_net"] += eng_net
|
|
|
|
# --- VÉRIFICATION ET MISE À JOUR AUTOMATIQUE (EN BASE ET EN MÉMOIRE) ---
|
|
try:
|
|
with db.engine.begin() as conn:
|
|
for ordre_id, groupe in groupes_dict.items():
|
|
if abs(groupe["total_quantite"]) < 1e-9:
|
|
total_eng_net = groupe["total_eng_net"]
|
|
|
|
max_date = max((p["date_op"] for p in groupe["pieds"] if p["date_op"] is not None), default=None)
|
|
|
|
conn.execute(
|
|
text(
|
|
"UPDATE ordre_pied SET Etat = :nouvel_etat WHERE Id_entete = :ordre_id"
|
|
),
|
|
{"nouvel_etat": "soldé", "ordre_id": ordre_id},
|
|
)
|
|
|
|
conn.execute(
|
|
text(
|
|
"UPDATE ordre SET PlusValue = :plus_value, date_plusvalue = :date_pv WHERE ID = :ordre_id"
|
|
),
|
|
{"plus_value": total_eng_net, "date_pv": max_date, "ordre_id": ordre_id},
|
|
)
|
|
|
|
groupe["plus_value"] = total_eng_net
|
|
groupe["date_plusvalue"] = max_date
|
|
for pied in groupe["pieds"]:
|
|
pied["etat"] = "soldé"
|
|
|
|
except Exception as e:
|
|
pass
|
|
|
|
groupes_ordres = list(groupes_dict.values())
|
|
|
|
annees_combo = ["Tout"] + [str(an) for an in range(2020, 2041)]
|
|
|
|
return render_template(
|
|
"gestion_ordres.html",
|
|
societes=societes,
|
|
selected_societe_id=selected_societe_id,
|
|
selected_societe=selected_societe,
|
|
groupes_ordres=groupes_ordres,
|
|
total_plus_value_globale=total_plus_value_globale,
|
|
annees_combo=annees_combo,
|
|
selected_annee=selected_annee,
|
|
)
|
|
|
|
|
|
# Route pour la creation d'un ordre
|
|
@main.route("/creer-ordre", methods=["GET", "POST"])
|
|
def creer_ordre():
|
|
if "user_id" not in session:
|
|
return redirect(url_for("main.index"))
|
|
return render_template("creer_ordre.html")
|
|
|
|
|
|
# Route pour vérifier l'existence d'un ISIN
|
|
@main.route("/api/verifier_isin", methods=["POST"])
|
|
def verifier_isin():
|
|
data = request.get_json()
|
|
isin = data.get("isin", "").strip()
|
|
|
|
query = text(
|
|
"SELECT company_name, ticker, currency FROM actions WHERE isin = :isin"
|
|
)
|
|
action = db.session.execute(query, {"isin": isin}).fetchone()
|
|
|
|
if action:
|
|
return jsonify({
|
|
"exists": True,
|
|
"company_name": action.company_name,
|
|
"ticker": action.ticker,
|
|
"currency": action.currency,
|
|
})
|
|
else:
|
|
return jsonify({"exists": False})
|
|
|
|
|
|
# Route pour la création d'un ordre de traitement
|
|
@main.route("/creer-ordre-traitement", methods=["POST"])
|
|
def creer_ordre_traitement():
|
|
if "user_id" not in session:
|
|
return redirect(url_for("main.index"))
|
|
|
|
isin = request.form.get("isin")
|
|
date_op = request.form.get("date_op")
|
|
selected_annee = request.form.get("annee", "tout")
|
|
|
|
try:
|
|
quantite = float(request.form.get("quantite") or 0)
|
|
prix_brut = float(request.form.get("prix_brut") or 0)
|
|
frais_brocker = float(request.form.get("frais_brocker") or 0)
|
|
frais_etat = float(request.form.get("frais_etat") or 0)
|
|
frais_autre = float(request.form.get("frais_autre") or 0)
|
|
except ValueError:
|
|
flash("Veuillez saisir des valeurs numériques valides pour les quantités, prix et frais.", "danger")
|
|
return redirect(url_for("main.gestion_ordres", annee=selected_annee))
|
|
|
|
ordre = request.form.get("ordre")
|
|
type_ordre = request.form.get("type_ordre")
|
|
etat = request.form.get("etat")
|
|
|
|
# Récupération de l'id_societe courant stocké dans la session
|
|
id_societe = session.get("selected_societe_id")
|
|
|
|
if not id_societe:
|
|
flash(
|
|
"Veuillez sélectionner une société active avant de créer un ordre.",
|
|
"danger",
|
|
)
|
|
return redirect(url_for("main.gestion_ordres", annee=selected_annee))
|
|
|
|
try:
|
|
# 1. Insertion dans la table 'ordre' avec l'isin et l'id_societe
|
|
query_ordre = text(
|
|
"INSERT INTO ordre (id_societe, isin) VALUES (:id_societe, :isin)"
|
|
)
|
|
db.session.execute(query_ordre, {"id_societe": id_societe, "isin": isin})
|
|
|
|
# Récupérer l'ID généré pour l'en-tête (Id_entete)
|
|
result_id = db.session.execute(
|
|
text("SELECT LAST_INSERT_ID()")
|
|
).scalar()
|
|
if not result_id:
|
|
result_id = db.session.execute(
|
|
text(
|
|
"SELECT ID FROM ordre WHERE isin = :isin ORDER BY ID DESC LIMIT 1"
|
|
),
|
|
{"isin": isin},
|
|
).fetchone()[0]
|
|
|
|
# 2. Insertion dans la table 'ordre_pied' avec les 3 frais distincts
|
|
query_pied = text("""
|
|
INSERT INTO ordre_pied (Id_entete, Date_op, Quantite, Prix_brut, Frais_brocker, Frais_etat, Frais_autre, Ordre, Type, Etat)
|
|
VALUES (:id_entete, :date_op, :quantite, :prix_brut, :frais_brocker, :frais_etat, :frais_autre, :ordre, :type_ordre, :etat)
|
|
""")
|
|
db.session.execute(
|
|
query_pied,
|
|
{
|
|
"id_entete": result_id,
|
|
"date_op": date_op,
|
|
"quantite": quantite,
|
|
"prix_brut": prix_brut,
|
|
"frais_brocker": frais_brocker,
|
|
"frais_etat": frais_etat,
|
|
"frais_autre": frais_autre,
|
|
"ordre": ordre,
|
|
"type_ordre": type_ordre,
|
|
"etat": etat,
|
|
},
|
|
)
|
|
|
|
db.session.commit()
|
|
flash("Ordre créé avec succès !", "success")
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
flash(f"Erreur lors de la création de l'ordre : {e}", "danger")
|
|
|
|
return redirect(url_for("main.gestion_ordres", annee=selected_annee))
|
|
|
|
|
|
# Route pour créer une nouvelle ligne rattachée à un ordre existant (depuis le bouton "Créer")
|
|
@main.route("/ordre/<int:ordre_id>/ajouter-ligne", methods=["GET", "POST"])
|
|
def ajouter_ligne_ordre(ordre_id):
|
|
if "user_id" not in session:
|
|
return redirect(url_for("main.index"))
|
|
|
|
# Récupérer l'en-tête de l'ordre, l'ISIN et le nom de la société associée
|
|
query_info = text("""
|
|
SELECT o.ID, o.isin, a.company_name
|
|
FROM ordre o
|
|
JOIN actions a ON o.isin = a.isin
|
|
WHERE o.ID = :ordre_id
|
|
""")
|
|
ordre_info = db.session.execute(query_info, {"ordre_id": ordre_id}).fetchone()
|
|
|
|
if not ordre_info:
|
|
flash("Ordre introuvable.", "danger")
|
|
return redirect(url_for("main.gestion_ordres"))
|
|
|
|
selected_annee = request.args.get("annee", "tout")
|
|
|
|
if request.method == "POST":
|
|
selected_annee = request.form.get("annee", "tout")
|
|
try:
|
|
quantite = float(request.form.get("quantite") or 0)
|
|
prix_brut = float(request.form.get("prix_brut") or 0)
|
|
frais_brocker = float(request.form.get("frais_brocker") or 0)
|
|
frais_etat = float(request.form.get("frais_etat") or 0)
|
|
frais_autre = float(request.form.get("frais_autre") or 0)
|
|
except ValueError:
|
|
flash("Veuillez saisir des valeurs numériques valides.", "danger")
|
|
return redirect(url_for("main.ajouter_ligne_ordre", ordre_id=ordre_id, annee=selected_annee))
|
|
|
|
try:
|
|
query_pied = text("""
|
|
INSERT INTO ordre_pied (Id_entete, Date_op, Quantite, Prix_brut, Frais_brocker, Frais_etat, Frais_autre, Ordre, Type, Etat)
|
|
VALUES (:id_entete, :date_op, :quantite, :prix_brut, :frais_brocker, :frais_etat, :frais_autre, :ordre, :type_ordre, :etat)
|
|
""")
|
|
db.session.execute(query_pied, {
|
|
"id_entete": ordre_id,
|
|
"date_op": request.form.get("date_op"),
|
|
"quantite": quantite,
|
|
"prix_brut": prix_brut,
|
|
"frais_brocker": frais_brocker,
|
|
"frais_etat": frais_etat,
|
|
"frais_autre": frais_autre,
|
|
"ordre": request.form.get("ordre"),
|
|
"type_ordre": request.form.get("type_ordre"),
|
|
"etat": request.form.get("etat"),
|
|
})
|
|
db.session.commit()
|
|
flash("Ligne ajoutée avec succès à l'ordre !", "success")
|
|
return redirect(url_for("main.gestion_ordres", annee=selected_annee))
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
flash(f"Erreur lors de l'ajout : {e}", "danger")
|
|
|
|
contexte_form = {
|
|
"mode": "creer_ligne",
|
|
"isin": ordre_info.isin,
|
|
"company_name": ordre_info.company_name,
|
|
"action_url": url_for("main.ajouter_ligne_ordre", ordre_id=ordre_id)
|
|
}
|
|
return render_template("form_ordre_ligne.html", data=contexte_form, ligne=None)
|
|
|
|
|
|
# Route pour modifier une ligne spécifique (depuis le bouton "Modifier" de la ligne)
|
|
@main.route("/modifier-ligne-ordre/<int:pied_id>", methods=["GET", "POST"])
|
|
def modifier_ligne_ordre(pied_id):
|
|
if "user_id" not in session:
|
|
return redirect(url_for("main.index"))
|
|
|
|
# Récupérer la ligne du pied, l'ISIN et le nom de l'entreprise
|
|
query_get = text("""
|
|
SELECT op.*, o.isin, a.company_name
|
|
FROM ordre_pied op
|
|
JOIN ordre o ON op.Id_entete = o.ID
|
|
JOIN actions a ON o.isin = a.isin
|
|
WHERE op.ID = :pied_id
|
|
""")
|
|
ligne = db.session.execute(query_get, {"pied_id": pied_id}).fetchone()
|
|
|
|
if not ligne:
|
|
flash("Ligne d'ordre introuvable.", "danger")
|
|
return redirect(url_for("main.gestion_ordres"))
|
|
|
|
selected_annee = request.args.get("annee", "tout")
|
|
|
|
if request.method == "POST":
|
|
selected_annee = request.form.get("annee", "tout")
|
|
try:
|
|
quantite = float(request.form.get("quantite") or 0)
|
|
prix_brut = float(request.form.get("prix_brut") or 0)
|
|
frais_brocker = float(request.form.get("frais_brocker") or 0)
|
|
frais_etat = float(request.form.get("frais_etat") or 0)
|
|
frais_autre = float(request.form.get("frais_autre") or 0)
|
|
except ValueError:
|
|
flash("Veuillez saisir des valeurs numériques valides.", "danger")
|
|
return redirect(url_for("main.modifier_ligne_ordre", pied_id=pied_id, annee=selected_annee))
|
|
|
|
try:
|
|
query_update = text("""
|
|
UPDATE ordre_pied
|
|
SET Date_op = :date_op, Quantite = :quantite, Prix_brut = :prix_brut,
|
|
Frais_brocker = :frais_brocker, Frais_etat = :frais_etat, Frais_autre = :frais_autre,
|
|
Ordre = :ordre, Type = :type_ordre, Etat = :etat
|
|
WHERE ID = :pied_id
|
|
""")
|
|
db.session.execute(query_update, {
|
|
"date_op": request.form.get("date_op"),
|
|
"quantite": quantite,
|
|
"prix_brut": prix_brut,
|
|
"frais_brocker": frais_brocker,
|
|
"frais_etat": frais_etat,
|
|
"frais_autre": frais_autre,
|
|
"ordre": request.form.get("ordre"),
|
|
"type_ordre": request.form.get("type_ordre"),
|
|
"etat": request.form.get("etat"),
|
|
"pied_id": pied_id
|
|
})
|
|
db.session.commit()
|
|
flash("Ligne modifiée avec succès !", "success")
|
|
return redirect(url_for("main.gestion_ordres", annee=selected_annee))
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
flash(f"Erreur lors de la modification : {e}", "danger")
|
|
|
|
contexte_form = {
|
|
"mode": "modifier_ligne",
|
|
"isin": ligne.isin,
|
|
"company_name": ligne.company_name,
|
|
"action_url": url_for("main.modifier_ligne_ordre", pied_id=pied_id)
|
|
}
|
|
return render_template("form_ordre_ligne.html", data=contexte_form, ligne=ligne) |