Lane-urile citire-xlsx si mapare: cititorul de foaie SAGA si pasul de mapare

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
This commit is contained in:
2026-09-21 15:59:56 +03:00
parent f23b3367d8
commit ba4cbbf62c
6 changed files with 898 additions and 0 deletions

172
citire_xlsx.py Normal file
View File

@@ -0,0 +1,172 @@
"""Cititor pentru balantele SAGA exportate ca foaie de calcul (.xls / .xlsx).
Produce balanta_<FIRMA>.csv cu contractul intern, aceleasi coloane ca extract_balanta.py:
pagina, cont, denumire, prec_d, prec_c, rulaj_d, rulaj_c, total_d, total_c,
sold_d, sold_c, este_total, tip.
.xlsx cu openpyxl, .xls cu xlrd (openpyxl nu citeste BIFF). Se citeste prima foaie a
registrului, fara cautare dupa nume (numele difera: balanta / xl / Sheet1). Numele de
coloana se potrivesc exact si case-insensitive, niciodata pe substring (TOTAL_DEB si
TOTAL_DEB_1 sunt coloane diferite). CONT se citeste ca text: 401.00002 citit ca numar
se strica. pagina si este_total raman goale (exportul nu are rand de totaluri).
Utilizare:
py citire_xlsx.py "<fisier.xls sau .xlsx>" "<iesire.csv>"
"""
import csv
import os
import sys
import openpyxl
import xlrd
COLOANE = ["pagina", "cont", "denumire", "prec_d", "prec_c", "rulaj_d", "rulaj_c",
"total_d", "total_c", "sold_d", "sold_c", "este_total", "tip"]
MAPARE_SUME = {
"DEB_PREC": "prec_d",
"CRED_PREC": "prec_c",
"RULAJ_D": "rulaj_d",
"RULAJ_C": "rulaj_c",
"TOTAL_DEB": "total_d",
"TOTAL_CRED": "total_c",
"FIN_D": "sold_d",
"FIN_C": "sold_c",
}
COLOANE_SURSA = ["CONT", "DENUMIRE", "TIP"] + list(MAPARE_SUME)
def _index_antet(antet):
"""Indexul fiecarei coloane cerute, pe potrivire exacta case-insensitive."""
normal = [("" if h is None else str(h)).strip().upper() for h in antet]
idx = {}
for i, nume in enumerate(normal):
if nume == "":
continue
if nume in idx:
raise ValueError("coloana %s apare de doua ori in antet" % nume)
idx[nume] = i
lipsa = [c for c in COLOANE_SURSA if c not in idx]
if lipsa:
raise ValueError("coloane lipsa in antetul SAGA: %s" % ", ".join(lipsa))
return idx
def _randuri_xlsx(cale):
wb = openpyxl.load_workbook(cale, data_only=True, read_only=True)
try:
ws = wb[wb.sheetnames[0]]
it = ws.iter_rows(values_only=True)
try:
antet = list(next(it))
except StopIteration:
raise ValueError("fisier fara antet: %s" % cale)
randuri = [list(r) for r in it]
finally:
wb.close()
return antet, randuri
def _randuri_xls(cale):
# Exportul SAGA (VFP) nu are inregistrare CODEPAGE; fara override xlrd cade pe
# iso-8859-1 si strica diacriticele ("FURNIZORI \x97 DEBITORI"). SAGA scrie WIN1250.
book = xlrd.open_workbook(cale, encoding_override="cp1250")
sh = book.sheet_by_index(0)
if sh.nrows == 0:
raise ValueError("fisier fara antet: %s" % cale)
antet = [sh.cell_value(0, c) for c in range(sh.ncols)]
randuri = [[sh.cell_value(r, c) for c in range(sh.ncols)]
for r in range(1, sh.nrows)]
return antet, randuri
def _randuri(cale):
ext = os.path.splitext(cale)[1].lower()
if ext == ".xlsx":
return _randuri_xlsx(cale)
if ext == ".xls":
return _randuri_xls(cale)
raise ValueError("extensie nesuportata pentru %s (astept .xls sau .xlsx)" % cale)
def _camp(raw, i):
return raw[i] if i < len(raw) else None
def _text(v):
if v is None:
return ""
if isinstance(v, str):
return v.strip()
return str(v).strip()
def _cont(v):
"""CONT ca text: un float intreg nu primeste zecimale, unul cu zecimale ramane exact."""
if v is None:
return ""
if isinstance(v, str):
return v.strip()
if isinstance(v, bool):
return str(v)
if isinstance(v, float):
return str(int(v)) if v.is_integer() else repr(v)
return str(v)
def _num(v):
if v is None:
return 0.0
if isinstance(v, bool):
return float(int(v))
if isinstance(v, (int, float)):
return float(v)
if isinstance(v, str):
s = v.strip()
if s in ("", "-"):
return 0.0
return float(s)
raise ValueError("valoare numerica neasteptata: %r" % (v,))
def citeste_balanta(cale):
"""Randurile balantei din prima foaie, in contractul intern."""
antet, brute = _randuri(cale)
idx = _index_antet(antet)
rows = []
for raw in brute:
cont = _cont(_camp(raw, idx["CONT"]))
denumire = _text(_camp(raw, idx["DENUMIRE"]))
if cont == "" and denumire == "" and \
all(_camp(raw, idx[n]) in (None, "") for n in MAPARE_SUME):
continue
rand = {"pagina": "", "cont": cont, "denumire": denumire}
for nume, col in MAPARE_SUME.items():
rand[col] = _num(_camp(raw, idx[nume]))
rand["este_total"] = ""
rand["tip"] = _text(_camp(raw, idx["TIP"]))
rows.append(rand)
return rows
def scrie_csv(rows, cale):
with open(cale, "w", newline="", encoding="utf-8-sig") as f:
wr = csv.DictWriter(f, fieldnames=COLOANE, extrasaction="ignore")
wr.writeheader()
wr.writerows(rows)
def main():
if len(sys.argv) != 3:
print(__doc__)
sys.exit(1)
cale, iesire = sys.argv[1], sys.argv[2]
rows = citeste_balanta(cale)
scrie_csv(rows, iesire)
print("scris %s: %d randuri" % (iesire, len(rows)))
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,103 @@
# Raport lane `citire-xlsx` - cititorul de foaie de calcul SAGA
Data: 21.09.2026. Lane: `citire-xlsx`, in paralel cu `mapare`.
## 1. Facut
### Livrabil nou: `citire_xlsx.py`
Cititor pentru balantele SAGA `.xls` / `.xlsx`, scrie `balanta_<FIRMA>.csv` cu contractul
intern (13 coloane). Puncte:
- `citire_xlsx.py:24` - `COLOANE` = contractul intern, identic cu `extract_balanta.py` plus `tip`.
- `citire_xlsx.py:27` - `MAPARE_SUME`: `DEB_PREC/CRED_PREC -> prec_d/prec_c`,
`RULAJ_D/RULAJ_C -> rulaj_d/rulaj_c`, `TOTAL_DEB/TOTAL_CRED -> total_d/total_c`,
`FIN_D/FIN_C -> sold_d/sold_c`.
- `citire_xlsx.py:41` `_index_antet` - potrivire **exacta, case-insensitive** (`strip().upper()`),
niciodata pe substring; coloana ceruta lipsa sau duplicata -> `ValueError` cu numele coloanei.
- `citire_xlsx.py:57` `_randuri_xlsx` - openpyxl, `data_only=True`, `read_only=True`.
- `citire_xlsx.py:72` `_randuri_xls` - xlrd 2.0.2.
- `citire_xlsx.py:83` `_randuri` - alege dupa extensie; alta extensie -> `ValueError`.
Ambele iau **prima foaie** (`sheetnames[0]` / `sheet_by_index(0)`), fara cautare dupa nume.
- `citire_xlsx.py:104` `_cont` - CONT ramane text; un float intreg (`401.0`) devine `"401"`,
unul cu zecimale (`401.00002`) devine `"401.00002"`, niciodata numar.
- `citire_xlsx.py:117` `_num` - `None`/gol -> `0.0`; alt text neconvertibil -> `ValueError`.
- `citire_xlsx.py:132` `citeste_balanta` - produce randurile; `pagina` si `este_total` raman
goale (exportul nu are rand de totaluri); sare doar peste randurile complet goale.
- `citire_xlsx.py:152` `scrie_csv` - `utf-8-sig`, `newline=""`, exact ca `extract_balanta.py`.
- `citire_xlsx.py:159` `main` - CLI: `py citire_xlsx.py <sursa> <iesire.csv>`.
### Test nou: `tests/test_citire_xlsx.py`
`unittest` din stdlib, fara dependinte noi. Cazuri:
- `:85` `test_xlsx_si_xls_identice` - `.xlsx` (openpyxl) si `.xls` (xlrd) produc acelasi rezultat.
- `:102` `test_relatii_si_frunze` - pe ambele fisiere: `total_d = prec_d + rulaj_d`,
`total_c = prec_c + rulaj_c`, `sold_d - sold_c = total_d - total_c` pe toate randurile,
plus `SUM(sold_d) = SUM(sold_c)` pe frunze. Toleranta 0.01.
- `:110` `test_cont_ramane_text` - toate valorile `cont` sunt `str`.
- `:116` `test_danube_analitice` - `saga2roa_danube\balanta.xls` (341 randuri, 263 analitice);
daca lipseste, `skipTest`. Verifica explicit ca `105.10` ramane text, nu numar.
- `:133` `test_cont_numeric_devine_text` - celula `CONT` scrisa ca numar (`401.00002`) iese text.
- `:148` `test_coloana_lipsa_si_fara_substring` - antet cu `TOTAL_DEB_1` dar fara `TOTAL_DEB`
da `ValueError` care numeste `TOTAL_DEB`: nici coloana lipsa nu trece tacut, nici `_1` nu se
confunda cu coloana principala.
## 2. Verificari
Comanda: `py -m unittest discover -s tests -v` (din radacina proiectului). Iesire exacta:
```
test_coloana_lipsa_si_fara_substring (test_citire_xlsx.TestCitireXlsx.test_coloana_lipsa_si_fara_substring) ... ok
test_cont_numeric_devine_text (test_citire_xlsx.TestCitireXlsx.test_cont_numeric_devine_text) ... ok
test_cont_ramane_text (test_citire_xlsx.TestCitireXlsx.test_cont_ramane_text) ... ok
test_danube_analitice (test_citire_xlsx.TestCitireXlsx.test_danube_analitice) ... ok
test_relatii_si_frunze (test_citire_xlsx.TestCitireXlsx.test_relatii_si_frunze) ... ok
test_xlsx_si_xls_identice (test_citire_xlsx.TestCitireXlsx.test_xlsx_si_xls_identice) ... ok
test_fundatia (test_regresie.TestRegresieEtalon.test_fundatia) ... ok
test_master (test_regresie.TestRegresieEtalon.test_master) ... ok
----------------------------------------------------------------------
Ran 8 tests in 0.542s
OK
```
`test_regresie.py` existent trece neatinse. (Pe stderr apar avertismentele cunoscute ale xlrd
`OLE2 inconsistency` / `No CODEPAGE` si cel al openpyxl despre stil - nu afecteaza rezultatul.)
Verificare CLI capat-la-capat, iesire in director temporar:
```
py citire_xlsx.py "exemple/saga-sqlite-balanta-09-2026.xlsx" "$TEMP/out_test.csv"
-> scris ...out_test.csv: 10 randuri
-> pagina,cont,denumire,prec_d,prec_c,rulaj_d,rulaj_c,total_d,total_c,sold_d,sold_c,este_total,tip
,371,MARFURI,2615.0,0.0,0.0,0.0,2615.0,0.0,2615.0,0.0,,A
```
Cifre pe cele trei fisiere (relatii de sume, frunze): xlsx 10 randuri / 0 esec / 10 frunze;
xls identic; DANUBE 341 randuri / 0 esec / 320 frunze, balanta frunzelor 19018742.48 = 19018742.48.
Fisiere noi: `citire_xlsx.py` (170 linii) si `tests/test_citire_xlsx.py` (137 linii), ambele
**ASCII curat (0 octeti > 0x7F)** si LF (conventia fisierelor `.py` din proiect). Nu sunt `.vc2`
/ `.sc2` / `.prg`, deci regula de cens cp1250 nu se aplica.
`git status`: doar `?? citire_xlsx.py` si `?? tests/test_citire_xlsx.py`. Nimic altceva atins.
## 3. Nefacut / blocat
- Interfata catre `genereaza_xlsx.py` (maparea) nu e atinsa: lane-ul `mapare` face asta.
`genereaza_xlsx.citeste_balanta` foloseste `csv.DictReader` si citeste doar
`cont/denumire/total_d/total_c/este_total`, deci coloana noua `tip` nu il deranjeaza.
- `extract_balanta.py` nu a fost atins (interzis), deci CSV-ul de PDF are inca 12 coloane, fara
`tip`. Contractul din plan cere 13 coloane pentru toate cele 3 cititoare; alinierea PDF-ului
ramane de facut intr-un alt lane, cand se decide forma lui `tip` pe drumul PDF.
- Intersectia cu `citire-fdb` (CSV identic xlsx/FDB pe firma de joaca) ramane pe lane-ul `citire-fdb`.
- Nu am adaugat `xlrd` in `CLAUDE.md` la `Mediu` (il cere planul, dar e in afara scopului acestui
lane; xlrd era deja instalat si documentat in `docs/raport_sursa_saga_xlsx.md`).
## 4. Stare periculoasa
Niciuna. Fara `.vc2`/`.sc2` editat fara write-back, fara procese lasate pornite, fara date de test
consumate (iesirile de test au mers in `%TEMP%`). Fara commit, fara `git add`, fara `sed -i`.
Nu s-a modificat nimic sub `D:\ROA\ROACONT`, `D:\SAGA250909` sau `saga2roa*`.

