from __future__ import annotations

import base64
import io
import math
import sys
import zipfile
from dataclasses import dataclass
from datetime import date
from pathlib import Path
from typing import Any

import pandas as pd

SERVER_DIR = Path(__file__).resolve().parent
ROOT_DIR = SERVER_DIR.parent
if str(SERVER_DIR) not in sys.path:
    sys.path.insert(0, str(SERVER_DIR))

from core_logic import (  # noqa: E402
    APP_VERSION,
    FIELD_LABELS,
    FIELD_SYNONYMS,
    FULL_WEIGHT_DAILY_MODE,
    HALF_WEIGHT_DAILY_MODE,
    ROLE_FIELDS,
    ROLE_LABELS,
    STOCK_DAILY_MODE,
    attach_config_artifacts,
    auto_assign_files,
    available_current_value_modes,
    available_daily_modes,
    available_ytd_modes,
    clean_text,
    create_result_workbook,
    execute_calculation,
    norm_header,
)
from config_logic import (  # noqa: E402
    BusinessConfig,
    create_default_config_workbook,
    parse_business_config,
)


APP_TITLE = "固有资金每日亏损核算与邮件汇报"
SAMPLE2_DIR = ROOT_DIR / "sample_data" / "sample2"


@dataclass
class UploadedFile:
    name: str
    data: bytes


class AppInputError(Exception):
    pass


def _json_safe(value: Any) -> Any:
    if isinstance(value, dict):
        return {str(k): _json_safe(v) for k, v in value.items()}
    if isinstance(value, (list, tuple, set)):
        return [_json_safe(v) for v in value]
    if isinstance(value, pd.Timestamp):
        return value.strftime("%Y-%m-%d")
    if hasattr(value, "item"):
        try:
            return _json_safe(value.item())
        except Exception:
            pass
    if isinstance(value, float):
        if math.isnan(value) or math.isinf(value):
            return None
        return value
    if pd.isna(value) if not isinstance(value, (list, tuple, dict, set)) else False:
        return None
    return value


def _records(df: pd.DataFrame, limit: int | None = None) -> dict[str, Any]:
    if df is None or df.empty:
        return {"rows": [], "total": 0, "truncated": False}
    frame = df.copy()
    total = len(frame)
    truncated = False
    if limit is not None and total > limit:
        frame = frame.head(limit)
        truncated = True
    frame = frame.astype(object).where(pd.notna(frame), None)
    rows = [_json_safe(row) for row in frame.to_dict(orient="records")]
    return {"rows": rows, "total": total, "truncated": truncated}


def _load_sample2_files() -> list[tuple[str, bytes]]:
    if not SAMPLE2_DIR.exists():
        raise AppInputError("未找到样例2测试数据，请检查部署包中的测试数据目录。")
    files: list[tuple[str, bytes]] = []
    for path in sorted(SAMPLE2_DIR.glob("*.xls*")):
        if path.name.startswith(".") or path.name.startswith("~$"):
            continue
        files.append((path.name, path.read_bytes()))
    if not files:
        raise AppInputError("样例2测试数据目录中没有可用 Excel 文件。")
    return files


def _normalize_files(use_sample: bool, uploaded_files: list[UploadedFile]) -> list[tuple[str, bytes]]:
    if use_sample:
        return _load_sample2_files()
    files = [(item.name, item.data) for item in uploaded_files if item.name and item.data]
    if not files:
        raise AppInputError("请先上传当天相关 Excel 文件。")
    return files


def _parse_config(config_file: UploadedFile | None) -> tuple[BusinessConfig | None, dict[str, Any]]:
    if not config_file or not config_file.data:
        return None, {
            "uploaded": False,
            "enabled": False,
            "valid": True,
            "sourceName": f"系统内置默认口径 v{APP_VERSION}",
            "messages": [],
        }

    config = parse_business_config(config_file.data, config_file.name, FIELD_LABELS, FIELD_SYNONYMS)
    messages = [
        {
            "级别": message.level,
            "工作表": message.sheet,
            "行号": "" if message.row is None else message.row,
            "说明": message.message,
        }
        for message in config.messages
    ]
    status = {
        "uploaded": True,
        "enabled": config.enabled,
        "valid": config.valid,
        "sourceName": config.source_name,
        "messages": messages,
    }
    if not config.enabled:
        return None, status
    return config, status


