from __future__ import annotations

from datetime import date, datetime, time
from decimal import Decimal, InvalidOperation
from itertools import groupby
from pathlib import Path
import re
from typing import Any, Iterable

from .models import NormalizedTransaction, SheetResult, StatementBatch
from .reader import RawSheet, read_workbook
from .templates import BankTemplate, TEMPLATES, normalize_header


class StatementParseError(RuntimeError):
    pass


class UnknownTemplateError(StatementParseError):
    pass


class AmbiguousTemplateError(StatementParseError):
    pass


SCAN_LIMIT = 50


def detect_header(
    rows: tuple[tuple[Any, ...], ...], scan_limit: int = SCAN_LIMIT
) -> tuple[BankTemplate, int, dict[str, int]]:
    candidates: list[tuple[int, BankTemplate, int, dict[str, int]]] = []
    for row_index, row in enumerate(rows[:scan_limit]):
        normalized = [normalize_header(value) for value in row]
        for template in TEMPLATES:
            column_map: dict[str, int] = {}
            for field, aliases in template.columns.items():
                for column_index, header in enumerate(normalized):
                    if header and header in aliases:
                        column_map[field] = column_index
                        break
            if all(field in column_map for field in template.required):
                candidates.append((len(column_map), template, row_index, column_map))

    if not candidates:
        inspected = min(len(rows), scan_limit)
        candidate_rows = _header_candidate_rows(rows[:scan_limit])
        detail = f"；候选表头：{'；'.join(candidate_rows)}" if candidate_rows else ""
        raise UnknownTemplateError(
            f"未识别到受支持的银行表头（已扫描前 {inspected} 行）{detail}。"
        )

    candidates.sort(key=lambda item: item[0], reverse=True)
    best_score = candidates[0][0]
    best = [candidate for candidate in candidates if candidate[0] == best_score]
    identities = {(item[1].template_id, item[2]) for item in best}
    if len(identities) > 1:
        matches = ", ".join(f"{item[1].bank_name}@{item[2] + 1}" for item in best)
        raise AmbiguousTemplateError(f"Ambiguous bank header signatures: {matches}")
    _, template, row_index, column_map = best[0]
    return template, row_index, column_map


def parse_statement(path: str | Path) -> tuple[StatementBatch, ...]:
    batches = tuple(result.batch for result in analyze_workbook(path) if result.batch)
    if not batches:
        raise UnknownTemplateError(f"{Path(path).name}: 工作簿中没有可解析的工作表。")
    return batches


def analyze_workbook(path: str | Path) -> tuple[SheetResult, ...]:
    """Parse every worksheet into an independent result.

    Each worksheet yields exactly one :class:`SheetResult` whose ``outcome``
    is ``parsed``, ``exception`` or ``ignored``. Unreadable workbooks raise
    ``CorruptWorkbookError``; a workbook that reads but contains no parsable
    sheet still returns one result per sheet so the UI can surface the
    filename / sheet / scanned range / candidate headers evidence.
    """
    source = Path(path)
    return tuple(_analyze_sheet(source, sheet) for sheet in read_workbook(source))


def _analyze_sheet(source: Path, sheet: RawSheet) -> SheetResult:
    rows = sheet.rows
    scanned = min(len(rows), SCAN_LIMIT)
    if not rows:
        return SheetResult(
            sheet_name=sheet.name,
            outcome="ignored",
            message=f"工作表「{sheet.name}」为空，已跳过。",
            scanned_rows=0,
        )
    if not any(any(_text(value) for value in row) for row in rows):
        return SheetResult(
            sheet_name=sheet.name,
            outcome="ignored",
            message=f"工作表「{sheet.name}」无有效内容，已跳过。",
            scanned_rows=scanned,
        )
    try:
        batch = _parse_sheet(source, sheet)
    except UnknownTemplateError as exc:
        return SheetResult(
            sheet_name=sheet.name,
            outcome="exception",
            message=_clean_sheet_message(str(exc), source),
            scanned_rows=scanned,
            candidate_headers=_header_candidate_rows(rows),
        )
    except (AmbiguousTemplateError, StatementParseError) as exc:
        return SheetResult(
            sheet_name=sheet.name,
            outcome="exception",
            message=_clean_sheet_message(str(exc), source),
            scanned_rows=scanned,
        )
    return SheetResult(
        sheet_name=sheet.name,
        outcome="parsed",
        scanned_rows=scanned,
        batch=batch,
    )


def _clean_sheet_message(message: str, source: Path) -> str:
    return message.replace(str(source), source.name).replace(source.name, "本文件")


def parse_directory(path: str | Path) -> tuple[StatementBatch, ...]:
    root = Path(path)
    files = sorted(
        file
        for file in root.rglob("*")
        if file.is_file()
        and file.suffix.lower() in {".xls", ".xlsx"}
        and not file.name.startswith("~$")
        and file.stat().st_size > 0
    )
    return tuple(batch for file in files for batch in parse_statement(file))


