from __future__ import annotations

from calendar import monthrange
from datetime import date, timedelta
from decimal import Decimal
import hashlib
import json
from sqlite3 import Connection
from typing import Any
from urllib.parse import urlencode

from fastapi import HTTPException

from jarvis_finance.services.budget_overview import get_budget_status_by_category
from jarvis_finance.services.household_financials import (
    FINANCIAL_SEMANTICS_VERSION,
    financial_quality_metadata,
    list_household_financial_effects,
    validate_budget_account_ids,
)
from jarvis_finance.services.household_review_corrections import transaction_token

_ALLOWED_GRANULARITIES = {"auto", "day", "week", "month"}
_ALLOWED_COMPARISONS = {"none", "previous_period", "previous_year"}
_ZERO = Decimal("0")
_CENT = Decimal("0.01")


def _money(value: Any) -> str:
    return format(Decimal(str(value or "0")).quantize(_CENT), ".2f")


def _percent(value: Decimal | None) -> str | None:
    return None if value is None else format(value.quantize(_CENT), ".2f")


def _parse_date(value: str, field: str) -> date:
    try:
        return date.fromisoformat(value)
    except ValueError:
        raise HTTPException(status_code=422, detail=f"{field} must use YYYY-MM-DD") from None


def _validate_filters(
    date_from: str,
    date_to: str,
    granularity: str,
    comparison: str,
    account_ids: list[str] | None,
) -> tuple[date, date]:
    start = _parse_date(date_from, "date_from")
    end = _parse_date(date_to, "date_to")
    if start > end:
        raise HTTPException(status_code=422, detail="date_from must not be after date_to")
    if (end - start).days > 730:
        raise HTTPException(status_code=422, detail="household cockpit range must not exceed 731 days")
    if granularity not in _ALLOWED_GRANULARITIES:
        raise HTTPException(status_code=422, detail="invalid cockpit granularity")
    if comparison not in _ALLOWED_COMPARISONS:
        raise HTTPException(status_code=422, detail="invalid cockpit comparison")
    return start, end


def _resolved_granularity(start: date, end: date, requested: str) -> str:
    if requested != "auto":
        return requested
    days = (end - start).days + 1
    if days <= 45:
        return "day"
    if days <= 120:
        return "week"
    return "month"


def _comparison_range(start: date, end: date, comparison: str) -> tuple[date, date] | None:
    if comparison == "none":
        return None
    if comparison == "previous_period":
        duration = end - start
        return start - duration - timedelta(days=1), start - timedelta(days=1)

    def previous_year(value: date) -> date:
        return value.replace(year=value.year - 1, day=min(value.day, monthrange(value.year - 1, value.month)[1]))

    return previous_year(start), previous_year(end)


def _bucket_key(value: date, granularity: str) -> str:
    if granularity == "day":
        return value.isoformat()
    if granularity == "week":
        return (value - timedelta(days=value.weekday())).isoformat()
    return value.strftime("%Y-%m")


def _bucket_keys(start: date, end: date, granularity: str) -> list[str]:
    keys: list[str] = []
    cursor = start
    while cursor <= end:
        key = _bucket_key(cursor, granularity)
        if not keys or keys[-1] != key:
            keys.append(key)
        cursor += timedelta(days=1)
    return keys


def _summary_rows(rows: list[dict[str, Any]]) -> tuple[Decimal, Decimal, Decimal]:
    income = sum((Decimal(str(row["income_effect"] or "0")) for row in rows), _ZERO)
    expense = sum((Decimal(str(row["expense_effect"] or "0")) for row in rows), _ZERO)
    return income, expense, income - expense


def _require_reconciliation(differences: list[Decimal]) -> Decimal:
    max_difference = max((abs(value) for value in differences), default=_ZERO)
    if max_difference > _CENT:
        raise HTTPException(
            status_code=409,
            detail="household cockpit reconciliation exceeds CHF 0.01",
        )
    return max_difference