def _assignment_payload(assignment: Any) -> dict[str, Any]:
    role_rows = []
    for role, item in [
        ("holding", assignment.holding),
        ("o32", assignment.o32),
        ("dimension", assignment.dimension),
    ]:
        role_rows.append(
            {
                "角色": ROLE_LABELS[role],
                "文件": item.parsed.source_name if item else "未识别",
                "工作表": item.parsed.sheet_name if item else "",
                "表头行": "" if not item else str(item.parsed.header_row),
                "状态": "已识别" if item and item.valid else "需调整",
            }
        )
    if assignment.daily_source:
        role_rows.append(
            {
                "角色": "每日损益来源",
                "文件": assignment.daily_source.parsed.source_name,
                "工作表": assignment.daily_source.parsed.sheet_name,
                "表头行": str(assignment.daily_source.parsed.header_row),
                "状态": assignment.daily_source.label,
            }
        )
    return {"roles": role_rows, "issues": list(assignment.issues)}


def _company_options(holding_df: pd.DataFrame, fields: dict[str, str | None]) -> list[dict[str, str]]:
    company_col = fields.get("company")
    values: list[str] = []
    if company_col and company_col in holding_df.columns:
        values = [v for v in holding_df[company_col].map(clean_text).drop_duplicates().tolist() if v]
    return [{"label": "全量", "value": ""}] + [{"label": value, "value": value} for value in values]


def _columns(df: pd.DataFrame) -> list[str]:
    return [str(col) for col in df.columns if str(col) != "_源文件行号"]


def _field_items(field_keys: list[str], values: dict[str, str | None]) -> list[dict[str, Any]]:
    return [
        {
            "key": key,
            "label": FIELD_LABELS.get(key, key),
            "value": values.get(key) or "",
        }
        for key in field_keys
    ]


def _field_override_payload(
    assignment: Any,
    h_fields: dict[str, str | None],
    d_fields: dict[str, str | None] | None,
) -> dict[str, Any]:
    groups = [
        {
            "role": "holding",
            "title": "持仓主表字段",
            "source": f"{assignment.holding.parsed.source_name} / {assignment.holding.parsed.sheet_name}",
            "columns": _columns(assignment.holding.parsed.df),
            "fields": _field_items(ROLE_FIELDS["holding"], h_fields),
        }
    ]

    if assignment.daily_source:
        if assignment.daily_source.kind == "standard_o32":
            daily_keys = ROLE_FIELDS["o32"]
        else:
            daily_keys = ["daily_return", "daily_pnl", "o32_stock_return", "o32_return", "o32_bond_return"]
        groups.append(
            {
                "role": "daily_source",
                "title": "每日损益字段",
                "source": f"{assignment.daily_source.parsed.source_name} / {assignment.daily_source.parsed.sheet_name}",
                "columns": _columns(assignment.daily_source.parsed.df),
                "fields": _field_items(daily_keys, dict(assignment.daily_source.fields)),
            }
        )

    if assignment.dimension and d_fields is not None:
        groups.append(
            {
                "role": "dimension",
                "title": "产品信息表字段",
                "source": f"{assignment.dimension.parsed.source_name} / {assignment.dimension.parsed.sheet_name}",
                "columns": _columns(assignment.dimension.parsed.df),
                "fields": _field_items(ROLE_FIELDS["dimension"], d_fields),
            }
        )

    return {"groups": groups}


_NO_FIELD_OVERRIDE = object()


def _resolve_manual_column(value: Any, columns: list[str], label: str) -> str | None | object:
    text = clean_text(value)
    if not text:
        return _NO_FIELD_OVERRIDE
    if text in {"不使用", "（不使用）", "(不使用)", "none", "None", "NONE", "null", "NULL"}:
        return None

    by_clean = {clean_text(col): col for col in columns}
    if text in by_clean:
        return by_clean[text]

    normalized = norm_header(text)
    by_norm = {norm_header(col): col for col in columns}
    if normalized in by_norm:
        return by_norm[normalized]

    raise AppInputError(f"手动字段“{label}”填写为“{text}”，但对应工作表中没有这个列名。请复制 Excel 表头原文。")


def _apply_field_override_dict(
    *,
    raw_values: dict[str, Any],
    fields: dict[str, str | None],
    columns: list[str],
) -> None:
    for key, value in (raw_values or {}).items():
        if key not in FIELD_LABELS:
            continue
        resolved = _resolve_manual_column(value, columns, FIELD_LABELS.get(key, key))
        if resolved is _NO_FIELD_OVERRIDE:
            continue
        fields[key] = resolved  # type: ignore[assignment]


def _same_parsed_sheet(left: Any, right: Any) -> bool:
    if not left or not right:
        return False
    return (
        left.parsed.source_name == right.parsed.source_name
        and left.parsed.sheet_name == right.parsed.sheet_name
    )


