solduri2roa/citire_xlsx.py

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