"""东方财富 API 客户端（资金流向数据）"""

import asyncio
import httpx
import json
import os
import re
from datetime import datetime
from typing import Optional, List

from services.cache import get_cache, set_cache

# ---- API Key 轮询 ----

_api_keys: List[str] = []
_key_index: int = -1


def _load_api_keys() -> List[str]:
    """收集所有可用 MX API key（支持逗号分隔和 MX_APIKEY_1 编号后缀）"""
    keys: List[str] = []
    seen: set = set()
    def add(k: str):
        k = k.strip()
        if k and k not in seen:
            seen.add(k)
            keys.append(k)
    raw = os.environ.get("MX_APIKEY", "")
    if raw:
        for k in raw.split(","):
            add(k)
    for i in range(1, 10):
        raw = os.environ.get(f"MX_APIKEY_{i}", "")
        if raw:
            for k in raw.split(","):
                add(k)
    return keys


def _ensure_keys() -> bool:
    global _api_keys, _key_index
    if not _api_keys:
        _api_keys = _load_api_keys()
    if _key_index == -1:
        val = get_cache("_mx_key_index")
        _key_index = int(val) if val else 0
    return bool(_api_keys)


def _get_key() -> str:
    """获取当前使用的 API key"""
    if not _ensure_keys():
        return ""
    return _api_keys[_key_index % len(_api_keys)]


def _rotate_key():
    """切换到下一个 key 并持久化当前索引"""
    global _key_index
    _key_index += 1
    set_cache("_mx_key_index", str(_key_index), ttl_hours=24)


# ---- 工具函数 ----


def get_eastmoney_market(code: str) -> str:
    """获取东方财富格式的市场标识"""
    if code.startswith("688"):
        return "6"
    if code.startswith("60"):
        return "1"
    return "0"


def parse_amount(text) -> float:
    """解析金额文本（支持 1.432亿元, -9915万元, 0元）"""
    if isinstance(text, (int, float)):
        return float(text)
    if not text or not isinstance(text, str):
        return 0
    text = text.strip()
    if not text:
        return 0
    sign = -1 if text.startswith("-") else 1
    clean = text.lstrip("+-").strip()
    m = re.match(r"^([\d.]+)\s*(亿|万|元)?$", clean)
    if not m:
        return 0
    value = float(m.group(1)) if m.group(1) else 0
    unit = m.group(2) or "元"
    if unit == "亿":
        return sign * value * 100000000
    elif unit == "万":
        return sign * value * 10000
    return sign * value


# ---- MX API ----


async def _call_mx_api(api_key: str, name: str, days: int) -> Optional[dict]:
    """执行一次 MX API 调用，返回原始 JSON"""
    url = "https://mkapi2.dfcfs.com/finskillshub/api/claw/query"
    payload = {"toolQuery": f"{name}最近{days}天主力资金流向和成交额"}
    async with httpx.AsyncClient() as client:
        try:
            resp = await client.post(
                url,
                json=payload,
                headers={
                    "Content-Type": "application/json",
                    "apikey": api_key,
                },
                timeout=15,
            )
            if resp.status_code != 200:
                return None
            return resp.json()
        except Exception as e:
            print(f"[eastmoney] MX API error: {e}")
            return None


def _parse_mx_response(result: dict) -> Optional[List[dict]]:
    """解析 MX API 返回，提取资金流向数据列表"""
    if result.get("status") != 0:
        return None
    dto_list = result.get("data", {}).get("data", {}).get("searchDataResultDTO", {}).get("dataTableDTOList", [])
    if not dto_list:
        return None
    dto = dto_list[0]
    name_map = dto.get("nameMap", {})
    raw_table = dto.get("rawTable", {})
    table = dto.get("table", {})
    dates = table.get("headName", [])
    if not dates:
        return None

    name_to_id = {}
    for kid, v in name_map.items():
        if isinstance(v, str):
            name_to_id[v] = kid

    main_net_col = name_to_id.get("(区间)主力净流入资金")
    amount_col = name_to_id.get("区间成交额")
    data_list = []

    for i, date_str in enumerate(dates):
        date_str = str(date_str).strip()
        m1 = re.match(r"^(\d{4})[/-](\d{1,2})[/-](\d{1,2})", date_str)
        if m1:
            date_str = f"{m1.group(1)}-{int(m1.group(2)):02d}-{int(m1.group(3)):02d}"
        else:
            m2 = re.match(r"^(\d{1,2})[/-](\d{1,2})", date_str)
            if m2:
                year = datetime.now().year
                date_str = f"{year}-{int(m2.group(1)):02d}-{int(m2.group(2)):02d}"
            else:
                continue

        main_net = 0
        if main_net_col and raw_table.get(main_net_col) and i < len(raw_table[main_net_col]):
            main_net = parse_amount(raw_table[main_net_col][i])

        amount = 0
        if amount_col and raw_table.get(amount_col) and i < len(raw_table[amount_col]):
            amount = parse_amount(raw_table[amount_col][i])

        data_list.append({"date": date_str, "mainNetInflow": main_net, "amount": amount})

    return data_list


