#!/usr/bin/env python3 """Import DBF ROAAUTO -> SQLite gateway (T5). Plan.md sect. 7: dry-run + raport intai (randuri valide, mapari orfane, coduri necunoscute in nomenclator), apoi scrie in SQLite. Surse: - prestatii_rar.DBF (COD_PREST, NUME_PREST) -> nomenclator_rar - mapare_prestatii.DBF (COD_OP, DESCR_OP, COD_RAR, AUTO_SEND) -> operations_mapping (rar_log.DBF NU se migreaza — jurnalul nou e `submissions` + live din RAR.) Importul e IDEMPOTENT (upsert pe cheile UNIQUE), deci rularea repetata nu duplica. Default = dry-run (raport, fara scriere). `--commit` scrie efectiv. Utilizare: python -m tools.import_dbf # dry-run + raport python -m tools.import_dbf --commit # scrie in SQLite Necesita: pip install dbfread """ from __future__ import annotations import argparse import sys from pathlib import Path from typing import Any from dbfread import DBF from app.config import ROOT from app.db import get_connection, init_db from app.mapping import DEFAULT_ACCOUNT_ID # DBF-urile vin din arhiva ROAAUTO (legacy VFP), mutata in legacy-vfp/. MAPARE_DBF = ROOT / "legacy-vfp" / "mapare_prestatii.DBF" PREST_DBF = ROOT / "legacy-vfp" / "prestatii_rar.DBF" # Language driver al DBF-urilor = 0x03 (Windows ANSI / cp1252). Diacriticele # scrise ca literal '?' sunt in sursa, nu un artefact de encoding. DBF_ENCODING = "cp1252" # --------------------------------------------------------------------------- # # Citire DBF -> randuri normalizate (pur, fara DB) # # --------------------------------------------------------------------------- # def _field(rec: dict, *names: str) -> Any: """Primul camp negol dintr-o lista de nume alternative (tolerant la schema).""" for n in names: if n in rec and rec[n] is not None: return rec[n] return None def read_nomenclator(path: Path, *, encoding: str = DBF_ENCODING) -> dict[str, Any]: """Citeste prestatii_rar.DBF. Intoarce raport + randuri valide. {rows: [{cod_prestatie, nume_prestatie}], duplicates: [cod...], blanks: int} cod_prestatie normalizat strip().upper(); duplicate = acelasi cod de 2+ ori (pastram prima aparitie). """ rows: list[dict[str, str]] = [] seen: set[str] = set() duplicates: list[str] = [] blanks = 0 for rec in DBF(str(path), encoding=encoding, char_decode_errors="replace"): cod = str(_field(rec, "COD_PREST", "COD_PRESTATIE", "COD") or "").strip().upper() nume = str(_field(rec, "NUME_PREST", "NUME_PRESTATIE", "NUME") or "").strip() if not cod: blanks += 1 continue if cod in seen: duplicates.append(cod) continue seen.add(cod) rows.append({"cod_prestatie": cod, "nume_prestatie": nume}) return {"rows": rows, "duplicates": duplicates, "blanks": blanks} def read_mapari(path: Path, *, encoding: str = DBF_ENCODING) -> dict[str, Any]: """Citeste mapare_prestatii.DBF. Intoarce raport + randuri valide. {rows: [{cod_op_service, denumire, cod_prestatie, auto_send}], duplicates: [cod_op...], blanks: int} Rand valid = are si COD_OP si COD_RAR. blanks = randuri carora le lipseste unul din ele. duplicate = acelasi COD_OP de 2+ ori (pastram prima aparitie). """ rows: list[dict[str, Any]] = [] seen: set[str] = set() duplicates: list[str] = [] blanks = 0 for rec in DBF(str(path), encoding=encoding, char_decode_errors="replace"): op = str(_field(rec, "COD_OP", "COD_OP_SERVICE") or "").strip() cod = str(_field(rec, "COD_RAR", "COD_PRESTATIE", "COD_PREST") or "").strip().upper() denumire = str(_field(rec, "DESCR_OP", "DENUMIRE", "DESCRIERE") or "").strip() auto = _field(rec, "AUTO_SEND") auto_send = bool(auto) if auto is not None else True if not op or not cod: blanks += 1 continue if op in seen: duplicates.append(op) continue seen.add(op) rows.append( {"cod_op_service": op, "denumire": denumire, "cod_prestatie": cod, "auto_send": auto_send} ) return {"rows": rows, "duplicates": duplicates, "blanks": blanks} def find_orphans(mapari: list[dict], known_codes: set[str]) -> list[dict]: """Mapari al caror cod_prestatie nu exista in nomenclator (nu pot fi trimise).""" return [m for m in mapari if m["cod_prestatie"] not in known_codes] # --------------------------------------------------------------------------- # # Scriere SQLite (idempotenta) # # --------------------------------------------------------------------------- # def write_nomenclator(conn, rows: list[dict]) -> int: conn.executemany( "INSERT INTO nomenclator_rar (cod_prestatie, nume_prestatie, updated_at) " "VALUES (?, ?, datetime('now')) " "ON CONFLICT(cod_prestatie) DO UPDATE SET nume_prestatie=excluded.nume_prestatie, " "updated_at=datetime('now')", [(r["cod_prestatie"], r["nume_prestatie"]) for r in rows], ) return len(rows) def write_mapari(conn, rows: list[dict], account_id: int) -> int: conn.executemany( "INSERT INTO operations_mapping (account_id, cod_op_service, cod_prestatie, auto_send) " "VALUES (?, ?, ?, ?) " "ON CONFLICT(account_id, cod_op_service) DO UPDATE SET " "cod_prestatie=excluded.cod_prestatie, auto_send=excluded.auto_send", [(account_id, r["cod_op_service"], r["cod_prestatie"], 1 if r["auto_send"] else 0) for r in rows], ) return len(rows) # --------------------------------------------------------------------------- # # Raport + CLI # # --------------------------------------------------------------------------- # def build_report( nomenclator: dict, mapari: dict, orphans: list[dict], *, account_id: int ) -> str: lines: list[str] = [] lines.append("=== Import DBF ROAAUTO -> SQLite (raport) ===") lines.append(f"Cont tinta: account_id={account_id}") lines.append("") lines.append("nomenclator_rar (<- prestatii_rar.DBF):") lines.append(f" randuri valide : {len(nomenclator['rows'])}") lines.append(f" duplicate cod : {len(nomenclator['duplicates'])} {sorted(set(nomenclator['duplicates'])) or ''}".rstrip()) lines.append(f" randuri goale : {nomenclator['blanks']}") lines.append("") lines.append("operations_mapping (<- mapare_prestatii.DBF):") lines.append(f" randuri valide : {len(mapari['rows'])}") lines.append(f" duplicate COD_OP: {len(mapari['duplicates'])} {sorted(set(mapari['duplicates'])) or ''}".rstrip()) lines.append(f" randuri goale : {mapari['blanks']} (lipsa COD_OP sau COD_RAR)") lines.append(f" mapari ORFANE : {len(orphans)} (cod_prestatie necunoscut in nomenclator)") for m in orphans: lines.append(f" - {m['cod_op_service']} -> {m['cod_prestatie']} ({m['denumire'] or 'fara denumire'})") return "\n".join(lines) def run( *, commit: bool, account_id: int = DEFAULT_ACCOUNT_ID, mapare_path: Path = MAPARE_DBF, prest_path: Path = PREST_DBF, encoding: str = DBF_ENCODING, ) -> dict[str, Any]: """Citeste DBF-urile, construieste raportul si (optional) scrie in SQLite. Intoarce {report, nomenclator, mapari, orphans, written:{nomenclator,mapari}}. """ missing = [str(p) for p in (prest_path, mapare_path) if not p.exists()] if missing: raise FileNotFoundError("DBF lipsa: " + ", ".join(missing)) nomenclator = read_nomenclator(prest_path, encoding=encoding) mapari = read_mapari(mapare_path, encoding=encoding) init_db() conn = get_connection() try: # Coduri cunoscute = nomenclatorul ce urmeaza importat + ce e deja in DB # (seed fallback / live din worker). Asa orfanele sunt detectate corect # chiar daca prestatii_rar.DBF nu acopera toate codurile. db_codes = {r["cod_prestatie"] for r in conn.execute("SELECT cod_prestatie FROM nomenclator_rar")} known = db_codes | {r["cod_prestatie"] for r in nomenclator["rows"]} orphans = find_orphans(mapari["rows"], known) written = {"nomenclator": 0, "mapari": 0} if commit: conn.execute("BEGIN IMMEDIATE") try: written["nomenclator"] = write_nomenclator(conn, nomenclator["rows"]) written["mapari"] = write_mapari(conn, mapari["rows"], account_id) conn.execute("COMMIT") except Exception: conn.execute("ROLLBACK") raise report = build_report(nomenclator, mapari, orphans, account_id=account_id) return { "report": report, "nomenclator": nomenclator, "mapari": mapari, "orphans": orphans, "written": written, } finally: conn.close() def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Import DBF ROAAUTO -> SQLite gateway (T5)") parser.add_argument("--commit", action="store_true", help="scrie in SQLite (implicit: doar raport)") parser.add_argument("--account-id", type=int, default=DEFAULT_ACCOUNT_ID, help="cont tinta pentru mapari") parser.add_argument("--mapare", type=Path, default=MAPARE_DBF, help="cale mapare_prestatii.DBF") parser.add_argument("--nomenclator", type=Path, default=PREST_DBF, help="cale prestatii_rar.DBF") parser.add_argument("--encoding", default=DBF_ENCODING, help=f"encoding DBF (implicit {DBF_ENCODING})") args = parser.parse_args(argv) try: result = run( commit=args.commit, account_id=args.account_id, mapare_path=args.mapare, prest_path=args.nomenclator, encoding=args.encoding, ) except FileNotFoundError as exc: print(f"EROARE: {exc}", file=sys.stderr) return 2 print(result["report"]) print("") if args.commit: w = result["written"] print(f"COMMIT: scris {w['nomenclator']} coduri nomenclator, {w['mapari']} mapari (idempotent).") else: print("DRY-RUN: nimic scris. Reia cu --commit dupa ce verifici raportul.") return 0 if __name__ == "__main__": sys.exit(main())