def _apply_field_overrides(
    raw_overrides: dict[str, Any] | None,
    assignment: Any,
    h_fields: dict[str, str | None],
    o_fields: dict[str, str | None],
    d_fields: dict[str, str | None] | None,
) -> None:
    overrides = raw_overrides or {}
    if not isinstance(overrides, dict):
        return

    if isinstance(overrides.get("holding"), dict):
        _apply_field_override_dict(
            raw_values=overrides["holding"],
            fields=h_fields,
            columns=_columns(assignment.holding.parsed.df),
        )

    daily_source = assignment.daily_source
    if daily_source and isinstance(overrides.get("daily_source"), dict):
        daily_fields = dict(daily_source.fields)
        _apply_field_override_dict(
            raw_values=overrides["daily_source"],
            fields=daily_fields,
            columns=_columns(daily_source.parsed.df),
        )
        daily_source.fields = daily_fields

        if daily_source.kind == "standard_o32":
            o_fields.update(daily_fields)
        elif _same_parsed_sheet(daily_source, assignment.holding):
            for key in ["daily_pnl", "daily_return", "o32_stock_return", "o32_return", "o32_bond_return"]:
                if key in daily_fields:
                    h_fields[key] = daily_fields.get(key)
            has_embedded_rate = bool(daily_fields.get("daily_return") or daily_fields.get("o32_stock_return"))
            if daily_source.kind == "embedded_amount" and has_embedded_rate:
                daily_source.kind = "embedded_rate"
                daily_source.label = "表内当日收益率"
            elif daily_source.kind == "embedded_rate" and not has_embedded_rate and daily_fields.get("daily_pnl"):
                daily_source.kind = "embedded_amount"
                daily_source.label = "表内当日损益金额"

    if assignment.dimension and d_fields is not None and isinstance(overrides.get("dimension"), dict):
        _apply_field_override_dict(
            raw_values=overrides["dimension"],
            fields=d_fields,
            columns=_columns(assignment.dimension.parsed.df),
        )


def _default_settings(
    assignment: Any,
    h_fields: dict[str, str | None],
    o_fields: dict[str, str | None],
    config: BusinessConfig | None,
) -> dict[str, Any]:
    hdf = assignment.holding.parsed.df
    current_modes = available_current_value_modes(hdf, h_fields, config)
    ytd_modes = available_ytd_modes(hdf, h_fields, config)
    daily_modes = available_daily_modes(assignment.daily_source, dict(assignment.daily_source.fields), config)
    return {
        "company_value": "",
        "current_value_mode": current_modes[0],
        "daily_rate_mode": daily_modes[0],
        "ytd_mode": ytd_modes[0],
        "ytd_scope": "与核心投资规模一致",
        "rate_unit_mode": "自动识别",
        "active_mode": "不按Y/N筛选",
        "exclude_non_six_product_code": True,
        "exclude_money": False,
        "exclude_bank": True,
        "daily_weight_mode": HALF_WEIGHT_DAILY_MODE,
        "scale_decimals": 2,
        "statistics_date": date.today().strftime("%Y-%m-%d"),
        "zero_keywords": [],
    }


def _build_default_cfg(cfg: dict[str, Any], daily_source_kind: str) -> dict[str, Any]:
    default_cfg = dict(cfg)
    default_cfg["business_config"] = None
    default_cfg["current_value_mode"] = "自动选择最可靠口径"
    default_cfg["ytd_mode"] = "自动选择最可靠口径"
    default_cfg["daily_rate_mode"] = STOCK_DAILY_MODE if daily_source_kind == "standard_o32" else "表内默认口径"
    default_cfg["ytd_scope"] = "与核心投资规模一致"
    default_cfg["rate_unit_mode"] = "自动识别"
    default_cfg["active_mode"] = "不按Y/N筛选"
    default_cfg["exclude_non_six_product_code"] = True
    default_cfg["exclude_money"] = False
    default_cfg["exclude_bank"] = True
    default_cfg["daily_weight_mode"] = HALF_WEIGHT_DAILY_MODE
    return default_cfg


def _pick(value: Any, options: list[Any], fallback: Any) -> Any:
    return value if value in options else fallback


def _bool(value: Any, default: bool = False) -> bool:
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        return value.strip().lower() in {"1", "true", "yes", "y", "on", "是"}
    return default


def _int(value: Any, default: int) -> int:
    try:
        return int(value)
    except Exception:
        return default


