from __future__ import annotations

import hashlib
import json
from collections.abc import Iterable, Mapping
from dataclasses import asdict, dataclass
from datetime import UTC, datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any, Protocol

from google.cloud import storage

from .netsuite_revenue_config import RevenueSyncConfig


class CandidateContractError(ValueError):
    pass


@dataclass(frozen=True)
class CandidateRow:
    source_key: str
    source_identity: dict[str, str | None]
    payload: dict[str, Any]
    payload_hash: str
    raw_account_id: str | None
    display_account_id: str | None
    admission_evidence_available: bool

    @property
    def account_id(self) -> str | None:
        # The RESTlet preserves these fields before its Excel compatibility
        # conversion. Literal "44" remains a legitimate nonblank ID here.
        return _nonblank(self.raw_account_id) or _nonblank(self.display_account_id)


@dataclass(frozen=True)
class CandidateManifest:
    run_id: str
    posting_period_id: str
    source_total: int
    source_hash: str
    page_count: int
    headers: tuple[str, ...]
    control_totals: dict[str, str]
    fixture: bool
    created_at: str
    object_prefix: str


class CandidateStore(Protocol):
    @property
    def prefix(self) -> str: ...

    def write_json(self, name: str, value: Any) -> str: ...

    def write_jsonl(self, name: str, values: Iterable[Mapping[str, Any]]) -> str: ...

    def read_json(self, name: str) -> Any: ...

    def read_jsonl(self, name: str) -> list[dict[str, Any]]: ...


class LocalCandidateStore:
    def __init__(self, root: str | Path, *, run_id: str, create: bool = True) -> None:
        self.root = Path(root).resolve() / f"run_id={run_id}"
        if create:
            self.root.mkdir(parents=True, exist_ok=False)
        elif not self.root.is_dir():
            raise FileNotFoundError(f"candidate run directory does not exist: {self.root}")

    @classmethod
    def from_run_directory(cls, directory: str | Path) -> LocalCandidateStore:
        path = Path(directory).resolve()
        instance = cls.__new__(cls)
        instance.root = path
        if not instance.root.is_dir():
            raise FileNotFoundError(f"candidate run directory does not exist: {instance.root}")
        return instance

    @property
    def prefix(self) -> str:
        return str(self.root)

    def write_json(self, name: str, value: Any) -> str:
        path = self.root / name
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("x", encoding="utf-8") as handle:
            json.dump(value, handle, sort_keys=True, ensure_ascii=False, separators=(",", ":"), default=str)
            handle.write("\n")
        return str(path)

    def write_jsonl(self, name: str, values: Iterable[Mapping[str, Any]]) -> str:
        path = self.root / name
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("x", encoding="utf-8") as handle:
            for value in values:
                handle.write(json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":"), default=str))
                handle.write("\n")
        return str(path)

    def read_json(self, name: str) -> Any:
        return json.loads((self.root / name).read_text())

    def read_jsonl(self, name: str) -> list[dict[str, Any]]:
        return [json.loads(line) for line in (self.root / name).read_text().splitlines() if line]


class GcsCandidateStore:
    def __init__(self, uri: str, *, run_id: str, client: storage.Client | None = None) -> None:
        if not uri.startswith("gs://"):
            raise ValueError("candidate GCS URI must begin with gs://")
        bucket_and_prefix = uri.removeprefix("gs://").rstrip("/").split("/", 1)
        self.bucket_name = bucket_and_prefix[0]
        self.base_prefix = bucket_and_prefix[1] if len(bucket_and_prefix) > 1 else ""
        self.run_prefix = "/".join(piece for piece in (self.base_prefix, f"run_id={run_id}") if piece)
        self.client = client or storage.Client()
        self.bucket = self.client.bucket(self.bucket_name)

    @property
    def prefix(self) -> str:
        return f"gs://{self.bucket_name}/{self.run_prefix}/"

    def _blob(self, name: str):
        return self.bucket.blob(f"{self.run_prefix}/{name}")

    def write_json(self, name: str, value: Any) -> str:
        payload = json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":"), default=str) + "\n"
        blob = self._blob(name)
        blob.upload_from_string(payload, content_type="application/json", if_generation_match=0)
        return f"gs://{self.bucket_name}/{blob.name}"

    def write_jsonl(self, name: str, values: Iterable[Mapping[str, Any]]) -> str:
        payload = "".join(
            json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":"), default=str) + "\n"
            for value in values
        )
        blob = self._blob(name)
        blob.upload_from_string(payload, content_type="application/x-ndjson", if_generation_match=0)
        return f"gs://{self.bucket_name}/{blob.name}"

    def read_json(self, name: str) -> Any:
        return json.loads(self._blob(name).download_as_text())

    def read_jsonl(self, name: str) -> list[dict[str, Any]]:
        return [json.loads(line) for line in self._blob(name).download_as_text().splitlines() if line]