def _trend(rows: list[dict[str, Any]], start: date, end: date, granularity: str) -> list[dict[str, Any]]:
    buckets = {
        key: {"income": _ZERO, "expense": _ZERO}
        for key in _bucket_keys(start, end, granularity)
    }
    for row in rows:
        key = _bucket_key(date.fromisoformat(str(row["transaction_date"])[:10]), granularity)
        if key not in buckets:
            continue
        buckets[key]["income"] += Decimal(str(row["income_effect"] or "0"))
        buckets[key]["expense"] += Decimal(str(row["expense_effect"] or "0"))
    return [
        {
            "bucket": key,
            "label": key,
            "income_chf": _money(values["income"]),
            "expense_chf": _money(values["expense"]),
            "net_chf": _money(values["income"] - values["expense"]),
        }
        for key, values in buckets.items()
    ]


def _category_rows(
    rows: list[dict[str, Any]],
    comparison_rows: list[dict[str, Any]],
    *,
    date_from: str,
    date_to: str,
    accounts: list[str],
) -> list[dict[str, Any]]:
    current: dict[str, dict[str, Any]] = {}
    previous: dict[str, Decimal] = {}
    for row in comparison_rows:
        amount = Decimal(str(row["expense_effect"] or "0"))
        if amount:
            key = str(row["effective_category_id"] or "uncategorized")
            previous[key] = previous.get(key, _ZERO) + amount
    for row in rows:
        amount = Decimal(str(row["expense_effect"] or "0"))
        if not amount:
            continue
        key = str(row["effective_category_id"] or "uncategorized")
        item = current.setdefault(
            key,
            {
                "category_id": key,
                "category": str(row["effective_category_name"] or "Unkategorisiert"),
                "expense": _ZERO,
                "transaction_count": 0,
            },
        )
        item["expense"] += amount
        item["transaction_count"] += 1
    total = sum((item["expense"] for item in current.values()), _ZERO)
    result = []
    for key, item in current.items():
        old = previous.get(key, _ZERO)
        change_percent = None if old == 0 else ((item["expense"] - old) / old) * Decimal("100")
        params: list[tuple[str, str]] = [
            ("date_from", date_from),
            ("date_to", date_to),
            ("category_id", key),
            ("financial_effect", "expense"),
        ]
        if accounts:
            params.extend(("account_id", account_id) for account_id in accounts)
        result.append({
            "category_id": key,
            "category": item["category"],
            "expense_chf": _money(item["expense"]),
            "share_percent": _percent((item["expense"] / total * Decimal("100")) if total else _ZERO),
            "transaction_count": item["transaction_count"],
            "comparison_expense_chf": _money(old),
            "change_chf": _money(item["expense"] - old),
            "change_percent": _percent(change_percent),
            "drilldown_url": "/household/transactions?" + urlencode(params),
        })
    return sorted(result, key=lambda item: Decimal(item["expense_chf"]), reverse=True)


def _selected_month_count(start: date, end: date, year: int) -> int:
    return len({
        (cursor.year, cursor.month)
        for offset in range((end - start).days + 1)
        if (cursor := start + timedelta(days=offset)).year == year
    })


