GTI_GERENCIADOR_FISCAL/fiscal/central/app/datasets/bootstrap_v1_extra.py

"""Parsers for the official NF-e/RTC spreadsheets the operator placed in
sources/raw/bootstrap_v1/ after the Portal NF-e/SPED downloads were blocked
in earlier missions. Each parser validates the workbook's own header/sheet
structure before trusting it — never trusts the filename alone.

Domains: CFOP, FCP, ANP (cProdANP), NCM x uTrib, RTC biofuel index
(pBioObrigatorio), RTC cCredPres.
"""
from dataclasses import dataclass, field
from hashlib import sha256
from io import BytesIO


def _digits(value):
    return "".join(c for c in str(value or "") if c.isdigit())


def _iso(value):
    if value is None:
        return None
    return value.isoformat() if hasattr(value, "isoformat") else str(value)


def _load(raw_bytes):
    import openpyxl
    return openpyxl.load_workbook(BytesIO(raw_bytes), read_only=True, data_only=True)


def _rows(ws, header_len):
    rows = ws.iter_rows(values_only=True)
    header = next(rows)
    return header[:header_len], rows


@dataclass(frozen=True)
class Payload:
    sha256: str
    entries: tuple
    parser_version: str
    extra_entries: tuple = ()


# --- CFOP -------------------------------------------------------------
_CFOP_EXPECTED_HEADER = ("CFOP", "Início de vigência", "Fim de vigência", "indNFe")


def parse_cfop(raw_bytes: bytes) -> Payload:
    wb = _load(raw_bytes)
    if "CFOP" not in wb.sheetnames:
        raise ValueError(f"expected a 'CFOP' sheet, found {wb.sheetnames}")
    header, rows = _rows(wb["CFOP"], 14)
    if _digits("".join(str(h) for h in header[:1])) or "CFOP" not in str(header[0]):
        raise ValueError(f"unexpected CFOP header: {header}")
    entries = []
    for row in rows:
        if row[0] is None:
            continue
        code = str(int(row[0])) if isinstance(row[0], (int, float)) else str(row[0]).strip()
        entries.append({
            "code": code, "effective_from": _iso(row[1]), "effective_to": _iso(row[2]),
            "ind_nfe": row[3], "ind_comunica": row[4], "ind_transp": row[5], "ind_devol": row[6],
            "ind_retor": row[7], "ind_anula": row[8], "ind_remes": row[9], "ind_comb": row[10],
            "ind_exc_ibscbs": row[11], "title": row[12] or "", "description": row[13],
        })
    if len(entries) < 100:
        raise ValueError(f"expected at least 100 CFOP codes, found {len(entries)}")
    return Payload(sha256(raw_bytes).hexdigest(), tuple(entries), "cfop-xlsx-v1")


# --- FCP ----------------------------------------------------------------
def parse_fcp(raw_bytes: bytes) -> Payload:
    wb = _load(raw_bytes)
    if "FCP" not in wb.sheetnames:
        raise ValueError(f"expected an 'FCP' sheet, found {wb.sheetnames}")
    header, rows = _rows(wb["FCP"], 8)
    if "UF" not in [str(h) for h in header]:
        raise ValueError(f"unexpected FCP header: {header}")
    entries = []
    for row in rows:
        if not row[1]:
            continue
        entries.append({
            "uf_code": str(row[0]), "uf": str(row[1]).strip(), "uf_name": str(row[2] or "").strip(),
            "aliq_1": row[3], "aliq_2": row[4], "aliq_3": row[5], "observacao": row[7],
        })
    if len(entries) < 20:
        raise ValueError(f"expected at least 20 UF rows (26 states + DF), found {len(entries)}")
    return Payload(sha256(raw_bytes).hexdigest(), tuple(entries), "fcp-xlsx-v1")


# --- ANP ------------------------------------------------------------------
def parse_anp(raw_bytes: bytes) -> Payload:
    wb = _load(raw_bytes)
    if "cProdANP" not in wb.sheetnames:
        raise ValueError(f"expected a 'cProdANP' sheet, found {wb.sheetnames}")
    ws = wb["cProdANP"]
    rows = ws.iter_rows(values_only=True)
    header = next(rows)
    if "cProdANP" not in str(header[0]):
        raise ValueError(f"unexpected ANP header: {header}")
    entries = []
    seen = set()
    for row in rows:
        if row[0] is None:
            continue
        code = str(row[0]).strip()
        if code in seen:
            continue
        seen.add(code)
        entries.append({"cprod_anp": code})
    if len(entries) < 100:
        raise ValueError(f"expected at least 100 cProdANP codes, found {len(entries)}")
    return Payload(sha256(raw_bytes).hexdigest(), tuple(entries), "anp-xlsx-v1")