def _build_cfg(
    raw_settings: dict[str, Any],
    assignment: Any,
    h_fields: dict[str, str | None],
    o_fields: dict[str, str | None],
    config: BusinessConfig | None,
) -> dict[str, Any]:
    defaults = _default_settings(assignment, h_fields, o_fields, config)
    hdf = assignment.holding.parsed.df
    daily_modes = available_daily_modes(assignment.daily_source, dict(assignment.daily_source.fields), config)
    current_modes = available_current_value_modes(hdf, h_fields, config)
    ytd_modes = available_ytd_modes(hdf, h_fields, config)
    company_values = [option["value"] for option in _company_options(hdf, h_fields)]
    if not daily_modes:
        raise AppInputError("未找到可用的当日收益率字段。请在“手动调整字段识别”中填写当日股票收益率、单位净值涨跌幅或当日债券收益率列名。")

    settings = {**defaults, **(raw_settings or {})}
    cfg = {
        "company_value": _pick(settings.get("company_value"), company_values, ""),
        "current_value_mode": _pick(settings.get("current_value_mode"), current_modes, current_modes[0]),
        "daily_rate_mode": _pick(settings.get("daily_rate_mode"), daily_modes, daily_modes[0]),
        "ytd_mode": _pick(settings.get("ytd_mode"), ytd_modes, ytd_modes[0]),
        "ytd_scope": _pick(
            settings.get("ytd_scope"),
            ["与核心投资规模一致", "目标公司全量（历史口径）"],
            "与核心投资规模一致",
        ),
        "rate_unit_mode": _pick(
            settings.get("rate_unit_mode"),
            ["自动识别", "百分数（例如 -1.25 表示 -1.25%）", "小数（例如 -0.0125 表示 -1.25%）"],
            "自动识别",
        ),
        "active_mode": _pick(
            settings.get("active_mode"),
            ["不按Y/N筛选", "仅当日损益保留Y/是", "规模、当日损益和累计损益均仅保留Y/是"],
            "不按Y/N筛选",
        ),
        "exclude_non_six_product_code": _bool(settings.get("exclude_non_six_product_code"), True),
        "exclude_money": _bool(settings.get("exclude_money"), False),
        "exclude_bank": _bool(settings.get("exclude_bank"), True),
        "daily_weight_mode": _pick(
            settings.get("daily_weight_mode"),
            [HALF_WEIGHT_DAILY_MODE, FULL_WEIGHT_DAILY_MODE],
            HALF_WEIGHT_DAILY_MODE,
        ),
        "scale_decimals": _pick(_int(settings.get("scale_decimals"), 2), [1, 2], 2),
        "statistics_date": clean_text(settings.get("statistics_date")) or date.today().strftime("%Y-%m-%d"),
        "zero_keywords": [],
        "business_config": config,
    }
    if assignment.daily_source.kind not in {"embedded_amount", "embedded_rate"}:
        cfg["statistics_date"] = ""
    return cfg


def _prepare_assignment(files: list[tuple[str, bytes]], config: BusinessConfig | None) -> Any:
    assignment = auto_assign_files(files, config)
    if not assignment.holding or not assignment.daily_source:
        issue_text = "；".join(assignment.issues) if assignment.issues else "未能识别持仓表或每日损益来源。"
        raise AppInputError(issue_text)
    return assignment


def prepare_payload(
    *,
    use_sample: bool,
    uploaded_files: list[UploadedFile],
    config_file: UploadedFile | None,
) -> dict[str, Any]:
    files = _normalize_files(use_sample, uploaded_files)
    config, config_status = _parse_config(config_file)
    if config_status["uploaded"] and config_status["enabled"] and not config_status["valid"]:
        return {
            "ok": False,
            "appTitle": APP_TITLE,
            "version": APP_VERSION,
            "config": config_status,
            "error": "配置文件存在错误，请按提示修改后重新上传。",
        }

    assignment = _prepare_assignment(files, config)
    h_fields = dict(assignment.holding.fields)
    o_fields = dict(assignment.o32.fields) if assignment.o32 else dict(assignment.daily_source.fields)
    d_fields = dict(assignment.dimension.fields) if assignment.dimension else None
    hdf = assignment.holding.parsed.df

    return {
        "ok": True,
        "appTitle": APP_TITLE,
        "version": APP_VERSION,
        "config": config_status,
        "assignment": _assignment_payload(assignment),
        "diagnostics": {
            "持仓字段": {FIELD_LABELS.get(k, k): v for k, v in h_fields.items() if v},
            "每日损益字段": {FIELD_LABELS.get(k, k): v for k, v in assignment.daily_source.fields.items() if v},
            "产品信息字段": {FIELD_LABELS.get(k, k): v for k, v in (d_fields or {}).items() if v},
        },
        "fieldOverrides": _field_override_payload(assignment, h_fields, d_fields),
        "options": {
            "companies": _company_options(hdf, h_fields),
            "dailyModes": available_daily_modes(assignment.daily_source, dict(assignment.daily_source.fields), config),
            "currentValueModes": available_current_value_modes(hdf, h_fields, config),
            "ytdModes": available_ytd_modes(hdf, h_fields, config),
            "dailyWeightModes": [HALF_WEIGHT_DAILY_MODE, FULL_WEIGHT_DAILY_MODE],
            "ytdScopes": ["与核心投资规模一致", "目标公司全量（历史口径）"],
            "rateUnitModes": ["自动识别", "百分数（例如 -1.25 表示 -1.25%）", "小数（例如 -0.0125 表示 -1.25%）"],
            "activeModes": ["不按Y/N筛选", "仅当日损益保留Y/是", "规模、当日损益和累计损益均仅保留Y/是"],
            "scaleDecimals": [1, 2],
            "needsStatisticsDate": assignment.daily_source.kind in {"embedded_amount", "embedded_rate"},
        },
        "defaults": _default_settings(assignment, h_fields, o_fields, config),
        "sampleFiles": [{"文件名": name, "大小KB": round(len(data) / 1024, 1)} for name, data in files] if use_sample else [],
    }


