EMWebApi/api.py

"""
东方财富财务摘要 API (FastAPI 封装)
GET /fin/summary?stocks=603233.SH:10500736,600519.SH:xxxxxx
GET /fin/summary?stocks=603233.SH  (只传代码, 自动查 ORG_CODE, 不稳)

返回 JSON: {success, count, columns, data, errors}
"""
from __future__ import annotations

from typing import Optional

from fastapi import FastAPI, Query, HTTPException
from fastapi.responses import JSONResponse
import pandas as pd

# 复用 fin_summary.py 里的核心逻辑
from utils.fin_summary import get_fin_summary, _fetch_org_codes
from utils.fin_quarterly import get_fin_quarterly_summary
from utils.fin_main_business import get_main_business
from utils.fin_balance import get_balance
from utils.fin_income import get_income
from utils.fin_cashflow import get_cashflow
from utils.fin_profitability import get_profitability
from utils.fin_capital_structure import get_capital_structure
from utils.fin_operate import get_operate_ability
from utils.fin_growth import get_growth
from utils.fin_dupont import get_dupont
from utils.fin_dividend import get_dividend_detail, get_dividend_statics
from utils.fin_free_holders import get_free_holders
from utils.fin_holders import get_holders
from utils.fin_holder_num import get_holder_num
from utils.fin_mda import get_mda
from utils.fin_equity import get_equity
from utils.fin_company import get_company_info, get_industry
from utils.fin_belong_index import get_index_classif, get_belong_index, get_index_membership
from utils.fin_major_events import (
    get_acquisitions,
    get_equity_incentive,
    get_litigation,
    get_related_transaction,
    get_violation,
    get_guarantee,
)


app = FastAPI(
    title="EMWeb API",
    description="东方财富 CHOICE API 数据抓取服务",
    version="0.1.0",
)


def _parse_stocks(raw: str) -> tuple[list[str], Optional[dict[str, str]]]:
    """
    解析 ?stocks= 参数.
      "603233.SH:10500736,600519.SH:xxxxxx" -> (codes, {code: org_code})
      "603233.SH,600519.SH"                 -> (codes, None) -> 自动查
      "603233.SH:10500736"                  -> (codes, {code: org_code}) 单只
    """
    codes: list[str] = []
    org_map: dict[str, str] = {}
    has_any_org = False
    for part in raw.split(","):
        part = part.strip()
        if not part:
            continue
        if ":" in part:
            code, org = part.split(":", 1)
            code = code.strip()
            org = org.strip()
            codes.append(code)
            if org:
                org_map[code] = org
                has_any_org = True
        else:
            codes.append(part.strip())
    if has_any_org and len(org_map) != len(codes):
        # 部分给了 ORG_CODE 部分没给 -> 退化为全部自动查
        return codes, None
    return codes, (org_map if has_any_org else None)


def _df_to_json_payload(df: pd.DataFrame) -> dict:
    """DataFrame -> JSON-safe dict."""
    import math
    safe = df.copy()
    # 处理 NaN / NaT / Timestamp / inf -> None
    for col in safe.columns:
        if pd.api.types.is_datetime64_any_dtype(safe[col]):
            safe[col] = safe[col].astype(str).replace("NaT", None)
    safe = safe.where(pd.notnull(safe), None)

    # 二次清理: inf / -inf / 极大值
    records = safe.to_dict(orient="records")
    cleaned = []
    for row in records:
        new_row = {}
        for k, v in row.items():
            if v is None:
                new_row[k] = None
            elif isinstance(v, float):
                if math.isnan(v) or math.isinf(v):
                    new_row[k] = None
                else:
                    new_row[k] = v
            else:
                new_row[k] = v
        cleaned.append(new_row)
    return {
        "success": True,
        "count": int(len(cleaned)),
        "columns": list(safe.columns),
        "data": cleaned,
    }


def _empty_response(message: str, codes: list[str], **extra) -> JSONResponse:
    """无数据时的统一 404 返回."""
    content = {
        "success": False,
        "count": 0,
        "message": message,
        "codes_requested": codes,
        "data": [],
    }
    content.update(extra)
    return JSONResponse(status_code=404, content=content)


