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

import asyncio
import time
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 轮询（MX 备选源用）----

_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 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)

    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


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


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

# 从东方财富 bkzj/list.js 逆向的字段映射
# f62=主力净流入, f184=主力净流入占比
# f66=超大单净流入, f69=超大单净流入占比
# f72=大单净流入, f75=大单净流入占比
# f78=中单净流入, f81=中单净流入占比
# f84=小单净流入, f87=小单净流入占比
# f70=成交额
SECTOR_FIELDS = "f12,f14,f2,f3,f62,f184,f66,f69,f72,f75,f78,f81,f84,f87,f70"

# 东方财富板块类型映射
SECTOR_MEDIA_MAP = {
    "industry": "m:90+s:4",
    "concept": "m:90+t:3",
}

# 东方财富 UT 令牌管理
_em_ut: str = "8dec03ba335b81bf4ebdf7b29ec27d15"
_em_ut_lock = asyncio.Lock()


async def _refresh_em_ut() -> str:
    """
    从东方财富前端 JS 中提取最新的 ut 令牌。
    按优先级尝试：
    1. bkzj/list.js（板块页专用）
    2. common/emdataview.js（通用数据组件）
    """
    urls = [
        "https://data.eastmoney.com/newstatic/js/bkzj/list.js",
        "https://data.eastmoney.com/newstatic/js/common/emdataview.js",
    ]
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
        "Referer": "https://data.eastmoney.com/bkzj/hy.html",
    }

    async with httpx.AsyncClient() as client:
        for url in urls:
            try:
                resp = await client.get(url, headers=headers, timeout=10)
                if resp.status_code != 200:
                    continue
                # 匹配 ut: 'xxxx' 或 ut:'xxxx' 或 ut: "xxxx"
                m = re.search(r"""ut['"]?\s*:\s*['"]([a-f0-9]{32})['"]""", resp.text)
                if m:
                    token = m.group(1)
                    print(f"[eastmoney] 已刷新 UT 令牌: {token[:8]}...")
                    return token
            except Exception as e:
                print(f"[eastmoney] 获取 UT 失败({url}): {e}")
    return _em_ut  # 保底返回当前值


async def get_em_ut(force_refresh: bool = False) -> str:
    """获取当前 UT，必要时刷新"""
    global _em_ut
    if force_refresh:
        async with _em_ut_lock:
            _em_ut = await _refresh_em_ut()
    return _em_ut


# curl_cffi 模拟 Chrome TLS 指纹
from curl_cffi.requests import AsyncSession

_sector_session: Optional[AsyncSession] = None


def _get_sector_session() -> AsyncSession:
    global _sector_session
    if _sector_session is None:
        _sector_session = AsyncSession(
            impersonate="chrome120",
            headers={
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
                "Referer": "https://data.eastmoney.com/bkzj/hy.html",
                "Accept": "*/*",
                "Accept-Language": "zh-CN,zh;q=0.9",
            },
            timeout=10,
        )
    return _sector_session


# 内存缓存
_sector_cache: dict[str, tuple[list[dict], float]] = {}
_SECTOR_CACHE_TTL = 60


async def fetch_sector_list(sector_type: str) -> list[dict]:
    now = time.time()
    if sector_type in _sector_cache:
        data, ts = _sector_cache[sector_type]
        if now - ts < _SECTOR_CACHE_TTL:
            return data

    data = await _fetch_push2(sector_type)
    if data:
        _sector_cache[sector_type] = (data, now)
        return data

    data = await _fetch_akshare(sector_type)
    if data:
        _sector_cache[sector_type] = (data, now)
    return data