def freeze_restlet_candidate(
    config: RevenueSyncConfig,
    *,
    client: Any,
    store: CandidateStore,
    run_id: str,
    posting_period_id: str,
) -> tuple[list[CandidateRow], CandidateManifest]:
    cursor: str | None = None
    seen_cursors: set[str] = set()
    seen_keys: set[str] = set()
    rows: list[CandidateRow] = []
    page_count = 0
    while True:
        envelope = client.fetch_page(
            posting_period_id=posting_period_id,
            cursor=cursor,
            page_size=int(config.source["page_size"]),
        )
        validated_rows, next_cursor = validate_restlet_envelope(
            config,
            envelope,
            posting_period_id=posting_period_id,
            expected_cursor=cursor,
            seen_keys=seen_keys,
        )
        page_count += 1
        store.write_json(f"pages/page-{page_count:04d}.json", envelope)
        rows.extend(validated_rows)
        if next_cursor is None:
            break
        if next_cursor in seen_cursors or next_cursor == cursor:
            raise CandidateContractError("NetSuite RESTlet repeated a cursor")
        seen_cursors.add(next_cursor)
        cursor = next_cursor

    return _persist_candidate(
        config,
        store=store,
        run_id=run_id,
        posting_period_id=posting_period_id,
        rows=rows,
        page_count=page_count,
    )


def freeze_workbook_fixture(
    config: RevenueSyncConfig,
    *,
    workbook: str | Path,
    store: CandidateStore,
    run_id: str,
    posting_period_id: str,
) -> tuple[list[CandidateRow], CandidateManifest]:
    """Create a source-candidate fixture only; it cannot be committed to NexusCRM.

    The XLSX has the verified 35 business values but deliberately lacks RESTlet
    raw/display identity evidence and source keys, so it is only suitable for
    local candidate-generation acceptance.
    """
    from openpyxl import load_workbook

    book = load_workbook(workbook, read_only=True, data_only=True)
    try:
        sheet = book.active
        values = sheet.iter_rows(values_only=True)
        headers = tuple(_text(value) for value in next(values))
        expected_headers = tuple(item.header for item in config.mappings)
        if headers != expected_headers:
            raise CandidateContractError("workbook header order differs from the 35-field mapping")
        rows: list[CandidateRow] = []
        for position, value_row in enumerate(values, start=1):
            if len(value_row) != 35:
                raise CandidateContractError(f"workbook row {position} does not have 35 fields")
            payload = {mapping.field: value for mapping, value in zip(config.mappings, value_row, strict=True)}
            payload_hash = _payload_hash(payload)
            synthetic_key = f"fixture:{position}:{payload_hash[:16]}"
            rows.append(
                CandidateRow(
                    source_key=synthetic_key,
                    source_identity={"transactionInternalId": None, "lineUniqueKey": None, "accountingBookId": None},
                    payload=payload,
                    payload_hash=payload_hash,
                    raw_account_id=None,
                    display_account_id=None,
                    admission_evidence_available=False,
                )
            )
    finally:
        book.close()
    store.write_json("pages/workbook-fixture.json", {"workbook": Path(workbook).name, "row_count": len(rows)})
    return _persist_candidate(
        config,
        store=store,
        run_id=run_id,
        posting_period_id=posting_period_id,
        rows=rows,
        page_count=1,
        fixture=True,
    )


