You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
38 lines
1.5 KiB
38 lines
1.5 KiB
"""东方财富 CHOICE 接口公共请求辅助 (仅内部模块复用, 不对外暴露)。"""
|
|
from __future__ import annotations
|
|
|
|
import requests
|
|
|
|
from utils.fin_summary import HEADERS, TIMEOUT
|
|
|
|
|
|
# emchoicews2 的 f9 页面接口, 返回带 urls 的跳转结构
|
|
_EMWS_URL = "https://emchoicews2.eastmoney.com"
|
|
# datacenter-choice 报表数据接口
|
|
_BASE_URL = "https://datacenter-choice.eastmoney.com/api/data/v1/get"
|
|
|
|
|
|
def get_data_url(path: str, params: dict[str, str]) -> str:
|
|
"""调用 emchoicews2 的 f9 路径, 返回真正的报表数据接口 URL."""
|
|
r = requests.get(_EMWS_URL + path, headers=HEADERS, params=params, timeout=TIMEOUT)
|
|
r.raise_for_status()
|
|
payload = r.json()
|
|
urls = payload.get("urls") or []
|
|
if not urls:
|
|
raise RuntimeError(f"{path}: 接口未返回数据 URL: {payload}")
|
|
return urls[0]
|
|
|
|
|
|
def fetch_report(url_or_params: str | dict[str, str]) -> list[dict]:
|
|
"""拉取 datacenter-choice 报表数据, 返回行 dict 列表.
|
|
支持直接传完整 URL (由 get_data_url 得到), 或传 reportName 等参数 dict.
|
|
"""
|
|
if isinstance(url_or_params, dict):
|
|
r = requests.get(_BASE_URL, headers=HEADERS, params=url_or_params, timeout=TIMEOUT)
|
|
else:
|
|
r = requests.get(url_or_params, headers=HEADERS, timeout=TIMEOUT)
|
|
r.raise_for_status()
|
|
payload = r.json()
|
|
if not payload.get("success"):
|
|
raise RuntimeError(f"接口返回失败: {payload}")
|
|
return payload.get("result", {}).get("data") or []
|
|
|