"""Parser for NetSuite financial-report exports (SpreadsheetML / "Excel").

Why this exists: NetSuite computes consolidation (intercompany elimination +
cumulative translation adjustment) inside its report engine. That cannot be
faithfully reproduced from raw GL via SuiteQL — verified 2026-06-27 that a raw
``SUM(transactionaccountingline.amount)`` over-counts ~6× because intercompany
transfers/capital injections don't net the way consolidation does. So the
source of truth for the flux figures is the report's own export, taken at the
consolidated subsidiary context (PingCap Ltd. 已合并) in USD.

The "Excel" export NetSuite produces is **SpreadsheetML 2003 XML**
(``urn:schemas-microsoft-com:office:spreadsheet``): ``Worksheet/Table/Row/Cell/
Data``. Hierarchy (section → line → GL account detail → subtotal) is conveyed by
row order + per-cell StyleID. The consolidated, USD, per-finance-line values are
exactly the rows whose label is a clean finance-line name (not a GL account —
those start with a digit — and not a ``总计`` GL subtotal) carrying a numeric
amount. Verified against the live May-2026 consolidated BS: Cash = 32,228,212.53
(matches the legacy portal to the cent), Total assets balances Total L&E.
"""
from __future__ import annotations

import re
from dataclasses import dataclass, field
from decimal import Decimal, InvalidOperation
from xml.etree import ElementTree as ET

from defusedxml.common import DefusedXmlException

# Uploads are untrusted input — defusedxml hardens parsing against
# billion-laughs / external-entity (XXE) attacks; ``ET`` is kept only for the
# Element API + the ParseError type.
from defusedxml.ElementTree import fromstring as _safe_fromstring

from app.domains.finance_flux import constants as c

_SS = "urn:schemas-microsoft-com:office:spreadsheet"
_NS = {"ss": _SS}
_STYLE_ATTR = f"{{{_SS}}}StyleID"

# month-name → 2-digit month, for the period header ("May,2026 末")
_MONTHS = {
    m.lower(): f"{i:02d}" for i, m in enumerate(
        ["January", "February", "March", "April", "May", "June", "July",
         "August", "September", "October", "November", "December"], start=1)
}
_PERIOD_RE = re.compile(r"([A-Za-z]+)\s*,\s*(\d{4})")

# A label is GL-account-level detail (skip) when it starts with a digit
# (account number) or is a NetSuite ``总计``/``Total - <account>`` GL roll-up.
_ACCOUNT_PREFIX_RE = re.compile(r"^\s*\d")
_GL_SUBTOTAL_RE = re.compile(r"^\s*(总计|Total\s*-)\s*", re.IGNORECASE)
# Consolidation sub-components inside a line (elimination entries, SWC coded
# sub-rows like ``S924 - …`` / ``Elim-1704 - …``). They net into the line total
# the report already shows, so they're below the flux-line view — skip them.
_SUBCOMPONENT_RE = re.compile(r"^\s*(Elim[-\s]|S\d+\s*-)", re.IGNORECASE)
# A finance-line-level row is a statement total/subtotal when its label reads
# like one. Covers BS ("Total …", "… assets/liabilities"), PL ("Gross profit",
# "Operating loss/income", "Net loss/income"), and the equity roll-ups.
_TOTAL_RE = re.compile(
    r"^\s*(total\b|grand total\b|gross profit\b|"
    r"operating (loss|income|profit)\b|net (loss|income|profit)\b|"
    r"(loss|profit) before tax\b|"
    r"(current |non-current )?(assets|liabilities)\b|"
    r"(total )?stockholders|accumulated deficit\b)",
    re.IGNORECASE)
# ``总计 - <X>`` is a NetSuite roll-up. The P&L groups its category totals this
# way (``总计 -   Revenue- License&Subscription``); when <X> is a finance-line
# name it IS the line, when <X> is a GL account (digit-prefixed) it's detail.
_TOTAL_PREFIX_RE = re.compile(r"^\s*总计\s*[-－]\s*(.*)$")
# Section banners (no amount) that open a block; tracked for display grouping.
_SECTION_RE = re.compile(
    r"^\s*(assets|liabilities|stockholders|equity|revenue|cost of revenue|"
    r"operating expenses|other income|current assets|non-current assets|"
    r"current liabilities|non-current liabilities|"
    r"liabilities and stockholders)", re.IGNORECASE)