@app.get("/")
def root():
    return {
        "name": "EMWeb API",
        "version": "0.1.0",
        "endpoints": {
            "GET /fin/summary": "拉取财务摘要(报告期)",
            "GET /fin/quarterly-summary": "拉取季度财务摘要",
            "GET /fin/main-business": "拉取主营构成(按产品/按地区)",
            "GET /fin/balance": "拉取资产负债表",
            "GET /fin/income": "拉取利润表",
            "GET /fin/cashflow": "拉取现金流量表",
            "GET /fin/profitability": "拉取盈利能力与收益质量",
            "GET /fin/capital-structure": "拉取资本结构与偿债能力",
            "GET /fin/operate-ability": "拉取营运能力",
            "GET /fin/growth": "拉取成长能力",
            "GET /fin/dupont": "拉取杜邦分析",
            "GET /fin/dividend-detail": "拉取分红明细",
            "GET /fin/dividend-statics": "拉取分红统计",
            "GET /fin/free-holders": "拉取十大流通股东",
            "GET /fin/holders": "拉取十大股东明细",
            "GET /fin/holder-num": "拉取股东户数",
            "GET /fin/mda": "拉取管理层讨论与分析(年报/中报)",
            "GET /fin/equity": "拉取股本结构",
            "GET /fin/company-info": "拉取公司介绍",
            "GET /fin/industry": "拉取所属行业",
            "GET /fin/index-classif": "拉取所属指数分类",
            "GET /fin/belong-index": "拉取纳入指数统计",
            "GET /fin/index-membership": "拉取指数成分权重",
            "GET /fin/acquisitions": "拉取并购事件",
            "GET /fin/equity-incentive": "拉取股权激励",
            "GET /fin/litigation": "拉取诉讼仲裁",
            "GET /fin/related-transaction": "拉取关联交易",
            "GET /fin/violation": "拉取违规处罚",
            "GET /fin/guarantee": "拉取对外担保",
            "GET /health": "健康检查",
        },
        "usage": {
            "fin_summary": "/fin/summary?stocks=603233.SH:10500736&years=10",
            "fin_quarterly_summary": "/fin/quarterly-summary?stocks=603233.SH&years=3",
            "fin_main_business": "/fin/main-business?stocks=603233.SH&classify_type=产品&years=3",
            "fin_balance": "/fin/balance?stocks=603233.SH&years=3",
            "fin_dividend_detail": "/fin/dividend-detail?stocks=603233.SH&years=5",
            "fin_mda": "/fin/mda?stocks=603233.SH&years=5",
            "fin_company_info": "/fin/company-info?stocks=603233.SH",
        },
    }


@app.get("/health")
def health():
    return {"status": "ok"}


@app.get("/fin/summary")
def fin_summary(
    stocks: str = Query(
        ...,
        description=(
            "股票列表, 逗号分隔. 支持仅代码自动查询；其中 '603233.SH' "
            "会自动解析为 ORG_CODE '10500736'。也可使用代码:ORG_CODE，"
            "如 '603233.SH:10500736,600519.SH:xxxxxx'"
        ),
        examples=["603233.SH", "603233.SH:10500736"],
    ),
    years: int = Query(10, ge=1, le=20, description="拉取多少年的历史"),
    date_type_codes: str = Query(
        "1,5,6",
        description="报告期代码, 逗号分隔. 1=一季报, 5=中报, 6=年报",
    ),
    is_newest: bool = Query(True, description="是否只要最新合并报表"),
):
    """拉取多只股票的财务摘要数据, 返回 JSON."""
    codes, org_map = _parse_stocks(stocks)
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    # 自动查 ORG_CODE
    if org_map is None:
        try:
            org_map = _fetch_org_codes(codes)
        except Exception as e:  # noqa: BLE001
            raise HTTPException(
                400,
                detail=(
                    f"自动查询失败: {e}. 请改用 '代码:ORG_CODE' 格式, "
                    f"ORG_CODE 可在 Reqable 抓包切换股票时拿到."
                ),
            )

    df = get_fin_summary(
        org_map,
        years=years,
        date_type_codes=date_type_codes,
        is_newest=is_newest,
        verbose=False,
    )
    if df.empty:
        return JSONResponse(
            status_code=404,
            content={
                "success": False,
                "count": 0,
                "message": "无数据, 请检查 ORG_CODE 或放宽筛选条件",
                "codes_requested": codes,
                "data": [],
            },
        )

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["years"] = years
    payload["date_type_codes"] = date_type_codes
    return payload


@app.get("/fin/quarterly-summary")
def fin_quarterly_summary(
    stocks: str = Query(
        ...,
        description="股票代码，逗号分隔，例如 '603233.SH,600519.SH'。仅支持代码，不需要 ORG_CODE。",
        examples=["603233.SH"],
    ),
    years: int = Query(3, ge=1, le=20, description="未传 report_dates 时拉取最近多少年的季度数据"),
    report_dates: Optional[str] = Query(
        None,
        description=(
            "可选，逗号分隔的 YYYY-MM-DD 报告期。未传时取最近 years 年季度数据；"
            "传入时精确筛选报告期，并按股票代码和报告期自动去重。"
        ),
        examples=["2025-03-31,2024-12-31"],
    ),
):
    """拉取季度财务摘要；指定报告期时执行精确报告期筛选。"""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    dates = None
    if report_dates is not None:
        dates = [date.strip() for date in report_dates.split(",") if date.strip()]
        if not dates:
            raise HTTPException(400, detail="report_dates 参数不能为空")

    df = get_fin_quarterly_summary(codes, years=years, report_dates=dates)
    if df.empty:
        return JSONResponse(
            status_code=404,
            content={
                "success": False,
                "count": 0,
                "message": "无季度财务摘要数据, 请检查股票代码或筛选条件",
                "codes_requested": codes,
                "years": years,
                "data": [],
            },
        )

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["years"] = years
    return payload