async def _fetch_push2(sector_type: str) -> list[dict]:
    """东方财富 push2 API（curl_cffi 模拟浏览器 TLS 指纹）"""
    fs = SECTOR_MEDIA_MAP.get(sector_type)
    if not fs:
        return []

    session = _get_sector_session()
    ut = await get_em_ut()
    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"
        f"&invt=2&ut={ut}"
    )

    for attempt in range(2):
        try:
            resp = await session.get(url)
            if resp.status_code != 200:
                if attempt == 0:
                    await asyncio.sleep(1)
                    continue
                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": None,
                    "mainNetInflow": item.get("f62", 0) or 0,
                    "mainNetInflowPercent": item.get("f184", 0),
                    "superLargeInflow": item.get("f66", 0) or 0,
                    "superLargeInflowPercent": item.get("f69", 0),
                    "largeInflow": item.get("f72", 0) or 0,
                    "largeInflowPercent": item.get("f75", 0),
                    "mediumInflow": item.get("f78", 0) or 0,
                    "mediumInflowPercent": item.get("f81", 0),
                    "smallInflow": item.get("f84", 0) or 0,
                    "smallInflowPercent": item.get("f87", 0),
                    "turnover": item.get("f70", 0) or 0,
                })
            return items
        except Exception as e:
            err = str(e)
            print(f"[eastmoney] push2 获取{sector_type}板块失败(attempt {attempt+1}): {err[:80]}")
            # UT 可能过期，尝试刷新
            if "disconnect" in err.lower() or "refused" in err.lower() or attempt == 1:
                await get_em_ut(force_refresh=True)
                ut = _em_ut
                # 重建会话（TLS 指纹可能会被缓存）
                global _sector_session
                _sector_session = None
                session = _get_sector_session()
            if attempt == 0:
                await asyncio.sleep(1)
    return []


import akshare as ak


async def _fetch_akshare(sector_type: str) -> list[dict]:
    """akshare 降级方案（同花顺数据源）"""
    loop = asyncio.get_event_loop()

    def _get():
        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

        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)
        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()
            inflow = float(row.get("流入资金", 0) or 0) * 100000000
            outflow = float(row.get("流出资金", 0) or 0) * 100000000
            items.append({
                "code": code_map.get(name, ""),
                "name": name,
                "level": float(row.get("行业指数") or 0),
                "changePercent": float(row.get("行业-涨跌幅") or 0),
                "changeAmount": None,
                "mainNetInflow": float(row.get("净额", 0) or 0) * 100000000,
                "mainNetInflowPercent": None,
                "superLargeInflow": None,
                "superLargeInflowPercent": None,
                "largeInflow": None,
                "largeInflowPercent": None,
                "mediumInflow": None,
                "mediumInflowPercent": None,
                "smallInflow": None,
                "smallInflowPercent": None,
                "turnover": inflow + outflow,
            })
        return items
    except Exception as e:
        print(f"[eastmoney] akshare 获取{sector_type}板块失败: {e}")
        return []


# ---- 公司概况 ----

_F10_MARKET_MAP = {"6": "SH", "0": "SZ", "3": "SZ"}
_DATACENTER_MARKET_MAP = {"6": "SH", "0": "SZ", "3": "SZ"}


def _code_to_f10(code: str) -> str:
    """6位代码 → 东方财富 F10 格式 (SH600519 / SZ000001)"""
    prefix = code[0]
    mkt = _F10_MARKET_MAP.get(prefix, "SH")
    return f"{mkt}{code}"


def _code_to_secucode(code: str) -> str:
    """6位代码 → 数据中心格式 (600519.SH / 000001.SZ)"""
    prefix = code[0]
    mkt = _DATACENTER_MARKET_MAP.get(prefix, "SH")
    return f"{code}.{mkt}"


