381 lines
14 KiB
Python
381 lines
14 KiB
Python
import csv
|
|
import io
|
|
from datetime import datetime
|
|
|
|
from flask import (
|
|
Blueprint,
|
|
Response,
|
|
current_app,
|
|
flash,
|
|
redirect,
|
|
render_template,
|
|
request,
|
|
url_for,
|
|
)
|
|
from sqlalchemy import text
|
|
|
|
from app import db
|
|
from app.decorators import login_required
|
|
|
|
historique_bp = Blueprint("historique", __name__)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Import / Export des actions CSV
|
|
# ---------------------------------------------------------------------------
|
|
@historique_bp.route("/gestion-import-export-actions-csv", methods=["GET"])
|
|
@login_required
|
|
def gestion_import_export_actions_csv():
|
|
return render_template("gestion_import_export_actions_csv.html")
|
|
|
|
|
|
@historique_bp.route("/export-actions-csv", methods=["GET"])
|
|
@login_required
|
|
def export_actions_csv():
|
|
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:
|
|
current_app.logger.error("Erreur export CSV : %s", e, exc_info=True)
|
|
flash("Erreur lors de l'exportation du fichier CSV.", "danger")
|
|
return redirect(url_for("historique.gestion_import_export_actions_csv"))
|
|
|
|
|
|
@historique_bp.route("/import-actions-csv", methods=["POST"])
|
|
@login_required
|
|
def import_actions_csv():
|
|
if "file" not in request.files:
|
|
flash("Aucun fichier sélectionné.", "danger")
|
|
return redirect(url_for("historique.gestion_import_export_actions_csv"))
|
|
|
|
file = request.files["file"]
|
|
|
|
if file.filename == "":
|
|
flash("Aucun fichier sélectionné.", "danger")
|
|
return redirect(url_for("historique.gestion_import_export_actions_csv"))
|
|
|
|
if not file.filename.endswith(".csv"):
|
|
flash("Veuillez fournir un fichier au format .csv valide.", "warning")
|
|
return redirect(url_for("historique.gestion_import_export_actions_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("historique.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()
|
|
current_app.logger.error("Erreur import CSV : %s", e, exc_info=True)
|
|
flash(f"Erreur lors de l'importation du fichier : {str(e)}", "danger")
|
|
|
|
return redirect(url_for("historique.gestion_import_export_actions_csv"))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Import historique CSV (cotation d'une action)
|
|
# ---------------------------------------------------------------------------
|
|
@historique_bp.route("/gestion-import-historique-csv", methods=["GET"])
|
|
@login_required
|
|
def gestion_import_historique_csv():
|
|
return render_template("gestion_import_historique_csv.html")
|
|
|
|
|
|
def _convertir_vol(valeur):
|
|
"""Convertit une valeur de volume du CSV en entier.
|
|
Suffixes gérés : 'K' (x1000), 'M' (x1 000 000). Sinon entier brut.
|
|
"""
|
|
if valeur is None:
|
|
return None
|
|
s = str(valeur).strip()
|
|
if s == "" or s.upper() == "N/A" or s.upper() == "NA":
|
|
return None
|
|
try:
|
|
if s.upper().endswith("K"):
|
|
return int(float(s[:-1]) * 1000)
|
|
if s.upper().endswith("M"):
|
|
return int(float(s[:-1]) * 1000000)
|
|
return int(float(s))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _to_decimal(valeur):
|
|
"""Convertit une valeur CSV en float (Decimal), None si vide/invalide.
|
|
Retire le suffixe '%' (ex: '0.7%' -> 0.7) et les séparateurs de milliers."""
|
|
if valeur is None:
|
|
return None
|
|
s = str(valeur).strip()
|
|
if s == "" or s.upper() in ("N/A", "NA"):
|
|
return None
|
|
# Suppression d'un éventuel suffixe '%' (et espaces autour)
|
|
if s.endswith("%"):
|
|
s = s[:-1].strip()
|
|
# Suppression d'un éventuel séparateur de milliers (virgule ou espace)
|
|
s = s.replace(" ", "").replace(",", "")
|
|
try:
|
|
return float(s)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
@historique_bp.route("/import-historique-csv", methods=["POST"])
|
|
@login_required
|
|
def import_historique_csv():
|
|
isin = (request.form.get("isin") or "").strip().upper()
|
|
|
|
# Vérification que l'ISIN existe bien dans le référentiel actions
|
|
if not isin:
|
|
flash("Veuillez saisir un code ISIN.", "danger")
|
|
return redirect(url_for("historique.gestion_import_historique_csv"))
|
|
|
|
action = db.session.execute(
|
|
text("SELECT company_name FROM actions WHERE isin = :isin"), {"isin": isin}
|
|
).fetchone()
|
|
if not action:
|
|
flash(f"L'ISIN {isin} n'existe pas dans le référentiel actions.", "danger")
|
|
return redirect(url_for("historique.gestion_import_historique_csv"))
|
|
|
|
if "file" not in request.files:
|
|
flash("Aucun fichier sélectionné.", "danger")
|
|
return redirect(url_for("historique.gestion_import_historique_csv"))
|
|
|
|
file = request.files["file"]
|
|
if file.filename == "":
|
|
flash("Aucun fichier sélectionné.", "danger")
|
|
return redirect(url_for("historique.gestion_import_historique_csv"))
|
|
|
|
if not file.filename.lower().endswith(".csv"):
|
|
flash("Veuillez fournir un fichier au format .csv valide.", "warning")
|
|
return redirect(url_for("historique.gestion_import_historique_csv"))
|
|
|
|
try:
|
|
# Lecture du fichier brut avec fallback d'encodage (utf-8 puis latin-1)
|
|
raw_bytes = file.stream.read()
|
|
# Suppression d'un éventuel BOM UTF-8 (\xef\xbb\xbf) en début de fichier
|
|
if raw_bytes.startswith(b"\xef\xbb\xbf"):
|
|
raw_bytes = raw_bytes[3:]
|
|
try:
|
|
content = raw_bytes.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
content = raw_bytes.decode("latin-1")
|
|
stream = io.StringIO(content)
|
|
|
|
# Auto-détection du séparateur (, ; ou tabulation) sur la 1re ligne
|
|
premiere_ligne = stream.readline()
|
|
stream.seek(0)
|
|
for delim in (",", ";", "\t"):
|
|
nb = len(premiere_ligne.split(delim))
|
|
if nb > 1:
|
|
separateur = delim
|
|
break
|
|
else:
|
|
separateur = ","
|
|
|
|
csv_reader = csv.reader(stream, delimiter=separateur)
|
|
|
|
header = next(csv_reader, None)
|
|
if not header:
|
|
flash("Le fichier CSV est vide.", "danger")
|
|
return redirect(url_for("historique.gestion_import_historique_csv"))
|
|
|
|
# Normalisation des en-têtes : strip + suppression des quotes + lower
|
|
def _norm_header(h):
|
|
h = h.strip()
|
|
if len(h) >= 2 and h[0] in ('"', "'") and h[-1] == h[0]:
|
|
h = h[1:-1]
|
|
return h.strip().lower()
|
|
|
|
header_norm = [_norm_header(h) for h in header]
|
|
|
|
# Mapping alias -> nom canonique (couvre les variantes EODHD, Yahoo, etc.)
|
|
alias = {
|
|
"date": "date",
|
|
"price": "price", "close": "price", "last": "price",
|
|
"adj_close": "price", "adjusted_close": "price", "adj. close": "price",
|
|
"open": "open",
|
|
"hight": "hight", "high": "hight",
|
|
"low": "low",
|
|
"vol": "vol", "vol.": "vol", "volume": "vol",
|
|
"change": "change", "change %": "change", "change_pct": "change",
|
|
"pct_change": "change", "change_percent": "change", "chg": "change",
|
|
}
|
|
# Construction du dictionnaire nom_canonique -> index
|
|
idx = {}
|
|
for i, h in enumerate(header_norm):
|
|
canon = alias.get(h)
|
|
if canon and canon not in idx:
|
|
idx[canon] = i
|
|
|
|
def col(row, name):
|
|
"""Récupère la valeur d'une colonne par son nom canonique."""
|
|
if name not in idx:
|
|
return None
|
|
i = idx[name]
|
|
return row[i] if i < len(row) else None
|
|
|
|
# Vérification : la colonne 'date' est obligatoire
|
|
if "date" not in idx:
|
|
flash(
|
|
f"Pas OK - Colonne 'date' introuvable dans l'en-tête CSV. "
|
|
f"En-tête détecté : {','.join(header)}",
|
|
"danger",
|
|
)
|
|
return redirect(url_for("historique.gestion_import_historique_csv"))
|
|
|
|
# Requête d'insertion (INSERT IGNORE grâce à l'unique key isin+date)
|
|
insert = text(
|
|
"INSERT IGNORE INTO historique "
|
|
"(isin, date, price, `open`, hight, low, vol, `change`) "
|
|
"VALUES (:isin, :date, :price, :open, :hight, :low, :vol, :change)"
|
|
)
|
|
|
|
count_insere = 0
|
|
count_ignore = 0
|
|
count_erreurs = 0
|
|
exemples_erreurs = []
|
|
|
|
for row in csv_reader:
|
|
date_raw = col(row, "date")
|
|
if not date_raw:
|
|
count_erreurs += 1
|
|
if len(exemples_erreurs) < 5:
|
|
exemples_erreurs.append(f"Date vide | ligne: {row}")
|
|
continue
|
|
|
|
# Normalisation de la date -> objet date (pour insertion YYYY-MM-DD)
|
|
# L'ordre compte : %Y/%d/%m (AAAA/JJ/MM) AVANT %Y/%m/%d pour éviter
|
|
# l'ambiguïté (un jour >12 invaliderait %Y/%m/%d et tomberait juste sur %Y/%d/%m).
|
|
date_str = date_raw.strip()[:10]
|
|
date_val = None
|
|
for fmt in ("%Y-%m-%d", "%Y/%d/%m", "%Y/%m/%d", "%m/%d/%Y", "%d/%m/%Y", "%d-%m-%Y"):
|
|
try:
|
|
date_val = datetime.strptime(date_str, fmt).date()
|
|
break
|
|
except ValueError:
|
|
continue
|
|
if date_val is None:
|
|
count_erreurs += 1
|
|
if len(exemples_erreurs) < 10:
|
|
exemples_erreurs.append(
|
|
f"Date invalide '{date_raw}' (nb colonnes ligne={len(row)}) | ligne: {row}"
|
|
)
|
|
continue
|
|
|
|
price = _to_decimal(col(row, "price"))
|
|
open_ = _to_decimal(col(row, "open"))
|
|
hight = _to_decimal(col(row, "hight"))
|
|
low = _to_decimal(col(row, "low"))
|
|
vol = _convertir_vol(col(row, "vol"))
|
|
change = _to_decimal(col(row, "change"))
|
|
|
|
result = db.session.execute(
|
|
insert,
|
|
{
|
|
"isin": isin,
|
|
"date": date_val,
|
|
"price": price,
|
|
"open": open_,
|
|
"hight": hight,
|
|
"low": low,
|
|
"vol": vol,
|
|
"change": change,
|
|
},
|
|
)
|
|
# rowcount == 1 si réellement inséré, 0 si ignoré (doublon unique key)
|
|
if result.rowcount == 1:
|
|
count_insere += 1
|
|
else:
|
|
count_ignore += 1
|
|
|
|
db.session.commit()
|
|
detail = ""
|
|
if exemples_erreurs:
|
|
detail = " | Exemples d'erreurs: " + " ; ".join(exemples_erreurs)
|
|
flash(
|
|
f"OK - Importation réussie pour {action.company_name} ({isin}) : "
|
|
f"{count_insere} ligne(s) insérée(s), "
|
|
f"{count_ignore} ligne(s) déjà existante(s) ignorée(s), "
|
|
f"{count_erreurs} ligne(s) en erreur. "
|
|
f"[Séparateur: '{separateur}' | En-tête: {','.join(header)}]"
|
|
f"{detail}",
|
|
"success",
|
|
)
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
current_app.logger.error("Erreur import historique CSV : %s", e, exc_info=True)
|
|
flash(f"Pas OK - Erreur lors de l'importation du fichier : {str(e)}", "danger")
|
|
|
|
return redirect(url_for("historique.gestion_import_historique_csv")) |