@app.get("/fin/main-business")
def fin_main_business(
    stocks: str = Query(
        ...,
        description="股票代码，逗号分隔，例如 '603233.SH,600519.SH'。仅支持代码，不需要 ORG_CODE。",
        examples=["603233.SH"],
    ),
    classify_type: str = Query(
        "产品",
        description="分类方式: 产品 或 地区",
        examples=["产品", "地区"],
    ),
    years: int = Query(3, ge=1, le=20, description="拉取最近多少年的主营构成数据"),
):
    """拉取主营构成(按产品或按地区)。"""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    df = get_main_business(codes, classify_type=classify_type, years=years)
    if df.empty:
        return JSONResponse(
            status_code=404,
            content={
                "success": False,
                "count": 0,
                "message": "无主营构成数据, 请检查股票代码或筛选条件",
                "codes_requested": codes,
                "classify_type": classify_type,
                "data": [],
            },
        )

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["classify_type"] = classify_type
    payload["years"] = years
    return payload


@app.get("/fin/balance")
def fin_balance(
    stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]),
    years: int = Query(3, ge=1, le=20, description="拉取最近多少年的资产负债表数据"),
):
    """拉取资产负债表。"""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    df = get_balance(codes, years=years)
    if df.empty:
        return _empty_response("无资产负债表数据, 请检查股票代码或筛选条件", codes, years=years)

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["years"] = years
    return payload


@app.get("/fin/income")
def fin_income(
    stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]),
    years: int = Query(3, ge=1, le=20, description="拉取最近多少年的利润表数据"),
):
    """拉取利润表。"""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    df = get_income(codes, years=years)
    if df.empty:
        return _empty_response("无利润表数据, 请检查股票代码或筛选条件", codes, years=years)

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["years"] = years
    return payload


@app.get("/fin/cashflow")
def fin_cashflow(
    stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]),
    years: int = Query(3, ge=1, le=20, description="拉取最近多少年的现金流量表数据"),
):
    """拉取现金流量表。"""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    df = get_cashflow(codes, years=years)
    if df.empty:
        return _empty_response("无现金流量表数据, 请检查股票代码或筛选条件", codes, years=years)

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["years"] = years
    return payload


@app.get("/fin/profitability")
def fin_profitability(
    stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]),
    years: int = Query(3, ge=1, le=20, description="拉取最近多少年的盈利能力与收益质量数据"),
):
    """拉取盈利能力与收益质量。"""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    df = get_profitability(codes, years=years)
    if df.empty:
        return _empty_response("无盈利能力与收益质量数据, 请检查股票代码或筛选条件", codes, years=years)

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["years"] = years
    return payload


@app.get("/fin/capital-structure")
def fin_capital_structure(
    stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]),
    years: int = Query(3, ge=1, le=20, description="拉取最近多少年的资本结构与偿债能力数据"),
):
    """拉取资本结构与偿债能力。"""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    df = get_capital_structure(codes, years=years)
    if df.empty:
        return _empty_response("无资本结构与偿债能力数据, 请检查股票代码或筛选条件", codes, years=years)

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["years"] = years
    return payload


@app.get("/fin/operate-ability")
def fin_operate_ability(
    stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]),
    years: int = Query(3, ge=1, le=20, description="拉取最近多少年的营运能力数据"),
):
    """拉取营运能力。"""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    df = get_operate_ability(codes, years=years)
    if df.empty:
        return _empty_response("无营运能力数据, 请检查股票代码或筛选条件", codes, years=years)

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["years"] = years
    return payload


@app.get("/fin/growth")
def fin_growth(
    stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]),
    years: int = Query(3, ge=1, le=20, description="拉取最近多少年的成长能力数据"),
):
    """拉取成长能力。"""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    df = get_growth(codes, years=years)
    if df.empty:
        return _empty_response("无成长能力数据, 请检查股票代码或筛选条件", codes, years=years)

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["years"] = years
    return payload


@app.get("/fin/dupont")
def fin_dupont(
    stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]),
    years: int = Query(3, ge=1, le=20, description="拉取最近多少年的杜邦分析数据"),
):
    """拉取杜邦分析。"""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    df = get_dupont(codes, years=years)
    if df.empty:
        return _empty_response("无杜邦分析数据, 请检查股票代码或筛选条件", codes, years=years)

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["years"] = years
    return payload