async def fetch_company_profile(code: str) -> Optional[dict]:
    """东方财富 F10 - 公司概况（缓存24小时）"""
    cache_key = f"company_profile:{code}"
    cached = get_cache(cache_key)
    if cached is not None:
        return json.loads(cached)

    em_code = _code_to_f10(code)
    url = (
        f"https://emweb.securities.eastmoney.com/PC_HSF10"
        f"/CompanySurvey/CompanySurveyAjax?code={em_code}"
    )
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Referer": (
            f"https://emweb.securities.eastmoney.com/PC_HSF10"
            f"/CompanySurvey/Index?type=web&code={em_code}"
        ),
    }

    async with httpx.AsyncClient() as client:
        try:
            resp = await client.get(url, headers=headers, timeout=15)
            if resp.status_code != 200:
                return None
            raw = resp.json()
        except Exception as e:
            print(f"[eastmoney] fetch_company_profile error: {e}")
            return None

    jbzl = raw.get("jbzl", {})
    fxxg = raw.get("fxxg", {})

    profile = {
        "code": code,
        "name": jbzl.get("agjc", ""),
        "fullName": jbzl.get("gsmc", ""),
        "englishName": jbzl.get("ywmc", ""),
        "boardType": jbzl.get("zqlb", ""),
        "industry": jbzl.get("sshy", ""),
        "exchange": jbzl.get("ssjys", ""),
        "industryDetail": jbzl.get("sszjhhy", ""),

        # 高管
        "chairman": jbzl.get("dsz", ""),
        "generalManager": jbzl.get("zjl", ""),
        "secretary": jbzl.get("dm", ""),

        # 联系方式
        "phone": jbzl.get("lxdh", ""),
        "email": jbzl.get("dzxx", ""),
        "fax": jbzl.get("cz", ""),
        "website": jbzl.get("gswz", ""),

        # 地址
        "officeAddress": jbzl.get("bgdz", ""),
        "registeredAddress": jbzl.get("zcdz", ""),
        "region": jbzl.get("qy", ""),
        "postalCode": jbzl.get("yzbm", ""),

        # 注册信息
        "registeredCapital": jbzl.get("zczb", ""),
        "unifiedCreditCode": jbzl.get("gsdj", ""),
        "employees": jbzl.get("gyrs", 0),
        "managementCount": jbzl.get("glryrs", 0),

        # 中介机构
        "lawFirm": jbzl.get("lssws", ""),
        "auditor": jbzl.get("kjssws", ""),

        # 公司简介&业务
        "businessScope": jbzl.get("jyfw", ""),
        "companyProfile": jbzl.get("gsjj", ""),

        # 发行信息
        "establishmentDate": fxxg.get("clrq", ""),
        "listingDate": fxxg.get("ssrq", ""),
        "listingPrice": fxxg.get("mgfxj", ""),
        "issuePriceEarningsRatio": fxxg.get("fxsyl", ""),
        "issueAmount": fxxg.get("fxl", ""),
        "netProceeds": fxxg.get("mjzjje", ""),
        "issueMethod": fxxg.get("fxfs", ""),
    }

    set_cache(cache_key, json.dumps(profile, ensure_ascii=False), ttl_hours=24)
    return profile


# ---- 财务数据 ----

# 财务指标关键字段映射（英文 → 中文）
FINANCIAL_FIELD_LABELS: dict[str, str] = {
    "EPSJB": "基本每股收益",
    "EPSKCJB": "扣非每股收益",
    "BPS": "每股净资产",
    "MGWFPLR": "每股未分配利润",
    "MGJYXJJE": "每股经营现金流",
    "TOTALOPERATEREVE": "营业总收入",
    "MLR": "毛利润",
    "PARENTNETPROFIT": "归母净利润",
    "KCFJCXSYJLR": "扣非净利润",
    "TOTALOPERATEREVETZ": "营收同比增长",
    "PARENTNETPROFITTZ": "净利润同比增长",
    "KCFJCXSYJLRTZ": "扣非净利润同比增长",
    "ROEJQ": "加权净资产收益率",
    "ROEKCJQ": "扣非ROE",
    "ZZCJLL": "总资产净利率",
    "XSJLL": "销售净利率",
    "XSMLL": "销售毛利率",
    "LD": "流动比率",
    "SD": "速动比率",
    "ZCFZL": "资产负债率",
    "XJLLB": "现金流比率",
    "YSZKZZTS": "应收账款周转天数",
    "CHZZTS": "存货周转天数",
    "ZZCZZTS": "总资产周转天数",
    "TOAZZL": "总资产周转率",
    "CHZZL": "存货周转率",
    "ROIC": "投入资本回报率",
    "QYCS": "权益乘数",
    "CQBL": "产权比率",
    "TAXRATE": "实际税率",
    "TOTAL_SHARE": "总股本",
    "A_FREE_SHARE": "流通A股",
    "DJD_TOI_YOY": "单季营收同比",
    "DJD_DPNP_YOY": "单季净利润同比",
    "EPSJBTZ": "每股收益同比增长",
    "BPSTZ": "每股净资产同比增长",
    "ROEJQTZ": "ROE同比增长",
    # 利润表补充
    "OPERATE_PROFIT_PK": "营业利润",
    "PER_TOI": "每股营业收入",
    "PER_EBIT": "每股EBIT",
    # 资产负债表补充
    "TOTAL_ASSETS_PK": "总资产",
    "LIABILITY": "总负债",
    "TOTAL_EQUITY_PK": "所有者权益",
    "MGZBGJ": "每股资本公积",
    "CA_TA": "流动资产占比",
    "NCA_TA": "非流动资产占比",
    "CL_TL": "流动负债占比",
    # 现金流量表补充
    "NETCASH_OPERATE_PK": "经营活动现金流净额",
    "NETCASH_INVEST_PK": "投资活动现金流净额",
    "NETCASH_FINANCE_PK": "筹资活动现金流净额",
    "JYXJLYYSR": "经营现金流占营收比",
    "XSJXLYYSR": "销售现金流占营收比",
    "FCFF_FORWARD": "企业自由现金流",
    # 营运能力补充
    "OPERATE_CYCLE": "营业周期",
    "CURRENT_ASSET_TR": "流动资产周转率",
    "FIXED_ASSET_TR": "固定资产周转率",
    "ACCOUNTS_PAYABLE_TR": "应付账款周转率",
    "PREPAID_ACCOUNTS_RATIO": "预收账款占营收比",
    "PAYABLE_TDAYS": "应付账款周转天数",
}