def _budget(
    conn: Connection,
    rows: list[dict[str, Any]],
    categories: list[dict[str, Any]],
    start: date,
    end: date,
) -> dict[str, Any]:
    budget_total = _ZERO
    annual_forecast = _ZERO
    budgeted_categories: set[str] = set()
    forecast_statuses: list[str] = []
    years = range(start.year, end.year + 1)
    for year in years:
        month_count = _selected_month_count(start, end, year)
        year_effects = list_household_financial_effects(
            conn,
            date_from=f"{year}-01-01",
            date_to=f"{year}-12-31",
        )
        for item in get_budget_status_by_category(
            conn, year=str(year), _effect_rows=year_effects
        ):
            if item.get("category_type") != "expense" or item.get("is_rollup"):
                continue
            monthly_budget = Decimal(str(item.get("budget_month") or "0"))
            if item.get("has_budget") and monthly_budget:
                budget_total += monthly_budget * month_count
                budgeted_categories.add(str(item["category_id"]))
            annual_forecast += Decimal(str(item.get("forecast_year") or "0"))
            forecast_statuses.append(str(item.get("status") or "Noch keine Daten"))
    actual = sum((Decimal(str(row["expense_effect"] or "0")) for row in rows), _ZERO)
    unbudgeted = sum(
        (
            Decimal(str(item["expense_chf"]))
            for item in categories
            if str(item["category_id"]) not in budgeted_categories
        ),
        _ZERO,
    )
    reliable = bool(annual_forecast) and any(status != "Noch keine Daten" for status in forecast_statuses)
    return {
        "budget_chf": _money(budget_total),
        "actual_expense_chf": _money(actual),
        "remaining_chf": _money(budget_total - actual),
        "unbudgeted_expense_chf": _money(unbudgeted),
        "forecast_chf": _money(annual_forecast) if reliable else None,
        "forecast_status": "available" if reliable else "not_reliable",
        "forecast_scope": "annual_all_accounts",
        "forecast_method": (
            "Bestehende kanonische Budgetplanung: Ist bis Stichtag plus verbleibende Monate × Monatsbudget; Jahreswert, nicht im Cockpit neu berechnet."
            if reliable
            else "Noch nicht belastbar: zu wenig kanonische Ist-/Budgetdaten."
        ),
    }


def _transaction_row(row: dict[str, Any]) -> dict[str, Any]:
    tx_id = str(row["budget_transaction_id"])
    return {
        "transaction_token": transaction_token(tx_id),
        "date": str(row["transaction_date"]),
        "merchant": str(row["merchant_display_name"] or "Quelle"),
        "category": str(row["effective_category_name"] or "Unkategorisiert"),
        "transaction_type": str(row["transaction_type"]),
        "amount_chf": _money(row["effective_amount_chf"]) if row["effective_amount_chf"] is not None else None,
        "amount_original": str(row["amount_original"]),
        "currency": str(row.get("currency_original") or "CHF"),
        "income_effect_chf": _money(row["income_effect"]),
        "expense_effect_chf": _money(row["expense_effect"]),
    }


def _merchant_rows(
    rows: list[dict[str, Any]], *, date_from: str, date_to: str, account_ids: list[str]
) -> list[dict[str, Any]]:
    merchants: dict[str, dict[str, Any]] = {}
    for row in rows:
        amount = Decimal(str(row["expense_effect"] or "0"))
        if amount <= 0:
            continue
        name = str(row["merchant_display_name"] or "Quelle")
        item = merchants.setdefault(name, {"expense": _ZERO, "count": 0})
        item["expense"] += amount
        item["count"] += 1
    result = []
    for name, item in sorted(
        merchants.items(), key=lambda pair: pair[1]["expense"], reverse=True
    )[:8]:
        params = [("date_from", date_from), ("date_to", date_to), ("merchant", name)]
        params.extend(("account_id", account_id) for account_id in account_ids)
        result.append({
            "merchant": name,
            "expense_chf": _money(item["expense"]),
            "transaction_count": item["count"],
            "drilldown_url": "/household/transactions?" + urlencode(params),
        })
    return result


def _account_options(conn: Connection) -> list[dict[str, str]]:
    return [
        {
            "budget_account_id": str(row["budget_account_id"]),
            "account_name": str(row["name"]),
            "account_type": str(row["account_type"]),
            "currency": str(row["currency"]),
        }
        for row in conn.execute(
            "SELECT budget_account_id,name,account_type,currency FROM budget_accounts "
            "WHERE is_active=1 ORDER BY name,budget_account_id"
        ).fetchall()
    ]


_COCKPIT_SMALL_DEPENDENCIES = (
    "budget_accounts",
    "budget_categories",
    "budget_plan_items",
    "budget_category_baselines",
    "budget_transfers",
    "budget_recurring_payments",
    "budget_fixed_costs",
    "household_migros_links",
    "household_credit_card_settlements",
)


