"""Normalization: provider frames -> canonical OHLCV schema + content hashes."""
from __future__ import annotations

import hashlib
import json
from typing import Any

import pandas as pd

SCHEMA_VERSION = "1"
NORMALIZATION_VERSION = "1"

# Canonical instrument identity format: "<EXCHANGE>#<CODE>", e.g. SH#600000.
# Prices are never substituted across canonical symbols.

# AKShare (Chinese) -> canonical column names. Non-mapped fields are preserved
# verbatim as raw provider fields (units/precision untouched).
COLUMN_MAP = {
    "日期": "date",
    "开盘": "open",
    "收盘": "close",
    "最高": "high",
    "最低": "low",
    "成交量": "volume",
}

REQUIRED = ["date", "open", "high", "low", "close", "volume"]


class NormalizeError(ValueError):
    def __init__(self, code: str, message: str, details: Any = None):
        super().__init__(message)
        self.code = code
        self.details = details


def map_columns(df: pd.DataFrame) -> pd.DataFrame:
    return df.rename(columns=COLUMN_MAP)


def normalize_frame(df: pd.DataFrame, symbol: str) -> pd.DataFrame:
    if df is None or df.empty:
        raise NormalizeError("empty_response", "provider returned no rows")
    missing_req = {c for c in REQUIRED if c not in df.columns}
    if missing_req:
        raise NormalizeError(
            "missing_columns",
            f"provider frame missing standard columns: {sorted(missing_req)}",
            {"columns": list(df.columns)},
        )
    out = df.copy()
    out["date"] = pd.to_datetime(out["date"], errors="coerce")
    if out["date"].isna().any():
        bad = out.loc[out["date"].isna()].index.tolist()
        raise NormalizeError(
            "bad_date", "unparseable date values", {"rows": [int(i) for i in bad[:5]]}
        )
    for col in ("open", "high", "low", "close", "volume"):
        out[col] = pd.to_numeric(out[col], errors="coerce")
        if out[col].isna().any():
            rows = out.loc[out[col].isna()].index.tolist()
            first = out.loc[rows[0], "date"].date().isoformat()
            raise NormalizeError(
                "missing_ohlcv",
                f"missing or non-numeric '{col}' on {first} — gap kept, never imputed",
                {"column": col, "row_count": len(rows), "rows": [int(i) for i in rows[:5]]},
            )
    out["date"] = out["date"].dt.strftime("%Y-%m-%d")
    out = out.sort_values("date", kind="mergesort")
    out = out.drop_duplicates(subset="date", keep="last")
    # canonical identity, prices never substituted across symbols
    out["symbol"] = symbol
    return out.reset_index(drop=True)


def frame_records(df: pd.DataFrame) -> list[dict]:
    return json.loads(df.to_json(orient="records", force_ascii=False))


def compute_object_hash(df: pd.DataFrame) -> str:
    """Content hash over canonical records; independent of request/user ids."""
    recs = frame_records(df)
    return hashlib.sha256(
        json.dumps(recs, sort_keys=True, ensure_ascii=False).encode("utf-8")
    ).hexdigest()


def coverage_summary(df_or_dates) -> dict:
    if isinstance(df_or_dates, pd.DataFrame):
        dates = pd.to_datetime(df_or_dates["date"]).tolist()
    else:
        dates = pd.to_datetime(list(df_or_dates)).tolist()
    if not dates:
        return {"actual_start": None, "actual_end": None, "segments": [], "gaps": []}
    segs = [[dates[0], dates[0]]]
    gaps: list[dict] = []
    for prev, cur in zip(dates, dates[1:]):
        delta = cur - prev
        if delta.days == 1:
            segs[-1][1] = cur
        else:
            segs.append([cur, cur])
            gaps.append({"after": prev.date().isoformat(), "before": cur.date().isoformat(),
                         "calendar_days": delta.days - 1})
    fmt = lambda d: d.date().isoformat()
    return {
        "actual_start": fmt(dates[0]),
        "actual_end": fmt(dates[-1]),
        "segments": [[fmt(a), fmt(b)] for a, b in segs],
        "gaps": gaps,
    }


def build_manifest_entry(
    df: pd.DataFrame,
    *,
    instrument: dict,
    provider: str,
    endpoint: str,
    params: dict,
    adjustment: str,
    requested_start: str,
    requested_end: str,
    raw_object_hash: str,
    warnings: list,
    schema_version: str = SCHEMA_VERSION,
    normalization_version: str = NORMALIZATION_VERSION,
    path: str,
    fetched_at: str,
    akshare_version: str,
) -> dict:
    cov = coverage_summary(df)
    return {
        "instrument": {
            "symbol": instrument["symbol"],
            "market": instrument["market"],
            "asset_type": instrument["asset_type"],
            **({"name": instrument["name"]} if instrument.get("name") else {}),
        },
        "object_hash": compute_object_hash(df),
        "path": path,  # internal-only: backend rewrites before client exposure
        "raw_object_hash": raw_object_hash,
        "provider": provider,
        "endpoint": endpoint,
        "params": params,
        "akshare_version": akshare_version,
        "fetched_at": fetched_at,
        "schema_version": schema_version,
        "normalization_version": normalization_version,
        "adjustment": adjustment,
        "requested_start": requested_start,
        "requested_end": requested_end,
        "actual_start": cov["actual_start"],
        "actual_end": cov["actual_end"],
        "coverage": {"segments": cov["segments"], "gaps": cov["gaps"]},
        "row_count": int(len(df)),
        "columns": list(df.columns),
        "warnings": list(warnings),
        "immutable": True,
    }