# 小数比率字段（后端返回小数，前端需显示为百分比）
_DECIMAL_RATIO_FIELDS = {
    "JYXJLYYSR", "XSJXLYYSR", "PREPAID_ACCOUNTS_RATIO",
    "NCO_OP", "NCO_FIXED", "SALE_ER_PK",
}


def _parse_financial_record(record: dict) -> dict:
    """将原始财务记录转为整洁的输出"""
    result = {}
    for field, label in FINANCIAL_FIELD_LABELS.items():
        val = record.get(field)
        if val is not None:
            if isinstance(val, (int, float)):
                # 小数比率字段转为百分比
                if field in _DECIMAL_RATIO_FIELDS and abs(val) < 1:
                    result[label] = round(val * 100, 2)
                    result[label + "(%)"] = True
                elif abs(val) > 1_0000_0000:
                    result[label] = round(val / 1_0000_0000, 2)
                    result[label + "(亿)"] = True
                elif abs(val) > 1_0000:
                    result[label] = round(val / 1_0000, 2)
                    result[label + "(万)"] = True
                else:
                    result[label] = round(val, 2)
            else:
                result[label] = val
    return result


async def fetch_financial_data_main(code: str, years: int = 5) -> Optional[list[dict]]:
    """东方财富数据中心 - 主要财务指标（缓存6小时）"""
    cache_key = f"financial_main:{code}:{years}"
    cached = get_cache(cache_key)
    if cached is not None:
        return json.loads(cached)

    secucode = _code_to_secucode(code)
    url = "https://datacenter.eastmoney.com/securities/api/data/v1/get"
    params = {
        "reportName": "RPT_F10_FINANCE_MAINFINADATA",
        "columns": "ALL",
        "filter": f'(SECUCODE="{secucode}")',
        "pageNumber": 1,
        "pageSize": max(years * 4, 4),  # 每年最多4份报告
        "sortTypes": -1,
        "sortColumns": "REPORT_DATE",
    }
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Referer": "https://datacenter.eastmoney.com/",
    }

    async with httpx.AsyncClient() as client:
        try:
            resp = await client.get(url, params=params, headers=headers, timeout=15)
            if resp.status_code != 200:
                return None
            body = resp.json()
            if not body.get("result") or not body["result"].get("data"):
                return None
            records = body["result"]["data"]
        except Exception as e:
            print(f"[eastmoney] fetch_financial_data error: {e}")
            return None

    items = []
    for r in records:
        items.append({
            "reportDate": r.get("REPORT_DATE", ""),
            "reportType": r.get("REPORT_TYPE", ""),
            "reportDateName": r.get("REPORT_DATE_NAME", ""),
            "noticeDate": r.get("NOTICE_DATE", ""),
            "indicators": _parse_financial_record(r),
        })

    set_cache(cache_key, json.dumps(items, ensure_ascii=False), ttl_hours=6)
    return items


# ---- 经营分析（主营构成/收入构成）----

_SEGMENT_TYPE_MAP = {
    "industry": "1",   # 按行业
    "product": "2",    # 按产品
    "region": "3",     # 按地区
}