class ReportParseError(ValueError):
    """Raised when an upload isn't a parseable NetSuite report export."""


@dataclass(frozen=True)
class ReportRow:
    """One finance-line-level row from the report (line / subtotal / total)."""

    label: str
    amount: Decimal
    row_type: str          # c.ROW_LINE / ROW_SUBTOTAL / ROW_TOTAL
    section: str | None     # nearest enclosing section header
    order: int              # position in the statement (for sort_order)
    # GL/sub-account breakdown beneath this line (drill-down). Mutable list on a
    # frozen row: we append during parse but never reassign the attribute.
    details: list[tuple[str, Decimal]] = field(default_factory=list)


@dataclass
class ParsedReport:
    period_label: str
    period_key: str               # "YYYY-MM"
    statement: str                # c.STATEMENT_BS / STATEMENT_PL
    subsidiary_context: str | None
    rows: list[ReportRow] = field(default_factory=list)
    # True when this came from a multi-period columnar export (the FY P&L),
    # so the importer knows to apply the closed-periods-only filter even if
    # only one column survived parsing. Single-period exports leave it False.
    from_columnar: bool = False


def _cell_texts(row: ET.Element) -> list[str | None]:
    out: list[str | None] = []
    for cell in row.findall("ss:Cell", _NS):
        data = cell.find("ss:Data", _NS)
        out.append(data.text if data is not None else None)
    return out


def _sanitize_xml(content: bytes) -> bytes:
    """Escape bare ``&`` so the SpreadsheetML parses.

    NetSuite embeds raw ``&`` in report titles/labels (``Consolidated P&L``,
    ``Revenue- License&Subscription``) without escaping it, which makes the
    XML malformed — ``defusedxml`` raises and the upload 422s. Escape any
    ``&`` that isn't already the start of an entity (``&amp;`` / ``&#39;``).
    """
    text = content.decode("utf-8", "replace")
    text = re.sub(r"&(?!#?\w+;)", "&amp;", text)
    return text.encode("utf-8")


def _to_decimal(raw: str) -> Decimal | None:
    """Parse a report amount (incl. scientific notation like 5.0E-8)."""
    try:
        return Decimal(raw.strip().replace(",", "")).quantize(Decimal("0.01"))
    except (InvalidOperation, AttributeError):
        return None


def _detect_statement(title: str, override: str | None) -> str:
    if override in c.STATEMENTS:
        return override
    t = (title or "").lower()
    if "income" in t or "profit" in t or "p&l" in t or "损益" in t or "利润" in t:
        return c.STATEMENT_PL
    # default to BS (covers "Balance Sheet" and the analysis-detail variant)
    return c.STATEMENT_BS


def _classify(label: str) -> str:
    return c.ROW_TOTAL if _TOTAL_RE.match(label) else c.ROW_LINE