def _header_candidate_rows(
    rows: tuple[tuple[Any, ...], ...], limit: int = 3
) -> tuple[str, ...]:
    """Preview up to ``limit`` plausible header rows.

    Every non-empty row counts as a candidate, including single-cell rows, so
    an ambiguous workbook always has explicit scan evidence for the UI.
    """
    known_headers = {
        alias
        for template in TEMPLATES
        for aliases in template.columns.values()
        for alias in aliases
    }
    candidates: list[tuple[int, int, int, tuple[str, ...]]] = []
    for row_index, row in enumerate(rows):
        values = tuple(_text(value) for value in row if _text(value))
        if not values:
            continue
        matched = sum(normalize_header(value) in known_headers for value in values)
        candidates.append((matched, len(values), row_index, values))

    candidates.sort(key=lambda item: (item[0], item[1]), reverse=True)
    summaries: list[str] = []
    for _, _, row_index, values in candidates[:limit]:
        preview = "、".join(value[:24] for value in values[:8])
        if len(values) > 8:
            preview += "……"
        summaries.append(f"第 {row_index + 1} 行「{preview}」")
    return tuple(summaries)


def _parse_sheet(source: Path, sheet: RawSheet) -> StatementBatch:
    template, header_index, columns = detect_header(sheet.rows)
    metadata = _extract_metadata(sheet.rows[:header_index])
    transactions: list[NormalizedTransaction] = []
    warnings: list[str] = []

    for row_index, row in enumerate(sheet.rows[header_index + 1 :], header_index + 2):
        transaction_at = _transaction_datetime(row, columns)
        if transaction_at is None:
            continue
        income = _decimal(_value(row, columns.get("income")))
        expense = _decimal(_value(row, columns.get("expense")))
        if income == 0 and expense == 0:
            continue
        if income > 0 and expense > 0:
            warnings.append(f"Row {row_index} has both income and expense.")

        transactions.append(
            NormalizedTransaction(
                source_file=source,
                sheet_name=sheet.name,
                source_row=row_index,
                transaction_at=transaction_at,
                income=income,
                expense=expense,
                balance=_optional_decimal(_value(row, columns.get("balance"))),
                own_account=_optional_text(_value(row, columns.get("own_account"))),
                own_name=_optional_text(_value(row, columns.get("own_name"))),
                counterparty_account=_optional_text(
                    _value(row, columns.get("counterparty_account"))
                ),
                counterparty_name=_optional_text(
                    _value(row, columns.get("counterparty_name"))
                ),
                counterparty_bank=_optional_text(
                    _value(row, columns.get("counterparty_bank"))
                ),
                summary=_optional_text(_value(row, columns.get("summary"))),
                purpose=_optional_text(_value(row, columns.get("purpose"))),
                reference=_optional_text(_value(row, columns.get("reference"))),
                currency=_optional_text(_value(row, columns.get("currency"))),
            )
        )

    if not transactions:
        raise StatementParseError(
            f"{source.name}/{sheet.name}: header found but no transaction rows parsed."
        )

    own_account = metadata.get("own_account") or _first_value(
        transaction.own_account for transaction in transactions
    )
    own_name = metadata.get("own_name") or _first_value(
        transaction.own_name for transaction in transactions
    )
    warnings.extend(_balance_warnings(transactions))

    return StatementBatch(
        source_file=source,
        sheet_name=sheet.name,
        bank_name=template.bank_name,
        template_id=template.template_id,
        header_row=header_index + 1,
        own_account=own_account if isinstance(own_account, str) else None,
        own_name=own_name if isinstance(own_name, str) else None,
        period_start=metadata.get("period_start")
        if isinstance(metadata.get("period_start"), date)
        else None,
        period_end=metadata.get("period_end")
        if isinstance(metadata.get("period_end"), date)
        else None,
        transactions=tuple(transactions),
        warnings=tuple(warnings),
        template_version=template.version,
    )


def _extract_metadata(rows: tuple[tuple[Any, ...], ...]) -> dict[str, str | date]:
    result: dict[str, str | date] = {}
    labels = {
        "账号": "own_account",
        "账/卡号": "own_account",
        "户名": "own_name",
        "账户名称": "own_name",
        "起始日期": "period_start",
        "开始日期": "period_start",
        "截止日期": "period_end",
        "结束日期": "period_end",
    }

    for row in rows:
        for index, value in enumerate(row):
            text = _text(value)
            if not text:
                continue
            key_text, inline_value = _split_label(text)
            key = labels.get(key_text)
            if key:
                candidate: Any = inline_value or _next_nonempty(row, index + 1)
                if key.startswith("period_"):
                    parsed = _date_value(candidate)
                    if parsed:
                        result[key] = parsed
                elif candidate is not None:
                    result[key] = _text(candidate)

            if "起止日期" in text:
                dates = _dates_in_text(text)
                if len(dates) >= 2:
                    result["period_start"], result["period_end"] = dates[:2]
    return result


