from __future__ import annotations

import html
import json
from collections import defaultdict
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from sqlite3 import Connection
from typing import Any

from jarvis_finance.config.settings import find_repo_root, load_settings
from jarvis_finance.services.budget_imports import candidate_user_count_summary
from jarvis_finance.services.portfolio_service import get_overview

REPORT_TYPES = {"budget_monthly", "cashflow_household", "budget_review_status", "crypto_status", "portfolio_status"}
FORMATS = {"html", "markdown", "pdf"}


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _d(value: Any) -> Decimal:
    try:
        return Decimal(str(value or "0").replace("'", ""))
    except (InvalidOperation, ValueError):
        return Decimal("0")


def _money(value: Any) -> str:
    return f"{_d(value):.2f}"


def _abs_money(value: Any) -> str:
    return f"{abs(_d(value)):.2f}"


def _month_bounds(period: str) -> tuple[str, str, str]:
    month = period if len(period) == 7 else f"{period[:4]}-01"
    year = int(month[:4]); mon = int(month[5:7])
    next_year = year + (1 if mon == 12 else 0)
    next_mon = 1 if mon == 12 else mon + 1
    return month, f"{year:04d}-{mon:02d}-01", f"{next_year:04d}-{next_mon:02d}-01"


def _reports_dir(reports_dir: Path | None = None, repo_root: Path | None = None) -> Path:
    if reports_dir is not None:
        return reports_dir.resolve()
    return load_settings(repo_root=repo_root or find_repo_root()).runtime_paths.reports_dir.resolve()


def _assert_runtime_path(path: Path, reports_dir: Path, repo_root: Path) -> None:
    if repo_root == reports_dir or repo_root in reports_dir.parents:
        raise ValueError("reports_dir_must_be_outside_repo")
    if reports_dir not in path.resolve().parents:
        raise ValueError("report_path_must_stay_under_runtime_reports_dir")


def _actual_totals(conn: Connection, period: str) -> dict[str, str]:
    month, start, end = _month_bounds(period)
    rows = conn.execute(
        """
        SELECT transaction_type, COALESCE(SUM(ABS(CAST(amount_chf AS REAL))), 0)
        FROM budget_transactions
        WHERE status='confirmed'
          AND date(transaction_date) >= date(?) AND date(transaction_date) < date(?)
          AND COALESCE(source_type, '') NOT IN ('transfer', 'credit_card_payment', 'investment_transfer')
          AND transaction_type IN ('income','expense')
        GROUP BY transaction_type
        """,
        (start, end),
    ).fetchall()
    totals = {r[0]: _d(r[1]) for r in rows}
    income = totals.get("income", Decimal("0")); expense = totals.get("expense", Decimal("0")); net = income - expense
    return {
        "period": month,
        "income_chf": _money(income),
        "expense_chf": _money(expense),
        "net_cashflow_chf": _money(net),
        "savings_rate_percent": _money((net / income * 100) if income else Decimal("0")),
        "fixed_cost_ratio_percent": "0.00",
        "budget_consumption_percent": "0.00",
    }


def _top_categories(conn: Connection, period: str) -> list[dict[str, Any]]:
    _month, start, end = _month_bounds(period)
    rows = conn.execute(
        """
        SELECT COALESCE(c.name, 'Unkategorisiert') AS category, COALESCE(SUM(ABS(CAST(t.amount_chf AS REAL))), 0) AS total
        FROM budget_transactions t LEFT JOIN budget_categories c ON c.category_id=t.category_id
        WHERE t.status='confirmed' AND t.transaction_type='expense'
          AND COALESCE(t.source_type, '') NOT IN ('transfer', 'credit_card_payment', 'investment_transfer')
          AND date(t.transaction_date) >= date(?) AND date(t.transaction_date) < date(?)
        GROUP BY category ORDER BY total DESC LIMIT 5
        """,
        (start, end),
    ).fetchall()
    return [{"category": r[0], "amount_chf": _abs_money(r[1])} for r in rows]


def _top_merchants(conn: Connection, period: str) -> list[dict[str, Any]]:
    _month, start, end = _month_bounds(period)
    rows = conn.execute(
        """
        SELECT COALESCE(NULLIF(description,''), 'Unbekannt') AS merchant, COALESCE(SUM(ABS(CAST(amount_chf AS REAL))), 0) AS total
        FROM budget_transactions
        WHERE status='confirmed' AND transaction_type='expense'
          AND COALESCE(source_type, '') NOT IN ('transfer', 'credit_card_payment', 'investment_transfer')
          AND date(transaction_date) >= date(?) AND date(transaction_date) < date(?)
        GROUP BY merchant ORDER BY total DESC LIMIT 5
        """,
        (start, end),
    ).fetchall()
    return [{"merchant": r[0], "amount_chf": _abs_money(r[1])} for r in rows]


