"""Import a parsed NetSuite report export into the finance-flux tables.

Idempotent per (period, statement): re-importing the same report replaces that
period+statement's line values wholesale, so a corrected re-export is safe. The
report is the source of truth, so the statement-layout catalog
(``finance_flux_line_defs``) is reconciled to whatever lines the report
actually contains — labels, sections, ordering, and line/subtotal/total roles
all come from the export, not a hand-maintained guess.
"""
from __future__ import annotations

import re
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime
from zoneinfo import ZoneInfo

from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.finance_flux import constants as c
from app.domains.finance_flux.models import (
    FinanceFluxLine,
    FinanceFluxLineDef,
    FinanceFluxLineDetail,
    FinanceFluxPeriod,
)
from app.domains.finance_flux.report_parser import (
    ParsedReport,
    ReportParseError,
    parse_report,
    parse_reports,
    slug_for_label,
)


@dataclass
class ImportResult:
    period_key: str
    period_label: str
    statement: str
    line_count: int
    created_period: bool


async def _upsert_period(
    db: AsyncSession, parsed: ParsedReport,
) -> tuple[FinanceFluxPeriod, bool]:
    row = (await db.execute(
        select(FinanceFluxPeriod).where(
            FinanceFluxPeriod.period_key == parsed.period_key)
    )).scalar_one_or_none()
    now = datetime.now(UTC)
    if row is None:
        row = FinanceFluxPeriod(
            id=uuid.uuid4(),
            period_key=parsed.period_key,
            label=parsed.period_label,
            is_closed=True,            # an exported report is a closed snapshot
            sync_status=c.SYNC_SYNCED,
            last_synced_at=now,
        )
        db.add(row)
        await db.flush()
        return row, True
    row.label = parsed.period_label
    row.sync_status = c.SYNC_SYNCED
    row.last_synced_at = now
    row.last_sync_error = None
    return row, False


# Sign-variant P&L summary lines slug differently each period ("Net loss" vs
# "Net income", "Operating profit" vs "Operating loss"), which would break the
# KPI ``source_line_key`` references and re-imports across profit/loss months.
# Map them to stable canonical keys (the ones the seed KPI defs already point
# at) so the KPI cards always resolve regardless of the period's sign.
_CANONICAL_KEYS = {
    "net income": "net_income", "net loss": "net_income", "net profit": "net_income",
    "operating income": "operating_income", "operating loss": "operating_income",
    "operating profit": "operating_income",
    "income before tax": "income_before_tax", "loss before tax": "income_before_tax",
    "profit before tax": "income_before_tax",
}


def _canonical_key(label: str) -> str | None:
    s = re.sub(r"[^a-z ]+", " ", label.lower())
    s = re.sub(r"\s+", " ", s).strip()
    return _CANONICAL_KEYS.get(s)


async def _resolve_keys(
    db: AsyncSession, parsed: ParsedReport,
) -> dict[int, str]:
    """Map each parsed row (by id) → its line_key.

    A report label that matches an existing line_def's label reuses that
    def's key (so the PR1 seed catalog / KPI source refs keep matching); a
    sign-variant P&L summary line maps to its canonical key; otherwise a
    genuinely new line gets a slug. Reconciles line_defs to the report
    (label/section/order/role are report-truth) while preserving finance's
    owner + per-line threshold overrides.
    """
    existing = (await db.execute(
        select(FinanceFluxLineDef).where(
            FinanceFluxLineDef.statement == parsed.statement)
    )).scalars().all()
    by_label = {d.label.strip().casefold(): d for d in existing}
    by_key = {d.line_key: d for d in existing}

    keys: dict[int, str] = {}
    for r in parsed.rows:
        # Canonical key wins FIRST for sign-variant P&L summary lines, even over
        # a label match — otherwise a prior import's "Operating profit" def
        # (key operating_profit) would shadow the canonical operating_income the
        # KPIs reference. Else match by label; else reuse a def already
        # occupying the slug (seeded catalog / slug collision — e.g. seed
        # "Total stockholders' equity" vs report "Total stockholders equity"
        # both slug to ``total_stockholders_equity``); else create under slug.
        canon = _canonical_key(r.label)
        if canon:
            key = canon
            d = by_key.get(canon)
        else:
            d = by_label.get(r.label.strip().casefold())
            if d is None:
                d = by_key.get(slug_for_label(r.label))
            key = d.line_key if d is not None else slug_for_label(r.label)
        keys[id(r)] = key
        if d is None:
            d = FinanceFluxLineDef(
                id=uuid.uuid4(), statement=parsed.statement, line_key=key,
                label=r.label, section=r.section, row_type=r.row_type,
                sort_order=r.order, is_active=True,
            )
            db.add(d)
            by_label[r.label.strip().casefold()] = d
            by_key[key] = d
        else:
            d.label = r.label
            d.section = r.section
            d.row_type = r.row_type
            d.sort_order = r.order
            by_label[r.label.strip().casefold()] = d
    await db.flush()
    return keys