def parse_report(
    content: bytes, *, statement_override: str | None = None,
) -> ParsedReport:
    """Parse a NetSuite SpreadsheetML report export into finance-line rows.

    ``statement_override`` forces BS/PL when the title is ambiguous (e.g. the
    BS Analysis Detail report). Raises :class:`ReportParseError` on anything
    that isn't a NetSuite report export.
    """
    try:
        root = _safe_fromstring(_sanitize_xml(content))
    except (ET.ParseError, DefusedXmlException) as e:
        raise ReportParseError(f"not a SpreadsheetML document: {e}") from e
    rows_xml = root.findall(".//ss:Worksheet/ss:Table/ss:Row", _NS)
    if not rows_xml:
        raise ReportParseError("no worksheet rows found")

    title = ""
    period_label = period_key = ""
    subsidiary_context: str | None = None
    # First few rows are metadata: company, subsidiary context, title, period.
    for row in rows_xml[:6]:
        texts = [t for t in _cell_texts(row) if t]
        if not texts:
            continue
        head = texts[0].strip()
        if "已合并" in head or "Consolidated" in head or ":" in head and (
            "PingCap" in head or "母公司" in head
        ):
            subsidiary_context = subsidiary_context or head
        if not title and ("Sheet" in head or "Statement" in head
                          or "Income" in head or "Balance" in head):
            title = head
        m = _PERIOD_RE.search(head)
        if m and not period_key:
            mon = _MONTHS.get(m.group(1).lower())
            if mon:
                period_label = f"{m.group(1)},{m.group(2)}"
                period_key = f"{m.group(2)}-{mon}"
    if not period_key:
        raise ReportParseError(
            "could not find a 'Month,YYYY' period header in the export")

    statement = _detect_statement(title, statement_override)

    out_rows: list[ReportRow] = []
    section: str | None = None
    order = 0
    # Leaf GL-account rows accumulate here until the finance line they belong
    # to (which follows them, carrying the net total) is emitted — then they're
    # attached as that line's drill-down detail.
    detail_buf: list[tuple[str, Decimal]] = []
    for row in rows_xml:
        texts = _cell_texts(row)
        if not texts or not texts[0]:
            continue
        label = texts[0].strip()
        amount_raw = texts[1] if len(texts) >= 2 else None
        amount = _to_decimal(amount_raw) if amount_raw else None

        # Leaf GL-account detail (digit-prefixed, not a 总计 roll-up) → buffer as
        # the current line's sub-account breakdown.
        if _ACCOUNT_PREFIX_RE.match(label) and not _GL_SUBTOTAL_RE.match(label):
            if amount is not None:
                detail_buf.append((label, amount))
            continue
        # GL roll-ups (总计 - <gl>) and consolidation sub-components: drop.
        if _GL_SUBTOTAL_RE.match(label) or _SUBCOMPONENT_RE.match(label):
            continue

        if amount is None:
            # No amount → a section banner that opens a block. Track only true
            # section banners so it groups lines correctly (a line's own opener
            # header repeats with an amount later, so it isn't mistaken here).
            if _SECTION_RE.match(label):
                section = label
                detail_buf = []
            continue

        row_type = _classify(label)
        out_rows.append(ReportRow(
            label=label, amount=amount, row_type=row_type, section=section,
            order=order,
            # Only finance lines carry a sub-account breakdown; subtotals/totals
            # don't absorb stray GL rows (which would be misattributed).
            details=detail_buf if row_type == c.ROW_LINE else [],
        ))
        detail_buf = []
        order += 10

    if not out_rows:
        raise ReportParseError("no finance-line rows with amounts were found")
    return ParsedReport(
        period_label=period_label, period_key=period_key, statement=statement,
        subsidiary_context=subsidiary_context, rows=out_rows,
    )


def _detect_period_columns(
    rows_xml: list[ET.Element],
) -> dict[int, tuple[str, str]]:
    """Find a columnar period header (one column per accounting period).

    The multi-period exports (e.g. the FY P&L) carry a header row like
    ``Finance Line | Department… | April,2026 | May,2026 | …``. Returns
    ``{col_index: (label, period_key)}`` for the header row with the most
    ``Month,YYYY`` cells beyond column 0. Empty when the export is
    single-period (the period lives in a metadata cell, not column headers).
    """
    best: dict[int, tuple[str, str]] = {}
    for row in rows_xml[:15]:
        cols: dict[int, tuple[str, str]] = {}
        for idx, t in enumerate(_cell_texts(row)):
            if idx == 0 or not t:
                continue
            m = _PERIOD_RE.search(t)
            if not m:
                continue
            mon = _MONTHS.get(m.group(1).lower())
            if mon:
                cols[idx] = (f"{m.group(1)},{m.group(2)}", f"{m.group(2)}-{mon}")
        if len(cols) > len(best):
            best = cols
    # Two+ period columns ⇒ genuinely multi-period; a lone match is more likely
    # a single-period report whose header merely echoes the period.
    return best if len(best) >= 2 else {}


def parse_reports(
    content: bytes, *, statement_override: str | None = None,
) -> list[ParsedReport]:
    """Parse a NetSuite export into one :class:`ParsedReport` per period.

    Single-period exports (the BS "no comparatives") yield one report;
    multi-period columnar exports (the FY P&L, one column per month) yield one
    report per period column. Raises :class:`ReportParseError` on non-reports.
    """
    try:
        root = _safe_fromstring(_sanitize_xml(content))
    except (ET.ParseError, DefusedXmlException) as e:
        raise ReportParseError(f"not a SpreadsheetML document: {e}") from e
    rows_xml = root.findall(".//ss:Worksheet/ss:Table/ss:Row", _NS)
    if not rows_xml:
        raise ReportParseError("no worksheet rows found")
    period_cols = _detect_period_columns(rows_xml)
    if not period_cols:
        return [parse_report(content, statement_override=statement_override)]
    return _parse_columnar(rows_xml, period_cols, statement_override)


