from __future__ import annotations

import argparse
import time
from pathlib import Path
from typing import Any

import pandas as pd
import requests

API_URL = "https://api.jiucaishuo.com/v2/fundsale/pc-new-fundsale"
DEFAULT_TIMEOUT = 15
DEFAULT_RETRIES = 3
UI_FIELDS = {"is_join"}
TEXT_FIELDS = {
    "fund_int_code",
    "name",
    "gaikuang_genzongzhishu",
    "gaikuang_fund_company",
    "setup_date",
    "guimo_date",
    "income_date",
    "sg_min_money",
    "sh_status",
}
PLACEHOLDER_VALUES = {"", "-", "--", "None", "-9999", "9999", "-99999"}
PLACEHOLDER_MAP = {value: pd.NA for value in PLACEHOLDER_VALUES}
HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/127.0.0.0 Safari/537.36"
    )
}


def format_fund_code(value: Any) -> Any:
    """把 ETF 代码补成带交易所后缀的格式。"""
    if pd.isna(value):
        return pd.NA

    code = str(value).strip().upper()
    if not code:
        return pd.NA
    if code.endswith((".SH", ".SZ")):
        return code

    suffix = ".SH" if code.startswith("5") else ".SZ" if code.startswith("1") else ""
    return f"{code}{suffix}"


def build_column_specs(table_list: list[dict[str, Any]]) -> list[tuple[str, str]]:
    """提取原始字段名和展示列名。"""
    specs = []
    for item in table_list:
        raw_name = item.get("text")
        if not raw_name or raw_name in UI_FIELDS:
            continue

        display_name = item.get("name") or raw_name
        if item.get("dw") and item.get("name"):
            display_name = f"{display_name}({item['dw']})"
        specs.append((raw_name, display_name))
    return specs


def coerce_value_column(series: pd.Series) -> pd.Series:
    """把看起来像数值的列尽量转换成数值，否则保留为字符串。"""
    values = series.astype("string").str.strip().replace(PLACEHOLDER_MAP)
    cleaned = values.str.replace(",", "", regex=False).str.replace("%", "", regex=False)
    numeric = pd.to_numeric(cleaned, errors="coerce")
    non_null_count = values.notna().sum()
    if non_null_count and numeric.notna().sum() / non_null_count >= 0.9:
        return numeric
    return values


def post_json(
    session: requests.Session,
    url: str,
    payload: dict[str, Any],
    timeout: int = DEFAULT_TIMEOUT,
    retries: int = DEFAULT_RETRIES,
) -> dict[str, Any]:
    """请求 JSON API，并在失败时做轻量重试。"""
    last_error: Exception | None = None
    for attempt in range(1, retries + 1):
        try:
            response = session.post(url, json=payload, headers=HEADERS, timeout=timeout)
            response.raise_for_status()
            result = response.json()
            if result.get("code") != 0:
                raise RuntimeError(f"接口返回异常: {result.get('msg') or result}")

            data = result.get("data")
            if not isinstance(data, dict):
                raise RuntimeError("接口返回缺少 data 字段")
            return data
        except (requests.RequestException, ValueError, RuntimeError) as exc:
            last_error = exc
            if attempt == retries:
                break
            time.sleep(attempt)

    raise RuntimeError(f"请求失败: {last_error}") from last_error


def normalize_table(data: dict[str, Any]) -> pd.DataFrame:
    """根据接口返回的列定义和行数据生成 DataFrame。"""
    table_list = data.get("table_list") or data.get("big_table_list") or []
    rows = data.get("lists") or []
    column_specs = build_column_specs(table_list)
    if not column_specs:
        raise RuntimeError("未找到可用表头")

    raw_columns = [raw_name for raw_name, _ in column_specs]
    frame = pd.DataFrame(rows).reindex(columns=raw_columns)

    for raw_name in raw_columns:
        if raw_name == "fund_int_code":
            frame[raw_name] = frame[raw_name].map(format_fund_code).astype("string")
        elif raw_name in TEXT_FIELDS:
            frame[raw_name] = frame[raw_name].astype("string")
        else:
            frame[raw_name] = coerce_value_column(frame[raw_name])

    return frame.rename(columns=dict(column_specs))


def fetch_etf_table(
    session: requests.Session,
    etf_id: int = 1,
    fund_code: str | None = None,
) -> pd.DataFrame:
    """抓取 ETF 表格。"""
    payload = {"etf_id": etf_id}
    if fund_code:
        payload["fund_code"] = fund_code
    return normalize_table(post_json(session, API_URL, payload))


def default_output_path(etf_id: int, fund_code: str | None) -> Path:
    suffix = fund_code or f"etf_{etf_id}"
    return Path(__file__).resolve().with_name(f"funddb_{suffix}.xlsx")


def save_table(df: pd.DataFrame, out_path: str | Path) -> Path:
    """统一导出为 Excel 文件。"""
    path = Path(out_path).expanduser().resolve().with_suffix(".xlsx")
    path.parent.mkdir(parents=True, exist_ok=True)

    try:
        df.to_excel(path, index=False)
    except ImportError as exc:
        raise RuntimeError("导出 xlsx 需要先安装 openpyxl") from exc
    return path


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="抓取 funddb.cn 的 ETF 表格")
    parser.add_argument("--etf-id", type=int, default=1, choices=[1, 2, 3, 4], help="表格类型，默认 1")
    parser.add_argument("--code", help="按 ETF 代码筛选，例如 510300")
    parser.add_argument("--out", help="输出文件名，统一导出为 .xlsx")
    return parser


def main() -> None:
    args = build_parser().parse_args()
    out_path = args.out or default_output_path(args.etf_id, args.code)

    with requests.Session() as session:
        df = fetch_etf_table(session, etf_id=args.etf_id, fund_code=args.code)

    saved_to = save_table(df, out_path)
    print(f"抓取完成，共 {len(df)} 条，已保存到: {saved_to}")
    print(f"表头: {', '.join(df.columns)}")


if __name__ == "__main__":
    main()