async def fetch_mx_api(name: str, days: int) -> Optional[List[dict]]:
    """调用东方财富妙想MX API获取资金流向（缓存6小时，多key轮询）"""
    cache_key = f"mx_fund_flow:{name}:{days}"

    # 缓存命中
    cached = get_cache(cache_key)
    if cached is not None:
        return json.loads(cached)

    # 多 key 轮询：按序尝试，遇到 113(超限) 自动切下一个 key
    if not _ensure_keys():
        return None

    for attempt in range(len(_api_keys)):
        key = _get_key()
        result = await _call_mx_api(key, name, days)
        if result is None:
            _rotate_key()
            continue

        status = result.get("status", -1)
        if status == 113:
            print(f"[eastmoney] key {_key_index % len(_api_keys)} 已达每日上限，切换到下一个")
            _rotate_key()
            continue

        if status == 114:
            print(f"[eastmoney] key {_key_index % len(_api_keys)} 无效(114)，跳过")
            _rotate_key()
            continue

        if status != 0:
            _rotate_key()
            continue

        # 成功
        data_list = _parse_mx_response(result)
        if data_list:
            set_cache(cache_key, json.dumps(data_list, ensure_ascii=False))
        return data_list

    print(f"[eastmoney] 所有 {len(_api_keys)} 个 MX API key 均已耗尽")
    return None


# ---- 板块数据 ----

# 东方财富板块类型映射
SECTOR_TYPE_MAP = {
    "industry": "m:90+t:2",  # 行业板块
    "concept": "m:90+t:3",   # 概念板块
}

# 板块列表字段：f12=代码, f14=名称, f3=涨跌幅%, f62=主力净流入, f184=主力净流入占比
# f66=超大单净流入, f69=超大单净流入占比, f70=成交额, f78=小单净流入
SECTOR_FIELDS = "f12,f14,f2,f3,f4,f62,f184,f66,f69,f70,f78"


# ---- 板块数据（通过 akshare 调用同花顺数据源）----

import akshare as ak
import pandas as pd


async def fetch_sector_list(sector_type: str) -> list[dict]:
    """
    获取板块资金流向数据
    sector_type: "industry" 或 "concept"
    返回按主力净流入降序排列的板块列表
    编码优先东方财富（BKxxxx），降级同花顺（6位数字）
    """
    loop = asyncio.get_event_loop()

    def _get_data():
        # 1) 板块编码映射：东方财富 BK 编码 → 降级同花顺编码
        code_map = {}
        try:
            if sector_type == "industry":
                code_df = ak.stock_board_industry_name_em()
            else:
                code_df = ak.stock_board_concept_name_em()
            if code_df is not None and not code_df.empty:
                for _, r in code_df.iterrows():
                    code_map[str(r.get("f14", ""))] = str(r.get("f12", ""))
        except Exception:
            try:
                if sector_type == "industry":
                    code_df = ak.stock_board_industry_name_ths()
                else:
                    code_df = ak.stock_board_concept_name_ths()
                if code_df is not None and not code_df.empty:
                    for _, r in code_df.iterrows():
                        code_map[str(r.get("name", ""))] = str(r.get("code", ""))
            except Exception:
                pass

        # 2) 资金流向（10jqka 同花顺数据源）
        if sector_type == "industry":
            df = ak.stock_fund_flow_industry()
        else:
            df = ak.stock_fund_flow_concept()

        return code_map, df

    try:
        code_map, df = await loop.run_in_executor(None, _get_data)

        if df is None or df.empty:
            return []

        df = df.sort_values("净额", ascending=False)

        items = []
        for _, row in df.iterrows():
            name = str(row.get("行业", "")).strip()
            items.append({
                "code": code_map.get(name, ""),
                "name": name,
                "level": _safe_float(row.get("行业指数")),
                "changePercent": _safe_float(row.get("行业-涨跌幅")),
                "changeAmount": None,
                "mainNetInflow": _safe_float(row.get("净额", 0)) * 100000000,  # 亿→元
                "mainNetInflowPercent": None,
                "superLargeInflow": None,
                "superLargeInflowPercent": None,
                "turnover": _safe_float(row.get("流入资金", 0)) * 100000000 + _safe_float(row.get("流出资金", 0)) * 100000000,
                "smallNetInflow": None,
            })
        return items
    except Exception as e:
        print(f"[eastmoney] 获取{sector_type}板块失败: {e}")
        return []


