GTI_GERENCIADOR_FISCAL/scripts/map_product_ui_fields.py

"""Assisted single-session mapping of the SGV "Cadastro de Produto" fiscal screen (READ-ONLY tool).

    python scripts/map_product_ui_fields.py --product <CODIGO>

For each UI field the tool shows CURRENT and SUGGESTED_TEMP, waits for ENTER while the operator changes ONLY that
field in the SGV and saves, re-reads the whole C000025 row, diffs it against the previous baseline and records
which column(s) changed. The tool never writes to Firebird and never touches the GUI. State is persisted after every
step (docs/SGV_PRODUCT_FISCAL_UI_DIFF_SESSION_V1.json) so the session can be resumed.

Binding classes: DIRECT_BINDING (one non-technical column changed) | DERIVED/COUPLED_FIELD (several changed;
ALL recorded, none chosen) | AUTOMATIC_METADATA (only timestamp-like technical columns) | AMBIGUOUS (nothing changed
or several changed without the expected one).
"""
import argparse
import json
import os
import sys
from datetime import datetime
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "agent" / "src")); sys.path.insert(0, str(ROOT / "scripts"))

SESSION = ROOT / "docs" / "SGV_PRODUCT_FISCAL_UI_DIFF_SESSION_V1.json"
# Columns the ERP rewrites by itself on every save. Only these are ignored, and they are still listed in the record.
TECHNICAL = {"ULTIMA_ALTERACAO", "DATA_ALTERACAO", "DATA_ULTIMA_ALTERACAO", "USUARIO_ALTERACAO"}

# ui_label, hypothesised column (never trusted), candidate temporary values (first one different from CURRENT wins)
SEQUENCE = [
    ("NCM", "CLASSIFICACAO_FISCAL", ["22030000", "19022000"]),
    ("CFOP entrada", "CFOP_ENTRADA", ["1102", "1949"]),
    ("CFOP saída", "CFOP", ["5405", "5102", "5949"]),
    ("CST ICMS entrada", "CST_ENTRADA", ["040", "000", "060"]),
    ("CST ICMS saída", "CST", ["041", "000", "060"]),
    ("Alíquota ICMS", "ALIQUOTA", ["12", "18", "17"]),
    ("CEST", "CEST", ["0601100", "1708400"]),
    ("CSOSN", "CSOSN", ["103", "500", "102"]),
    ("PIS entrada", "COD_ENTRADA_PIS", ["73", "98", "50"]),
    ("PIS saída", "COD_SAIDA_PIS", ["04", "01", "49"]),
    ("Alíquota PIS", "ALIQ_PIS", ["0,65", "1,65"]),
    ("COFINS entrada", "COD_ENTRADA_COFINS", ["73", "98", "50"]),
    ("COFINS saída", "COD_SAIDA_COFINS", ["04", "01", "49"]),
    ("Alíquota COFINS", "ALIQ_COFINS", ["3", "7,6"]),
    ("IPI", "IPI", ["5", "10", "15"]),
    ("CST IPI", "CST_IPI", ["50", "00", "99"]),
    ("EX TIPI", "EX_TIPI", ["1", "2"]),
    ("Origem", "ORIGEM", ["1", "2", "0"]),
    ("IAT", "IAT", ["T", "A"]),
    ("Incidência PIS/COFINS", "INCIDENCIA_PISCOFINS", ["MONOFÁSICA", "SUBST.TRIBUTÁRIA", "NENHUMA"]),
    ("IPPT", "IPPT", ["P", "T"]),
]


def _same(a, b):
    if a is None or b is None:
        return a is None and b is None
    try:
        return float(str(a).replace(",", ".")) == float(str(b).replace(",", "."))
    except ValueError:
        return str(a).strip().upper() == str(b).strip().upper()


def suggest_temp(current, candidates):
    """First candidate that differs from the current value (never the same value, never blank)."""
    for c in candidates:
        if not _same(current, c):
            return c
    return None


def classify(changes, expected_column):
    """changes: [{'FIELD','BEFORE','AFTER','technical'}]. Never picks a column when several changed."""
    real = [c["FIELD"] for c in changes if not c["technical"]]
    tech = [c["FIELD"] for c in changes if c["technical"]]
    if not real:
        return "AUTOMATIC_METADATA" if tech else "AMBIGUOUS", "no fiscal column changed" if not tech else "only technical timestamp columns changed"
    if len(real) == 1:
        return "DIRECT_BINDING", "exactly one column changed" + ("" if real[0] == expected_column else f" (differs from hypothesis {expected_column})")
    if expected_column in real:
        return "DERIVED/COUPLED_FIELD", f"{len(real)} columns changed together; none selected"
    return "AMBIGUOUS", f"{len(real)} columns changed and none is the hypothesis {expected_column}"


def compute_diff(before, after):
    return [{"FIELD": f, "BEFORE": before.get(f), "AFTER": after.get(f), "technical": f in TECHNICAL}
            for f in sorted(set(before) | set(after)) if before.get(f) != after.get(f)]