def get_household_cockpit_data_version(conn: Connection) -> str:
    """Return a compact revision aggregate over every cockpit input table."""
    digest = hashlib.sha256()
    for table in ("budget_transactions", "budget_transaction_candidates"):
        columns = {str(row[1]) for row in conn.execute(f"PRAGMA table_info({table})")}
        updated = "updated_at" if "updated_at" in columns else "created_at"
        row = conn.execute(
            f"SELECT COUNT(*),MAX(rowid),MAX(COALESCE({updated},'')),"
            f"SUM(length(COALESCE(CAST(rowid AS TEXT),''))) FROM {table}"
        ).fetchone()
        digest.update(json.dumps([table, *row], default=str).encode())
    existing = {
        str(row[0])
        for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
    }
    for table in _COCKPIT_SMALL_DEPENDENCIES:
        if table not in existing:
            continue
        cursor = conn.execute(f"SELECT * FROM {table} ORDER BY rowid")
        digest.update(table.encode())
        for row in cursor:
            digest.update(json.dumps(list(row), default=str, sort_keys=True).encode())
    return digest.hexdigest()[:20]


def _review_scope(
    conn: Connection, start: date, end: date, account_ids: list[str]
) -> dict[str, Any]:
    date_params: list[Any] = [start.isoformat(), end.isoformat()]
    base = (
        "household_batch_id IS NOT NULL AND status='needs_review' "
        "AND transaction_date BETWEEN ? AND ?"
    )
    missing = int(conn.execute(
        f"SELECT COUNT(*) FROM budget_transaction_candidates WHERE {base} "
        "AND (account_source IS NULL OR trim(account_source)='')",
        date_params,
    ).fetchone()[0])
    if account_ids:
        placeholders = ",".join("?" for _ in account_ids)
        count = int(conn.execute(
            f"SELECT COUNT(*) FROM budget_transaction_candidates WHERE {base} "
            f"AND account_source IN ({placeholders})",
            (*date_params, *account_ids),
        ).fetchone()[0])
        return {
            "count": count,
            "scope": "unavailable_global" if missing else "date_and_accounts",
            "partial": bool(count or missing),
            "missing_binding_count": missing,
        }
    count = int(conn.execute(
        f"SELECT COUNT(*) FROM budget_transaction_candidates WHERE {base}", date_params
    ).fetchone()[0])
    return {
        "count": count,
        "scope": "date_all_accounts",
        "partial": bool(count),
        "missing_binding_count": missing,
    }


def _summary_from_rows(rows: list[dict[str, Any]]) -> dict[str, Any]:
    """Derive cockpit quality and neutral-transfer totals from its one canonical effect scan."""
    unavailable_chf_count = sum(int(row["unavailable_chf"] or 0) for row in rows)
    unlinked_refund_count = sum(int(row["unlinked_refund"] or 0) for row in rows)
    transfer_conflict_count = sum(int(row["transfer_membership_conflict"] or 0) for row in rows)
    transfer_units: dict[str, dict[str, Any]] = {}
    for row in rows:
        if not int(row["neutral_transfer"] or 0):
            continue
        unit = transfer_units.setdefault(
            str(row["transfer_unit_id"]), {"unavailable": False, "volume": _ZERO}
        )
        unit["unavailable"] = bool(unit["unavailable"] or row["unavailable_transfer"])
        if row["effective_amount_chf"] is not None:
            unit["volume"] = max(
                Decimal(str(unit["volume"])), abs(Decimal(str(row["effective_amount_chf"])))
            )
    unavailable_transfer_count = sum(int(unit["unavailable"]) for unit in transfer_units.values())
    quality = financial_quality_metadata(
        unavailable_chf_count=unavailable_chf_count,
        unlinked_refund_count=unlinked_refund_count,
        transfer_membership_conflict_count=transfer_conflict_count,
    )
    return {
        **quality,
        "neutral_transfer_count": sum(int(row["neutral_transfer"] or 0) for row in rows),
        "neutral_transfer_volume_chf": (
            None
            if unavailable_transfer_count
            else _money(sum((Decimal(str(unit["volume"])) for unit in transfer_units.values()), _ZERO))
        ),
    }