def _transaction_datetime(row: tuple[Any, ...], columns: dict[str, int]) -> datetime | None:
    if "transaction_at" in columns:
        return _datetime_value(_value(row, columns["transaction_at"]))

    transaction_date = _date_value(_value(row, columns.get("transaction_date")))
    if transaction_date is None:
        return None
    transaction_time = _time_value(_value(row, columns.get("transaction_time")))
    return datetime.combine(transaction_date, transaction_time or time.min)


def _datetime_value(value: Any) -> datetime | None:
    if isinstance(value, datetime):
        return value
    if isinstance(value, date):
        return datetime.combine(value, time.min)
    text = _text(value)
    if not text:
        return None
    normalized = text.replace("年", "-").replace("月", "-").replace("日", " ")
    normalized = " ".join(normalized.split())
    for pattern in (
        "%Y-%m-%d %H:%M:%S",
        "%Y/%m/%d %H:%M:%S",
        "%Y%m%d %H:%M:%S",
        "%Y-%m-%d",
        "%Y/%m/%d",
        "%Y%m%d",
    ):
        try:
            return datetime.strptime(normalized, pattern)
        except ValueError:
            continue
    return None


def _date_value(value: Any) -> date | None:
    parsed = _datetime_value(value)
    return parsed.date() if parsed else None


def _time_value(value: Any) -> time | None:
    if isinstance(value, datetime):
        return value.time()
    if isinstance(value, time):
        return value
    text = _text(value)
    if not text:
        return None
    for pattern in ("%H:%M:%S", "%H:%M"):
        try:
            return datetime.strptime(text, pattern).time()
        except ValueError:
            continue
    return None


def _dates_in_text(value: str) -> list[date]:
    date_strings = re.findall(
        r"20\d{2}(?:年|-|/)?\d{1,2}(?:月|-|/)?\d{1,2}日?", value
    )
    return [parsed for item in date_strings if (parsed := _date_value(item))]


def _decimal(value: Any) -> Decimal:
    if value is None:
        return Decimal("0")
    if isinstance(value, Decimal):
        return value
    if isinstance(value, (int, float)):
        return Decimal(str(value))
    text = _text(value)
    if not text:
        return Decimal("0")
    text = text.replace(",", "").replace("¥", "").replace("￥", "")
    text = text.removesuffix("元")
    negative = text.startswith("(") and text.endswith(")")
    if negative:
        text = text[1:-1]
    try:
        amount = Decimal(text)
    except InvalidOperation as exc:
        raise StatementParseError(f"Invalid amount: {value!r}") from exc
    return -amount if negative else amount


def _optional_decimal(value: Any) -> Decimal | None:
    if value is None or _text(value) == "":
        return None
    return _decimal(value)


def _balance_warnings(
    transactions: list[NormalizedTransaction],
) -> list[str]:
    warnings: list[str] = []
    if len(transactions) < 2:
        return warnings

    descending_source = transactions[0].transaction_at > transactions[-1].transaction_at
    ordered = sorted(transactions, key=lambda item: item.transaction_at)
    groups = [
        list(items)
        for _, items in groupby(ordered, key=lambda item: item.transaction_at)
    ]

    def source_endpoint(group: list[NormalizedTransaction]) -> NormalizedTransaction:
        key = lambda item: item.source_row
        return min(group, key=key) if descending_source else max(group, key=key)

    previous_balance = source_endpoint(groups[0]).balance
    for group in groups[1:]:
        if previous_balance is None:
            previous_balance = source_endpoint(group).balance
            continue

        expected = previous_balance + sum(
            (item.income - item.expense for item in group), Decimal("0")
        )
        matching_balance = next(
            (
                item.balance
                for item in group
                if item.balance is not None
                and abs(expected - item.balance) <= Decimal("0.01")
            ),
            None,
        )
        if matching_balance is not None:
            previous_balance = matching_balance
            continue

        rows = ",".join(str(item.source_row) for item in group)
        warnings.append(f"Rows {rows} do not reconcile by balance.")
        previous_balance = source_endpoint(group).balance
    return warnings


def _value(row: tuple[Any, ...], column: int | None) -> Any:
    if column is None or column >= len(row):
        return None
    return row[column]


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


def _optional_text(value: Any) -> str | None:
    text = _text(value)
    return text or None


def _split_label(value: str) -> tuple[str, str]:
    parts = re.split(r"[:：]", value, maxsplit=1)
    key = parts[0].strip()
    return key, parts[1].strip() if len(parts) == 2 else ""


def _next_nonempty(row: tuple[Any, ...], start: int) -> Any:
    for value in row[start:]:
        if _text(value):
            return value
    return None


def _first_value(values: Iterable[str | None]) -> str | None:
    return next((value for value in values if value), None)