def _review_source_counts(conn: Connection, year: str) -> list[dict[str, Any]]:
    rows = conn.execute(
        """
        SELECT COALESCE(source_type,'manual') source, COUNT(*)
        FROM budget_transaction_candidates
        WHERE substr(transaction_date,1,4)=?
        GROUP BY source ORDER BY COUNT(*) DESC
        """,
        (year,),
    ).fetchall()
    wanted = {"visa_credit_card": "VISA", "migros_purchase": "Migros", "akb_bank": "AKB", "raiffeisen_bank": "Raiffeisen", "manual": "Manuell"}
    out = [{"source": wanted.get(str(r[0]), str(r[0]) or "Manuell"), "count": int(r[1] or 0)} for r in rows]
    for label in ["VISA", "Migros", "AKB", "Raiffeisen", "Manuell"]:
        if not any(x["source"] == label for x in out):
            out.append({"source": label, "count": 0})
    return out


def _cashflow_series(conn: Connection, year: str) -> list[dict[str, Any]]:
    rows = conn.execute(
        """
        SELECT substr(transaction_date,1,7) month, transaction_type, COALESCE(SUM(ABS(CAST(amount_chf AS REAL))),0)
        FROM budget_transactions
        WHERE status='confirmed' AND substr(transaction_date,1,4)=?
          AND COALESCE(source_type, '') NOT IN ('transfer', 'credit_card_payment', 'investment_transfer')
          AND transaction_type IN ('income','expense')
        GROUP BY month, transaction_type ORDER BY month
        """,
        (year,),
    ).fetchall()
    by_month: dict[str, dict[str, Decimal]] = defaultdict(lambda: {"income": Decimal("0"), "expense": Decimal("0")})
    for month, tx_type, total in rows:
        by_month[str(month)][str(tx_type)] = _d(total)
    return [{"month": m, "income_chf": _money(v["income"]), "expense_chf": _money(v["expense"]), "net_cashflow_chf": _money(v["income"] - v["expense"])} for m, v in sorted(by_month.items())]


def _portfolio_sections(conn: Connection) -> dict[str, Any]:
    try:
        overview = get_overview(conn).model_dump()
    except Exception:
        overview = {}
    return {
        "summary": {
            "total_portfolio_chf": str(overview.get("total_value_chf") or "0.00"),
            "equity_value_chf": str(overview.get("equity_value_chf") or "0.00"),
            "cash_chf": str(overview.get("cash_value_chf") or "0.00"),
            "crypto_chf": str(overview.get("crypto_value_chf") or "0.00"),
            "missing_prices": int(overview.get("unpriced_positions_count") or 0),
            "missing_fx": int(overview.get("missing_fx_count") or 0),
            "last_price_update": overview.get("last_price_update"),
        },
        "largest_positions": overview.get("largest_positions") or [],
        "data_quality_warnings": overview.get("warnings") or [],
    }


def preview_report_v1(conn: Connection, *, report_type: str, period: str) -> dict[str, Any]:
    if report_type not in REPORT_TYPES:
        raise ValueError("unsupported_report_type")
    year = period[:4]
    totals = _actual_totals(conn, period if len(period) == 7 else f"{year}-01")
    review = candidate_user_count_summary(conn, budget_year=year)
    sections: dict[str, Any] = {
        "top_categories": _top_categories(conn, totals["period"]),
        "top_merchants": _top_merchants(conn, totals["period"]),
        "budget_overruns": [],
        "categories_without_budget": [],
        "duplicates_and_covered_hint": "Duplikate und covered_by_source werden separat ausgewiesen.",
        "data_quality": {"status": "ok"},
    }
    if report_type == "cashflow_household":
        sections.update({"monthly_cashflow": _cashflow_series(conn, year), "transfers_separate": True, "investment_transfers_separate": True, "variable_expenses": []})
    elif report_type == "budget_review_status":
        sections.update({"candidates_by_source": _review_source_counts(conn, year), "merchant_rule_status": [], "next_review_actions": ["needs_review prüfen", "Duplikate separat prüfen"]})
    elif report_type == "crypto_status":
        p = _portfolio_sections(conn); sections.update({"crypto_summary": p["summary"], "wallet_verification": "lokale Daten", "crypto_report_link": None})
    elif report_type == "portfolio_status":
        p = _portfolio_sections(conn); sections.update({"portfolio_summary": p["summary"], "largest_positions": p["largest_positions"], "data_quality_warnings": p["data_quality_warnings"]})
    return {"purpose": "finance_report_v1_preview", "report_type": report_type, "period": period, "totals": totals, "review": review, "sections": sections, "metadata": {"generated_at": _now(), "runtime_only": True}}