def _safe_float(val) -> float:
    if val is None:
        return 0.0
    try:
        return float(val)
    except (ValueError, TypeError):
        return 0.0


async def fetch_sector_list_direct(sector_type: str) -> list[dict]:
    """
    直接从东方财富 push2 API 获取板块列表（备用，当 akshare 不可用时）
    """
    fs = SECTOR_TYPE_MAP.get(sector_type)
    if not fs:
        return []

    url = (
        f"https://push2.eastmoney.com/api/qt/clist/get"
        f"?fs={fs}&fields={SECTOR_FIELDS}"
        f"&fid=f62&po=1&pz=500&pn=1&np=1&fltt=2"
    )
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Referer": "https://quote.eastmoney.com/",
    }

    async with httpx.AsyncClient() as client:
        try:
            resp = await client.get(url, headers=headers, timeout=10)
            if resp.status_code != 200:
                return []
            result = resp.json()
            if result.get("rc") != 0:
                return []
            diff = result.get("data", {}).get("diff", [])
            items = []
            for item in diff:
                items.append({
                    "code": item.get("f12", ""),
                    "name": item.get("f14", ""),
                    "level": item.get("f2"),
                    "changePercent": item.get("f3"),
                    "changeAmount": item.get("f4"),
                    "mainNetInflow": item.get("f62", 0),
                    "mainNetInflowPercent": item.get("f184", 0),
                    "superLargeInflow": item.get("f66", 0),
                    "superLargeInflowPercent": item.get("f69", 0),
                    "turnover": item.get("f70", 0),
                    "smallNetInflow": item.get("f78", 0),
                })
            return items
        except Exception as e:
            print(f"[eastmoney] 获取{sector_type}板块失败: {e}")
            return []


async def fetch_push2his(code: str, days: int) -> Optional[list]:
    """回退到东方财富 push2his 接口（自动重试一次）"""
    market = get_eastmoney_market(code)
    secid = f"{market}.{code}"
    url = (
        f"https://push2his.eastmoney.com/api/qt/stock/fflow/daykline/get"
        f"?lmt={days}&fields1=f1,f2,f3,f7"
        f"&fields2=f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61,f62,f63,f64,f65,f66,f67,f68,f69"
        f"&ut=b2884a393a59ad64002292a3e90d46a5&secid={secid}"
    )
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Referer": "https://quote.eastmoney.com/",
    }

    for attempt in range(2):
        async with httpx.AsyncClient() as client:
            try:
                resp = await client.get(url, headers=headers, timeout=10)
                if resp.status_code != 200:
                    if attempt == 0:
                        continue
                    print(f"[eastmoney] push2his HTTP {resp.status_code}")
                    return None
                result = resp.json()
                klines = result.get("data", {}).get("klines", [])
                if not klines:
                    if attempt == 0:
                        continue
                    print(f"[eastmoney] push2his no klines in response")
                    return None
                return klines
            except Exception as e:
                if attempt == 0:
                    continue
                print(f"[eastmoney] push2his error: {e}")
                return None
    return None