# --- NCM x uTrib ------------------------------------------------------
def parse_ncm_utrib(raw_bytes: bytes) -> Payload:
    wb = _load(raw_bytes)
    sheet_name = next((s for s in wb.sheetnames if "uTrib" in s or "utrib" in s.lower()), None)
    if sheet_name is None:
        raise ValueError(f"expected a sheet with 'uTrib' in its name, found {wb.sheetnames}")
    ws = wb[sheet_name]
    rows = ws.iter_rows(values_only=True)
    header = next(rows)
    if "NCM" not in str(header[0]):
        raise ValueError(f"unexpected NCM x uTrib header: {header}")
    entries = []
    for row in rows:
        if row[0] is None:
            continue
        entries.append({
            "ncm": _digits(row[0]), "effective_from": _iso(row[1]), "effective_to": _iso(row[2]),
            "utrib_abbreviation": row[3], "utrib_description": row[4],
        })
    if len(entries) < 1000:
        raise ValueError(f"expected at least 1000 NCM x uTrib rows, found {len(entries)}")
    return Payload(sha256(raw_bytes).hexdigest(), tuple(entries), "ncm-utrib-xlsx-v1")


# --- RTC biofuel index (pBioObrigatorio) --------------------------------
def parse_rtc_biofuel_index(raw_bytes: bytes) -> Payload:
    wb = _load(raw_bytes)
    current_sheet = next((s for s in wb.sheetnames if "Historico" not in s and "pBio" in s), None)
    historical_sheet = next((s for s in wb.sheetnames if "Historico" in s), None)
    if current_sheet is None:
        raise ValueError(f"expected a 'pBioObrigatorio' sheet, found {wb.sheetnames}")

    entries = []
    for sheet_name, is_historical in ((current_sheet, False), (historical_sheet, True)):
        if sheet_name is None:
            continue
        ws = wb[sheet_name]
        rows = ws.iter_rows(values_only=True)
        header = next(rows)
        if "cProdANP" not in str(header[0]):
            raise ValueError(f"unexpected biofuel index header in {sheet_name}: {header}")
        last_code, last_desc = None, None
        for row in rows:
            if row[0] is None and row[1] is None and row[2] is None:
                continue
            code = str(row[0]).strip() if row[0] is not None else last_code
            desc = row[1] if row[1] is not None else last_desc
            last_code, last_desc = code, desc
            entries.append({
                "cprod_anp": code, "description": desc, "p_bio_percent": row[2],
                "effective_from": _iso(row[3]), "effective_to": _iso(row[4]),
                "is_historical": is_historical,
            })
    if len(entries) < 5:
        raise ValueError(f"expected at least 5 biofuel index rows, found {len(entries)}")
    return Payload(sha256(raw_bytes).hexdigest(), tuple(entries), "rtc-biofuel-index-xlsx-v1")


# --- RTC cCredPres --------------------------------------------------------
def parse_rtc_cred_pres(raw_bytes: bytes) -> Payload:
    wb = _load(raw_bytes)
    if "cCredPres" not in wb.sheetnames:
        raise ValueError(f"expected a 'cCredPres' sheet, found {wb.sheetnames}")
    header, rows = _rows(wb["cCredPres"], 18)
    if "cCredPres" not in str(header[0]):
        raise ValueError(f"unexpected cCredPres header: {header}")
    entries = []
    for row in rows:
        if row[0] is None:
            continue
        entries.append({
            "code": str(int(row[0])) if isinstance(row[0], (int, float)) else str(row[0]).strip(),
            "description": row[1] or "", "legal_basis": row[2],
            "apropria_via_nf": bool(row[3]) if row[3] is not None else None,
            "apropria_via_evento": bool(row[4]) if row[4] is not None else None,
            "ind_deduz_cred_pres": row[5], "ind_g_cbs_cred_pres": row[6], "ind_g_ibs_cred_pres": row[7],
            "aliquota_cbs_text": row[8], "aliquota_ibs_text": row[9],
            "p_aliq_cred_pres_cbs": row[10], "p_aliq_cred_pres_ibs": row[11], "p_red_transicao_ibs": row[12],
            "cclass_nota_referenciada": str(row[13]) if row[13] is not None else None,
            "d_ini_vig_cbs": _iso(row[14]), "d_fim_vig_cbs": _iso(row[15]),
            "d_ini_vig_ibs": _iso(row[16]), "d_fim_vig_ibs": _iso(row[17]),
        })
    if len(entries) < 1:
        raise ValueError("expected at least 1 cCredPres row, found 0")
    return Payload(sha256(raw_bytes).hexdigest(), tuple(entries), "rtc-cred-pres-xlsx-v1")