def _render_markdown(data: dict[str, Any]) -> str:
    return "\n".join([
        f"# {data['report_type']} {data['period']}", "", "## Metadaten", f"- Runtime-only: {data['metadata']['runtime_only']}", "", "## Totals",
        *[f"- {k}: {v}" for k, v in data["totals"].items()], "", "## Review", *[f"- {k}: {v}" for k, v in data["review"].items()], "", "## Sections", "```json", json.dumps(data["sections"], ensure_ascii=False, indent=2), "```", "", "## Datenqualität", str(data["sections"].get("data_quality", {"status": "ok"})), "", "## Disclaimer", "Lokaler Runtime-Report. Keine Steuer-, Anlage- oder Rechtsberatung.",
    ])


def _render_html(data: dict[str, Any]) -> str:
    body = html.escape(json.dumps(data, ensure_ascii=False, indent=2))
    return f"<!doctype html><html lang='de'><head><meta charset='utf-8'><title>{html.escape(data['report_type'])}</title><style>body{{font-family:Inter,system-ui,sans-serif;margin:2rem;line-height:1.45}}pre{{background:#f8fafc;padding:1rem;border-radius:1rem;white-space:pre-wrap}}</style></head><body><h1>{html.escape(data['report_type'])} {html.escape(data['period'])}</h1><p>Runtime-only Report. Datenqualität ausgewiesen.</p><h2>Datenqualität</h2><p>{html.escape(str(data['sections'].get('data_quality', {'status':'ok'})))}</p><pre>{body}</pre></body></html>"


def _write_event(reports_dir: Path, entry: dict[str, Any]) -> None:
    with (reports_dir / "report_events.jsonl").open("a", encoding="utf-8") as fh:
        fh.write(json.dumps(entry, ensure_ascii=False, sort_keys=True) + "\n")


def generate_report_v1(conn: Connection, *, report_type: str, period: str, format: str = "html", reports_dir: Path | None = None, repo_root: Path | None = None) -> dict[str, Any]:
    if format not in FORMATS:
        raise ValueError("unsupported_report_format")
    repo = (repo_root or find_repo_root()).resolve()
    out_dir = _reports_dir(reports_dir, repo).resolve()
    out_dir.mkdir(parents=True, exist_ok=True)
    render_format = "html" if format == "pdf" else format
    data = preview_report_v1(conn, report_type=report_type, period=period)
    data["metadata"].update({"report_type": report_type, "format": render_format, "generated_at": _now()})
    safe_period = period.replace("/", "-").replace("..", "")
    ext = "md" if render_format == "markdown" else "html"
    path = (out_dir / f"{report_type}_{safe_period}_{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}.{ext}").resolve()
    _assert_runtime_path(path, out_dir, repo)
    content = _render_markdown(data) if render_format == "markdown" else _render_html(data)
    path.write_text(content, encoding="utf-8")
    event = {"event": "report_generated", "report_type": report_type, "period": period, "format": render_format, "path": str(path), "created_at": _now()}
    _write_event(out_dir, event)
    return {"status": "generated", "runtime_only": True, "path": str(path), "format": render_format, "metadata": data["metadata"], "report_id": path.name, "preview": data}


def list_reports_v1(*, reports_dir: Path | None = None, repo_root: Path | None = None) -> list[dict[str, Any]]:
    out_dir = _reports_dir(reports_dir, repo_root).resolve()
    if not out_dir.exists():
        return []
    rows = []
    for p in sorted([p for p in out_dir.rglob("*") if p.is_file() and p.suffix.lower() in {".html", ".md", ".pdf"}], key=lambda x: x.stat().st_mtime, reverse=True)[:100]:
        rows.append({"report_id": str(p.relative_to(out_dir)), "title": p.stem.replace("_", " ").title(), "report_type": p.stem.split("_")[0], "format": p.suffix.lstrip("."), "generated_at": datetime.fromtimestamp(p.stat().st_mtime, timezone.utc).isoformat(), "data_quality_status": "ok", "path_display": str(p), "download_url": f"/api/reports/files/{p.relative_to(out_dir).as_posix()}"})
    return rows