Marius Mutu
ba4cbbf62c
citire_xlsx.py citeste .xlsx (openpyxl) si .xls (xlrd) in acelasi contract intern
ca extract_balanta.py. Prima foaie, nu dupa nume; potrivire exacta a coloanelor,
niciodata pe substring (TOTAL_DEB vs TOTAL_DEB_1); CONT ramane text.
mapare.py: propunere automata, corectii_<FIRMA>.xlsx detinut de utilizator si
mapare_<FIRMA>.xlsx regenerat de unealta, .TOATE/.RESTUL in ordinea exact ->
RESTUL -> TOATE, redenumirea sinteticului. Se opreste cu eroare la cont/acont
peste 4 caractere, coliziune accidentala si cont_saga inexistent in corectii.
Corectat in orchestrare: .xls-ul SAGA nu are inregistrare CODEPAGE, deci xlrd
cadea pe iso-8859-1 si strica denumirile ("FURNIZORI \x97 DEBITORI"). Citirea se
face cu encoding_override="cp1250", cu o verificare in test care pica daca revine.
18 teste trec, inclusiv poarta de regresie pe etalon.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EYeAtVxeS8m4oXekjX8Am2
261 lines
8.9 KiB
Python
261 lines
8.9 KiB
Python
"""Propunere automata de mapare SAGA -> ROA, cu corectii manuale, per firma.
|
|
|
|
La fiecare rulare:
|
|
1. se citeste balanta (rows);
|
|
2. se propune automat `cont`/`acont` (regula din docs/plan_etapa2_convertor.md);
|
|
3. se suprapun corectiile din `corectii_<FIRMA>.xlsx`, daca exista;
|
|
4. se scrie `mapare_<FIRMA>.xlsx` complet.
|
|
|
|
Doua fisiere, proprietari diferiti:
|
|
- `mapare_<FIRMA>.xlsx` - scris DOAR de unealta, regenerat complet;
|
|
- `corectii_<FIRMA>.xlsx` - scris DOAR de utilizator; se creeaza o singura data,
|
|
gol, cu antet si un exemplu comentat. Unealta nu scrie niciodata in el, deci
|
|
regenerarea nu poate distruge corectiile.
|
|
|
|
Propunerea automata: `cont` = sinteticul SAGA, `acont` = restul analiticului
|
|
(`1171.01.12` -> `cont=1171`, `acont=0112`). La conturile cu parteneri `acont`
|
|
NU se pre-completa: forma lui o decide `genereaza_xlsx.py` (pct. 16 din
|
|
docs/decizii_import.md).
|
|
|
|
Conventii in `corectii_`, pe coloana `cont_saga`:
|
|
- `<cod>` exact;
|
|
- `<sintetic>.TOATE` - toate conturile sinteticului merg pe acelasi `cont`/`acont`;
|
|
- `<sintetic>.RESTUL` - la fel, doar pentru cele fara potrivire exacta.
|
|
Ordinea de cautare: potrivire exacta -> `.RESTUL` -> `.TOATE`.
|
|
"""
|
|
|
|
import csv
|
|
import os
|
|
import re
|
|
|
|
from openpyxl import Workbook, load_workbook
|
|
|
|
COLOANE_MAPARE = ["cont_saga", "denumire", "sold", "cont", "acont", "sursa"]
|
|
COLOANE_CORECTII = ["cont_saga", "cont", "acont", "exclus", "motiv"]
|
|
|
|
SURSA_AUTO = "AUTO"
|
|
SURSA_CORECTAT = "CORECTAT"
|
|
|
|
MAX = 4
|
|
|
|
PATTERN_GRUP = re.compile(r"^(.+)\.(TOATE|RESTUL)$")
|
|
|
|
CONFIG_PARTENERI = os.path.join(
|
|
os.path.dirname(os.path.abspath(__file__)), "config", "conturi_parteneri.csv")
|
|
|
|
# Randul-exemplu din corectii: incepe cu '#' pe cont_saga, deci e ignorat la citire.
|
|
EXEMPLU = ["# exemplu (sterge-l): 409.TOATE", "# 4091", "# (gol)", "",
|
|
"# randurile cu '#' in cont_saga sunt comentarii; gol = pastreaza propunerea automata"]
|
|
|
|
|
|
class Stop(Exception):
|
|
"""Conditie neprevazuta: nu ghicim, raportam."""
|
|
|
|
|
|
def citeste_parteneri(path=CONFIG_PARTENERI):
|
|
"""Sinteticele cu parteneri din config/conturi_parteneri.csv (doar citire)."""
|
|
if not os.path.exists(path):
|
|
raise Stop("lipseste fisierul de configurare a partenerilor: %s" % path)
|
|
conturi = set()
|
|
with open(path, encoding="utf-8-sig", newline="") as f:
|
|
for r in csv.DictReader(f):
|
|
cont = (r.get("cont") or "").strip()
|
|
if cont:
|
|
conturi.add(cont)
|
|
return conturi
|
|
|
|
|
|
def separa(cont_saga):
|
|
"""(sintetic, analitic): partea dinaintea punctului si restul, fara punct."""
|
|
seg = cont_saga.split(".")
|
|
return seg[0], "".join(seg[1:])
|
|
|
|
|
|
def propune(cont_saga, partener=False):
|
|
"""Propunerea automata; la parteneri acont ramane gol (decide generatorul)."""
|
|
cont, acont = separa(cont_saga)
|
|
if partener:
|
|
return cont, None
|
|
return cont, (acont or None)
|
|
|
|
|
|
def _text(v):
|
|
"""Celula ca text curatat; gol -> None. Numerele intregi devin text fara .0."""
|
|
if v is None or isinstance(v, bool):
|
|
return None
|
|
if isinstance(v, float) and v.is_integer():
|
|
v = int(v)
|
|
s = str(v).strip()
|
|
return s or None
|
|
|
|
|
|
def _adevar(v):
|
|
if isinstance(v, bool):
|
|
return v
|
|
text = _text(v)
|
|
return text.lower() in ("1", "da", "true", "x", "yes") if text else False
|
|
|
|
|
|
def creeaza_corectii(path):
|
|
"""Creeaza corectii_<FIRMA>.xlsx gol, cu antet si un exemplu comentat."""
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "corectii"
|
|
ws.append(COLOANE_CORECTII)
|
|
ws.append(EXEMPLU)
|
|
wb.save(path)
|
|
|
|
|
|
def citeste_corectii(path):
|
|
"""Corectiile din xlsx, cheie = cont_saga. Randurile '#' sunt comentarii."""
|
|
if not os.path.exists(path):
|
|
return {}
|
|
wb = load_workbook(path, read_only=True, data_only=True)
|
|
try:
|
|
ws = wb.active
|
|
header = [_text(c.value) for c in ws[1]]
|
|
while header and header[-1] is None:
|
|
header.pop()
|
|
if header != COLOANE_CORECTII:
|
|
raise Stop("antet neasteptat in %s: %r"
|
|
% (os.path.basename(path), header))
|
|
out = {}
|
|
for row in ws.iter_rows(min_row=2, values_only=True):
|
|
valori = dict(zip(COLOANE_CORECTII,
|
|
list(row) + [None] * (len(COLOANE_CORECTII) - len(row))))
|
|
cheie = _text(valori["cont_saga"])
|
|
if cheie is None or cheie.startswith("#"):
|
|
continue
|
|
if cheie in out:
|
|
raise Stop("rand duplicat in corectii pentru %s" % cheie)
|
|
out[cheie] = {
|
|
"cont": _text(valori["cont"]),
|
|
"acont": _text(valori["acont"]),
|
|
"exclus": _adevar(valori["exclus"]),
|
|
"motiv": valori["motiv"],
|
|
}
|
|
return out
|
|
finally:
|
|
wb.close()
|
|
|
|
|
|
def _regula(cont_saga, corectii):
|
|
"""(corectie, cheia_regulii) dupa exact -> .RESTUL -> .TOATE; (None, None) daca nu."""
|
|
if cont_saga in corectii:
|
|
return corectii[cont_saga], cont_saga
|
|
sintetic = cont_saga.split(".")[0]
|
|
for sufix in ("RESTUL", "TOATE"):
|
|
cheie = "%s.%s" % (sintetic, sufix)
|
|
if cheie in corectii:
|
|
return corectii[cheie], cheie
|
|
return None, None
|
|
|
|
|
|
def _sold(r):
|
|
if r.get("sold") is not None:
|
|
return round(float(r["sold"]), 2)
|
|
return round(float(r.get("total_d", 0.0)) - float(r.get("total_c", 0.0)), 2)
|
|
|
|
|
|
def _verifica_lungime(cont, acont, cont_saga):
|
|
if cont and len(cont) > MAX:
|
|
raise Stop("cont peste 4 caractere pentru %s: %r" % (cont_saga, cont))
|
|
if acont and len(acont) > MAX:
|
|
raise Stop("acont peste 4 caractere pentru %s: %r" % (cont_saga, acont))
|
|
|
|
|
|
def _verifica_corectii(rows, corectii):
|
|
existente = {r["cont"] for r in rows}
|
|
sintetice = {c.split(".")[0] for c in existente}
|
|
for cheie in corectii:
|
|
m = PATTERN_GRUP.match(cheie)
|
|
if m:
|
|
if m.group(1) not in sintetice:
|
|
raise Stop("corectie pentru un sintetic inexistent in balanta: %s" % cheie)
|
|
elif cheie not in existente:
|
|
raise Stop("corectie pentru un cont inexistent in balanta: %s" % cheie)
|
|
|
|
|
|
def _verifica_coliziuni(rezultat, parteneri):
|
|
"""Doua conturi SAGA pe acelasi (cont, acont) sunt legale doar sub acelasi .TOATE/.RESTUL."""
|
|
grupe = {}
|
|
for r in rezultat:
|
|
if r["cont_saga"].split(".")[0] in parteneri or r["exclus"]:
|
|
continue
|
|
grupe.setdefault((r["cont"], r["acont"]), []).append(r)
|
|
for (cont, acont), randuri in grupe.items():
|
|
if len(randuri) < 2:
|
|
continue
|
|
reguli = {r["_regula"] for r in randuri}
|
|
singura = next(iter(reguli)) if len(reguli) == 1 else None
|
|
if not (singura and PATTERN_GRUP.match(singura)):
|
|
conturi = sorted(r["cont_saga"] for r in randuri)
|
|
raise Stop("coliziune pe (%r, %r) intre %s; foloseste .TOATE/.RESTUL "
|
|
"daca agregarea e intentionata" % (cont, acont, conturi))
|
|
|
|
|
|
def aplica(rows, corectii, parteneri=None):
|
|
"""Rezultatul final per cont: cont_saga, denumire, sold, cont, acont, sursa, exclus."""
|
|
if parteneri is None:
|
|
parteneri = citeste_parteneri()
|
|
_verifica_corectii(rows, corectii)
|
|
|
|
rezultat = []
|
|
for r in rows:
|
|
cont_saga = r["cont"]
|
|
partener = cont_saga.split(".")[0] in parteneri
|
|
cont_auto, acont_auto = propune(cont_saga, partener)
|
|
|
|
cor, cheie = _regula(cont_saga, corectii)
|
|
if cor is None:
|
|
cont, acont = cont_auto, acont_auto
|
|
sursa, exclus = SURSA_AUTO, False
|
|
else:
|
|
if cor["exclus"]:
|
|
cont, acont = cont_auto, acont_auto
|
|
else:
|
|
cont = cor["cont"] or cont_auto
|
|
acont = cor["acont"]
|
|
sursa, exclus = SURSA_CORECTAT, cor["exclus"]
|
|
|
|
_verifica_lungime(cont, acont, cont_saga)
|
|
rezultat.append({
|
|
"cont_saga": cont_saga,
|
|
"denumire": r.get("denumire") or "",
|
|
"sold": _sold(r),
|
|
"cont": cont,
|
|
"acont": acont,
|
|
"sursa": sursa,
|
|
"exclus": exclus,
|
|
"_regula": cheie,
|
|
})
|
|
|
|
_verifica_coliziuni(rezultat, parteneri)
|
|
for r in rezultat:
|
|
r.pop("_regula")
|
|
return rezultat
|
|
|
|
|
|
def scrie_mapare(rezultat, path):
|
|
"""Scrie mapare_<FIRMA>.xlsx complet, doar din rezultatul final."""
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "mapare"
|
|
ws.append(COLOANE_MAPARE)
|
|
for r in rezultat:
|
|
ws.append([r[c] for c in COLOANE_MAPARE])
|
|
wb.save(path)
|
|
|
|
|
|
def genereaza_mapare(rows, firma, dir_iesire="."):
|
|
"""Scrie mapare_<FIRMA>.xlsx, creeaza corectii_<FIRMA>.xlsx doar daca lipseste."""
|
|
if not os.path.isdir(dir_iesire):
|
|
os.makedirs(dir_iesire)
|
|
cale_corectii = os.path.join(dir_iesire, "corectii_%s.xlsx" % firma)
|
|
cale_mapare = os.path.join(dir_iesire, "mapare_%s.xlsx" % firma)
|
|
if not os.path.exists(cale_corectii):
|
|
creeaza_corectii(cale_corectii)
|
|
rezultat = aplica(rows, citeste_corectii(cale_corectii))
|
|
scrie_mapare(rezultat, cale_mapare)
|
|
return rezultat
|