View File

@@ -0,0 +1,59 @@
# Raport lane `mapare` - solduri2roa etapa 2
## 1. Facut
Doua fisiere noi, nimic altceva atins.
`mapare.py` (nou):
- `mapare.py:33-48` contractul: coloanele `mapare_` / `corectii_`, sursele, regex
`.TOATE`/`.RESTUL`, config parteneri, randul-exemplu comentat.
- `mapare.py:55-65` `citeste_parteneri` - doar citire din `config/conturi_parteneri.csv`.
- `mapare.py:68-79` `separa` / `propune` - propunerea automata; partener -> `acont` gol.
- `mapare.py:99-106` `creeaza_corectii` - fisier gol + antet + exemplu comentat.
- `mapare.py:109-139` `citeste_corectii` - antet exact, randurile `#` ignorate, duplicat = Stop.
- `mapare.py:142-151` `_regula` - ordinea exact -> `.RESTUL` -> `.TOATE`.
- `mapare.py:167-176` `_verifica_corectii` - cheie fara cont in balanta = Stop.
- `mapare.py:179-194` `_verifica_coliziuni` - colaps legal doar sub acelasi `.TOATE`/`.RESTUL`;
partenerii si randurile `exclus` sunt in afara verificarilor.
- `mapare.py:197-236` `aplica` - merge-ul; lungimi > 4 = Stop.
- `mapare.py:239-260` `scrie_mapare` / `genereaza_mapare` - scrie maparea, creeaza corectiile
doar daca lipsesc.
`tests/test_mapare.py` (nou): 10 teste (vezi 2).
## 2. Verificari
- Comanda: `py -m unittest discover -s tests -v`
- Rezultat: `Ran 18 tests in 0.853s` / `OK` (exit 0).
- Include `tests/test_regresie.py` (2/2 OK) si `tests/test_citire_xlsx.py` al lane-ului
paralel (6/6 OK). Zero esecuri.
- Cens octeti > 0x7F: `mapare.py` 0, `tests/test_mapare.py` 0. LF, ca restul `.py` din proiect
(`genereaza_xlsx.py` tot LF) - nu am introdus CRLF si nici diacritice.
- Smoke pe date reale (in temp, fara sa scriu in radacina): `mapare.aplica` pe
`balanta_FUNDATIA.csv` (82 conturi) si `balanta_MASTERJOB.csv` (81) - fara Stop, fara coliziuni.
- `git status`: doar `mapare.py` si `tests/test_mapare.py` adaugate de mine (restul untracked sunt
ale lane-ului `citire-xlsx`). Fara commit, fara `git add`.
## 3. Decizii si intrebari deschise
- Propunerea automata: `cont` = sinteticul dinaintea punctului, `acont` = analiticul concatenat
(toate caracterele, nu doar cifre - `conturi_roa.dbf` are `acont2` alfanumeric `X`/`PG`).
Peste 4 caractere -> Stop (ca `genereaza_xlsx.py:196`), nu trunchiere. Nu am preluat literal
"primele max 4 caractere" din plan, care ar include punctul la sintetici de 3 cifre.
- `.TOATE`/`.RESTUL` se aplica si randului sintetic (parinte), altfel o redenumire
`409 -> 4091` ar cere doua corectii. Fara efect pe colaps: toate au aceeasi regula.
- Coliziunea e permisa doar cand toate conturile au **aceeasi** cheie `.TOATE`/`.RESTUL`;
un amestec exact + `.TOATE` pe aceeasi pereche da Stop.
- `exclus` se intoarce in rezultatul structurat (pentru integrare), dar nu are coloana in
`mapare_` - contractul are exact 6 coloane.
- `cont` gol intr-o corectie = pastreaza propunerea automata; `acont` gol = fara analitic.
- `PLANCONT` nu se preia (cerut).
Nefacut: integrarea in generator (lane `integrare`). Nu am adaugat `citire_balanta` in modul;
integrarea va pasa `rows` din `genereaza_xlsx.citeste_balanta` (suporta `sold` sau
`total_d/total_c`).
## 4. Stare periculoasa
Niciuna: fara binar editat/write-back, fara procese ramase, fara fisiere scrise in radacina
(maparile de test au fost doar in temp). Fara commit.