@app.get("/fin/dividend-detail")
def fin_dividend_detail(
    stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]),
    years: int = Query(5, ge=1, le=20, description="拉取最近多少年的分红明细"),
):
    """拉取分红明细。"""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    df = get_dividend_detail(codes, years=years)
    if df.empty:
        return _empty_response("无分红明细数据, 请检查股票代码或筛选条件", codes, years=years)

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["years"] = years
    return payload


@app.get("/fin/dividend-statics")
def fin_dividend_statics(
    stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]),
    years: int = Query(5, ge=1, le=20, description="拉取最近多少年的分红统计"),
):
    """拉取分红统计。"""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    df = get_dividend_statics(codes, years=years)
    if df.empty:
        return _empty_response("无分红统计数据, 请检查股票代码或筛选条件", codes, years=years)

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["years"] = years
    return payload


@app.get("/fin/free-holders")
def fin_free_holders(
    stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]),
    years: int = Query(3, ge=1, le=20, description="拉取最近多少年的十大流通股东"),
):
    """拉取十大流通股东。"""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    df = get_free_holders(codes, years=years)
    if df.empty:
        return _empty_response("无十大流通股东数据, 请检查股票代码或筛选条件", codes, years=years)

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["years"] = years
    return payload


@app.get("/fin/holders")
def fin_holders(
    stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]),
    years: int = Query(3, ge=1, le=20, description="拉取最近多少年的十大股东明细"),
):
    """拉取十大股东明细。"""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    df = get_holders(codes, years=years)
    if df.empty:
        return _empty_response("无十大股东明细数据, 请检查股票代码或筛选条件", codes, years=years)

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["years"] = years
    return payload


@app.get("/fin/holder-num")
def fin_holder_num(
    stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]),
    years: int = Query(3, ge=1, le=20, description="拉取最近多少年的股东户数"),
):
    """拉取股东户数。"""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")

    df = get_holder_num(codes, years=years)
    if df.empty:
        return _empty_response("无股东户数数据, 请检查股票代码或筛选条件", codes, years=years)

    payload = _df_to_json_payload(df)
    payload["codes_requested"] = codes
    payload["years"] = years
    return payload


def _split_codes(stocks: str) -> list[str]:
    """解析逗号分隔的股票代码参数."""
    codes = [code.strip() for code in stocks.split(",") if code.strip()]
    if not codes:
        raise HTTPException(400, detail="stocks 参数不能为空")
    return codes


def _simple_endpoint(path: str, fetch, label: str, default_years: int, has_years: bool = True):
    """生成统一的 GET 端点: 只有 stocks (+years) 参数的接口."""
    def endpoint(
        stocks: str = Query(..., description="股票代码，逗号分隔，例如 '603233.SH'。仅支持代码，不需要 ORG_CODE。", examples=["603233.SH"]),
        years: Optional[int] = Query(default_years if has_years else None, ge=1, le=20, description=f"拉取最近多少年的{label}"),
    ):
        codes = _split_codes(stocks)
        df = fetch(codes) if years is None else fetch(codes, years=years)
        if df.empty:
            extra = {"years": years} if years is not None else {}
            return _empty_response(f"无{label}数据, 请检查股票代码或筛选条件", codes, **extra)
        payload = _df_to_json_payload(df)
        payload["codes_requested"] = codes
        if years is not None:
            payload["years"] = years
        return payload

    app.get(path)(endpoint)
    return endpoint


# ---- 管理层讨论与分析 / 股本结构 (无 years) ----
_simple_endpoint("/fin/mda", get_mda, "管理层讨论与分析", default_years=5)
_simple_endpoint("/fin/equity", get_equity, "股本结构", default_years=3, has_years=False)
_simple_endpoint("/fin/company-info", get_company_info, "公司介绍", default_years=3, has_years=False)
_simple_endpoint("/fin/industry", get_industry, "所属行业", default_years=3, has_years=False)
_simple_endpoint("/fin/index-classif", get_index_classif, "所属指数分类", default_years=3, has_years=False)
_simple_endpoint("/fin/belong-index", get_belong_index, "纳入指数统计", default_years=3, has_years=False)
_simple_endpoint("/fin/index-membership", get_index_membership, "指数成分权重", default_years=3, has_years=False)

# ---- 重大事项六类 ----
_simple_endpoint("/fin/acquisitions", get_acquisitions, "并购事件", default_years=3)
_simple_endpoint("/fin/equity-incentive", get_equity_incentive, "股权激励", default_years=3)
_simple_endpoint("/fin/litigation", get_litigation, "诉讼仲裁", default_years=3)
_simple_endpoint("/fin/related-transaction", get_related_transaction, "关联交易", default_years=3)
_simple_endpoint("/fin/violation", get_violation, "违规处罚", default_years=3)
_simple_endpoint("/fin/guarantee", get_guarantee, "对外担保", default_years=3)


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8765, log_level="info")