async def fetch_business_segments(code: str, by_type: str = "product", years: int = 3) -> Optional[list[dict]]:
    """东方财富经营分析 - 主营构成（按产品/行业/地区），缓存6小时

    Args:
        code: 6位股票代码
        by_type: "product"|"industry"|"region"
        years: 返回最近N年的数据
    """
    seg_type = _SEGMENT_TYPE_MAP.get(by_type, "2")
    cache_key = f"business_segments:{code}:{by_type}:{years}"
    cached = get_cache(cache_key)
    if cached is not None:
        return json.loads(cached)

    secucode = _code_to_secucode(code)
    url = "https://datacenter.eastmoney.com/securities/api/data/v1/get"
    params = {
        "reportName": "RPT_F10_FN_SEGMENTSV",
        "columns": (
            "SECUCODE,SECURITY_CODE,SECURITY_NAME_ABBR,MAINOP_TYPE,ITEM_NAME,"
            "ITEM_CODE,ITEM_PARENT_CODE,MAIN_BUSINESS_INCOME,MAIN_BUSINESS_RPOFIT,"
            "GROSS_RPOFIT_RATIO,MBI_RATIO,MBR_RATIO,REPORT_DATE,REPORT_NAME,"
            "REPORT_DATE_CODE,MAIN_BUSINESS_COST,MBC_RATIO"
        ),
        "filter": f'(SECUCODE="{secucode}")(MAINOP_TYPE="{seg_type}")(ITEM_PARENT_CODE="999999")',
        "pageNumber": 1,
        "pageSize": 200,
        "sortTypes": "1,1",
        "sortColumns": "MAINOP_TYPE,RANK3",
    }
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Referer": "https://datacenter.eastmoney.com/",
    }

    async with httpx.AsyncClient() as client:
        try:
            resp = await client.get(url, params=params, headers=headers, timeout=15)
            if resp.status_code != 200:
                return None
            body = resp.json()
            if not body.get("result") or not body["result"].get("data"):
                return None
            records = body["result"]["data"]
        except Exception as e:
            print(f"[eastmoney] fetch_business_segments error: {e}")
            return None

    # 按报告期分组
    from collections import OrderedDict
    period_map: dict[str, list[dict]] = OrderedDict()
    for r in records:
        report_name = r.get("REPORT_NAME", "")
        report_date = r.get("REPORT_DATE", "")
        item_name = r.get("ITEM_NAME", "")
        income = r.get("MAIN_BUSINESS_INCOME")
        cost = r.get("MAIN_BUSINESS_COST")
        profit = r.get("MAIN_BUSINESS_RPOFIT")
        gross_ratio = r.get("GROSS_RPOFIT_RATIO")
        mbi_ratio = r.get("MBI_RATIO")
        mbr_ratio = r.get("MBR_RATIO")
        mbc_ratio = r.get("MBC_RATIO")

        entry = {
            "itemName": item_name,
            "income": income,
            "cost": cost,
            "profit": profit,
            "grossRatio": round(gross_ratio * 100, 2) if gross_ratio is not None else None,
            "incomeRatio": round(mbi_ratio * 100, 2) if mbi_ratio is not None else None,
            "profitRatio": round(mbr_ratio * 100, 2) if mbr_ratio is not None else None,
            "costRatio": round(mbc_ratio * 100, 2) if mbc_ratio is not None else None,
        }

        if report_name not in period_map:
            period_map[report_name] = []
        period_map[report_name].append(entry)

    # 补充 profitRatio：若 API 未返回，用该期总利润推算各项目利润占比
    for pname, items in period_map.items():
        total_profit = sum(it["profit"] for it in items if it["profit"] is not None)
        if total_profit != 0:
            for it in items:
                if it["profitRatio"] is None and it["profit"] is not None:
                    it["profitRatio"] = round(it["profit"] / total_profit * 100, 2)

    # 取最近N年
    all_periods = list(period_map.keys())
    all_periods.sort(reverse=True)
    selected = all_periods[:max(years * 4, 4)]

    result = []
    for p in selected:
        result.append({
            "reportName": p,
            "items": period_map[p],
        })

    set_cache(cache_key, json.dumps(result, ensure_ascii=False), ttl_hours=6)
    return result


