412 lines
17 KiB
Python
412 lines
17 KiB
Python
from flask import Blueprint, render_template, request, redirect, url_for, session, flash, Response, session
|
|
import time
|
|
import requests
|
|
import mysql.connector
|
|
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'))
|
|
|
|
# Lecture exacte de la table login de votre base de données
|
|
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'))
|
|
|
|
# Récupération avec la nouvelle structure
|
|
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'))
|
|
|
|
# Lecture de la table societe
|
|
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')
|
|
|
|
@main.route('/export-actions-csv', methods=['GET'])
|
|
def export_actions_csv():
|
|
if 'user_id' not in session:
|
|
return redirect(url_for('main.index'))
|
|
|
|
try:
|
|
# Récupération de toutes les données de la table actions
|
|
query = text("SELECT id, isin, ticker, company_name, exchange, pays, currency, updated_at FROM actions")
|
|
result = db.session.execute(query)
|
|
|
|
# Création du fichier CSV en mémoire
|
|
output = io.StringIO()
|
|
writer = csv.writer(output, delimiter=';', quoting=csv.QUOTE_MINIMAL)
|
|
|
|
# Écriture de l'en-tête du CSV
|
|
writer.writerow(['id', 'isin', 'ticker', 'company_name', 'exchange', 'pays', 'currency', 'updated_at'])
|
|
|
|
# Écriture des lignes
|
|
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)
|
|
|
|
# Envoi du fichier en téléchargement
|
|
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'))
|
|
|
|
@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:
|
|
# Vérifie qu'on a au moins les 6 colonnes de base
|
|
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 # Ignore les lignes mal formatées
|
|
|
|
# Insertion ou mise à jour si le code ISIN existe déjà
|
|
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')) |