def validate_restlet_envelope(
    config: RevenueSyncConfig,
    envelope: Mapping[str, Any],
    *,
    posting_period_id: str,
    expected_cursor: str | None,
    seen_keys: set[str],
) -> tuple[list[CandidateRow], str | None]:
    source = _mapping(envelope.get("source"), "source")
    scope = _mapping(envelope.get("scope"), "scope")
    page = _mapping(envelope.get("page"), "page")
    if envelope.get("schemaVersion") != config.source["schema_version"]:
        raise CandidateContractError("NetSuite schemaVersion drift")
    if envelope.get("reportSchemaVersion") != config.source["report_schema_version"]:
        raise CandidateContractError("NetSuite reportSchemaVersion drift")
    if str(source.get("accountRealm")) != str(config.source["account_realm"]):
        raise CandidateContractError("NetSuite account realm drift")
    if source.get("savedSearchId") != config.source["saved_search_id"]:
        raise CandidateContractError("NetSuite Saved Search drift")
    if str(scope.get("postingPeriodId")) != str(posting_period_id):
        raise CandidateContractError("NetSuite posting-period scope drift")
    if scope.get("transactionType") != config.source["transaction_type"]:
        raise CandidateContractError("NetSuite transaction type scope drift")
    if str(scope.get("accountNumberPrefix")) != str(config.source["account_number_prefix"]):
        raise CandidateContractError("NetSuite account prefix scope drift")
    if int(page.get("pageSize", 0)) != int(config.source["page_size"]):
        raise CandidateContractError("NetSuite page size drift")
    actual_cursor = str(page.get("cursor"))
    required_cursor = expected_cursor if expected_cursor is not None else "0"
    if actual_cursor != required_cursor:
        raise CandidateContractError("NetSuite response cursor does not match the request")

    columns = envelope.get("columns")
    if not isinstance(columns, list):
        raise CandidateContractError("NetSuite columns are missing")
    headers = tuple(str(_mapping(item, "column").get("header", "")) for item in columns)
    expected_headers = tuple(mapping.header for mapping in config.mappings)
    if headers != expected_headers:
        raise CandidateContractError("NetSuite 35-column order drift")
    field_ids = {}
    for column in columns:
        descriptor = _mapping(column, "column")
        field_ids[str(descriptor["header"])] = str(descriptor.get("fieldId", ""))
    if not all(field_ids.values()):
        raise CandidateContractError("NetSuite columns lack raw/display field IDs")

    items = envelope.get("items")
    if not isinstance(items, list):
        raise CandidateContractError("NetSuite items are missing")
    mapped_rows = [_map_restlet_item(config, _mapping(item, "item"), field_ids, seen_keys) for item in items]
    next_cursor = page.get("nextCursor")
    if next_cursor is not None and (not isinstance(next_cursor, str) or not next_cursor):
        raise CandidateContractError("NetSuite nextCursor is invalid")
    return mapped_rows, next_cursor


def _map_restlet_item(
    config: RevenueSyncConfig,
    item: Mapping[str, Any],
    field_ids: Mapping[str, str],
    seen_keys: set[str],
) -> CandidateRow:
    source_key = _nonblank(item.get("sourceKey"))
    if not source_key:
        raise CandidateContractError("NetSuite source key is blank")
    if source_key in seen_keys:
        raise CandidateContractError("NetSuite source key is duplicated")
    seen_keys.add(source_key)
    values_by_header = _mapping(item.get("valuesByHeader"), "valuesByHeader")
    report_values = item.get("reportValues")
    if not isinstance(report_values, list) or len(report_values) != 35:
        raise CandidateContractError("NetSuite item does not have 35 report values")
    if tuple(values_by_header) != tuple(mapping.header for mapping in config.mappings):
        raise CandidateContractError("NetSuite valuesByHeader order drift")
    values = _mapping(item.get("values"), "values")
    display_values = _mapping(item.get("displayValues"), "displayValues")
    account_mapping = config.account_id_mapping
    account_field_id = field_ids[account_mapping.header]
    raw_account_id = _optional_text(values.get(account_field_id))
    display_account_id = _optional_text(display_values.get(account_field_id))
    payload = {mapping.field: values_by_header[mapping.header] for mapping in config.mappings}
    payload[account_mapping.field] = _nonblank(raw_account_id) or _nonblank(display_account_id) or ""
    identity = _mapping(item.get("sourceIdentity"), "sourceIdentity")
    transaction_id = _nonblank(identity.get("transactionInternalId"))
    line_unique_key = _nonblank(identity.get("lineUniqueKey"))
    if not transaction_id or not line_unique_key:
        raise CandidateContractError("NetSuite source identity is incomplete")
    return CandidateRow(
        source_key=source_key,
        source_identity={
            "transactionInternalId": transaction_id,
            "lineUniqueKey": line_unique_key,
            "accountingBookId": _optional_text(identity.get("accountingBookId")),
        },
        payload=payload,
        payload_hash=_payload_hash(payload),
        raw_account_id=raw_account_id,
        display_account_id=display_account_id,
        admission_evidence_available=True,
    )