def _txt(value):
    return None if value is None else str(value).strip()


def _code(value, width):
    text = _txt(value) or ""
    return text.zfill(width) if text.isdigit() else text


_CCLASS_COLUMNS = {
    "CST-IBS/CBS": "cst", "Descrição CST-IBS/CBS": "cst_description", "cClassTrib": "cclasstrib",
    "Nome cClassTrib": "name", "Descrição cClassTrib": "description", "LC Redação": "lc_redacao",
    "LC 214/25": "lc_214", "Regulamento CBS": "regulamento_cbs", "Regulamento IBS": "regulamento_ibs",
    "Tipo de Alíquota": "tipo_aliquota", "pRedIBS": "p_red_ibs", "pRedCBS": "p_red_cbs",
    "dIniVig": "d_ini_vig", "dFimVig": "d_fim_vig", "DataAtualização": "data_atualizacao",
    "ANEXO": "anexo", "Link": "link",
}


def parse_rtc_cclasstrib(raw_bytes: bytes) -> Payload:
    """entries = cClassTrib rows; extra_entries = CST-IBS/CBS sheet rows."""
    wb = _load(raw_bytes)
    cst_sheet = next((n for n in wb.sheetnames if n.startswith("CST")), None)
    cclass_sheet = next((n for n in wb.sheetnames if n.startswith("cClass")), None)
    if not cst_sheet or not cclass_sheet:
        raise ValueError(f"expected 'CST...' and 'cClass...' sheets, found {wb.sheetnames}")

    rows = wb[cst_sheet].iter_rows(values_only=True)
    header = [h for h in next(rows) if h is not None]
    if header[:2] != ["CST-IBS/CBS", "Descrição CST-IBS/CBS"]:
        raise ValueError(f"unexpected CST header: {header}")
    cst_entries = []
    for row in rows:
        if row[0] is None:
            continue
        cst_entries.append({
            "cst": _code(row[0], 3), "description": _txt(row[1]) or "",
            "indicators": {h.replace(" ", ""): _txt(v) for h, v in zip(header[2:], row[2:])},
        })

    rows = wb[cclass_sheet].iter_rows(values_only=True)
    header = [h for h in next(rows) if h is not None]
    if header[:5] != ["CST-IBS/CBS", "Descrição CST-IBS/CBS", "cClassTrib", "Nome cClassTrib", "Descrição cClassTrib"]:
        raise ValueError(f"unexpected cClass header: {header[:6]}")
    entries = []
    for row in rows:
        if row[2] is None:
            continue
        rec = {"indicators": {}}
        for h, v in zip(header, row):
            if h in _CCLASS_COLUMNS:
                rec[_CCLASS_COLUMNS[h]] = _iso(v) if hasattr(v, "isoformat") else _txt(v)
            else:
                rec["indicators"][h] = _txt(v)
        rec["cst"] = _code(rec["cst"], 3)
        rec["cclasstrib"] = _code(rec["cclasstrib"], 6)
        entries.append(rec)
    if len(entries) < 100 or len(cst_entries) < 10:
        raise ValueError(f"too few rows: cClass={len(entries)} CST={len(cst_entries)}")
    orphans = {e["cst"] for e in entries} - {e["cst"] for e in cst_entries}
    if orphans:
        raise ValueError(f"cClassTrib rows reference unknown CST codes: {sorted(orphans)}")
    return Payload(sha256(raw_bytes).hexdigest(), tuple(entries), "rtc-cclasstrib-xlsx-v1", tuple(cst_entries))


def parse_rtc_cbs_rates(raw_bytes: bytes) -> Payload:
    wb = _load(raw_bytes)
    rows = list(wb.worksheets[0].iter_rows(values_only=True))
    if len(rows) < 3 or "CBS" not in str(rows[0][0]).upper() or [str(c).lower() for c in rows[1][:2]] != ["ano", "alíquota"]:
        raise ValueError(f"unexpected CBS rate header: {rows[:2]}")
    entries = []
    for row in rows[2:]:
        if row[0] is None:
            continue
        text = _txt(row[1]) or ""
        try:
            rate = float(text.replace(",", "."))
        except ValueError:
            rate = None
        entries.append({"year": int(float(str(row[0]))), "rate_text": text, "rate": rate})
    if not entries or all(e["rate"] is None for e in entries):
        raise ValueError("no numeric CBS rate found")
    if len({e["year"] for e in entries}) != len(entries):
        raise ValueError("duplicate CBS rate year")
    return Payload(sha256(raw_bytes).hexdigest(), tuple(entries), "rtc-cbs-rates-xlsx-v1")