131 lines
4.8 KiB
Python
131 lines
4.8 KiB
Python
"""
|
|
Synchronisation Indy : récupère les transactions via l'API privée d'Indy et les
|
|
normalise au format d'import.
|
|
|
|
La « catégorie » Indy = un numéro de compte du plan comptable (PCG). On mappe ce
|
|
numéro vers nos catégories via le catalogue (categories_indy). Montants en
|
|
centimes, signés. `tva_intracom` => UE (autoliquidation).
|
|
|
|
Ventilation : une transaction peut être éclatée en plusieurs subdivisions de
|
|
charge (ex. un repas à 10 % + 20 %). Chaque subdivision de charge porte son HT
|
|
(`amount_in_cents`), son taux (`tva_rate`) et sa TVA (`tva_amount_in_cents`).
|
|
Indy ajoute aussi des subdivisions « miroir » de TVA (compte 445*, `is_tva:true`)
|
|
et la contrepartie bancaire (compte 512*) : on les ignore. On émet donc UNE ligne
|
|
par subdivision de charge, avec un identifiant de groupe commun (l'id de la
|
|
transaction) pour reconstituer l'opération d'origine.
|
|
|
|
API non officielle d'Indy : le token JWT expire (~1 h) et doit être renouvelé.
|
|
"""
|
|
from decimal import Decimal
|
|
|
|
import httpx
|
|
|
|
from .categories_indy import COMPTE_VERS_NOM, COMPTE_ACTIVITE
|
|
|
|
BASE_URL = "https://app.indy.fr"
|
|
TRANSACTIONS_PATH = "/api/transactions/transactions-list"
|
|
CENT = Decimal("0.01")
|
|
|
|
|
|
def _cat(num):
|
|
return COMPTE_VERS_NOM.get(str(num or ""))
|
|
|
|
|
|
def _act(num):
|
|
return COMPTE_ACTIVITE.get(str(num or ""), "")
|
|
|
|
|
|
def _charge_subdivisions(tx):
|
|
"""Subdivisions de charge/produit réelles : hors banque (512*) et hors
|
|
miroir de TVA (445* avec is_tva)."""
|
|
out = []
|
|
for s in tx.get("subdivisions", []):
|
|
num = str((s.get("accounting_account") or {}).get("number", ""))
|
|
if num.startswith("512"):
|
|
continue
|
|
if s.get("is_tva"):
|
|
continue
|
|
out.append((num, s))
|
|
return out
|
|
|
|
|
|
def lignes_transaction(tx):
|
|
"""Transaction Indy -> liste de lignes normalisées (une par subdivision de
|
|
charge). Liste vide si inexploitable."""
|
|
txid = tx.get("_id", "")
|
|
date = tx.get("date", "")
|
|
libelle = tx.get("description") or tx.get("raw_description") or "(sans libellé)"
|
|
subs = _charge_subdivisions(tx)
|
|
if not subs:
|
|
return []
|
|
# Mode de saisie de la TVA : en "tva_ttc" le montant inclut déjà la TVA.
|
|
mode = tx.get("tva_selected") or (tx.get("vat") or {}).get("tvaSelected") or "tva_ht"
|
|
inclut_tva = mode == "tva_ttc"
|
|
n = len(subs)
|
|
lignes = []
|
|
for i, (num, s) in enumerate(subs):
|
|
amt = Decimal(s.get("amount_in_cents", 0))
|
|
tva = Decimal(s.get("tva_amount_in_cents", 0))
|
|
ttc = (amt if inclut_tva else amt + tva) / 100
|
|
ttc = ttc.quantize(CENT)
|
|
intra = s.get("tva_intracom")
|
|
if intra and intra.get("type") == "eu":
|
|
regime, taux = "intracom", Decimal(intra.get("rate", 0)) / 100
|
|
else:
|
|
regime, taux = "nationale", Decimal(s.get("tva_rate", 0)) / 100
|
|
lignes.append({
|
|
"indy_id": f"{txid}#{i}" if n > 1 else txid,
|
|
"groupe": txid,
|
|
"n_parts": n,
|
|
"date": date,
|
|
"libelle": libelle,
|
|
"categorie_nom": _cat(num),
|
|
"activite": _act(num),
|
|
"taux": taux,
|
|
"regime": regime,
|
|
"montant": ttc,
|
|
"compte": num,
|
|
})
|
|
return lignes
|
|
|
|
|
|
def normaliser(tx):
|
|
"""Compat : première ligne de charge de la transaction (ou None)."""
|
|
lignes = lignes_transaction(tx)
|
|
return lignes[0] if lignes else None
|
|
|
|
|
|
def fetch_transactions(token, date_debut=None, date_fin=None, search="", page=1,
|
|
base_url=BASE_URL):
|
|
headers = {"Accept": "application/json", "Authorization": f"Bearer {token}",
|
|
"x-client-app-type": "web"}
|
|
params = {"search": search or "", "page": page}
|
|
if date_debut:
|
|
params["dateFrom"] = date_debut
|
|
if date_fin:
|
|
params["dateTo"] = date_fin
|
|
try:
|
|
with httpx.Client(base_url=base_url.rstrip("/"), headers=headers,
|
|
timeout=httpx.Timeout(12.0, connect=8.0), follow_redirects=True) as c:
|
|
r = c.get(TRANSACTIONS_PATH, params=params)
|
|
if r.status_code != 200:
|
|
return r.status_code, [], f"HTTP {r.status_code}"
|
|
if "json" not in r.headers.get("content-type", "").lower():
|
|
return r.status_code, [], "Reponse non-JSON (Cloudflare)."
|
|
return 200, r.json().get("transactions", []), None
|
|
except httpx.TimeoutException:
|
|
return None, [], "Delai depasse."
|
|
except Exception as ex:
|
|
return None, [], f"{type(ex).__name__}: {ex}"
|
|
|
|
|
|
def lignes_normalisees(token, date_debut=None, date_fin=None, search="", page=1,
|
|
base_url=BASE_URL):
|
|
status, txs, err = fetch_transactions(token, date_debut, date_fin, search, page, base_url)
|
|
if err:
|
|
return [], err
|
|
lignes = []
|
|
for t in txs:
|
|
lignes.extend(lignes_transaction(t))
|
|
return lignes, None
|