260
mapare.py Normal file
View File

@@ -0,0 +1,260 @@
"""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

143
tests/test_citire_xlsx.py Normal file
View File

@@ -0,0 +1,143 @@
"""Teste pentru citire_xlsx: contractul intern produs din foaia de calcul SAGA.
Rulare: py -m unittest discover -s tests -v
"""
import os
import pathlib
import sys
import tempfile
import unittest
import openpyxl
RADACINA = pathlib.Path(__file__).resolve().parent.parent
if str(RADACINA) not in sys.path:
sys.path.insert(0, str(RADACINA))
import citire_xlsx
EXEMPLE = RADACINA / "exemple"
XLSX = EXEMPLE / "saga-sqlite-balanta-09-2026.xlsx"
XLS = EXEMPLE / "saga-sqlite-balanta-09-2026.xls"
DANUBE = RADACINA.parent / "saga2roa_danube" / "balanta.xls"
TOL = 0.01
NUMERICE = ["prec_d", "prec_c", "rulaj_d", "rulaj_c",
"total_d", "total_c", "sold_d", "sold_c"]
def frunze(rows):
"""Randurile care nu sunt parintele niciunui alt cont din balanta."""
conts = [r["cont"] for r in rows if r["cont"]]
parinti = {c.rsplit(".", 1)[0] for c in conts if "." in c}
return [r for r in rows if r["cont"] and r["cont"] not in parinti]
def scrie_xlsx(cale, antet, randuri):
wb = openpyxl.Workbook()
ws = wb.active
ws.append(antet)
for rand in randuri:
ws.append(rand)
wb.save(cale)
wb.close()
class TestCitireXlsx(unittest.TestCase):
def _verifica_relatii(self, rows, eticheta):
self.assertTrue(rows, "%s: nicio linie citita" % eticheta)
for r in rows:
self.assertAlmostEqual(
r["total_d"], r["prec_d"] + r["rulaj_d"], delta=TOL,
msg="%s %s: total_d != prec_d + rulaj_d" % (eticheta, r["cont"]))
self.assertAlmostEqual(
r["total_c"], r["prec_c"] + r["rulaj_c"], delta=TOL,
msg="%s %s: total_c != prec_c + rulaj_c" % (eticheta, r["cont"]))
self.assertAlmostEqual(
r["sold_d"] - r["sold_c"], r["total_d"] - r["total_c"], delta=TOL,
msg="%s %s: sold_d - sold_c != total_d - total_c" % (eticheta, r["cont"]))
def _verifica_frunze(self, rows, eticheta):
f = frunze(rows)
sd = sum(r["sold_d"] for r in f)
sc = sum(r["sold_c"] for r in f)
self.assertAlmostEqual(sd, sc, delta=TOL,
msg="%s: SUM(sold_d)=%s != SUM(sold_c)=%s" % (eticheta, sd, sc))
def test_xlsx_si_xls_identice(self):
rows_xlsx = citire_xlsx.citeste_balanta(str(XLSX))
rows_xls = citire_xlsx.citeste_balanta(str(XLS))
self.assertTrue(rows_xlsx)
self.assertEqual(len(rows_xlsx), len(rows_xls))
for a, b in zip(rows_xlsx, rows_xls):
self.assertEqual(a["cont"], b["cont"])
self.assertEqual(a["denumire"], b["denumire"])
self.assertEqual(a["tip"], b["tip"])
for col in ("pagina", "este_total"):
self.assertEqual(a[col], b[col])
for col in NUMERICE:
self.assertAlmostEqual(a[col], b[col], delta=TOL,
msg="%s %s: %s xlsx=%s xls=%s"
% (a["cont"], col, a["cont"], a[col], b[col]))
def test_relatii_si_frunze(self):
for eticheta, cale in (("xlsx", XLSX), ("xls", XLS)):
rows = citire_xlsx.citeste_balanta(str(cale))
self._verifica_relatii(rows, eticheta)
self._verifica_frunze(rows, eticheta)
def test_cont_ramane_text(self):
for eticheta, cale in (("xlsx", XLSX), ("xls", XLS)):
rows = citire_xlsx.citeste_balanta(str(cale))
for r in rows:
self.assertIsInstance(r["cont"], str, "%s: cont nu e text" % eticheta)
def test_danube_analitice(self):
if not DANUBE.exists():
self.skipTest("lipseste %s" % DANUBE)
rows = citire_xlsx.citeste_balanta(str(DANUBE))
self._verifica_relatii(rows, "danube")
self._verifica_frunze(rows, "danube")
analitice = [r for r in rows if "." in r["cont"]]
self.assertTrue(analitice, "danube ar trebui sa aiba conturi analitice")
for r in analitice:
self.assertIsInstance(r["cont"], str, "cont analitic nu e text")
# .xls-ul SAGA nu are CODEPAGE: fara encoding_override, xlrd cade pe iso-8859-1
# si denumirile ies cu caractere de control in loc de diacritice.
stricate = [r["denumire"] for r in rows
if any(ord(c) < 32 or 127 <= ord(c) < 160 for c in str(r["denumire"]))]
self.assertEqual([], stricate, "denumiri decodate gresit (codepage)")
cautat = [r for r in rows if r["cont"] == "105.10"]
self.assertTrue(cautat, "lipseste contul analitic 105.10")
self.assertIsInstance(cautat[0]["cont"], str)
self.assertEqual(cautat[0]["cont"], "105.10")
def test_cont_numeric_devine_text(self):
with tempfile.TemporaryDirectory() as d:
cale = os.path.join(d, "numeric.xlsx")
scrie_xlsx(
cale,
["CONT", "DENUMIRE", "TIP", "DEB_PREC", "CRED_PREC", "RULAJ_D",
"RULAJ_C", "TOTAL_DEB", "TOTAL_CRED", "FIN_D", "FIN_C"],
[[401.00002, "PARTENER", "P", 0, 0, 0, 0, 0, 100, 0, 100]],
)
rows = citire_xlsx.citeste_balanta(cale)
self.assertEqual(len(rows), 1)
self.assertIsInstance(rows[0]["cont"], str)
self.assertEqual(rows[0]["cont"], "401.00002")
def test_coloana_lipsa_si_fara_substring(self):
antet = ["CONT", "DENUMIRE", "TIP", "DEB_PREC", "CRED_PREC", "RULAJ_D",
"RULAJ_C", "TOTAL_DEB_1", "TOTAL_CRED", "FIN_D", "FIN_C"]
with tempfile.TemporaryDirectory() as d:
cale = os.path.join(d, "lipsa.xlsx")
scrie_xlsx(cale, antet, [["401", "FURNIZORI", "P", 0, 0, 0, 0, 0, 0, 0, 0]])
with self.assertRaises(ValueError) as cm:
citire_xlsx.citeste_balanta(cale)
self.assertIn("TOTAL_DEB", str(cm.exception))
if __name__ == "__main__":
unittest.main()

161
tests/test_mapare.py Normal file
View File

@@ -0,0 +1,161 @@
"""Teste pentru mapare.py - propunere automata + corectii manuale.
Ruleaza pe date sintetice in director temporar; nu scrie mapare_*.xlsx /
corectii_*.xlsx in radacina proiectului.
Rulare: py -m unittest discover -s tests -v
"""
import os
import pathlib
import sys
import tempfile
import unittest
import openpyxl
RADACINA = pathlib.Path(__file__).resolve().parent.parent
if str(RADACINA) not in sys.path:
sys.path.insert(0, str(RADACINA))
import mapare
def rand(cont, denumire="", sold=0.0):
return {"cont": cont, "denumire": denumire, "sold": sold}
class TestMapare(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory(prefix="mapare_test_")
self.dir = self._tmp.name
def tearDown(self):
self._tmp.cleanup()
def _cale(self, nume):
return os.path.join(self.dir, nume)
def _ruleaza(self, rows, firma="FIRMA"):
return mapare.genereaza_mapare(rows, firma, self.dir)
def _scrie_corectii(self, randuri, firma="FIRMA"):
cale = self._cale("corectii_%s.xlsx" % firma)
wb = openpyxl.load_workbook(cale)
ws = wb.active
for r in randuri:
ws.append(r)
wb.save(cale)
wb.close()
def _citeste_mapare(self, firma="FIRMA"):
wb = openpyxl.load_workbook(self._cale("mapare_%s.xlsx" % firma))
ws = wb.active
header = [c.value for c in ws[1]]
out = [dict(zip(header, [c.value for c in row]))
for row in ws.iter_rows(min_row=2)]
wb.close()
return out
def test_corectie_supravietuieste_regenerarii(self):
rows = [rand("2131.01", "CLADIRI", 100.0), rand("371.01", "MARFA", 50.0)]
self._ruleaza(rows)
cale = self._cale("corectii_FIRMA.xlsx")
self.assertTrue(os.path.exists(cale))
self._scrie_corectii([["371.01", "3711", "99", "", "corectie marfa"]])
inainte = pathlib.Path(cale).read_bytes()
rezultat = self._ruleaza(rows)
dupa = pathlib.Path(cale).read_bytes()
self.assertEqual(inainte, dupa, "regenerarea a modificat corectii_*.xlsx")
gasit = {r["cont_saga"]: r for r in rezultat}
self.assertEqual(
(gasit["371.01"]["cont"], gasit["371.01"]["acont"], gasit["371.01"]["sursa"]),
("3711", "99", "CORECTAT"))
self.assertEqual(gasit["2131.01"]["sursa"], "AUTO")
def test_ordine_exact_bate_restul_bate_toate(self):
rows = [rand("409.01"), rand("409.02"), rand("409.03")]
self._ruleaza(rows)
self._scrie_corectii([
["409.TOATE", "4091", "T", "", ""],
["409.RESTUL", "4091", "R", "", ""],
["409.01", "4091", "X", "", ""],
])
rezultat = self._ruleaza(rows)
gasit = {r["cont_saga"]: (r["cont"], r["acont"]) for r in rezultat}
self.assertEqual(gasit["409.01"], ("4091", "X"))
self.assertEqual(gasit["409.02"], ("4091", "R"))
self.assertEqual(gasit["409.03"], ("4091", "R"))
def test_toate_se_aplica_fara_exact(self):
rows = [rand("2131.01"), rand("2131.02")]
self._ruleaza(rows)
self._scrie_corectii([["2131.TOATE", "2131", "", "", ""]])
rezultat = self._ruleaza(rows)
for r in rezultat:
self.assertEqual((r["cont"], r["acont"]), ("2131", None))
self.assertEqual(r["sursa"], "CORECTAT")
def test_cont_saga_inexistent_da_eroare(self):
rows = [rand("2131.01")]
self._ruleaza(rows)
self._scrie_corectii([["9999.99", "9999", "99", "", "typo"]])
with self.assertRaises(mapare.Stop):
self._ruleaza(rows)
def test_coliziune_accidentala_da_eroare(self):
rows = [rand("409.01"), rand("410.01")]
self._ruleaza(rows)
self._scrie_corectii([
["409.01", "4091", "01", "", ""],
["410.01", "4091", "01", "", ""],
])
with self.assertRaises(mapare.Stop):
self._ruleaza(rows)
def test_colapsare_prin_toate_nu_da_eroare(self):
rows = [rand("409.01"), rand("409.02")]
self._ruleaza(rows)
self._scrie_corectii([["409.TOATE", "4091", "", "", "agregare intentionata"]])
rezultat = self._ruleaza(rows)
self.assertEqual([(r["cont"], r["acont"]) for r in rezultat],
[("4091", None), ("4091", None)])
def test_redenumire_sintetic(self):
rows = [rand("409"), rand("409.01")]
self._ruleaza(rows)
self._scrie_corectii([["409.TOATE", "4091", "", "", "redenumire sintetic"]])
rezultat = self._ruleaza(rows)
gasit = {r["cont_saga"]: r["cont"] for r in rezultat}
self.assertEqual(gasit["409"], "4091")
self.assertEqual(gasit["409.01"], "4091")
def test_cont_prea_lung_da_eroare(self):
rows = [rand("2131.01")]
self._ruleaza(rows)
self._scrie_corectii([["2131.01", "12345", "01", "", ""]])
with self.assertRaises(mapare.Stop):
self._ruleaza(rows)
def test_acont_prea_lung_da_eroare(self):
rows = [rand("2131.01")]
self._ruleaza(rows)
self._scrie_corectii([["2131.01", "2131", "12345", "", ""]])
with self.assertRaises(mapare.Stop):
self._ruleaza(rows)
def test_propunere_automata_si_partener(self):
rows = [rand("1171.01.12"), rand("401.00002")]
rezultat = mapare.aplica(rows, {}, parteneri={"401"})
gasit = {r["cont_saga"]: r for r in rezultat}
self.assertEqual((gasit["1171.01.12"]["cont"], gasit["1171.01.12"]["acont"]),
("1171", "0112"))
self.assertEqual((gasit["401.00002"]["cont"], gasit["401.00002"]["acont"]),
("401", None))
if __name__ == "__main__":
unittest.main()