def new_session(product):
    return {"artifact": "SGV_PRODUCT_FISCAL_UI_DIFF_SESSION_V1", "database": "TESTESOFICIAIS.FDB", "product": product, "started": datetime.now().isoformat(timespec="seconds"),
            "initial_snapshot": None, "baseline": None, "results": [], "firebird_writes": 0}


def save(session):
    SESSION.write_text(json.dumps(session, ensure_ascii=False, indent=1, default=str), encoding="utf-8")


def load_env():
    for line in (ROOT / ".env").read_text(encoding="utf-8").splitlines():
        if "=" in line and not line.startswith("#"):
            k, _, v = line.partition("=")
            os.environ.setdefault(k.strip(), v.strip())


def run_step(session, ui_label, column, candidates, read_row, ask):
    baseline = session["baseline"]
    current = baseline.get(column) if column else None
    temp = suggest_temp(current, candidates) if candidates else None
    print(f"\nUI_FIELD={ui_label}\nCURRENT={current}\nSUGGESTED_TEMP={temp}\nHYPOTHESIS_COLUMN={column} (not trusted)")
    answer = ask("Change ONLY this field in the SGV and save, then press ENTER  [s=skip (not on screen), q=quit]: ").strip().lower()
    if answer == "q":
        return "quit"
    if answer == "s":
        session["results"].append({"ui_label": ui_label, "hypothesis_column": column, "status": "NOT_ON_SCREEN", "at": datetime.now().isoformat(timespec="seconds")})
        save(session)
        return "skipped"
    after = read_row()
    changes = compute_diff(baseline, after)
    binding, why = classify(changes, column)
    rec = {"ui_label": ui_label, "hypothesis_column": column, "current": current, "suggested_temp": temp, "status": "TESTED", "binding_type": binding, "reason": why,
           "db_fields_changed": [c["FIELD"] for c in changes if not c["technical"]], "technical_fields_changed": [c["FIELD"] for c in changes if c["technical"]],
           "changes": changes, "at": datetime.now().isoformat(timespec="seconds")}
    session["results"].append(rec)
    session["baseline"] = after  # new state is the baseline for the next field
    save(session)
    print(f"UI_FIELD={ui_label}\nDB_FIELDS_CHANGED={rec['db_fields_changed']}\nBEFORE={ {c['FIELD']: c['BEFORE'] for c in changes} }\nAFTER={ {c['FIELD']: c['AFTER'] for c in changes} }\nBINDING={binding} ({why})")
    return "tested"


def main(argv=None):
    ap = argparse.ArgumentParser()
    ap.add_argument("--product", required=True)
    ap.add_argument("--restart", action="store_true", help="discard a previous session file for this product")
    a = ap.parse_args(argv)
    load_env()
    from sgv_product_snapshot import read_row  # read-only single-row reader (read-only isolation)
    from gti_fiscal_agent.firebird_reader import parse_company_context, _connect
    path = os.environ.get("FIREBIRD_DATABASE_PATH", "")
    if Path(path).name.upper() != "TESTESOFICIAIS.FDB":
        raise SystemExit("this experiment is only for the disposable TESTESOFICIAIS.FDB (FIREBIRD_DATABASE_PATH)")
    reader = lambda: read_row(os.environ, a.product)  # noqa: E731

    if SESSION.exists() and not a.restart and json.loads(SESSION.read_text(encoding="utf-8")).get("product") == a.product:
        session = json.loads(SESSION.read_text(encoding="utf-8"))
        print(f"Resuming session ({len(session['results'])} results already recorded).")
        session["baseline"] = reader()  # honour any manual change since the last step
    else:
        session = new_session(a.product)
        session["initial_snapshot"] = session["baseline"] = reader()
        save(session)
    conn = _connect(os.environ)
    try:
        cur = conn.cursor(); cur.execute("SELECT UF, CRT FROM C000004")
        session["company_context_read_from_db"] = parse_company_context(cur.fetchall())
    finally:
        conn.close()
    done = {r["ui_label"] for r in session["results"]}
    for ui_label, column, candidates in SEQUENCE:
        if ui_label in done:
            continue
        if run_step(session, ui_label, column, candidates, reader, input) == "quit":
            break
    else:
        while True:
            extra = input("\nAny OTHER fiscal field visible on the screen? Type its label (or ENTER to finish): ").strip()
            if not extra:
                break
            if run_step(session, extra, None, [], reader, input) == "quit":
                break
    session["finished"] = datetime.now().isoformat(timespec="seconds")
    save(session)
    changed = {}
    for r in session["results"]:
        for c in r.get("changes", []):
            if not c["technical"]:
                changed.setdefault(c["FIELD"], session["initial_snapshot"].get(c["FIELD"]))
    print("\nSession saved:", SESSION)
    if changed:
        print("RESTORE_ORIGINAL_VALUES (set these back in the SGV):")
        for f, v in changed.items():
            print(f"  {f} = {v}")


if __name__ == "__main__":
    main()