import asyncio
import httpx
import math
import pandas as pd
from datetime import datetime, timezone, timedelta
import logging

logging.getLogger("httpx").setLevel(logging.WARNING)

HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/122.0.0.0 Safari/537.36"
    ),
    "Referer": "https://quote.eastmoney.com/center/gridlist.html#fund_etf",  # 合理来源页
    "Accept": "application/json, text/plain, */*",
    "Accept-Language": "zh-CN,zh;q=0.9",
    "Connection": "keep-alive",
}


async def fetch_page(client: httpx.AsyncClient, url: str, params: dict) -> pd.DataFrame:
    try:
        resp = await client.get(url, params=params, timeout=10)
        data_json = resp.json()
        if "data" in data_json and data_json["data"].get("diff"):
            return pd.DataFrame(data_json["data"]["diff"])
    except Exception as e:
        print(f"Error on page {params.get('pn')}: {e}")
    return pd.DataFrame()


async def fund_etf_spot_em_async() -> pd.DataFrame:
    url = "https://88.push2.eastmoney.com/api/qt/clist/get"
    params = {
        "pn": "1",
        "pz": "100",
        "po": "1",
        "np": "1",
        "ut": "bd1d9ddb04089700cf9c27f6f7426281",
        "fltt": "2",
        "invt": "2",
        "wbp2u": "|0|0|0|web",
        "fid": "f12",
        "fs": "b:MK0021,b:MK0022,b:MK0023,b:MK0024,b:MK0827",
        "fields": (
            "f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,"
            "f12,f13,f14,f15,f16,f17,f18,f20,f21,"
            "f23,f24,f25,f22,f11,f30,f31,f32,f33,"
            "f34,f35,f38,f62,f63,f64,f65,f66,f69,"
            "f72,f75,f78,f81,f84,f87,f115,f124,f128,"
            "f136,f152,f184,f297,f402,f441"
        ),
    }

    async with httpx.AsyncClient(headers=HEADERS, http2=True) as client:
        # 获取首页数据，确认总页数
        resp = await client.get(url, params=params)
        data_json = resp.json()
        if "data" not in data_json or not data_json["data"].get("diff"):
            return pd.DataFrame()

        total = data_json["data"]["total"]
        per_page = len(data_json["data"]["diff"])
        total_pages = math.ceil(total / per_page)

        tasks = []
        for page in range(1, total_pages + 1):
            new_params = params.copy()
            new_params["pn"] = str(page)
            tasks.append(fetch_page(client, url, new_params))

        dfs = await asyncio.gather(*tasks)
        temp_df = pd.concat([df for df in dfs if not df.empty], ignore_index=True)

        temp_df.rename(
            columns={
                "f12": "代码",
                "f14": "名称",
                "f2": "最新价",
                "f4": "涨跌额",
                "f3": "涨跌幅",
                "f5": "成交量",
                "f6": "成交额",
                "f7": "振幅",
                "f17": "开盘价",
                "f15": "最高价",
                "f16": "最低价",
                "f18": "昨收",
                "f8": "换手率",
                "f10": "量比",
                "f30": "现手",
                "f31": "买一",
                "f32": "卖一",
                "f33": "委比",
                "f34": "外盘",
                "f35": "内盘",
                "f62": "主力净流入-净额",
                "f184": "主力净流入-净占比",
                "f66": "超大单净流入-净额",
                "f69": "超大单净流入-净占比",
                "f72": "大单净流入-净额",
                "f75": "大单净流入-净占比",
                "f78": "中单净流入-净额",
                "f81": "中单净流入-净占比",
                "f84": "小单净流入-净额",
                "f87": "小单净流入-净占比",
                "f38": "最新份额",
                "f21": "流通市值",
                "f20": "总市值",
                "f402": "基金折价率",
                "f441": "IOPV实时估值",
                "f297": "数据日期",
                "f124": "更新时间",
            },
            inplace=True,
        )
        temp_df = temp_df[
            [
                "代码",
                "名称",
                "最新价",
                "IOPV实时估值",
                "基金折价率",
                "涨跌额",
                "涨跌幅",
                "成交量",
                "成交额",
                "开盘价",
                "最高价",
                "最低价",
                "昨收",
                "振幅",
                "换手率",
                "量比",
                "委比",
                "外盘",
                "内盘",
                "主力净流入-净额",
                "主力净流入-净占比",
                "超大单净流入-净额",
                "超大单净流入-净占比",
                "大单净流入-净额",
                "大单净流入-净占比",
                "中单净流入-净额",
                "中单净流入-净占比",
                "小单净流入-净额",
                "小单净流入-净占比",
                "现手",
                "买一",
                "卖一",
                "最新份额",
                "流通市值",
                "总市值",
                "数据日期",
                "更新时间",
            ]
        ].copy()
        temp_df["最新价"] = pd.to_numeric(temp_df["最新价"], errors="coerce")
        temp_df["涨跌额"] = pd.to_numeric(temp_df["涨跌额"], errors="coerce")
        temp_df["涨跌幅"] = pd.to_numeric(temp_df["涨跌幅"], errors="coerce")
        temp_df["成交量"] = pd.to_numeric(temp_df["成交量"], errors="coerce")
        temp_df["成交额"] = pd.to_numeric(temp_df["成交额"], errors="coerce")
        temp_df["开盘价"] = pd.to_numeric(temp_df["开盘价"], errors="coerce")
        temp_df["最高价"] = pd.to_numeric(temp_df["最高价"], errors="coerce")
        temp_df["最低价"] = pd.to_numeric(temp_df["最低价"], errors="coerce")
        temp_df["昨收"] = pd.to_numeric(temp_df["昨收"], errors="coerce")
        temp_df["换手率"] = pd.to_numeric(temp_df["换手率"], errors="coerce")
        temp_df["量比"] = pd.to_numeric(temp_df["量比"], errors="coerce")
        temp_df["委比"] = pd.to_numeric(temp_df["委比"], errors="coerce")
        temp_df["外盘"] = pd.to_numeric(temp_df["外盘"], errors="coerce")
        temp_df["内盘"] = pd.to_numeric(temp_df["内盘"], errors="coerce")
        temp_df["流通市值"] = pd.to_numeric(temp_df["流通市值"], errors="coerce")
        temp_df["总市值"] = pd.to_numeric(temp_df["总市值"], errors="coerce")
        temp_df["振幅"] = pd.to_numeric(temp_df["振幅"], errors="coerce")
        temp_df["现手"] = pd.to_numeric(temp_df["现手"], errors="coerce")
        temp_df["买一"] = pd.to_numeric(temp_df["买一"], errors="coerce")
        temp_df["卖一"] = pd.to_numeric(temp_df["卖一"], errors="coerce")
        temp_df["最新份额"] = pd.to_numeric(temp_df["最新份额"], errors="coerce")
        temp_df["IOPV实时估值"] = pd.to_numeric(temp_df["IOPV实时估值"], errors="coerce")
        temp_df["基金折价率"] = pd.to_numeric(temp_df["基金折价率"], errors="coerce")
        temp_df["主力净流入-净额"] = pd.to_numeric(
            temp_df["主力净流入-净额"], errors="coerce"
        )
        temp_df["主力净流入-净占比"] = pd.to_numeric(
            temp_df["主力净流入-净占比"], errors="coerce"
        )
        temp_df["超大单净流入-净额"] = pd.to_numeric(
            temp_df["超大单净流入-净额"], errors="coerce"
        )
        temp_df["超大单净流入-净占比"] = pd.to_numeric(
            temp_df["超大单净流入-净占比"], errors="coerce"
        )
        temp_df["大单净流入-净额"] = pd.to_numeric(
            temp_df["大单净流入-净额"], errors="coerce"
        )
        temp_df["大单净流入-净占比"] = pd.to_numeric(
            temp_df["大单净流入-净占比"], errors="coerce"
        )
        temp_df["中单净流入-净额"] = pd.to_numeric(
            temp_df["中单净流入-净额"], errors="coerce"
        )
        temp_df["中单净流入-净占比"] = pd.to_numeric(
            temp_df["中单净流入-净占比"], errors="coerce"
        )
        temp_df["小单净流入-净额"] = pd.to_numeric(
            temp_df["小单净流入-净额"], errors="coerce"
        )
        temp_df["小单净流入-净占比"] = pd.to_numeric(
            temp_df["小单净流入-净占比"], errors="coerce"
        )
        temp_df["数据日期"] = pd.to_datetime(
            temp_df["数据日期"], format="%Y%m%d", errors="coerce"
        )
        temp_df["更新时间"] = (
            pd.to_datetime(temp_df["更新时间"], unit="s", errors="coerce")
            .dt.tz_localize("UTC")
            .dt.tz_convert("Asia/Shanghai")
        )
    return temp_df


# 示例同步调用封装（如需在同步代码中使用）
def fund_etf_spot_em() -> pd.DataFrame:
    return asyncio.run(fund_etf_spot_em_async())


if __name__ == "__main__":
    df = fund_etf_spot_em()
    print(df.head())