async def fetch_kline_history(code: str, days: int = 90) -> Optional[list[dict]]:
    """从 push2his kline/get 获取日K线（curl_cffi 模拟浏览器 TLS 指纹）

    可能受 IP 限流影响，调用方应有降级。
    返回: [{date, open, close, high, low, volume(手), turnover, changePercent, amplitude, turnoverRate}]
    """
    market = get_eastmoney_market(code)
    secid = f"{market}.{code}"
    url = (
        f"https://push2his.eastmoney.com/api/qt/stock/kline/get"
        f"?secid={secid}"
        f"&ut=b2884a393a59ad64002292a3e90d46a5"
        f"&fields1=f1,f2,f3,f4,f5,f6"
        f"&fields2=f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61"
        f"&klt=101&fqt=1&end=20500101&lmt={days}"
    )

    from curl_cffi.requests import AsyncSession
    async with AsyncSession(impersonate="chrome120") as session:
        for attempt in range(2):
            try:
                resp = await session.get(url, headers={
                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
                    "Referer": "https://quote.eastmoney.com/",
                }, timeout=10)
                if resp.status_code != 200:
                    if attempt == 0:
                        continue
                    return None
                result = resp.json()
                if result.get("rc") != 0:
                    if attempt == 0:
                        continue
                    return None
                klines = result.get("data", {}).get("klines", [])
                if not klines:
                    if attempt == 0:
                        continue
                    return None

                parsed = []
                for line in klines:
                    fields = line.split(",")
                    if len(fields) < 11:
                        continue
                    def _f(i): return float(fields[i]) if fields[i] else 0
                    parsed.append({
                        "date": fields[0],
                        "open": _f(1),
                        "close": _f(2),
                        "high": _f(3),
                        "low": _f(4),
                        "volume": int(_f(5)),  # 手
                        "turnover": _f(6),
                        "amplitude": _f(7),
                        "changePercent": _f(8),
                        "change": _f(9),
                        "turnoverRate": _f(10),
                    })

                parsed.reverse()
                return parsed
            except Exception as e:
                if attempt == 0:
                    continue
                return None
    return None


async def fetch_fund_flow_daykline(code: str, days: int) -> Optional[list[dict]]:
    """从 push2his daykline/get 获取资金流向日线数据（结构化返回，替代 MX API）

    返回字段：date, mainNetInflow, superLargeInflow/Outflow, largeInflow/Outflow,
              mediumInflow/Outflow, smallInflow/Outflow, 各占比%, closePrice, changePercent,
              mainForceNet, retailNet
    """
    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"
        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
                    return None
                result = resp.json()
                klines = result.get("data", {}).get("klines", [])
                if not klines:
                    if attempt == 0:
                        continue
                    return None

                parsed = []
                for line in klines:
                    fields = line.split(",")
                    if len(fields) < 13:
                        continue
                    def _f(i): return float(fields[i]) if fields[i] else 0

                    # 字段索引: 0=f51(date), 1=f52(主力), 2=f53(小单), 3=f54(中单),
                    # 4=f55(大单), 5=f56(超大单), 6=f57(主力%), 7=f58(小单%),
                    # 8=f59(中单%), 9=f60(大单%), 10=f61(超大单%), 11=f62(收盘价), 12=f63(涨跌幅)
                    date = fields[0]
                    main_net = _f(1)
                    small_net = _f(2)
                    medium_net = _f(3)
                    large_net = _f(4)
                    super_large_net = _f(5)

                    parsed.append({
                        "date": date,
                        "mainNetInflow": main_net,
                        "superLargeInflow": max(0, super_large_net),
                        "superLargeOutflow": abs(min(0, super_large_net)),
                        "largeInflow": max(0, large_net),
                        "largeOutflow": abs(min(0, large_net)),
                        "mediumInflow": max(0, medium_net),
                        "mediumOutflow": abs(min(0, medium_net)),
                        "smallInflow": max(0, small_net),
                        "smallOutflow": abs(min(0, small_net)),
                        "mainNetInflowPercent": _f(6),
                        "superLargeInflowPercent": _f(10),
                        "superLargeOutflowPercent": 0,
                        "largeInflowPercent": _f(9),
                        "largeOutflowPercent": 0,
                        "mediumInflowPercent": _f(8),
                        "mediumOutflowPercent": 0,
                        "smallInflowPercent": _f(7),
                        "smallOutflowPercent": 0,
                        "turnover": 0,  # daykline 不含成交额，由调用方从 Tencent K 线合并
                        "mainForceNet": main_net,
                        "retailNet": small_net + medium_net,
                        "changePercent": _f(12),
                        "closePrice": _f(11),
                    })

                return parsed
            except Exception as e:
                if attempt == 0:
                    continue
                print(f"[eastmoney] fetch_fund_flow_daykline error: {e}")
                return None
    return None