def get_household_cockpit(
    conn: Connection,
    **kwargs: Any,
) -> dict[str, Any]:
    """Assemble one SQLite-consistent read snapshot; never commit caller-owned work."""
    owns_read_transaction = not conn.in_transaction
    if owns_read_transaction:
        conn.execute("BEGIN")
    try:
        return _assemble_household_cockpit(conn, **kwargs)
    finally:
        if owns_read_transaction:
            conn.rollback()


def _assemble_household_cockpit(
    conn: Connection,
    *,
    date_from: str,
    date_to: str,
    account_ids: list[str] | None = None,
    granularity: str = "auto",
    comparison: str = "none",
) -> dict[str, Any]:
    start, end = _validate_filters(
        date_from, date_to, granularity, comparison, account_ids
    )
    selected_accounts = validate_budget_account_ids(conn, account_ids)
    initial_data_version = get_household_cockpit_data_version(conn)
    resolved_granularity = _resolved_granularity(start, end, granularity)
    rows = list_household_financial_effects(
        conn,
        date_from=date_from,
        date_to=date_to,
        accounts=selected_accounts,
    )
    summary = _summary_from_rows(rows)
    previous_range = _comparison_range(start, end, comparison)
    previous_rows: list[dict[str, Any]] = []
    if previous_range:
        previous_rows = list_household_financial_effects(
            conn,
            date_from=previous_range[0].isoformat(),
            date_to=previous_range[1].isoformat(),
            accounts=selected_accounts,
        )
    income, expense, net = _summary_rows(rows)
    old_income, old_expense, old_net = _summary_rows(previous_rows)
    trend = _trend(rows, start, end, resolved_granularity)
    previous_trend = (
        _trend(previous_rows, previous_range[0], previous_range[1], resolved_granularity)
        if previous_range
        else []
    )
    categories = _category_rows(
        rows,
        previous_rows,
        date_from=date_from,
        date_to=date_to,
        accounts=selected_accounts,
    )
    if selected_accounts:
        actual = sum((Decimal(str(row["expense_effect"] or "0")) for row in rows), _ZERO)
        budget = {
            "budget_chf": None,
            "actual_expense_chf": _money(actual),
            "remaining_chf": None,
            "unbudgeted_expense_chf": None,
            "forecast_chf": None,
            "forecast_status": "unavailable_for_account_scope",
            "forecast_scope": "selected_accounts",
            "forecast_method": "Keine kanonische Budgetallokation auf Konten vorhanden.",
            "budget_status": "unavailable_for_account_scope",
            "budget_scope": "selected_accounts",
            "budget_method": "all_account_plan_not_allocated",
        }
    else:
        budget = _budget(conn, rows, categories, start, end)
        budget.update({
            "budget_status": "available",
            "budget_scope": "all_accounts",
            "budget_method": "canonical_planning_formula",
        })
    expense_rows = [row for row in rows if Decimal(str(row["expense_effect"] or "0")) > 0]
    latest_rows = sorted(rows, key=lambda row: (str(row["transaction_date"]), str(row["budget_transaction_id"])), reverse=True)
    largest_rows = sorted(expense_rows, key=lambda row: Decimal(str(row["expense_effect"])), reverse=True)
    category_total = sum((Decimal(item["expense_chf"]) for item in categories), _ZERO)
    trend_income = sum((Decimal(item["income_chf"]) for item in trend), _ZERO)
    trend_expense = sum((Decimal(item["expense_chf"]) for item in trend), _ZERO)
    trend_net = sum((Decimal(item["net_chf"]) for item in trend), _ZERO)
    differences = [
        abs(income - trend_income),
        abs(net - trend_net),
        abs(expense - category_total),
        abs(expense - trend_expense),
        abs(expense - Decimal(budget["actual_expense_chf"])),
    ]
    review = _review_scope(conn, start, end, selected_accounts)
    quality_warnings = list(summary["warnings"])
    if review["partial"]:
        quality_warnings.append({
            "code": "open_monetary_review_candidates",
            "count": int(review["count"]),
            "scope": review["scope"],
            "message": "Offene monetäre Prüfentscheidungen im Zeitraum machen die Auswertung teilweise verfügbar.",
        })
    savings_rate = ((net / income) * Decimal("100")) if income > 0 else None
    max_difference = _require_reconciliation(differences)
    base_drilldown = [("date_from", date_from), ("date_to", date_to)]
    base_drilldown.extend(("account_id", account_id) for account_id in selected_accounts)
    income_drilldown = urlencode([*base_drilldown, ("type", "income")])
    expense_drilldown = urlencode([*base_drilldown, ("financial_effect", "expense")])
    response = {
        "contract_version": "household_cockpit_v1",
        "semantics_version": FINANCIAL_SEMANTICS_VERSION,
        "data_version": initial_data_version,
        "filters": {
            "date_from": date_from,
            "date_to": date_to,
            "accounts": selected_accounts,
            "granularity": resolved_granularity,
            "comparison": comparison,
        },
        "account_options": _account_options(conn),
        "as_of": max((str(row["transaction_date"]) for row in rows), default=date_to),
        "data_status": {
            "status": "partial" if quality_warnings else "current",
            "label": "Teilweise verfügbar" if quality_warnings else "Aktuell",
            "warnings": quality_warnings,
            "unavailable_chf_count": int(summary["unavailable_chf_count"]),
            "unlinked_refund_count": int(summary["unlinked_refund_count"]),
        },
        "kpis": {
            "income_chf": _money(income),
            "expense_chf": _money(expense),
            "net_chf": _money(net),
            "savings_rate_percent": _percent(savings_rate),
            "budget_remaining_chf": budget["remaining_chf"],
            "unbudgeted_expense_chf": budget["unbudgeted_expense_chf"],
            "open_review_count": int(review["count"]),
            "review_scope": review["scope"],
            "income_change_chf": _money(income - old_income),
            "expense_change_chf": _money(expense - old_expense),
            "net_change_chf": _money(net - old_net),
            "income_drilldown_url": "/household/transactions?" + income_drilldown,
            "expense_drilldown_url": "/household/transactions?" + expense_drilldown,
        },
        "comparison": {
            "date_from": previous_range[0].isoformat() if previous_range else None,
            "date_to": previous_range[1].isoformat() if previous_range else None,
            "income_chf": _money(old_income),
            "expense_chf": _money(old_expense),
            "net_chf": _money(old_net),
            "trend": previous_trend,
        },
        "trend": trend,
        "expense_categories": categories,
        "budget": budget,
        "top_merchants": _merchant_rows(
            rows,
            date_from=date_from,
            date_to=date_to,
            account_ids=selected_accounts,
        ),
        "latest_transactions": [_transaction_row(row) for row in latest_rows[:8]],
        "largest_transactions": [_transaction_row(row) for row in largest_rows[:8]],
        "neutral_activity": {
            "transfer_count": int(summary["neutral_transfer_count"]),
            "transfer_volume_chf": summary["neutral_transfer_volume_chf"],
        },
        "reconciliation": {
            "status": "pass",
            "tolerance_chf": "0.01",
            "kpi_expense_chf": _money(expense),
            "kpi_income_chf": _money(income),
            "kpi_net_chf": _money(net),
            "trend_income_chf": _money(trend_income),
            "categories_expense_chf": _money(category_total),
            "trend_expense_chf": _money(trend_expense),
            "trend_net_chf": _money(trend_net),
            "budget_actual_expense_chf": budget["actual_expense_chf"],
            "max_difference_chf": _money(max_difference),
        },
    }
    return response