async def _replace_lines(
    db: AsyncSession, period: FinanceFluxPeriod, parsed: ParsedReport,
    keys: dict[int, str],
) -> int:
    await db.execute(
        delete(FinanceFluxLine).where(
            FinanceFluxLine.period_id == period.id,
            FinanceFluxLine.statement == parsed.statement,
        )
    )
    await db.execute(
        delete(FinanceFluxLineDetail).where(
            FinanceFluxLineDetail.period_id == period.id,
            FinanceFluxLineDetail.statement == parsed.statement,
        )
    )
    seen: set[str] = set()
    n = 0
    for r in parsed.rows:
        key = keys[id(r)]
        if key in seen:          # collapse accidental label collisions
            continue
        seen.add(key)
        db.add(FinanceFluxLine(
            id=uuid.uuid4(), period_id=period.id, statement=parsed.statement,
            line_key=key, value=r.amount,
        ))
        n += 1
        # Persist the line's GL/sub-account breakdown for drill-down.
        for i, (label, value) in enumerate(r.details):
            db.add(FinanceFluxLineDetail(
                id=uuid.uuid4(), period_id=period.id, statement=parsed.statement,
                parent_line_key=key, label=label[:255], value=value, sort_order=i * 10,
            ))
    await db.flush()
    return n


async def import_report(
    db: AsyncSession, content: bytes, *, statement_override: str | None = None,
) -> ImportResult:
    """Parse + import a NetSuite report export. Caller commits."""
    parsed = parse_report(content, statement_override=statement_override)
    period, created = await _upsert_period(db, parsed)
    keys = await _resolve_keys(db, parsed)
    n = await _replace_lines(db, period, parsed, keys)
    return ImportResult(
        period_key=parsed.period_key, period_label=parsed.period_label,
        statement=parsed.statement, line_count=n, created_period=created,
    )


async def import_reports(
    db: AsyncSession, content: bytes, *,
    statement_override: str | None = None,
    as_of: datetime | None = None,
) -> list[ImportResult]:
    """Import a single- or multi-period export. Caller commits.

    A multi-period columnar export (the FY P&L) imports one period per month
    column, but only **closed** periods — strictly before the current month —
    honouring the closed-periods-only scope (the FY report also carries the
    current/future months, which aren't closed yet). A single-period export is
    an explicit snapshot the user chose to upload, so it's imported as-is.
    """
    reports = parse_reports(content, statement_override=statement_override)
    # A columnar (multi-period FY) export carries the current/future months and
    # empty columns; a single-period export is an explicit snapshot the user
    # chose to upload, so neither the closed-period gate nor the all-zero skip
    # apply to it. Drive both off the parse origin, not the surviving count.
    columnar = bool(reports) and reports[0].from_columnar
    if as_of is None:
        # #1796: anchor "current month" to Asia/Shanghai, not UTC. A columnar
        # FY P&L uploaded on the 1st at 07:00 Beijing is still the previous
        # day in UTC, so a UTC ``current_key`` equalled the just-closed month
        # and the ``period_key >= current_key`` guard silently dropped that
        # freshly-closed month (import reported success, flux view stuck on
        # stale data). PR #626 fixed the same UTC/Shanghai month-boundary
        # class in ai_spending. Callers may still pass an explicit ``as_of``.
        as_of = datetime.now(ZoneInfo("Asia/Shanghai"))
    current_key = f"{as_of.year:04d}-{as_of.month:02d}"

    out: list[ImportResult] = []
    for parsed in reports:
        if columnar and parsed.period_key >= current_key:
            continue  # current month + future aren't closed
        if columnar and all(r.amount == 0 for r in parsed.rows):
            continue  # a period column with no activity
        period, created = await _upsert_period(db, parsed)
        keys = await _resolve_keys(db, parsed)
        n = await _replace_lines(db, period, parsed, keys)
        out.append(ImportResult(
            period_key=parsed.period_key, period_label=parsed.period_label,
            statement=parsed.statement, line_count=n, created_period=created,
        ))
    if not out:
        raise ReportParseError(
            "no closed periods with data were found in the export")
    return out