def _result_payload(result: Any, *, include_email: bool, include_workbook: bool) -> dict[str, Any]:
    summary = _json_safe(result.summary)
    payload = {
        "summary": summary,
        "warnings": _json_safe(result.warnings),
        "fieldNotes": _json_safe(result.field_notes),
        "candidateMetrics": [
            {"候选口径": key, "金额（亿元）": _json_safe(value / 1e8)}
            for key, value in result.candidate_metrics.items()
        ],
        "configComparison": _records(result.config_comparison),
        "exceptions": _records(result.exceptions, limit=1000),
    }
    if include_email:
        payload["emailText"] = result.email_text
    if include_workbook:
        workbook = create_result_workbook(result)
        payload["workbookBase64"] = base64.b64encode(workbook).decode("ascii")
        payload["workbookName"] = f"固有资金每日亏损核算_{summary.get('统计日期', '')}.xlsx"
    return payload


def calculate_payload(
    *,
    use_sample: bool,
    uploaded_files: list[UploadedFile],
    config_file: UploadedFile | None,
    settings: dict[str, Any],
    action: str,
) -> dict[str, Any]:
    files = _normalize_files(use_sample, uploaded_files)
    config, config_status = _parse_config(config_file)
    if config_status["uploaded"] and config_status["enabled"] and not config_status["valid"]:
        return {
            "ok": False,
            "config": config_status,
            "error": "配置文件存在错误，请按提示修改后重新上传。",
        }

    assignment = _prepare_assignment(files, config)
    h_fields = dict(assignment.holding.fields)
    o32_assess = assignment.o32
    dimension_assess = assignment.dimension
    o_fields = dict(o32_assess.fields) if o32_assess else dict(assignment.daily_source.fields)
    d_fields = dict(dimension_assess.fields) if dimension_assess else None
    _apply_field_overrides(settings.get("field_overrides"), assignment, h_fields, o_fields, d_fields)
    cfg = _build_cfg(settings, assignment, h_fields, o_fields, config)

    result = execute_calculation(
        assignment.holding.parsed.df,
        o32_assess.parsed.df if o32_assess else pd.DataFrame(),
        dimension_assess.parsed.df if dimension_assess else None,
        h_fields,
        o_fields,
        d_fields,
        cfg,
        assignment.daily_source,
    )

    default_result = None
    try:
        default_result = execute_calculation(
            assignment.holding.parsed.df,
            o32_assess.parsed.df if o32_assess else pd.DataFrame(),
            dimension_assess.parsed.df if dimension_assess else None,
            h_fields,
            o_fields,
            d_fields,
            _build_default_cfg(cfg, assignment.daily_source.kind),
            assignment.daily_source,
        )
    except Exception as exc:
        result.warnings.append(f"默认口径对比未生成：{exc}")
    result = attach_config_artifacts(result, config, default_result)

    return {
        "ok": True,
        "config": config_status,
        "result": _result_payload(
            result,
            include_email=action in {"email", "final"},
            include_workbook=action == "final",
        ),
    }


def template_workbook() -> bytes:
    return create_default_config_workbook(FIELD_LABELS, FIELD_SYNONYMS)


def sample_data_archive() -> bytes:
    buffer = io.BytesIO()
    files = _load_sample2_files()
    with zipfile.ZipFile(buffer, "w", compression=zipfile.ZIP_DEFLATED) as archive:
        for filename, data in files:
            archive.writestr(filename, data)
    return buffer.getvalue()