def _persist_candidate(
    config: RevenueSyncConfig,
    *,
    store: CandidateStore,
    run_id: str,
    posting_period_id: str,
    rows: list[CandidateRow],
    page_count: int,
    fixture: bool = False,
) -> tuple[list[CandidateRow], CandidateManifest]:
    if not rows:
        raise CandidateContractError("candidate is empty")
    source_hash = _rows_hash(rows)
    store.write_jsonl("source-candidate.jsonl", (candidate_row_to_dict(row) for row in rows))
    manifest = CandidateManifest(
        run_id=run_id,
        posting_period_id=posting_period_id,
        source_total=len(rows),
        source_hash=source_hash,
        page_count=page_count,
        headers=tuple(item.header for item in config.mappings),
        control_totals=_control_totals(config, rows),
        fixture=fixture,
        created_at=datetime.now(UTC).isoformat(),
        object_prefix=store.prefix,
    )
    store.write_json("source-manifest.json", asdict(manifest))
    return rows, manifest


def candidate_row_to_dict(row: CandidateRow) -> dict[str, Any]:
    return asdict(row)


def candidate_rows_from_dicts(rows: Iterable[Mapping[str, Any]]) -> list[CandidateRow]:
    return [
        CandidateRow(
            source_key=str(row["source_key"]),
            source_identity=dict(row["source_identity"]),
            payload=dict(row["payload"]),
            payload_hash=str(row["payload_hash"]),
            raw_account_id=_optional_text(row.get("raw_account_id")),
            display_account_id=_optional_text(row.get("display_account_id")),
            admission_evidence_available=bool(row.get("admission_evidence_available")),
        )
        for row in rows
    ]


def _control_totals(config: RevenueSyncConfig, rows: list[CandidateRow]) -> dict[str, str]:
    fields = {mapping.header: mapping.field for mapping in config.mappings}
    totals: dict[str, Decimal] = {"transaction_amount": Decimal("0"), "base_amount": Decimal("0")}
    for row in rows:
        totals["transaction_amount"] += _decimal(row.payload[fields["Transaction Amount"]])
        totals["base_amount"] += _decimal(row.payload[fields["Amount"]])
    return {name: str(value) for name, value in totals.items()}


def _rows_hash(rows: list[CandidateRow]) -> str:
    digest = hashlib.sha256()
    for row in sorted(rows, key=lambda value: value.source_key):
        digest.update(row.source_key.encode())
        digest.update(b"\0")
        digest.update(row.payload_hash.encode())
        digest.update(b"\n")
    return digest.hexdigest()


def _payload_hash(payload: Mapping[str, Any]) -> str:
    encoded = json.dumps(payload, sort_keys=True, ensure_ascii=False, separators=(",", ":"), default=str).encode()
    return hashlib.sha256(encoded).hexdigest()


def _decimal(value: Any) -> Decimal:
    try:
        return Decimal(str(value or 0).replace(",", ""))
    except (InvalidOperation, AttributeError) as error:
        raise CandidateContractError("amount field is not numeric") from error


def _mapping(value: Any, name: str) -> Mapping[str, Any]:
    if not isinstance(value, Mapping):
        raise CandidateContractError(f"NetSuite {name} is missing or invalid")
    return value


def _text(value: Any) -> str:
    return "" if value is None else str(value)


def _optional_text(value: Any) -> str | None:
    return None if value is None else str(value)


def _nonblank(value: Any) -> str | None:
    if value is None:
        return None
    text = str(value).strip()
    return text or None