def _parse_columnar(
    rows_xml: list[ET.Element],
    period_cols: dict[int, tuple[str, str]],
    statement_override: str | None,
) -> list[ParsedReport]:
    """Parse a multi-period columnar export into one report per period column.

    Differs from the single-period path in two ways: amounts are read from the
    period column (not column 1), and a ``总计 - <finance-line name>`` row is
    kept as that line (P&L groups its category totals this way) while
    ``总计 - <GL account>`` stays a skipped roll-up.
    """
    title = ""
    subsidiary_context: str | None = None
    for row in rows_xml[:8]:
        texts = [t for t in _cell_texts(row) if t]
        if not texts:
            continue
        head = texts[0].strip()
        if subsidiary_context is None and (
            "已合并" in head or "Consolidated" in head
            or (":" in head and ("PingCap" in head or "母公司" in head))
        ):
            subsidiary_context = head
        if not title and any(k in head for k in (
            "Sheet", "Statement", "Income", "Balance", "P&L", "Profit",
            "损益", "利润",
        )):
            title = head
    statement = _detect_statement(title, statement_override)

    per_period: dict[str, list[ReportRow]] = {
        key: [] for (_lbl, key) in period_cols.values()
    }
    section: str | None = None
    order = 0
    # Leaf GL rows buffer here (with their per-period values) until the line
    # they belong to is emitted, then attach as that line's drill-down detail.
    detail_buf: list[tuple[str, dict[str, Decimal | None]]] = []
    for row in rows_xml:
        texts = _cell_texts(row)
        if not texts or not texts[0]:
            continue
        label0 = texts[0].strip()
        if not label0 or label0 == "\xa0":
            continue
        # Consolidation sub-components (Elim-/S\d) are dropped outright.
        if _SUBCOMPONENT_RE.match(label0):
            continue
        # Leaf GL-account detail (digit-prefixed) → buffer per-period values.
        if _ACCOUNT_PREFIX_RE.match(label0):
            dvals = {
                key: (_to_decimal(texts[idx]) if idx < len(texts) and texts[idx] else None)
                for idx, (_l, key) in period_cols.items()
            }
            if any(v is not None for v in dvals.values()):
                detail_buf.append((label0, dvals))
            continue
        mt = _TOTAL_PREFIX_RE.match(label0)
        if mt:
            inner = mt.group(1).strip()
            if not inner or _ACCOUNT_PREFIX_RE.match(inner):
                continue  # ``总计 - <GL account>`` roll-up
            label = inner
        elif _GL_SUBTOTAL_RE.match(label0):
            continue  # English ``Total - <GL account>`` roll-up
        else:
            label = label0

        amounts: dict[str, Decimal | None] = {}
        any_amount = False
        for idx, (_lbl, key) in period_cols.items():
            raw = texts[idx] if idx < len(texts) else None
            val = _to_decimal(raw) if raw else None
            amounts[key] = val
            if val is not None:
                any_amount = True
        if not any_amount:
            if _SECTION_RE.match(label0):
                section = label0
                detail_buf = []
            continue

        row_type = _classify(label)
        for key, val in amounts.items():
            rr = ReportRow(
                label=label, amount=val if val is not None else Decimal("0.00"),
                row_type=row_type, section=section, order=order,
            )
            # Only finance lines carry a sub-account breakdown (per-period value).
            if row_type == c.ROW_LINE:
                for dlabel, dvals in detail_buf:
                    dv = dvals.get(key)
                    rr.details.append((dlabel, dv if dv is not None else Decimal("0.00")))
            per_period[key].append(rr)
        detail_buf = []
        order += 10

    reports: list[ParsedReport] = []
    for _idx, (plabel, key) in period_cols.items():
        rows = per_period[key]
        if rows:
            reports.append(ParsedReport(
                period_label=plabel, period_key=key, statement=statement,
                subsidiary_context=subsidiary_context, rows=rows,
                from_columnar=True,
            ))
    if not reports:
        raise ReportParseError("no finance-line rows with amounts were found")
    return reports


def slug_for_label(label: str) -> str:
    """Stable line_key from a report label (matches PR1 seed keys where possible)."""
    s = label.strip().lower()
    s = re.sub(r"[^a-z0-9]+", "_", s).strip("_")
    return s[:64] or "line"