admin
888bcfa213
Deploy via Portainer / deploy (push) Successful in 0s
new file: .gitea/workflows/deploy.yml modified: compta/templates/compta/base.html modified: compta/templates/compta/gestion.html modified: compta/templates/compta/import_csv.html modified: compta/templates/compta/saisie_paiement.html modified: compta/views_gestion.py modified: compta/views_import.py modified: compta/views_saisie.py
382 lines
16 KiB
Python
382 lines
16 KiB
Python
"""
|
||
Import d'opérations : CSV, OCR (capture), ou SYNC INDY (API).
|
||
Dans tous les cas : aperçu ÉDITABLE + anti-doublon (indy_id) + panneau debug OCR.
|
||
"""
|
||
import csv
|
||
import io
|
||
import re
|
||
import unicodedata
|
||
from decimal import Decimal, InvalidOperation
|
||
from datetime import datetime, date as date_cls
|
||
|
||
from django.contrib.auth.decorators import login_required
|
||
from django.shortcuts import render
|
||
|
||
from .models import (Entreprise, Client, Facture, LigneFacture, Paiement,
|
||
Depense, Categorie, Activite, RegimeTva)
|
||
from . import indy
|
||
|
||
CENT = Decimal("0.01")
|
||
MOIS = {"janvier": 1, "février": 2, "fevrier": 2, "mars": 3, "avril": 4, "mai": 5,
|
||
"juin": 6, "juillet": 7, "août": 8, "aout": 8, "septembre": 9,
|
||
"octobre": 10, "novembre": 11, "décembre": 12, "decembre": 12}
|
||
AMOUNT_RE = re.compile(r"^-?\d[\d .]*[.,]\d{2}$")
|
||
DATE_RE = re.compile(r"^\d{2}/\d{2}/\d{4}$")
|
||
|
||
|
||
def _dec(v, d="0"):
|
||
try:
|
||
s = str(v or d)
|
||
for ch in (" ", " ", " "):
|
||
s = s.replace(ch, "")
|
||
return Decimal(s.replace(",", ".").strip() or d)
|
||
except (InvalidOperation, AttributeError):
|
||
return Decimal(d)
|
||
|
||
|
||
def _date(s):
|
||
s = (s or "").strip()
|
||
for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%d-%m-%Y"):
|
||
try:
|
||
return datetime.strptime(s, fmt).date()
|
||
except ValueError:
|
||
continue
|
||
return None
|
||
|
||
|
||
def _activite(s):
|
||
s = (s or "").strip().lower()
|
||
return {"vente": "vente", "service_bic": "service_bic", "service_bnc": "service_bnc",
|
||
"bic": "service_bic", "bnc": "service_bnc"}.get(s, "")
|
||
|
||
|
||
def _regime(s):
|
||
s = (s or "").strip().lower()
|
||
if s in ("intracom", "intracommunautaire", "ue"):
|
||
return RegimeTva.INTRACOM
|
||
if s in ("import", "hors ue"):
|
||
return RegimeTva.IMPORT
|
||
return RegimeTva.NATIONALE
|
||
|
||
|
||
def _noacc(s):
|
||
return "".join(c for c in unicodedata.normalize("NFD", (s or "").lower())
|
||
if unicodedata.category(c) != "Mn")
|
||
|
||
|
||
def _match_cat(text, cats):
|
||
low = _noacc(text)
|
||
items = [(_noacc(nom), c) for nom, c in cats.items()]
|
||
for nom_low, c in items:
|
||
if nom_low and nom_low in low:
|
||
return c
|
||
for nom_low, c in items:
|
||
mots = [m for m in re.split(r"[ ,/.]+", nom_low) if len(m) > 4]
|
||
if mots and mots[0] in low:
|
||
return c
|
||
return None
|
||
|
||
|
||
def _row(dt, libelle, montant, taux=Decimal("0"), regime=RegimeTva.NATIONALE,
|
||
categorie=None, activite="", indy_id="", n=0):
|
||
neutre = bool(categorie and categorie.exclure_calculs)
|
||
if montant > 0:
|
||
typ = "Mouvement neutre" if neutre else "Vente"
|
||
else:
|
||
typ = "Dépense neutre" if neutre else "Dépense"
|
||
return {"n": n, "date": dt.isoformat() if dt else "", "libelle": libelle,
|
||
"categorie_id": categorie.pk if categorie else "",
|
||
"taux": taux, "regime": regime, "montant": montant, "type": typ,
|
||
"activite": activite, "indy_id": indy_id}
|
||
|
||
|
||
# ---- CSV ----------------------------------------------------------------
|
||
|
||
def _parse_csv(text, cats):
|
||
text = text.lstrip("")
|
||
delim = ";" if text.count(";") >= text.count(",") else ","
|
||
rows = list(csv.reader(io.StringIO(text), delimiter=delim))
|
||
lignes, erreurs = [], []
|
||
start = 1 if rows and rows[0] and rows[0][0].strip().lower() == "date" else 0
|
||
for n, raw in enumerate(rows[start:], start=start + 1):
|
||
if not any(c.strip() for c in raw):
|
||
continue
|
||
c = (raw + [""] * 7)[:7]
|
||
dt = _date(c[0])
|
||
montant = _dec(c[6])
|
||
if dt is None:
|
||
erreurs.append(f"Ligne {n} : date invalide « {c[0]} »."); continue
|
||
if montant == 0:
|
||
erreurs.append(f"Ligne {n} : montant nul."); continue
|
||
cat = cats.get((c[2] or "").strip().lower())
|
||
lignes.append(_row(dt, (c[1] or "").strip() or "(sans libellé)", montant,
|
||
_dec(c[4]), _regime(c[5]), cat, _activite(c[3]), "", n))
|
||
return lignes, erreurs, []
|
||
|
||
|
||
# ---- OCR ----------------------------------------------------------------
|
||
|
||
def _ocr_words(image_file):
|
||
try:
|
||
import pytesseract
|
||
from pytesseract import Output
|
||
from PIL import Image
|
||
except Exception:
|
||
return None, ["OCR indisponible (pytesseract / Pillow non installés)."]
|
||
try:
|
||
img = Image.open(image_file)
|
||
except Exception:
|
||
return None, ["Image illisible."]
|
||
if img.width < 1000:
|
||
r = 1400 / img.width
|
||
img = img.resize((int(img.width * r), int(img.height * r)))
|
||
try:
|
||
lang = "fra" if "fra" in pytesseract.get_languages(config="") else None
|
||
kw = {"output_type": Output.DICT}
|
||
if lang:
|
||
kw["lang"] = lang
|
||
data = pytesseract.image_to_data(img, **kw)
|
||
except Exception as ex:
|
||
return None, [f"OCR impossible ({ex}). Vérifie Tesseract, ou utilise le CSV."]
|
||
words = []
|
||
for i in range(len(data["text"])):
|
||
t = data["text"][i].strip()
|
||
try:
|
||
conf = int(float(data["conf"][i]))
|
||
except (ValueError, TypeError):
|
||
conf = -1
|
||
if t and conf > 20:
|
||
words.append({"x": data["left"][i],
|
||
"yc": data["top"][i] + data["height"][i] / 2, "t": t})
|
||
return words, []
|
||
|
||
|
||
def _debug_lines(words):
|
||
out, cur, top = [], [], None
|
||
for w in sorted(words, key=lambda w: (w["yc"], w["x"])):
|
||
if top is None or abs(w["yc"] - top) <= 14:
|
||
cur.append(w); top = w["yc"] if top is None else top
|
||
else:
|
||
out.append(" ".join(x["t"] for x in sorted(cur, key=lambda x: x["x"])))
|
||
cur = [w]; top = w["yc"]
|
||
if cur:
|
||
out.append(" ".join(x["t"] for x in sorted(cur, key=lambda x: x["x"])))
|
||
return out
|
||
|
||
|
||
def _parse_liste(words, annee, cats):
|
||
anchors = sorted([w for w in words if "%" not in w["t"] and AMOUNT_RE.match(w["t"])],
|
||
key=lambda w: w["yc"])
|
||
lignes, last = [], None
|
||
for n, a in enumerate(anchors, start=1):
|
||
band = sorted([w for w in words if abs(w["yc"] - a["yc"]) <= 22], key=lambda w: w["x"])
|
||
day = mois = None
|
||
taux = Decimal("0")
|
||
regime = RegimeTva.NATIONALE
|
||
reste = []
|
||
for w in band:
|
||
t = w["t"]
|
||
if w["x"] < 150 and t.isdigit() and 1 <= int(t) <= 31:
|
||
day = int(t); continue
|
||
if w["x"] < 150 and t.lower().strip(".,") in MOIS:
|
||
mois = MOIS[t.lower().strip(".,")]; continue
|
||
if "%" in t:
|
||
taux = _dec(t.replace("%", "")); continue
|
||
if t.upper() == "UE":
|
||
regime = RegimeTva.INTRACOM; continue
|
||
if w is a or (t == a["t"] and w["x"] == a["x"]):
|
||
continue
|
||
if t.upper() == "TVA":
|
||
continue
|
||
reste.append(t)
|
||
if day and mois:
|
||
dt = date_cls(annee, mois, day); last = dt
|
||
else:
|
||
dt = last
|
||
texte = " ".join(reste).strip()
|
||
cat = _match_cat(texte, cats)
|
||
libelle = texte
|
||
if cat:
|
||
libelle = re.sub(re.escape(cat.nom), "", texte, flags=re.I).strip(" -·,")
|
||
lignes.append(_row(dt, libelle or texte or "(sans libellé)", _dec(a["t"]),
|
||
taux, regime, cat, "", "", n))
|
||
return lignes
|
||
|
||
|
||
def _parse_detail(words, cats):
|
||
dtok = next((w for w in words if DATE_RE.match(w["t"])), None)
|
||
dt = _date(dtok["t"]) if dtok else None
|
||
amounts = [w for w in words if "%" not in w["t"] and AMOUNT_RE.match(w["t"])]
|
||
montant = _dec(amounts[0]["t"]) if amounts else Decimal("0")
|
||
y_ref = dtok["yc"] if dtok else 99999
|
||
haut = [w for w in words if w["x"] < 450 and w["yc"] < y_ref
|
||
and w["t"].lower() not in ("informations", "date", "libellé", "libelle")]
|
||
libelle = " ".join(w["t"] for w in sorted(haut, key=lambda w: (w["yc"], w["x"])))
|
||
libelle = libelle.strip(" ‘'\"") or "(à compléter)"
|
||
droite = [w for w in words if w["x"] > 480 and not AMOUNT_RE.match(w["t"]) and "%" not in w["t"]
|
||
and w["t"].lower() not in ("ventilation(s)", "ventilation", "catégorie",
|
||
"categorie", "montant", "ttc", "tva")]
|
||
cat = _match_cat(" ".join(w["t"] for w in sorted(droite, key=lambda w: (w["yc"], w["x"]))), cats)
|
||
if montant == 0:
|
||
return []
|
||
return [_row(dt, libelle, montant, Decimal("0"), RegimeTva.NATIONALE, cat, "", "", 1)]
|
||
|
||
|
||
def _parse_ocr(image_file, annee, cats):
|
||
words, erreurs = _ocr_words(image_file)
|
||
if words is None:
|
||
return [], erreurs, []
|
||
debug = _debug_lines(words)
|
||
allt = " ".join(w["t"] for w in words).lower()
|
||
is_detail = ("ventilation" in allt) or ("informations" in allt and "montant" in allt)
|
||
lignes = _parse_detail(words, cats) if is_detail else _parse_liste(words, annee, cats)
|
||
if not lignes:
|
||
erreurs.append("Aucune ligne exploitable détectée. Regarde le texte OCR ci-dessous "
|
||
"et complète à la main, ou utilise le CSV.")
|
||
return lignes, erreurs, debug
|
||
|
||
|
||
# ---- INDY ---------------------------------------------------------------
|
||
|
||
def _parse_indy(entreprise, token, date_debut, date_fin, cats_par_nom):
|
||
lignes_norm, err = indy.lignes_normalisees(token, date_debut or None, date_fin or None)
|
||
if err:
|
||
return [], [f"Indy : {err}"]
|
||
lignes = []
|
||
for n, ln in enumerate(lignes_norm, start=1):
|
||
dt = _date(ln["date"])
|
||
cat = cats_par_nom.get((ln["categorie_nom"] or "").strip().lower())
|
||
lignes.append(_row(dt, ln["libelle"], ln["montant"], ln["taux"],
|
||
_regime(ln["regime"]), cat, ln["activite"] or "",
|
||
ln["indy_id"], n))
|
||
return lignes, []
|
||
|
||
|
||
# ---- Anti-doublon -------------------------------------------------------
|
||
|
||
def _deja_importe(entreprise, indy_id):
|
||
if not indy_id:
|
||
return False
|
||
return (Facture.objects.filter(entreprise=entreprise, indy_id=indy_id).exists()
|
||
or Depense.objects.filter(entreprise=entreprise, indy_id=indy_id).exists()
|
||
or Paiement.objects.filter(entreprise=entreprise, indy_id=indy_id).exists())
|
||
|
||
|
||
# ---- Création -----------------------------------------------------------
|
||
|
||
def _creer(entreprise, dt, libelle, montant, taux, regime, categorie,
|
||
activite="", indy_id=""):
|
||
neutre = bool(categorie and categorie.exclure_calculs)
|
||
if montant > 0:
|
||
if neutre:
|
||
Paiement.objects.create(entreprise=entreprise, date=dt, libelle=libelle,
|
||
montant_total=montant, sens="credit",
|
||
categorie=categorie, indy_id=indy_id)
|
||
return "neutre"
|
||
client, _ = Client.objects.get_or_create(entreprise=entreprise, nom=libelle)
|
||
ht = (montant / (Decimal("1") + taux / Decimal("100"))).quantize(CENT)
|
||
annee = dt.year
|
||
base = Facture.objects.filter(entreprise=entreprise,
|
||
numero__startswith=f"{annee}-").count() + 1
|
||
numero = f"{annee}-{base:03d}"
|
||
while Facture.objects.filter(entreprise=entreprise, numero=numero).exists():
|
||
base += 1; numero = f"{annee}-{base:03d}"
|
||
f = Facture.objects.create(entreprise=entreprise, client=client, numero=numero,
|
||
activite=activite or entreprise.activite,
|
||
date_emission=dt, date_encaissement=dt,
|
||
statut="encaissee", indy_id=indy_id)
|
||
LigneFacture.objects.create(facture=f, designation=libelle, quantite=Decimal("1"),
|
||
prix_unitaire_ht=ht, taux_tva=taux)
|
||
f.recompute_totaux()
|
||
return "vente"
|
||
Depense.objects.create(entreprise=entreprise, date=dt, libelle=libelle,
|
||
categorie=categorie, activite=activite, regime_tva=regime,
|
||
taux_tva=taux, montant_ttc=-montant, indy_id=indy_id)
|
||
return "depense"
|
||
|
||
|
||
@login_required
|
||
def import_csv(request):
|
||
entreprises = Entreprise.accessibles(request.user)
|
||
cats = {c.nom.strip().lower(): c for c in Categorie.objects.all()}
|
||
ctx = {"entreprises": entreprises,
|
||
"categories": Categorie.objects.filter(actif=True),
|
||
"activites": Activite.choices, "regimes": RegimeTva.choices,
|
||
"annee_defaut": date_cls.today().year, "vue": "import"}
|
||
|
||
if request.method != "POST":
|
||
return render(request, "compta/import_csv.html", ctx)
|
||
|
||
action = request.POST.get("action")
|
||
ctx["entreprise_id"] = int(request.POST.get("entreprise") or 0)
|
||
entreprise = entreprises.filter(pk=request.POST.get("entreprise")).first()
|
||
|
||
if action == "save":
|
||
if not entreprise:
|
||
ctx["erreurs"] = ["Sélectionne une entreprise."]
|
||
return render(request, "compta/import_csv.html", ctx)
|
||
dates = request.POST.getlist("l_date")
|
||
libelles = request.POST.getlist("l_libelle")
|
||
cat_ids = request.POST.getlist("l_categorie")
|
||
taux = request.POST.getlist("l_taux")
|
||
regimes = request.POST.getlist("l_regime")
|
||
montants = request.POST.getlist("l_montant")
|
||
activites = request.POST.getlist("l_activite")
|
||
indy_ids = request.POST.getlist("l_indy_id")
|
||
cats_by_id = {str(c.pk): c for c in Categorie.objects.all()}
|
||
compteur = {}
|
||
ignores = 0
|
||
for i in range(len(libelles)):
|
||
dt = _date(dates[i]) if i < len(dates) else None
|
||
montant = _dec(montants[i] if i < len(montants) else "0")
|
||
iid = indy_ids[i] if i < len(indy_ids) else ""
|
||
if dt is None or montant == 0:
|
||
continue
|
||
if iid and _deja_importe(entreprise, iid):
|
||
ignores += 1
|
||
continue
|
||
k = _creer(entreprise, dt, (libelles[i] or "").strip() or "(sans libellé)",
|
||
montant, _dec(taux[i] if i < len(taux) else "0"),
|
||
_regime(regimes[i] if i < len(regimes) else ""),
|
||
cats_by_id.get(cat_ids[i] if i < len(cat_ids) else ""),
|
||
_activite(activites[i] if i < len(activites) else ""), iid)
|
||
compteur[k] = compteur.get(k, 0) + 1
|
||
if ignores:
|
||
compteur["déjà importées (ignorées)"] = ignores
|
||
ctx["resultat"] = compteur
|
||
return render(request, "compta/import_csv.html", ctx)
|
||
|
||
if action == "indy":
|
||
token = (request.POST.get("indy_token") or "").strip()
|
||
if entreprise and token:
|
||
entreprise.indy_token = token
|
||
entreprise.save(update_fields=["indy_token"])
|
||
elif entreprise and not token:
|
||
token = entreprise.indy_token
|
||
if not entreprise or not token:
|
||
ctx["erreurs"] = ["Choisis une entreprise et renseigne le token Indy."]
|
||
return render(request, "compta/import_csv.html", ctx)
|
||
lignes, erreurs = _parse_indy(entreprise, token, request.POST.get("date_debut"),
|
||
request.POST.get("date_fin"), cats)
|
||
ctx.update({"lignes": lignes, "erreurs": erreurs, "source": "indy"})
|
||
return render(request, "compta/import_csv.html", ctx)
|
||
|
||
# Aperçu CSV ou OCR
|
||
image = request.FILES.get("image")
|
||
if image:
|
||
annee = int(request.POST.get("annee") or date_cls.today().year)
|
||
lignes, erreurs, debug = _parse_ocr(image, annee, cats)
|
||
ctx["source"] = "ocr"
|
||
ctx["ocr_debug"] = debug
|
||
else:
|
||
texte = request.POST.get("texte", "")
|
||
fichier = request.FILES.get("fichier")
|
||
if fichier:
|
||
texte = fichier.read().decode("utf-8-sig", "ignore")
|
||
lignes, erreurs, debug = _parse_csv(texte, cats)
|
||
ctx["texte"] = texte
|
||
ctx["source"] = "csv"
|
||
ctx["lignes"] = lignes
|
||
ctx["erreurs"] = erreurs
|
||
return render(request, "compta/import_csv.html", ctx)
|