romfast/solduri2roa
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
solduri2roa/citire_fdb.py
Marius Mutu b5773c9822 Decizii LACERTA: data = ultima zi, 542 cu parteneri, 4428/01, backup .fbk 
- genereaza_xlsx: coloana data = ultima zi a lunii; etalonul de regresie
  actualizat doar pe aceasta coloana (verificat celula cu celula).
- config: 542 cont cu parteneri (acum identic cu REF); descrierea lui 4092
  corectata.
- extrage/citire_fdb: .fbk se restaureaza cu gbak -c in temporar; test ca
  da aceleasi fisiere ca FDB-ul.
- test_lacerta: 49/53 frunze identice cu REF, 4428.TOATE -> 4428/01.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EYeAtVxeS8m4oXekjX8Am2
2026-09-27 16:43:54 +03:00

350 lines
13 KiB
Python
Raw Permalink Blame History

"""Cititor pentru baza SAGA noua (Firebird): balanta, parteneri si facturi.
Produce, din copia unei baze `CONT_BAZA.FDB`, contractul intern al convertorului:
balanta_<FIRMA>.csv - pagina, cont, denumire, prec_d, prec_c, rulaj_d, rulaj_c,
total_d, total_c, sold_d, sold_c, este_total, tip
parteneri_<FIRMA>.csv - cont_analitic, cod, denumire, cod_fiscal
facturi_<FIRMA>.csv - cont_analitic, numar, data, scadent, sold
Baza se deschide NUMAI embedded (pe TCP 3060 parola implicita nu merge), in
tranzactie read-only, doar SELECT, rollback la final. Parametrul este calea catre o
COPIE a bazei (gbak sau fisier oprit), niciodata baza vie a clientului.
Sumele nu vin din soldurile stocate in CONTURI (sunt 0); se agrega REGISTRU pe trei
intervale: inainte de inceputul anului (sold de preluare), de la inceputul anului pana
la inceputul lunii cerute si luna ceruta. Precedentul = soldul de preluare netat pe
latura + rulajele din an, BRUTE (ca SAGA: netarea lor ar anula rulajul conturilor de
cheltuieli si venituri); rulajul lunii se ia brut, iar total = prec + rulaj.
Fiecare cont din REGISTRU intra in toate nivelurile lui: 401.00002 -> 401 si 401.00002,
deci balanta are sinteticele si analiticele, ca exportul SAGA cu analitice. Contul
colector `%` se exclude.
Baza primita se copiaza in temporar si copia se repune online (clientul o poate opri
cu `full shutdown` inainte de copiere); fisierul primit nu se atinge.
VERIFICARI: firma de joaca din `D:/SAGA250909/0001` (sinteticele = exportul .xlsx,
la octet) si LACERTA 06/2026 (tests/test_lacerta.py: 48 din 53 de frunze identice cu
importul facut din exportul SAGA, restul explicate in docs/analiza_lacerta_fdb.md;
facturile per document insumeaza netul fiecarui partener). La un client nou se
reruleaza o poarta pe datele lui.
Utilizare:
py citire_fdb.py "<copie CONT_BAZA.FDB>" "<FIRMA>" <an> <luna> [dir_iesire]
"""
import csv
import os
import shutil
import subprocess
import sys
import tempfile
from calendar import monthrange
from datetime import date
FBCLIENT = r"C:\Program Files\Firebird\Firebird30_Saga\fbclient.dll"
GFIX = os.path.join(os.path.dirname(FBCLIENT), "gfix.exe")
GBAK = os.path.join(os.path.dirname(FBCLIENT), "gbak.exe")
UTILIZATOR = "SYSDBA"
PAROLA = "masterkey"
CHARSET = "WIN1250"
COLOANE_BALANTA = ["pagina", "cont", "denumire", "prec_d", "prec_c", "rulaj_d",
"rulaj_c", "total_d", "total_c", "sold_d", "sold_c",
"este_total", "tip"]
COLOANE_PARTENERI = ["cont_analitic", "cod", "denumire", "cod_fiscal"]
COLOANE_FACTURI = ["cont_analitic", "numar", "data", "scadent", "sold"]
_SQL_BALANTA = """
SELECT CONT, LATURA,
SUM(CASE WHEN DATA < ? THEN SUMA ELSE 0 END) AS INIT,
SUM(CASE WHEN DATA >= ? AND DATA < ? THEN SUMA ELSE 0 END) AS AN_PREC,
SUM(CASE WHEN DATA >= ? AND DATA < ? THEN SUMA ELSE 0 END) AS LUNA
FROM (
SELECT TRIM(CONT_D) AS CONT, 'D' AS LATURA, SUMA, DATA
FROM REGISTRU
WHERE TRIM(CONT_D) NOT IN ('', '%') AND DATA < ?
UNION ALL
SELECT TRIM(CONT_C) AS CONT, 'C' AS LATURA, SUMA, DATA
FROM REGISTRU
WHERE TRIM(CONT_C) NOT IN ('', '%') AND DATA < ?
) A
GROUP BY CONT, LATURA
ORDER BY CONT, LATURA
"""
_API_INCARCAT = False
def _driver():
import firebird.driver as fdb
return fdb
def _incarca_clientul():
global _API_INCARCAT
fdb = _driver()
if not os.path.exists(FBCLIENT):
raise FileNotFoundError("lipseste fbclient.dll: %s" % FBCLIENT)
if not _API_INCARCAT:
fdb.load_api(FBCLIENT)
_API_INCARCAT = True
return fdb
def _conecteaza(cale):
"""Conexiune embedded read-only; se inchide cu rollback de catre apelant."""
fdb = _incarca_clientul()
if not os.path.exists(cale):
raise FileNotFoundError("lipseste baza Firebird: %s" % cale)
con = fdb.connect(database=cale, user=UTILIZATOR, password=PAROLA,
charset=CHARSET, no_gc=True)
con.begin(fdb.tpb(fdb.Isolation.READ_COMMITTED,
access_mode=fdb.TraAccessMode.READ))
return con
def _select(con, sql, params=()):
cur = con.cursor()
try:
cur.execute(sql, params)
return cur.fetchall()
finally:
cur.close()
def _text(v):
if v is None:
return ""
return str(v).strip()
def _num(v):
if v is None:
return 0.0
return float(v)
def _net(d, c):
"""(debitor, creditor) pe netul celor doua sume."""
if d >= c:
return d - c, 0
return 0, c - d
def _limite(an, luna):
an, luna = int(an), int(luna)
if not 1 <= luna <= 12:
raise ValueError("luna trebuie intre 1 si 12: %r" % luna)
urmator = date(an + 1, 1, 1) if luna == 12 else date(an, luna + 1, 1)
return date(an, 1, 1), date(an, luna, 1), urmator
def _ultima_zi(an, luna):
an, luna = int(an), int(luna)
return date(an, luna, monthrange(an, luna)[1])
def _conturi(con):
"""{cont: (denumire, tip)} din planul de conturi."""
info = {}
for cont, denumire, tip in _select(
con, "SELECT TRIM(CONT), TRIM(DENUMIRE), TRIM(TIP) FROM CONTURI"):
info[_text(cont)] = (_text(denumire), _text(tip))
return info
def citeste_balanta(cale, an, luna):
"""Randurile balantei in contractul intern, pentru luna ceruta."""
d_an, d_luna, d_urm = _limite(an, luna)
con = _conecteaza(cale)
try:
brute = _select(con, _SQL_BALANTA,
(d_an, d_an, d_luna, d_luna, d_urm, d_urm, d_urm))
conturi = _conturi(con)
finally:
con.rollback()
con.close()
sume = {}
for cont, latura, init, an_prec, luna_v in brute:
seg = _text(cont).split(".")
# contul intra in fiecare nivel al lui: 215.1 -> 215 si 215.1
for k in range(1, len(seg) + 1):
s = sume.setdefault(".".join(seg[:k]), {"i_d": 0, "i_c": 0, "a_d": 0,
"a_c": 0, "l_d": 0, "l_c": 0})
if latura == "D":
s["i_d"] += init or 0
s["a_d"] += an_prec or 0
s["l_d"] += luna_v or 0
else:
s["i_c"] += init or 0
s["a_c"] += an_prec or 0
s["l_c"] += luna_v or 0
rows = []
for cont in sorted(sume):
s = sume[cont]
# precedent = soldul de preluare pe net + rulajele din an, brute (ca SAGA:
# netarea rulajelor ar sterge rulajul conturilor de cheltuieli/venituri)
prec_d, prec_c = _net(s["i_d"], s["i_c"])
prec_d, prec_c = prec_d + s["a_d"], prec_c + s["a_c"]
rulaj_d, rulaj_c = s["l_d"], s["l_c"]
total_d = prec_d + rulaj_d
total_c = prec_c + rulaj_c
sold_d, sold_c = _net(total_d, total_c)
denumire, tip = conturi.get(cont) or ("", conturi.get(cont.split(".")[0], ("", ""))[1])
rows.append({
"pagina": "", "cont": cont, "denumire": denumire,
"prec_d": _num(prec_d), "prec_c": _num(prec_c),
"rulaj_d": _num(rulaj_d), "rulaj_c": _num(rulaj_c),
"total_d": _num(total_d), "total_c": _num(total_c),
"sold_d": _num(sold_d), "sold_c": _num(sold_c),
"este_total": "", "tip": tip,
})
return rows
def citeste_parteneri(cale):
"""Furnizori + clienti; codul fiscal lipsa cade pe codul contului analitic."""
con = _conecteaza(cale)
try:
brute = _select(
con,
"SELECT TRIM(ANALITIC), TRIM(COD), TRIM(DENUMIRE), TRIM(COD_FISCAL) "
"FROM FURNIZORI "
"UNION ALL "
"SELECT TRIM(ANALITIC), TRIM(COD), TRIM(DENUMIRE), TRIM(COD_FISCAL) "
"FROM CLIENTI")
finally:
con.rollback()
con.close()
rows = []
for analitic, cod, denumire, cod_fiscal in brute:
cont_analitic = _text(analitic)
if cont_analitic == "":
continue
rows.append({
"cont_analitic": cont_analitic,
"cod": _text(cod),
"denumire": _text(denumire),
"cod_fiscal": _text(cod_fiscal) or cont_analitic,
})
rows.sort(key=lambda r: (r["cont_analitic"], r["cod"]))
return rows
def _rand_factura(cont_analitic, numar, data, scadent, sold, scadent_implicit):
cont_analitic = _text(cont_analitic)
if cont_analitic == "":
return None
return {
"cont_analitic": cont_analitic,
"numar": _text(numar),
"data": data.isoformat() if data is not None else "",
"scadent": (scadent or scadent_implicit).isoformat(),
"sold": _num(sold),
}
def citeste_facturi(cale, an, luna):
"""Facturi furnizori (INTRD/INTRARI) + clienti (IESIRI), intr-un singur fisier."""
scadent_implicit = _ultima_zi(an, luna)
con = _conecteaza(cale)
try:
furnizori = _select(
con,
"SELECT ID_INTRARE, CONT_FUR, NR_INTRARE, DATA, SCADENT, NEACHITAT "
"FROM INTRD")
vazute = {row[0] for row in furnizori}
for row in _select(
con,
"SELECT ID_INTRARE, CONT_FUR, NR_INTRARE, DATA, SCADENT, NEACHITAT "
"FROM INTRARI"):
if row[0] not in vazute:
furnizori.append(row)
vazute.add(row[0])
clienti = _select(
con,
"SELECT CONT_CLI, NR_IESIRE, DATA, SCADENT, NEACHITAT FROM IESIRI")
finally:
con.rollback()
con.close()
rows = []
for _, cont, numar, data, scadent, sold in furnizori:
r = _rand_factura(cont, numar, data, scadent, sold, scadent_implicit)
if r:
rows.append(r)
for cont, numar, data, scadent, sold in clienti:
r = _rand_factura(cont, numar, data, scadent, sold, scadent_implicit)
if r:
rows.append(r)
rows.sort(key=lambda r: (r["cont_analitic"], r["data"], r["numar"], r["scadent"]))
return rows
def scrie_csv(rows, cale, coloane):
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 copie_online(cale_fdb, dir_tmp):
"""Copiaza baza in dir_tmp si o repune online, daca clientul a oprit-o inainte de
copiere (`full shutdown`). Un backup `.fbk` se restaureaza cu gbak in dir_tmp.
Fisierul primit nu se atinge. Pe o baza deja online gfix raspunde cu eroare,
inofensiva: daca repunerea chiar esueaza, conectarea urmatoare pica cu mesajul
real (`database ... shutdown`)."""
if not os.path.exists(cale_fdb):
raise FileNotFoundError("lipseste baza Firebird: %s" % cale_fdb)
copie = os.path.join(dir_tmp, "CONT_BAZA.FDB")
if cale_fdb.lower().endswith(".fbk"):
rez = subprocess.run([GBAK, "-c", cale_fdb, copie, "-user", UTILIZATOR,
"-password", PAROLA], capture_output=True, text=True)
if rez.returncode != 0:
raise RuntimeError("restaurarea %s a esuat: %s"
% (cale_fdb, (rez.stderr or rez.stdout).strip()))
return copie
shutil.copyfile(cale_fdb, copie)
if os.path.exists(GFIX):
subprocess.run([GFIX, "-online", "normal", copie, "-user", UTILIZATOR,
"-password", PAROLA], capture_output=True)
return copie
def produce(cale_fdb, firma, an, luna, dir_iesire="."):
"""Scrie cele trei fisiere ale contractului intern; intoarce randurile.
Citeste dintr-o copie temporara, repusa online (vezi copie_online)."""
if not os.path.isdir(dir_iesire):
os.makedirs(dir_iesire)
with tempfile.TemporaryDirectory(prefix="citire_fdb_") as tmp:
copie = copie_online(cale_fdb, tmp)
balanta = citeste_balanta(copie, an, luna)
parteneri = citeste_parteneri(copie)
facturi = citeste_facturi(copie, an, luna)
scrie_csv(balanta, os.path.join(dir_iesire, "balanta_%s.csv" % firma),
COLOANE_BALANTA)
scrie_csv(parteneri, os.path.join(dir_iesire, "parteneri_%s.csv" % firma),
COLOANE_PARTENERI)
scrie_csv(facturi, os.path.join(dir_iesire, "facturi_%s.csv" % firma),
COLOANE_FACTURI)
return {"balanta": balanta, "parteneri": parteneri, "facturi": facturi}
def main():
if len(sys.argv) not in (5, 6) or sys.argv[1] in ("-h", "--help"):
print(__doc__)
sys.exit(1)
cale, firma, an, luna = sys.argv[1:5]
dir_iesire = sys.argv[5] if len(sys.argv) == 6 else "."
rez = produce(cale, firma, an, luna, dir_iesire)
print("balanta_%s.csv: %d randuri" % (firma, len(rez["balanta"])))
print("parteneri_%s.csv: %d randuri" % (firma, len(rez["parteneri"])))
print("facturi_%s.csv: %d randuri" % (firma, len(rez["facturi"])))
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink