""" 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} 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)