Bigdata Skill

Security scan passed
Scanned at: 2026-06-13T19:44:41.069633
Tool: gitleaks + pattern-based validation
Content hash: 1dc770d556b9a8d204518f8435cd0ee63398ff7a544faac168b48f397238a9fc

"""bigdata_toolkit —— Bigdata.com 可复用工具库
================================================

一个 class 两种能力：SDK 高层封装 + ``bd._api`` REST 逃生舱直通。

基于对 ``bigdata-client`` SDK 的运行时实测（L4）构建，**不编 API**。
每个能力都标注走 SDK 还是走 REST 逃生舱，见各子模块 docstring。

快速开始
--------
>>> import os
>>> os.environ["BIGDATA_API_KEY"] = "bd_v2_xxx"        # doctest: +SKIP
>>> os.environ["HTTPS_PROXY"] = "http://127.0.0.1:8080"  # 仅在需要出站代理时  # doctest: +SKIP
>>> from bigdata_toolkit import BigdataClient, EntityResolver, StructuredDataREST
>>> client = BigdataClient()                            # doctest: +SKIP
>>> # 1) 实体解析（A 股用英文名/ISIN）
>>> resolver = EntityResolver(client)                   # doctest: +SKIP
>>> aapl = resolver.resolve_id("Apple")                 # doctest: +SKIP -> 'D8442A'
>>> # 2) SDK 没有的前瞻日历（走 REST 逃生舱）
>>> rest = StructuredDataREST(client)                   # doctest: +SKIP
>>> cal = rest.events_calendar(aapl, categories=["earnings-call"],
...                            start_date="2026-06-01", end_date="2026-12-31")  # doctest: +SKIP

模块速查
--------
- :mod:`~bigdata_toolkit.client`   —— 统一入口（SDK + REST 逃生舱）
- :mod:`~bigdata_toolkit.search`   —— 带标注 chunk 抽取（SDK）
- :mod:`~bigdata_toolkit.kg`       —— 实体解析 + ISIN crosswalk（SDK）
- :mod:`~bigdata_toolkit.rest_ext` —— SDK 缺失的结构化金融数据（REST 逃生舱）
- :mod:`~bigdata_toolkit.cost`     —— chunk 消耗追踪 + 配额意识
"""

from .client import BigdataClient, require_env
from .cost import (
    CHUNKS_PER_QUERY_UNIT,
    USD_PER_QUERY_UNIT,
    CostModel,
    CostTracker,
)
from .kg import EntityResolver, company_to_dict
from .rest_ext import BatchSearch, StructuredDataREST, fields_values_to_records
from .retry import RETRYABLE_MARKERS, rc, with_retry
from .search import (
    AnnotatedSearcher,
    chunk_to_dict,
    document_to_dict,
)

__version__ = "0.1.0"

__all__ = [
    # client
    "BigdataClient",
    "require_env",
    # search
    "AnnotatedSearcher",
    "chunk_to_dict",
    "document_to_dict",
    # kg
    "EntityResolver",
    "company_to_dict",
    # rest_ext
    "StructuredDataREST",
    "BatchSearch",
    "fields_values_to_records",
    # cost
    "CostTracker",
    "CostModel",
    "CHUNKS_PER_QUERY_UNIT",
    "USD_PER_QUERY_UNIT",
    # retry
    "rc",
    "with_retry",
    "RETRYABLE_MARKERS",
]

"""统一入口：一个 BigdataClient 暴露两种能力
=====================================================

1. **SDK 高层能力**（`self.bd`）—— 官方 `bigdata_client.Bigdata`
   封装好的 search / knowledge_graph / subscription / chat / watchlists。

2. **ad hoc REST 逃生舱**（`self.http`）—— `bd._api.http`
   （`RateLimitedHTTPWrapper`，base url = ``https://api.bigdata.com/``）。
   SDK 高层没暴露的 endpoint（events-calendar / analyst-estimates /
   latest-surprise / target-price / company-screener / quotas ...）全部
   走这里。认证（ApiKeyAuth 注 JWT）+ 代理（靠 ``HTTPS_PROXY`` 环境变量）
   **自动复用**，无需自己拼 header。

为什么需要逃生舱
----------------
``bigdata_client`` 的 ``BigdataConnection`` 只 wrap 了
search / chunks / knowledge-graph / chat / watchlists / uploads。
一整套 RavenPack 遗留的 ``/v1/*`` 结构化金融数据产品线（前瞻财报日历、
一致预期、财报 surprise、评级、目标价、screener）SDK 一个高层方法都没写，
但同一后端、同一 JWT，raw http 直达。

机制证据链（运行时 L4 实测，非文档推断）
-----------------------------------------
- ``bd._api`` 是 ``bigdata_client.connection.BigdataConnection``。
- 它持有 ``bd._api.http``（``RateLimitedHTTPWrapper``），``api_url='https://api.bigdata.com/'``。
- SDK 高层方法（query_chunks / by_ids / autosuggest / get_my_quota ...）
  全都内部 delegate 到 ``self.http.post(endpoint, json=...)`` /
  ``self.http.get(endpoint, params=...)``。
- 所以打 SDK 没暴露的 endpoint = 直接调 ``self.http.<verb>(相对路径, ...)``。

路由形态规律（避免踩坑）
------------------------
- 业务面是 ``POST /v1/<resource>/query``，**裸 ``GET /v1/<resource>`` 会 404**。
- 平台面（quotas）是 ``GET /v1/subscription/quotas``。
- ``403 'Missing Authentication Token'`` = API Gateway 说该路径上无此路由
  （不是权限拒绝），``404`` = 路径不存在。需用文档（docs.bigdata.com/llms.txt）
  确认确切 path，不要瞎猜。
"""

from __future__ import annotations

import inspect
import os
from typing import Any, Optional, Union

from bigdata_client import Bigdata

__all__ = ["BigdataClient", "require_env"]


def require_env(name: str) -> str:
    """读环境变量，缺失立即 fail-fast（NO FALLBACK 原则）。

    禁止用明文 default 兜底 secret —— 这正是会被 scanner 扫到的反模式。
    """
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(
            f"Missing required env var: {name}. "
            f"Set it before constructing BigdataClient, e.g. "
            f"`export {name}=...`"
        )
    return value


class BigdataClient:
    """Bigdata.com 统一客户端 —— SDK 高层 + REST 逃生舱二合一。

    Parameters
    ----------
    api_key:
        Bigdata API key（``bd_v2_...`` 形态）。默认从 ``BIGDATA_API_KEY``
        环境变量读取（**绝不**硬编码）。
    check_proxy:
        若为 True（默认）且未显式传 ``proxy``，构造时检查 ``HTTPS_PROXY``
        是否设置，未设则提醒（出站代理可走 env var 或 ``proxy=`` 参数，二选一）。
    verify_ssl:
        透传给官方 ``Bigdata(verify_ssl=...)``（``False`` 关校验，或传 CA 路径
        字符串，见 ssl_verification.md）。代理做 TLS 拦截 / 自签证书时，传代理
        CA 路径是正道，优于盲重试。默认 None（用 SDK 默认）。⚠️ 不建议 ``False``。
    proxy:
        透传给官方 ``Bigdata(proxy=...)``（构造方式见 proxy_configuration.md）；
        与 ``HTTPS_PROXY`` env 二选一。默认 None（走 env var 路径）。

    Examples
    --------
    >>> import os
    >>> os.environ["BIGDATA_API_KEY"] = "bd_v2_xxx"
    >>> os.environ["HTTPS_PROXY"] = "http://127.0.0.1:8080"  # 仅在需要出站代理时
    >>> client = BigdataClient()                  # doctest: +SKIP
    >>> companies = client.bd.knowledge_graph.find_companies("Apple")  # doctest: +SKIP
    >>> quota = client.get_quota_raw()            # doctest: +SKIP  # REST 逃生舱
    """

    #: REST 业务面路由前缀（仅文档用途，方法里仍传完整 endpoint）
    API_BASE = "https://api.bigdata.com/"

    def __init__(
        self,
        api_key: Optional[str] = None,
        *,
        check_proxy: bool = True,
        verify_ssl: Optional[Union[bool, str]] = None,
        proxy: Optional[Any] = None,
    ) -> None:
        self.api_key = api_key or require_env("BIGDATA_API_KEY")

        if check_proxy and proxy is None and not (
            os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy")
        ):
            # 不抛错 —— 海外/无代理环境本就不需要。仅提醒。
            import warnings

            warnings.warn(
                "HTTPS_PROXY 未设置。若你的网络需要出站代理才能访问 "
                "api.bigdata.com，可 `export HTTPS_PROXY=http://<host>:<port>`，"
                "或给 BigdataClient(proxy=...) 传官方 Proxy 对象（二选一，见 "
                "proxy_configuration.md）。WebSocket（chat）另需 WSS_PROXY。",
                stacklevel=2,
            )

        # ---- SDK 高层入口（verify_ssl / proxy 是官方构造器参数，仅在显式
        #      传入时透传，默认不改 SDK 行为）----
        sdk_kwargs: dict[str, Any] = {"api_key": self.api_key}
        if verify_ssl is not None:
            sdk_kwargs["verify_ssl"] = verify_ssl
        if proxy is not None:
            sdk_kwargs["proxy"] = proxy
        self.bd = Bigdata(**sdk_kwargs)

    # ------------------------------------------------------------------ #
    # REST 逃生舱直通                                                     #
    # ------------------------------------------------------------------ #
    @property
    def http(self):
        """``bd._api.http`` —— RateLimitedHTTPWrapper（REST 直通入口）。

        签名（运行时确认）::

            http.get(endpoint: str, params: dict = None) -> dict | list
            http.post(endpoint: str, json: dict | list[dict]) -> dict | list
            http.put(endpoint: str, json) -> dict | list
            http.patch(endpoint: str, json) -> dict | list
            http.delete(endpoint: str) -> dict | list
            http.get_chunks(endpoint: str, chunk_size: int) -> Iterable[bytes]
            http.async_get(list[...]) -> 并发 GET

        endpoint 是相对路径（如 ``"v1/events-calendar/query"``），内部
        ``urljoin(api_url, endpoint)`` 拼成绝对 URL；也接受绝对 URL。
        """
        return self.bd._api.http

    @property
    def conn(self):
        """``bd._api`` —— BigdataConnection（"for internal use only"）。

        除 ``.http`` 外，它还有一批高层封装方法可直接复用：
        ``query_chunks`` / ``by_ids`` / ``autosuggest`` /
        ``query_discovery_panel`` / ``download_annotated_dict`` /
        ``get_my_quota`` / ``get_companies_by_isin`` 等。需要时优先用这些
        （它们内部已处理 request/response model），而非自己拼 raw http。
        """
        return self.bd._api

    def rest_get(self, endpoint: str, params: Optional[dict] = None) -> Any:
        """REST GET（平台面，如 quotas）。薄封装 ``self.http.get``。

        endpoint 形如 ``"v1/subscription/quotas"``（不带前导斜杠）。
        """
        return self.http.get(endpoint, params)

    def rest_post(self, endpoint: str, json: Any) -> Any:
        """REST POST（业务面，统一 ``/v1/<resource>/query`` 形态）。

        薄封装 ``self.http.post``。endpoint 形如
        ``"v1/events-calendar/query"``。
        """
        return self.http.post(endpoint, json)

    # ------------------------------------------------------------------ #
    # 配额 / 计量（成本意识入口，详见 cost.py）                          #
    # ------------------------------------------------------------------ #
    def get_quota_raw(self) -> dict:
        """原始 chunk 级配额（走 SDK 的 ``get_my_quota`` 封装方法）。

        返回 ``MyBigdataQuotaResponse.model_dump()``，含
        ``organization_quota.contextual_units_max_read`` 和
        ``organization_consumed.contextual_units_read``。
        **这是 chunk 计数器**（1 query_unit = 10 chunks），是成本模型的
        承重字段。SDK 高层 ``subscription.get_details()`` 把它 ÷10 包装成
        query_unit 会丢失 chunk 语义，所以用这条 raw 路径。
        """
        resp = self.conn.get_my_quota()
        return resp.model_dump()

    def get_quotas_v1(self) -> Any:
        """实时细分配额（REST GET ``v1/subscription/quotas``）。

        返回结构 ``{'results': ..., 'errors': ..., 'metadata': ...}``，
        含 credits limit/usage/remaining + billing 周期 + units 细分
        （search:web_unit_read 等）。**这个 endpoint SDK 完全没有高层方法**，
        是逃生舱能打 SDK 缺失路由的直接证据。可在冷启动中途轮询做
        chunk→credit 换算率实测校准（免费，不按 chunk 计费）。
        """
        return self.rest_get("v1/subscription/quotas")

    # ------------------------------------------------------------------ #
    # 调试辅助                                                            #
    # ------------------------------------------------------------------ #
    def introspect_conn(self, max_chars: int = 2000) -> str:
        """introspect ``bd._api`` 的方法 + 源码（开新 REST 路由前先看）。

        新 endpoint 不确定怎么打时，先看 ``query_chunks`` 等已有方法是
        怎么 delegate 到 ``self.http`` 的，照葫芦画瓢。
        """
        methods = [m for m in dir(self.conn) if not m.startswith("_")]
        try:
            src = inspect.getsource(type(self.conn))[:max_chars]
        except (OSError, TypeError):
            src = "<source unavailable>"
        return f"methods={methods}\n\n--- source head ---\n{src}"

"""chunk 消耗追踪 + 配额意识（成本承重模块）
=============================================

Bigdata 按 **chunk** 计费。这个模块把计量单位、单价、配额查询、消耗
delta 追踪、冷启动成本外推全部固化下来，让任何批量任务都带成本意识。

计量单位（运行时三处独立佐证坐实）
----------------------------------
- ``1 query_unit = 10 chunks``（精确，SDK 把 raw chunk 数 ÷10 包装）。
  - ``search.py:166``: ``usage += query_chunks_response.chunks_count``（raw chunk 数）
  - ``search.py:157``: ``get_usage()`` return ``usage / 10``
  - ``subscription.py:40``: ``query_unit_used = contextual_units_read / 10``
- REST raw 计数器：``get_my_quota().organization_consumed.contextual_units_read``
  是 chunk 数（不是 query_unit）。

单价（来自 docs.bigdata.com 公开 pricing，均为 list price）
--------------------------------------------------------------------------
- Fast Search:   $0.015 / query_unit
- Smart Search:  $0.03  / query_unit
- Batch Search (async): $0.0075 / query_unit（50% 折扣）

成本铁律（NO FALLBACK 类风险）
------------------------------
``Search.run(limit)`` 的 ``limit`` 是 ``int`` 时走 **doc-limit**，按"每页
返回的 chunk 数"计费；``ChunkLimit(n)`` 才按 chunk 计费。我们实测曾见
run(1) ≈ run(10) ≈ ~52 query_unit 的差距——**但这是单点实测,官方计费文档
未印证此倍数**,当方向性参考即可（倍数随标的/窗口浮动）。规则本身成立:
``max_chunks`` 是官方计费单位,务必走 ``ChunkLimit`` 而非裸 int。

> 一个默认参数就能静默烧光配额。冷启动脚本必须 code-review 保证零裸
> ``run(int)``，全部走 ``ChunkLimit``。

trial 配额现实
--------------
一个典型的 1 周 full-content trial ≈ 67000 query_unit = 670000 chunks
≈ $1005（list price 名义值，以你账号实际 quota 为准）。机构级 universe
（100-200 标的）做一次多年回溯 backfill 即接近或超过整个 trial 配额
（3 年季度 100 标的 = 89.6%；200 标的 = 180%）。**trial 只够 PoC 级
抽样（≤20 标的单快照）**，全量上线需要更大的付费配额。
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Optional

from .client import BigdataClient

__all__ = ["CostTracker", "CostModel", "CHUNKS_PER_QUERY_UNIT", "USD_PER_QUERY_UNIT"]

#: 计量换算常量（运行时坐实）
CHUNKS_PER_QUERY_UNIT = 10

#: 单价表（USD / query_unit，public list price）
USD_PER_QUERY_UNIT = {
    "fast": 0.015,
    "smart": 0.03,
    "batch": 0.0075,
}


@dataclass
class CostModel:
    """成本外推模型（纯计算，无 IO）。

    用于冷启动 backfill 前估算配额消耗，**禁止用代码行数等无关指标**，
    只按 chunk 真实计费口径算。
    """

    chunk_limit_per_query: int = 500
    """每次 search 的 chunk 上限（必须配合 ChunkLimit 使用，否则估算无效）。"""

    tier: str = "fast"
    """计费档：fast / smart / batch。"""

    trial_query_units: int = 67000
    """trial 配额基数（query_unit），用于 ``pct_of_trial_quota`` 估算。默认
    67000 = 典型 1 周 full-content trial ≈ $1005 list。换成你账号的实际额度
    （看 ``CostTracker.quota()`` 的 max_query_units）即可让百分比准确。"""

    def query_units_per_query(self) -> float:
        """单次查询消耗的 query_unit = chunk_limit / 10。"""
        return self.chunk_limit_per_query / CHUNKS_PER_QUERY_UNIT

    def estimate(
        self,
        n_entities: int,
        n_windows: int = 1,
    ) -> dict:
        """估算一次冷启动 backfill 的总成本。

        Parameters
        ----------
        n_entities:
            标的数量。
        n_windows:
            时间窗数量（如 3 年 × 季度 = 12 个窗）。

        Returns
        -------
        dict
            total_query_units / total_chunks / usd / pct_of_trial_quota。
        """
        qu_per_query = self.query_units_per_query()
        total_queries = n_entities * n_windows
        total_qu = qu_per_query * total_queries
        unit_price = USD_PER_QUERY_UNIT.get(self.tier, USD_PER_QUERY_UNIT["fast"])
        return {
            "n_entities": n_entities,
            "n_windows": n_windows,
            "chunk_limit_per_query": self.chunk_limit_per_query,
            "query_units_per_query": qu_per_query,
            "total_queries": total_queries,
            "total_query_units": total_qu,
            "total_chunks": total_qu * CHUNKS_PER_QUERY_UNIT,
            "tier": self.tier,
            "usd": round(total_qu * unit_price, 2),
            "pct_of_trial_quota": round(total_qu / self.trial_query_units * 100, 1),
        }


@dataclass
class CostTracker:
    """实时配额追踪 + 消耗 delta 计量（带 IO，查真实配额）。

    用法：操作前 ``snapshot()`` 记起点，操作后 ``delta()`` 看实际烧了多少
    chunk —— 用真实用量校准 :class:`CostModel` 的外推（替代纯估算）。
    """

    client: BigdataClient
    _baseline_chunks: Optional[int] = field(default=None, init=False)

    # ------------------------------------------------------------------ #
    # 配额查询                                                            #
    # ------------------------------------------------------------------ #
    def quota(self) -> dict:
        """当前 chunk 级配额（走 SDK ``get_my_quota`` 封装）。

        Returns
        -------
        dict
            max_chunks / used_chunks / remaining_chunks / used_query_units /
            remaining_query_units / pct_used。
        """
        raw = self.client.get_quota_raw()
        max_chunks = raw["organization_quota"]["contextual_units_max_read"]
        used_chunks = raw["organization_consumed"]["contextual_units_read"]
        remaining = max_chunks - used_chunks
        return {
            "max_chunks": max_chunks,
            "used_chunks": used_chunks,
            "remaining_chunks": remaining,
            "used_query_units": round(used_chunks / CHUNKS_PER_QUERY_UNIT, 1),
            "remaining_query_units": round(remaining / CHUNKS_PER_QUERY_UNIT, 1),
            "max_query_units": round(max_chunks / CHUNKS_PER_QUERY_UNIT, 1),
            "pct_used": round(used_chunks / max_chunks * 100, 2) if max_chunks else None,
        }

    def quota_detailed_raw(self) -> Any:
        """实时细分配额（REST ``v1/subscription/quotas``，免费旁路）。

        含 billing 周期 + units 细分。可在冷启动中途轮询此 endpoint 实测
        chunk→credit 换算率做校准。**这个 endpoint SDK 没有高层方法**。
        """
        return self.client.get_quotas_v1()

    # ------------------------------------------------------------------ #
    # 消耗 delta 追踪                                                     #
    # ------------------------------------------------------------------ #
    def snapshot(self) -> int:
        """记录当前已用 chunk 数为基线，返回该值。"""
        self._baseline_chunks = self.quota()["used_chunks"]
        return self._baseline_chunks

    def delta(self) -> dict:
        """相对上次 ``snapshot()`` 的消耗 delta（chunk + query_unit + USD 估算）。

        必须先调 ``snapshot()``，否则抛错（不猜基线，NO FALLBACK）。
        """
        if self._baseline_chunks is None:
            raise RuntimeError("call snapshot() before delta()")
        now_chunks = self.quota()["used_chunks"]
        delta_chunks = now_chunks - self._baseline_chunks
        delta_qu = delta_chunks / CHUNKS_PER_QUERY_UNIT
        return {
            "delta_chunks": delta_chunks,
            "delta_query_units": round(delta_qu, 2),
            "usd_fast": round(delta_qu * USD_PER_QUERY_UNIT["fast"], 4),
            "usd_smart": round(delta_qu * USD_PER_QUERY_UNIT["smart"], 4),
            "baseline_chunks": self._baseline_chunks,
            "now_chunks": now_chunks,
        }

"""实体解析 + crosswalk（SDK 高层路径）
=======================================

封装 ``bd.knowledge_graph``，把"公司名 / ISIN / 交易所代码"解析成
**rp_entity_id**（6 位字母数字，如 Apple = ``D8442A``、贵州茅台 =
``914E1F``）。

为什么这是 gateway
------------------
Bigdata 几乎所有能力都以 rp_entity_id 为主键：
- search 的 ``Entity(id)``（见 search.py）
- rest_ext 的 events-calendar / analyst-estimates / surprise（见 rest_ext.py）

**A 股标的的唯一可用解析路径**（实测）：
- 中文名直查 ``find_companies('贵州茅台')`` → **0 命中**（数据源中文实体层空）
- 英文官方名 ``find_companies('Kweichow Moutai')`` → 命中 ``914E1F``
- ISIN crosswalk ``get_companies_by_isin(['CNE0000018R8'])`` → ``914E1F``

所以 A 股请用 **英文官方名** 或 **ISIN** 解析，不要用中文名。

方法签名（运行时确认）
----------------------
- ``kg.find_companies(values, /, type=None, country=None, limit=20)``
- ``kg.find_topics(values, /, limit=20)``
- ``kg.find_sources(values, /, limit=20, country=None, rank=None, retention=None)``
- ``kg.get_companies_by_isin(isins: list[str]) -> list[Company | None]``
- ``kg.get_companies_by_cusip / _by_sedol / _by_listing``
- ``kg.get_entities(ids: list[str], /)`` —— **同时解 COMP 实体 + TOPC 话题**
  （不是 ENTITY-only）
- ``kg.find_topics`` —— 中文话题（如 '人工智能'）实测 0 命中，TOPIC 解析
  同样卡在中文层

注意
----
``kg.autosuggest`` 在 API-key 模式下 ``NotImplementedError``（与 uploads
同族），交互式补全用不了；实体解析只能走 ``find_*`` 系列。
"""

from __future__ import annotations

from typing import Any, Optional

from .client import BigdataClient

__all__ = ["EntityResolver", "company_to_dict"]


def company_to_dict(company: Any) -> dict:
    """``Company`` 实体 → 精简 dict（id + 名称 + 国家 + ticker）。

    字段名做防御性提取（不同 SDK 版本属性名可能微调）。
    """
    if company is None:
        return {}
    return {
        "id": getattr(company, "id", None),               # rp_entity_id
        "name": getattr(company, "name", None),
        "ticker": getattr(company, "ticker", None),
        "country": getattr(company, "country", None),
        "sector": getattr(company, "sector", None),
        "industry": getattr(company, "industry", None),
        "isin": getattr(company, "isin", None),
        "entity_type": getattr(company, "entity_type", None),
    }


class EntityResolver:
    """公司 / 话题实体解析（SDK 高层）。"""

    def __init__(self, client: BigdataClient) -> None:
        self.client = client

    @property
    def kg(self):
        return self.client.bd.knowledge_graph

    # ------------------------------------------------------------------ #
    # 公司解析                                                            #
    # ------------------------------------------------------------------ #
    def find_companies(
        self,
        name: str,
        *,
        country: Optional[str] = None,
        limit: int = 5,
        as_dict: bool = True,
    ):
        """按名称解析公司。

        ⚠️ A 股请用 **英文官方名**（'Kweichow Moutai' 而非 '贵州茅台'）。
        ``country`` 用 ISO-2（'CN' / 'US' / 'HK'）。
        """
        result = self.kg.find_companies(name, country=country, limit=limit)
        # find_companies 单值返回 list；多值返回 dict[str, list]
        companies = result if isinstance(result, list) else result.get(name, [])
        if as_dict:
            return [company_to_dict(c) for c in companies]
        return companies

    def resolve_id(
        self,
        name: str,
        *,
        country: Optional[str] = None,
    ) -> Optional[str]:
        """解析公司名 → 单个 rp_entity_id（取第一个命中）。

        命中 0 个返回 None（**不猜、不 fallback**，NO FALLBACK 原则）。
        A 股中文名大概率返回 None —— 改用英文名或 ``resolve_by_isin``。
        """
        companies = self.find_companies(name, country=country, limit=1, as_dict=True)
        if not companies:
            return None
        return companies[0].get("id")

    # ------------------------------------------------------------------ #
    # crosswalk：ISIN / CUSIP / SEDOL / 交易所代码 → rp_entity_id         #
    # ------------------------------------------------------------------ #
    def resolve_by_isin(self, isins: list[str], *, as_dict: bool = True):
        """ISIN crosswalk（A 股最可靠路径，如 茅台 CNE0000018R8 → 914E1F）。

        返回与输入等长的 list（未命中位置为 None / {}）。
        """
        companies = self.kg.get_companies_by_isin(isins)
        if as_dict:
            return [company_to_dict(c) for c in companies]
        return companies

    def resolve_by_cusip(self, cusips: list[str], *, as_dict: bool = True):
        """CUSIP crosswalk（美股）。"""
        companies = self.kg.get_companies_by_cusip(cusips)
        return [company_to_dict(c) for c in companies] if as_dict else companies

    def resolve_by_sedol(self, sedols: list[str], *, as_dict: bool = True):
        """SEDOL crosswalk。"""
        companies = self.kg.get_companies_by_sedol(sedols)
        return [company_to_dict(c) for c in companies] if as_dict else companies

    # ------------------------------------------------------------------ #
    # 话题 / 通用实体解析                                                 #
    # ------------------------------------------------------------------ #
    def find_topics(self, values, *, limit: int = 5, as_dict: bool = False):
        """解析话题（TOPC）。⚠️ 中文话题实测 0 命中。"""
        result = self.kg.find_topics(values, limit=limit)
        return result

    def get_entities(self, ids: list[str]):
        """按 id 批量解实体。**同时解 COMP（公司）+ TOPC（话题）**，非 ENTITY-only。

        实测：传 dotted topic id 返回 ``Topic(...)``，传 rp_entity_id 返回
        ``Company(...)``。
        """
        return self.kg.get_entities(ids)

"""SDK 没有的能力：用 bd._api 打 REST（逃生舱）
================================================

``bigdata_client`` 的 ``BigdataConnection`` 只 wrap 了 search / chunks /
KG / chat / watchlists / uploads。一整套 RavenPack 遗留的 ``/v1/*``
**结构化金融数据产品线** SDK 一个高层方法都没写。本模块用
``client.http.post(endpoint, json)`` 直打这些 endpoint，复用 SDK 的
JWT 认证 + 代理。

覆盖的 endpoint（概览）
----------------------
本模块用 ``client.http.post`` 直打 SDK 没封的 ``/v1/*`` 结构化金融数据：前瞻
（estimates / events-calendar / surprise / ratings / target）、财报三表 + TTM
指标比率、公司画像、行情、分红、分部营收、entity-sentiment、co-mention、
company-screener，加 reporting_period 回填（``cqs/query-chunks``，特殊、按 chunk
计费）。**完整方法清单 + 精确签名 + 验证等级（L3/L4）以
``references/verified_api_signatures.md`` 为单一权威**——此处只给覆盖范围,不复述
具体路径与等级（避免改一处漏同步,本周期就因这张复制表漏改过 ratings/target 的等级）。

关键修正（推翻早期 "Bigdata 中文/A股一刀切失效" 结论）
------------------------------------------------------
**必须拆成两个数据面分开讲：**

1. **结构化金融面**（本模块的 /v1/* 全家桶）——对大陆 A 股 + 港股 **可用**：
   - 茅台 ``914E1F`` events-calendar 返回前瞻日历、analyst-estimates 返回
     consensus、latest-surprise 返回 reporting_date + actual vs estimated
     （实测 A 股结构面数据有近月更新，非历史陈值）。
   - 经 rp_entity_id（英文名 or ISIN crosswalk 解析，见 kg.py）。
   - ⚠️ A 股结构数据有空洞：茅台 price-target 只返回 entity 无
     target_high/low/consensus（美股 AAPL 完整）。

2. **非结构化中文 NLP 面**（中文新闻实体检测 / 中文情绪）—— **确认死路**，
   SDK 和 REST 都救不了，数据源真没有。

路由 / 报错坑（避免后续踩）
--------------------------
- 业务面只对 ``POST + /query`` 注册；裸 ``GET /v1/<resource>`` 会 404。
- ``403 'Missing Authentication Token'`` = 该 auth path 上无此路由（不是
  权限拒绝）；``404`` = 路径不存在。
- analyst-estimates 的 ``period=quarter`` 实测 ``limit`` 上限约 20，
  ``limit=30`` 直接 400（报错信息极不友好，别误判为 endpoint 挂了）。
- company-screener 的 filter **必须嵌套在 ``"filters"`` 对象**里
  （market_cap_more_than / sector / industry / country / exchange / is_etf）；
  ``limit`` 在 top-level,≤1000。平铺 filter 不报错但被静默忽略、返回未过滤结果。

成本
----
这些 analyst/events/quota endpoint **不按 chunk 计费**（只 search 的
query-chunks 才 usage += chunks_count）。本模块默认 limit 极小，近乎零成本。
唯一例外：``fetch_reporting_period`` 走 ``cqs/query-chunks``，**按 chunk 计费**，
故默认 chunk_limit 很小。

只读说明
--------
本模块全部走只读查询（GET / POST query），无写入、无上传。
"""

from __future__ import annotations

from typing import Any, Optional

from .client import BigdataClient

__all__ = ["StructuredDataREST", "fields_values_to_records", "BatchSearch"]


def _entity_id_body(rp_entity_id: str | list[str], key: str = "rp_entity_id") -> dict:
    """统一构造 entity 维度 body（events-calendar 用 rp_entity_id 数组）。"""
    ids = rp_entity_id if isinstance(rp_entity_id, list) else [rp_entity_id]
    return {key: ids}


def _ident(rp_entity_id: str | list[str]) -> dict:
    """单标的 endpoint 的 identifier 形态（analyst / financials / market-data 系列共用）。

    OpenAPI spec + contract test 确认:这一大批 endpoint 用
    ``{"identifier": {"type": "rp_entity_id", "value": id}}``,
    与 events-calendar 的 ``{"rp_entity_id": [...]}`` 数组形态不同（各自固定）。
    """
    return {"identifier": {"type": "rp_entity_id", "value": rp_entity_id}}


def fields_values_to_records(result: Any) -> Any:
    """把 ``{fields: [...], values: [[...]]}`` 拼成 ``[{field: val}]`` 方便消费。

    财报（income/balance/cash-flow）、行情（daily-prices）、分红、分部营收
    等 endpoint 的 ``results`` 是 ``{fields, values}``（单实体）或其 list（多实体）。
    TTM / profile 系列已是扁平 ``[{...}]``,原样返回。
    """
    res = result.get("results", result) if isinstance(result, dict) else result

    def _one(d):
        if isinstance(d, dict) and d.get("fields") and d.get("values") is not None:
            return [dict(zip(d["fields"], row)) for row in d["values"]]
        return d

    if isinstance(res, dict):
        return _one(res)
    if isinstance(res, list):
        records = [_one(d) for d in res]
        # 单实体（最常见的单标的查询，如 daily_prices / dividends 的
        # ``results`` 是含一个实体的 list）直接 flatten 成记录列表,与
        # dict-results（income/balance 等）行为一致;多实体才返回每实体一组。
        return records[0] if len(records) == 1 else records
    return res


class StructuredDataREST:
    """SDK 缺失的结构化金融数据（全部走 bd._api.http REST 逃生舱）。

    所有方法返回 **原始 dict/list**（不做强 schema 解析）——因为这些是
    半文档化的 RavenPack 遗留 endpoint，schema 可能随后端变更。消费者按
    实际返回的 key 取值，并自行加 schema 防御。
    """

    def __init__(self, client: BigdataClient) -> None:
        self.client = client

    @property
    def http(self):
        return self.client.http

    # ------------------------------------------------------------------ #
    # 1) 前瞻财报/电话会日历（L4 实打）                                   #
    # ------------------------------------------------------------------ #
    def events_calendar(
        self,
        rp_entity_id: Optional[str | list[str]] = None,
        *,
        categories: Optional[list[str]] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None,
        countries: Optional[list[str]] = None,
        limit: int = 5,
        cursor: Optional[str] = None,
    ) -> Any:
        """前瞻财报/电话会日历。``POST v1/events-calendar/query``。

        两种用法：
        - **单/多标的**：传 ``rp_entity_id``（前瞻该标的财报日历）。
        - **全市场广扫**：不传 entity，传 ``countries=['US']`` + 日期窗，
          配合 ``cursor`` 分页扫全市场（回答"下周谁发财报"/冷启动筛选）。

        Parameters
        ----------
        categories:
            如 ``['earnings-call']`` / ``['conference-call']``。
        start_date, end_date:
            ``'YYYY-MM-DD'``。
        limit:
            ≤1000。默认 5（成本意识，且本 endpoint 不按 chunk 计费）。

        Returns
        -------
        dict
            ``{'results': ..., 'pagination': ...}``（实测顶层 key）。
            event 项含 category / event_datetime / title / fiscal_year /
            fiscal_period / updated_at 等（前瞻数据，event_datetime 为未来日期）。
        """
        body: dict[str, Any] = {}
        if rp_entity_id is not None:
            body.update(_entity_id_body(rp_entity_id))
        if categories is not None:
            body["categories"] = categories
        if start_date is not None:
            body["start_date"] = start_date
        if end_date is not None:
            body["end_date"] = end_date
        if countries is not None:
            body["countries"] = countries
        if cursor is not None:
            body["cursor"] = cursor
        body["limit"] = limit
        return self.http.post("v1/events-calendar/query", body)

    # ------------------------------------------------------------------ #
    # 2) 前瞻一致预期（L4 实打）                                          #
    # ------------------------------------------------------------------ #
    def analyst_estimates(
        self,
        rp_entity_id: str,
        *,
        period: str = "quarter",
        limit: int = 5,
    ) -> Any:
        """前瞻逐期一致预期。``POST v1/analyst-estimates/query``。

        返回 FISCAL_PERIOD_END_DATE + REVENUE/EBITDA/EBIT/NET_INCOME/SGA/EPS
        各 LOW/HIGH/AVG + NUM_ANALYSTS_REVENUE/EPS。实测美股可给到 ~2.5 年
        前瞻（Apple 到 2028Q3）。

        Parameters
        ----------
        period:
            ``'quarter'`` 或 ``'annual'``。
        limit:
            ⚠️ ``period=quarter`` 实测上限约 20，超出报 400。默认 5。

        Notes
        -----
        OpenAPI spec + contract test 确认 identifier 形态为
        ``{"type": "rp_entity_id", "value": id}``（analyst / financials /
        market-data 系列统一,与 events-calendar 的 ``rp_entity_id`` 数组不同,
        各自固定——不是"两种都试")。
        """
        body = {
            "identifier": {"type": "rp_entity_id", "value": rp_entity_id},
            "period": period,
            "limit": limit,
        }
        return self.http.post("v1/analyst-estimates/query", body)

    # ------------------------------------------------------------------ #
    # 3) 最近一期财报 surprise（L4 实打）                                 #
    # ------------------------------------------------------------------ #
    def latest_surprise(self, rp_entity_id: str) -> Any:
        """最近一期财报 surprise。``POST v1/latest-surprise/query``。

        返回 reporting_date + eps_actual/eps_estimated/eps_surprise_pct +
        revenue_actual/revenue_estimated/revenue_surprise_pct + last_updated。

        ⚠️ 名字是 "latest" —— 只返回最近一期（实测单条）。历史 surprise
        序列本方法拿不到（可能需翻 analyst-estimates 历史行自算）。
        """
        body = {"identifier": {"type": "rp_entity_id", "value": rp_entity_id}}
        return self.http.post("v1/latest-surprise/query", body)

    # ------------------------------------------------------------------ #
    # 4) 分析师评级一致（L3 文档证实，未运行时实打）                      #
    # ------------------------------------------------------------------ #
    def analyst_ratings(self, rp_entity_id: str) -> Any:
        """买卖评级一致。``POST v1/analyst-ratings/query``。

        文档：返回 strong_buy/buy/hold/sell/strong_sell + consensus。
        验证等级 L4（2026-05-31 实跑确认返回 ``{results}``，升自 L3）。
        identifier 形态同 analyst_estimates（spec 确认的 identifier 对象）。
        """
        body = {"identifier": {"type": "rp_entity_id", "value": rp_entity_id}}
        return self.http.post("v1/analyst-ratings/query", body)

    # ------------------------------------------------------------------ #
    # 5) 目标价 consensus（L3 文档证实）                                  #
    # ------------------------------------------------------------------ #
    def price_target(self, rp_entity_id: str) -> Any:
        """目标价一致预期。``POST v1/price/target/query``。

        文档：返回 target high/low/consensus/median + currency。
        ⚠️ A 股有空洞：部分 A 股只返回 entity 无 target 数值（美股如 AAPL
        则完整返回 target high/low/consensus 数值）。
        验证等级 L4（2026-05-31 实跑，升自 L3）。identifier 形态同 analyst_estimates（spec 确认的 identifier 对象）。
        """
        body = {"identifier": {"type": "rp_entity_id", "value": rp_entity_id}}
        return self.http.post("v1/price/target/query", body)

    # ------------------------------------------------------------------ #
    # 6) universe 构建：公司筛选器（L4 endpoint 可达）                    #
    # ------------------------------------------------------------------ #
    def company_screener(
        self,
        *,
        market_cap_more_than: Optional[float] = None,
        market_cap_lower_than: Optional[float] = None,
        sector: Optional[str] = None,
        industry: Optional[str] = None,
        country: Optional[str] = None,
        exchange: Optional[str] = None,
        is_etf: Optional[bool] = None,
        limit: int = 10,
        **extra_filters: Any,
    ) -> Any:
        """股票池筛选器。``POST v1/company-screener/query``。

        ⚠️ filter **必须嵌套在 ``"filters"`` 对象**里
        （``{"filters": {...}, "limit": n}``）。平铺在 top-level **不报错但被后端
        静默忽略、返回未过滤结果**（实测裁决 2026-05-30,见 known_pitfalls #6）。
        ``limit`` 在 top-level,≤1000。

        ``extra_filters`` 透传其它 flat filter（如 ``beta_more_than`` 等，
        以文档为准）。
        """
        filters: dict[str, Any] = {}
        if market_cap_more_than is not None:
            filters["market_cap_more_than"] = market_cap_more_than
        if market_cap_lower_than is not None:
            filters["market_cap_lower_than"] = market_cap_lower_than
        if sector is not None:
            filters["sector"] = sector
        if industry is not None:
            filters["industry"] = industry
        if country is not None:
            filters["country"] = country
        if exchange is not None:
            filters["exchange"] = exchange
        if is_etf is not None:
            filters["is_etf"] = is_etf
        filters.update(extra_filters)
        # ⚠️ filter 必须嵌套在 "filters" 对象里；平铺 top-level 不报错但被后端
        # 静默忽略（返回未过滤结果）。实测裁决 2026-05-30,见 known_pitfalls #6。
        return self.http.post(
            "v1/company-screener/query", {"filters": filters, "limit": limit}
        )

    # ================================================================== #
    # 历史财务报表（三大表 + 分部营收，identifier + period + limit）       #
    # 不按 chunk 计费；contract-tested 2026-05-30；results 为             #
    # {fields, values}，用 fields_values_to_records() 拼成记录。         #
    # ================================================================== #
    def income_statement(self, rp_entity_id: str, *, period: str = "annual", limit: int = 5) -> Any:
        """历史利润表。``POST v1/income-statement/query``。

        ``results.fields`` 含 REVENUE / GROSS_PROFIT / OPERATING_INCOME /
        EBITDA / EBIT / NET_INCOME / R&D / SG&A 等。``period``: ``annual`` / ``quarter``。
        """
        body = _ident(rp_entity_id)
        body.update(period=period, limit=limit)
        return self.http.post("v1/income-statement/query", body)

    def balance_sheet(self, rp_entity_id: str, *, period: str = "annual", limit: int = 5) -> Any:
        """历史资产负债表。``POST v1/balance-sheet/query``。

        ``results.fields`` 含 TOTAL_ASSETS / TOTAL_LIABILITIES / TOTAL_DEBT /
        NET_DEBT / CASH_AND_CASH_EQUIVALENTS / TOTAL_STOCKHOLDERS_EQUITY 等。
        """
        body = _ident(rp_entity_id)
        body.update(period=period, limit=limit)
        return self.http.post("v1/balance-sheet/query", body)

    def cash_flow_statement(self, rp_entity_id: str, *, period: str = "annual", limit: int = 5) -> Any:
        """历史现金流量表。``POST v1/cash-flow-statement/query``。

        ``results.fields`` 含 OPERATING_CASH_FLOW / FREE_CASH_FLOW /
        CAPITAL_EXPENDITURE / NET_INCOME / STOCK_BASED_COMPENSATION 等。
        """
        body = _ident(rp_entity_id)
        body.update(period=period, limit=limit)
        return self.http.post("v1/cash-flow-statement/query", body)

    def revenue_geographic_segments(self, rp_entity_id: str, *, period: str = "annual", limit: int = 10) -> Any:
        """分地区营收。``POST v1/company-revenue-geographic-segments/query``。

        ``results.values`` 每行 ``[FISCAL_YEAR, PERIOD, CURRENCY, {地区: 营收}]``。
        """
        body = _ident(rp_entity_id)
        body.update(period=period, limit=limit)
        return self.http.post("v1/company-revenue-geographic-segments/query", body)

    def revenue_product_segments(self, rp_entity_id: str, *, period: str = "annual", limit: int = 10) -> Any:
        """分产品营收。``POST v1/company-revenue-product-segments/query``。

        ``results.values`` 每行 ``[FISCAL_YEAR, PERIOD, CURRENCY, {产品线: 营收}]``
        （如 NVDA 的 GPU / Tegra ...）。
        """
        body = _ident(rp_entity_id)
        body.update(period=period, limit=limit)
        return self.http.post("v1/company-revenue-product-segments/query", body)

    # ================================================================== #
    # TTM 指标 / 比率 / 公司画像（identifier；results 为扁平 [{...}]）    #
    # 不按 chunk 计费；contract-tested 2026-05-30。                       #
    # ================================================================== #
    def key_metrics_ttm(self, rp_entity_id: str) -> Any:
        """TTM 关键指标。``POST v1/key-metrics-ttm/query``。

        扁平字段含 enterprise_value_ttm / ev_to_ebitda_ttm /
        return_on_equity_ttm / return_on_invested_capital_ttm /
        free_cash_flow_yield_ttm / earnings_yield_ttm 等。
        """
        return self.http.post("v1/key-metrics-ttm/query", _ident(rp_entity_id))

    def company_ratios_ttm(self, rp_entity_id: str) -> Any:
        """TTM 财务比率。``POST v1/company-ratios-ttm/query``。

        扁平字段含 gross_profit_margin_ttm / net_profit_margin_ttm /
        price_to_earnings_ratio_ttm / price_to_book_ratio_ttm /
        debt_to_equity_ratio_ttm / dividend_yield_ttm 等。
        """
        return self.http.post("v1/company-ratios-ttm/query", _ident(rp_entity_id))

    def company_profile(self, rp_entity_id: str) -> Any:
        """公司画像。``POST v1/company-profile/query``。

        扁平字段含 company_name / ceo / sector / industry / website /
        description / full_time_employees / ipo_date / isin / cusip / exchange 等。
        """
        return self.http.post("v1/company-profile/query", _ident(rp_entity_id))

    # ================================================================== #
    # 市场数据（行情 / 分红，identifier + date_range）                    #
    # 不按 chunk 计费；contract-tested 2026-05-30；results 为 {fields, values}。#
    # ================================================================== #
    def daily_prices(self, rp_entity_id: str, *, start_date: str, end_date: str) -> Any:
        """日线 OHLC。``POST v1/price/daily/query``。

        ``results.fields`` = DATE / OPEN / LOW / HIGH / CLOSE / VOLUME /
        CHANGE / CHANGE_PERCENT / VWAP / CURRENCY。日期 ``YYYY-MM-DD``。
        """
        body = _ident(rp_entity_id)
        body["date_range"] = {"start": start_date, "end": end_date}
        return self.http.post("v1/price/daily/query", body)

    def dividends(self, rp_entity_id: str, *, start_date: str, end_date: str) -> Any:
        """分红历史。``POST v1/dividends/query``（注意:不是 ``v1/price/dividends``）。

        ``results.fields`` = DATE / DIVIDEND / ADJ_DIVIDEND / RECORD_DATE /
        PAYMENT_DATE / DECLARATION_DATE / YIELD / FREQUENCY。
        """
        body = _ident(rp_entity_id)
        body["date_range"] = {"start": start_date, "end": end_date}
        return self.http.post("v1/dividends/query", body)

    # ================================================================== #
    # 聚合情感时间序列（identifier + timestamp，不按 chunk 计费）          #
    # ================================================================== #
    def entity_sentiment(self, rp_entity_id: str, *, start_date: str, end_date: str) -> Any:
        """日频聚合情感时间序列。``POST v1/entity-sentiment/``（**尾斜杠，非 /query**）。

        ``results[].values`` 每点含 date / daily_sentiment（日均事件情感）/
        sentiment_pressure（异常情感强度）/ abnormal_media_attention（异常关注量）。
        **这是官方现成的日频情感序列——不必从 chunk 自聚合**。contract-tested 2026-05-30。
        """
        body = _ident(rp_entity_id)
        body["timestamp"] = {"start": start_date, "end": end_date}
        return self.http.post("v1/entity-sentiment/", body)

    # ================================================================== #
    # 实体共现关系图（⚠️ 走 search 系，**按 chunk 计费**！）              #
    # ================================================================== #
    def connected_entities(
        self,
        rp_entity_id: str,
        *,
        date_range: Optional[dict] = None,
        limit: int = 10,
    ) -> Any:
        """实体共现（co-mention）关系图。``POST v1/search/co-mentions/entities``。

        ``results`` 按类别（places / companies / organizations / people /
        products / concepts）分组,每个实体含 ``total_chunks_count`` /
        ``total_headlines_count``（按共现量排序）—— 用于建供应链 / 竞品 / 客户
        共现网络。**要某一类就直接读对应分组**（结果本就按类分好）。

        Parameters
        ----------
        date_range:
            可选,``{"start": "2024-01-01T00:00:00Z", "end": "2024-12-31T23:59:59Z"}``
            （ANSI date-time + UTC,**带时分秒,非 YYYY-MM-DD**）→ 透传到
            ``query.filters.timestamp``,看某时间窗内的共现（关系随时间演化）。

        ⚠️ **本方法按 chunk 计费**（响应含 ``usage.api_query_units``）,与
        financials / market-data 那批免费 endpoint 不同,默认 limit 小。

        Notes
        -----
        co-mentions/entities 的 body **没有 ``entity_categories`` 参数**（OpenAPI
        spec 确认；早期版本曾透传它做类别过滤,实测无效、已移除——按类别就直接读
        返回的分组）。``date_range`` 走 ``query.filters.timestamp``,contract-tested。
        """
        filters: dict[str, Any] = {"entity": {"any_of": [rp_entity_id]}}
        if date_range is not None:
            filters["timestamp"] = date_range
        body = {"query": {"filters": filters}, "limit": limit}
        return self.http.post("v1/search/co-mentions/entities", body)

    # ================================================================== #
    # 特殊:reporting_period 回填（SDK 砍字段，REST 有，**按 chunk 计费**！）#
    # ================================================================== #
    def fetch_reporting_period_raw(self, payload: dict) -> Any:
        """直打 ``cqs/query-chunks`` 拿原始 ``stories[].reportingPeriod``。

        **根因**：``reporting_period`` 在 REST wire 上存在且填充率约 75%
        （filings），但 SDK 的 ``ChunkedDocumentResponse`` 模型没有这个字段，
        pydantic 默认丢弃 → ``Document.reporting_period`` 永远 None。绕过
        SDK 模型、直读 raw JSON 的 ``reportingPeriod`` 才能回填。

        格式两类共存：绝对财年 ``'2026FY'`` + 相对财季 ``'FQ1'-'FQ4'``。
        ⚠️ ``'FQ1'`` 无年份锚点，需结合同 story 的 ``'YYYYFY'`` 或 timestamp
        二次推断，直接当结构化季度键有歧义。

        ⚠️ **本方法按 chunk 计费**（走 query-chunks）。``payload`` 请自行
        控制 chunk 规模。payload 即 ``POST cqs/query-chunks`` 的 body
        （可先 monkeypatch ``http.post`` 抓 SDK 真实 payload 拿 schema，
        见 examples）。

        Returns
        -------
        dict
            原始 query-chunks 响应；``stories[].reportingPeriod`` /
            ``reportingEntities`` / ``documentType`` 为 camelCase wire 字段。
        """
        return self.http.post("cqs/query-chunks", payload)


class BatchSearch:
    """Batch search 异步流程（chunk 计费，但比 fast 便宜 **50%**）。

    适合一次性打包大量 search query（如给单标的做多主题 × 多时间窗的海量
    证据流回填）—— N 个 search 打成一个 batch，单价从 ``$0.015`` 降到
    ``$0.0075`` / query_unit。

    流程（2026-05-31 contract-tested：``create_job`` / ``upload_input`` /
    ``get_status`` 轮询 = **L4 实跑**（含 upload 的 Content-Type 403 bug 修复，
    实测状态流转 pending→processing）；``download_results`` 是标准 S3 GET，因
    smart batch 处理 >10min 未跑到 completed 而**未端到端验**——用时若异常照本
    docstring 自查）::

        bs = BatchSearch(client)
        job = bs.create_job()                              # {batch_id, presigned_url}
        jsonl = bs.build_input_jsonl([{...search1...}, {...search2...}])
        bs.upload_input(job["presigned_url"], jsonl)       # PUT 到 S3 presigned
        st = bs.get_status(job["batch_id"])                # poll 到 completed
        rows = bs.download_results(st["output_file_url"])

    每行 jsonl = 一个 ``POST v1/search`` 的 body（``search_mode`` fast/smart +
    query/filters）。轮询 ``status`` 到 ``completed``（``output_file_url`` 可用）
    或 ``failed``。
    """

    def __init__(self, client: BigdataClient) -> None:
        self.client = client

    @property
    def http(self):
        return self.client.http

    def create_job(self) -> dict:
        """创建 batch job。``POST v1/search/batches``（无需 body）。

        返回 ``{batch_id, presigned_url}`` —— 把 .jsonl 输入 PUT 到 presigned_url。
        """
        return self.http.post("v1/search/batches", {})

    def get_status(self, batch_id: str) -> dict:
        """查 batch 状态。``GET v1/search/batches/{batch_id}``。

        返回 ``{batch_id, status, output_file_url}``。轮询直到 ``status`` 为
        ``completed``（此时 ``output_file_url`` 可用）或 ``failed``。
        """
        return self.http.get(f"v1/search/batches/{batch_id}")

    @staticmethod
    def build_input_jsonl(search_requests: list[dict]) -> str:
        """把 search request dict 列表拼成 .jsonl 文本（每行一个 ``v1/search`` body）。"""
        import json as _json

        return "\n".join(_json.dumps(r, ensure_ascii=False) for r in search_requests)

    @staticmethod
    def upload_input(presigned_url: str, jsonl_text: str) -> int:
        """PUT .jsonl 到 ``create_job`` 返回的 presigned_url（S3，非 Bigdata API）。

        ⚠️ **Content-Type 必须匹配 presigned 签名**：URL query 的 ``content-type``
        参数（实测后端签的是 ``application/jsonl``）正是 S3 计算签名用的值，PUT
        必须带完全相同的 Content-Type，否则 ``403 SignatureDoesNotMatch``。这里从
        URL query 解析出来用上，不写死（后端若换 type 也跟得上）。contract-tested
        2026-05-31。
        ⚠️ 走裸 ``requests``（presigned 是 S3 直连签名 URL，不经 Bigdata 认证层）；
        ``requests`` 读 ``HTTPS_PROXY`` 走代理，代理对 S3 大 host 偶发 ``503 Tunnel``，
        故内置瞬时重试（``rc()`` 的 marker 大小写匹配不到 ``ProxyError``，单独处理）。
        """
        import time
        import urllib.parse

        import requests

        qs = urllib.parse.parse_qs(urllib.parse.urlparse(presigned_url).query)
        content_type = qs.get("content-type", ["application/jsonl"])[0]
        last_exc: Optional[Exception] = None
        for _ in range(5):
            try:
                resp = requests.put(
                    presigned_url,
                    data=jsonl_text.encode("utf-8"),
                    headers={"Content-Type": content_type},
                    timeout=120,
                )
                resp.raise_for_status()
                return resp.status_code
            except requests.exceptions.ProxyError as exc:
                last_exc = exc  # 代理对 S3 偶发 503 Tunnel，退避重试
                time.sleep(2)
        raise last_exc  # type: ignore[misc]

    @staticmethod
    def download_results(output_file_url: str) -> list[dict]:
        """GET ``output_file_url``（status=completed 时）下载 .jsonl 结果，解析成 dict 列表。"""
        import json as _json

        import requests

        resp = requests.get(output_file_url, timeout=120)
        resp.raise_for_status()
        return [_json.loads(line) for line in resp.text.splitlines() if line.strip()]

"""瞬时网络/SSL 错误重试穿透（无 IO 纯工具）
==============================================

打海外 API（``api.bigdata.com``）时，**首发握手**常因网络抖动抛
``SSLError: UNEXPECTED_EOF`` / ``Connection reset`` / ``RemoteDisconnected``——
尤其经本地出站代理转发时。``bigdata-client`` 的 HTTP 层（直接依赖是
``requests`` + ``aiohttp``）对 **SSL 握手阶段的 EOF** 不做重试，官方异常
体系（见 sdk-reference/exceptions）也不建模这类握手错——所以一次抖动
就直接冒泡成异常，让整个 backfill 中断。

``rc()`` 在 SDK/REST 调用外面再包一层：识别这些**瞬时**错误标记就退避重试，
其它错误（400/认证失败/schema 错）立即抛出不掩盖（NO FALLBACK——只重试
确定瞬时的，不把真错误吞掉）。**限流（429）单独排除**：它该长退避或交给
调用方降速，固定 ``delay``×N 硬重试只会加剧限流，故命中限流标记直接抛。

为什么 marker 用子串匹配
------------------------
这些异常跨 ``ssl`` / ``urllib3`` / ``requests`` / ``http.client`` 多层包装，
类型不统一但 ``str(e)`` 里稳定带这些词。子串匹配比 except 一堆具体异常类
更鲁棒，也不会误吞业务错误（业务错误的文案里不含 'SSL'/'EOF' 这些词）。

用法
----
>>> from bigdata_toolkit import BigdataClient, EntityResolver, rc
>>> client = BigdataClient()                                  # doctest: +SKIP
>>> er = EntityResolver(client)                               # doctest: +SKIP
>>> nvda = rc(lambda: er.resolve_id("NVIDIA", country="US"))  # doctest: +SKIP

循环里注意闭包陷阱（务必绑定循环变量）::

    for kw, dr, lim in jobs:
        # ✅ 绑定，否则所有 lambda 共享最后一次的 kw/dr/lim
        docs = rc(lambda kw=kw, dr=dr, lim=lim:
                  searcher.search_entity(nvda, keyword=kw, chunk_limit=lim, date_range=dr))
"""

from __future__ import annotations

import time
from typing import Callable, TypeVar

__all__ = ["rc", "with_retry", "RETRYABLE_MARKERS"]

T = TypeVar("T")

#: ``str(exc)`` 命中任一即视为**瞬时**错误，可退避重试。
#: 来源：实测打 api.bigdata.com 首发握手抖动的异常文案（SSL/EOF/连接重置/超时）。
RETRYABLE_MARKERS = (
    "SSL",
    "EOF",
    "Connection",
    "Max retries",
    "timeout",
    "RemoteDisconnected",
)

#: 即使 ``str(exc)`` 里含上面的瞬时词，命中这些**限流/配额**标记也**不**重试——
#: 429 该长退避或交调用方降速，固定 ``delay``×N 硬重试会加剧限流；配额耗尽重试无意义。
NON_RETRYABLE_MARKERS = (
    "429",
    "Too Many Requests",
    "rate limit",
    "ratelimit",
    "quota",
)


def _is_retryable(exc: Exception) -> bool:
    msg = str(exc)
    # 限流/配额优先级高于瞬时：命中即不重试（避免硬重试加剧限流）。
    low = msg.lower()
    if any(m in low for m in NON_RETRYABLE_MARKERS):
        return False
    return any(marker in msg for marker in RETRYABLE_MARKERS)


def rc(fn: Callable[[], T], *, tries: int = 8, delay: float = 2.0) -> T:
    """跑 ``fn()``，遇到**瞬时**网络/SSL 错误就退避重试，其它错误立即抛。

    Parameters
    ----------
    fn:
        无参 thunk（用 ``lambda: ...`` 包住真正的调用；循环里记得绑定变量）。
    tries:
        最多尝试次数（含首次）。默认 8。
    delay:
        每次重试前 sleep 秒数（固定退避，够穿透首发抖动；不做指数退避避免
        长尾等待）。默认 2.0。

    Returns
    -------
    ``fn()`` 的返回值。

    Raises
    ------
    最后一次的原始异常（重试用尽），或任何**非瞬时**异常（立即抛，不掩盖）。
    """
    last_exc: Exception | None = None
    for i in range(tries):
        try:
            return fn()
        except Exception as exc:  # noqa: BLE001 —— 故意宽捕获后按 marker 区分
            last_exc = exc
            if i < tries - 1 and _is_retryable(exc):
                time.sleep(delay)
                continue
            raise
    # 理论不可达（循环里要么 return 要么 raise）；为类型完整性兜底。
    assert last_exc is not None
    raise last_exc


def with_retry(*, tries: int = 8, delay: float = 2.0):
    """装饰器版 :func:`rc`，把瞬时重试包到一个函数上。

    >>> @with_retry(tries=5)
    ... def pull(entity_id):                 # doctest: +SKIP
    ...     return rest.latest_surprise(entity_id)
    """

    def deco(func: Callable[..., T]) -> Callable[..., T]:
        def wrapper(*args, **kwargs) -> T:
            return rc(lambda: func(*args, **kwargs), tries=tries, delay=delay)

        wrapper.__name__ = getattr(func, "__name__", "wrapped")
        wrapper.__doc__ = func.__doc__
        return wrapper

    return deco

"""带标注的 chunk 抽取（SDK 高层路径）
=====================================

封装 ``bd.search`` 的语义检索，把每个返回 chunk 的 **sentiment + 实体
span + 定位** 拍平成普通 dict，方便下游消费。

走 SDK 高层（``bd.search.new(...).run(...)``），不是 REST 逃生舱。

⚠️ 中文/A股注意
----------------
实测：中文新闻实体检测 ≈0，CJK chunk 的 sentiment 是 doc-level 继承值
（chunk sentiment == doc sentiment），且 ``language`` 字段会把含中文的
filing 误标成 English。**这是数据源层硬伤，不是 SDK 封装问题**——
A 股标的请先用英文官方名 / ISIN 解析成 rp_entity_id（见 kg.py），再做
entity-scoped 检索；纯中文 keyword 检索基本 0 命中。结构化金融面
（前瞻日历 / 一致预期 / surprise）走 rest_ext.py，对 A 股可用。

字段来源（运行时确认的 SDK model）
-----------------------------------
- ``Document``: id, headline, sentiment, document_scope, source, timestamp,
  chunks, language, cluster, reporting_period, document_type,
  reporting_entities, url
  （注意 ``reporting_period`` 字段存在但 SDK 反序列化时不填，永远 None —
  要拿真值走 rest_ext.fetch_reporting_period）
- ``DocumentChunk``: text, chunk, entities, sentences, relevance, sentiment,
  section_metadata, speaker
- ``DocumentSentenceEntity``: key（rp_entity_id）, start, end（字符 span 位置）,
  query_type

成本铁律（详见 cost.py）
------------------------
``Search.run(limit)`` 接受 ``int``（doc-limit）或 ``ChunkLimit(n)``。
**doc-limit 不省钱**：run(1) 和 run(10) 成本几乎一样（~52 query_unit），
因为后端按"每页返回的 chunk 数"计费。真正省钱的是 ``ChunkLimit(n)``
（实测 ChunkLimit(10) 仅 1 query_unit，52x 差距）。所以本模块默认走
ChunkLimit，强制成本意识。
"""

from __future__ import annotations

from typing import Any, Optional, Union

from bigdata_client.daterange import AbsoluteDateRange, RollingDateRange
from bigdata_client.query import Entity, Keyword
from bigdata_client.search import ChunkLimit
from bigdata_client.models.search import DocumentType, SortBy

from .client import BigdataClient

__all__ = [
    "AnnotatedSearcher",
    "chunk_to_dict",
    "document_to_dict",
]


def _entity_to_dict(ent: Any) -> dict:
    """``DocumentSentenceEntity`` → dict（key + 字符 span + query_type）。"""
    return {
        "key": getattr(ent, "key", None),          # rp_entity_id
        "start": getattr(ent, "start", None),       # 字符起始位置
        "end": getattr(ent, "end", None),           # 字符结束位置
        "query_type": getattr(ent, "query_type", None),
    }


def chunk_to_dict(chunk: Any) -> dict:
    """``DocumentChunk`` → 拍平 dict，保留标注。

    每个 entity 的 ``(start, end)`` 是该 chunk ``text`` 内的字符 span，可用
    ``text[start:end]`` 取出被标注的实体表面词。
    """
    entities = [_entity_to_dict(e) for e in (getattr(chunk, "entities", None) or [])]
    return {
        "chunk_index": getattr(chunk, "chunk", None),
        "text": getattr(chunk, "text", None),
        "sentiment": getattr(chunk, "sentiment", None),
        "relevance": getattr(chunk, "relevance", None),
        "speaker": getattr(chunk, "speaker", None),
        "section_metadata": getattr(chunk, "section_metadata", None),
        "entities": entities,
    }


def document_to_dict(doc: Any, *, with_chunks: bool = True) -> dict:
    """``Document`` → 拍平 dict。

    ``source`` 是 ``DocumentSource``（key/name/rank），这里展开成可读字段。
    ``reporting_period`` 大概率是 None（SDK 解析 bug，见模块 docstring）。
    """
    source = getattr(doc, "source", None)
    out = {
        "id": getattr(doc, "id", None),
        "headline": getattr(doc, "headline", None),
        "sentiment": getattr(doc, "sentiment", None),
        "document_scope": getattr(doc, "document_scope", None),
        "document_type": getattr(doc, "document_type", None),
        "timestamp": getattr(doc, "timestamp", None),
        "language": getattr(doc, "language", None),
        "url": getattr(doc, "url", None),
        "reporting_period": getattr(doc, "reporting_period", None),  # 多半 None
        "reporting_entities": getattr(doc, "reporting_entities", None),
        "source_key": getattr(source, "key", None) if source else None,
        "source_name": getattr(source, "name", None) if source else None,
        "source_rank": getattr(source, "rank", None) if source else None,
    }
    if with_chunks:
        out["chunks"] = [chunk_to_dict(c) for c in (getattr(doc, "chunks", None) or [])]
    return out


class AnnotatedSearcher:
    """语义检索 + 标注抽取（SDK 高层）。

    Parameters
    ----------
    client:
        :class:`~bigdata_toolkit.client.BigdataClient` 实例。
    """

    def __init__(self, client: BigdataClient) -> None:
        self.client = client

    # ------------------------------------------------------------------ #
    # 查询构造辅助                                                        #
    # ------------------------------------------------------------------ #
    @staticmethod
    def entity_query(rp_entity_id: str, keyword: Optional[str] = None):
        """构造 "实体（+ 可选关键词）" 查询。

        entity-scoped 检索是 A 股唯一可用路径（纯 keyword 中文检索 ≈0）。
        rp_entity_id 由 kg.py 的实体解析得到。
        """
        ent = Entity(rp_entity_id)
        if keyword:
            # 用 `&` 运算符组合（SDK 重载的 __and__）；不要用 All(a, b)——
            # All 只接受单个 iterable，传两个位置参数会 TypeError。
            return ent & Keyword(keyword)
        return ent

    # ------------------------------------------------------------------ #
    # 检索执行                                                            #
    # ------------------------------------------------------------------ #
    def search(
        self,
        query,
        *,
        chunk_limit: int = 10,
        date_range: Optional[Union[AbsoluteDateRange, RollingDateRange]] = None,
        scope: DocumentType = DocumentType.ALL,
        sortby: SortBy = SortBy.RELEVANCE,
        rerank_threshold: Optional[float] = None,
        as_dict: bool = True,
    ):
        """跑一次检索，返回 Document 列表（默认拍平成 dict）。

        Parameters
        ----------
        query:
            ``bigdata_client.query`` 的 QueryComponent（用 ``entity_query``
            或自己拼 ``Entity`` / ``Keyword`` / ``All`` / ``Any``）。
        chunk_limit:
            **chunk 上限**（成本承重参数）。内部包成 ``ChunkLimit(n)``，
            即 n 个 chunk = n/10 query_unit。默认 10（≈1 qu）。
            **绝不**直接传整数 doc-limit（会 52x 超支，见模块 docstring）。
        date_range:
            ``AbsoluteDateRange(start, end)`` 或 ``RollingDateRange.*``。
            窗口越窄越省钱（同 limit 下 1 天窗 vs 周窗 ~2.6x 差距）。
        scope:
            ``DocumentType.ALL / NEWS / FILINGS / TRANSCRIPTS`` 等。
        as_dict:
            True 返回拍平 dict（含标注）；False 返回原始 Document 对象。

        Returns
        -------
        list[dict] | list[Document]
        """
        search = self.client.bd.search.new(
            query=query,
            date_range=date_range,
            scope=scope,
            sortby=sortby,
            rerank_threshold=rerank_threshold,
        )
        # 关键：用 ChunkLimit 而非裸 int，强制 chunk 计费而非整页计费
        docs = search.run(ChunkLimit(chunk_limit))

        if as_dict:
            return [document_to_dict(d) for d in docs]
        return docs

    def search_entity(
        self,
        rp_entity_id: str,
        *,
        keyword: Optional[str] = None,
        chunk_limit: int = 10,
        **kwargs,
    ):
        """便捷方法：按 rp_entity_id（+ 可选关键词）检索。"""
        query = self.entity_query(rp_entity_id, keyword)
        return self.search(query, chunk_limit=chunk_limit, **kwargs)

    # ------------------------------------------------------------------ #
    # 标注字典下载（SDK 封装方法，按 id 拉完整标注）                      #
    # ------------------------------------------------------------------ #
    def download_annotated_dict(self, search_id: str) -> dict:
        """对已保存的 search 拉完整标注字典。

        走 SDK 封装方法 ``bd._api.download_annotated_dict(id_)``（不是 raw
        http）。返回完整的 chunk-level 标注。适合先 save 一个 search 再批量
        回拉标注的场景。
        """
        return self.client.conn.download_annotated_dict(search_id)

"""bigdata_toolkit 可跑示例
============================

演示五大能力，全部小样本（成本意识），并用 CostTracker 量化每段消耗。

运行
----
::

    export BIGDATA_API_KEY=bd_v2_xxx            # 你的 Bigdata API key
    export HTTPS_PROXY=http://127.0.0.1:8080    # 仅在需要出站代理时
    # 用装好 bigdata-client 的 Python 环境跑（见 SKILL.md 装包步骤）：
    python scripts/probe_example.py            # 加 --with-search 额外测 chunk 检索

设计原则
--------
- 实体解析 / events / estimates / quota 这些 endpoint **不按 chunk 计费**，
  近乎零成本，所以默认演示它们。
- search（query-chunks）**按 chunk 计费**，示例用 ``chunk_limit=10``（≈1 qu），
  并用 ``--with-search`` 显式开启，避免误烧配额。
"""

from __future__ import annotations

import argparse
import json
import sys

from bigdata_toolkit import (
    AnnotatedSearcher,
    BigdataClient,
    CostModel,
    CostTracker,
    EntityResolver,
    StructuredDataREST,
    rc,
)


def _show(title: str, obj, max_chars: int = 600) -> None:
    print(f"\n{'='*60}\n{title}\n{'='*60}")
    try:
        s = json.dumps(obj, ensure_ascii=False, default=str)
    except TypeError:
        s = str(obj)
    print(s[:max_chars] + (" ..." if len(s) > max_chars else ""))


def main() -> int:
    ap = argparse.ArgumentParser(description="bigdata_toolkit probe example")
    ap.add_argument(
        "--with-search",
        action="store_true",
        help="额外跑一次 search（按 chunk 计费，~1 query_unit）",
    )
    args = ap.parse_args()

    # ---- 0. 统一客户端（key 从 env 读，绝不硬编码） ----
    client = BigdataClient()  # 缺 BIGDATA_API_KEY 会 fail-fast
    tracker = CostTracker(client)
    resolver = EntityResolver(client)
    rest = StructuredDataREST(client)

    quota0 = rc(lambda: tracker.quota())
    _show("[0] 起始配额（chunk 级，1 qu = 10 chunks）", quota0)

    # ---- 1. 实体解析（SDK KG）：美股英文名 + A 股 ISIN crosswalk ----
    aapl_id = rc(lambda: resolver.resolve_id("Apple"))
    print(f"\n[1a] Apple -> rp_entity_id = {aapl_id}")

    # A 股：中文名直查会 0 命中，演示用 ISIN crosswalk（茅台）
    moutai = rc(lambda: resolver.resolve_by_isin(["CNE0000018R8"]))
    _show("[1b] 贵州茅台 ISIN(CNE0000018R8) crosswalk", moutai)
    moutai_id = moutai[0].get("id") if moutai and moutai[0] else None

    # ---- 2. SDK 缺失的前瞻财报日历（REST 逃生舱） ----
    if aapl_id:
        cal = rc(lambda: rest.events_calendar(
            aapl_id,
            categories=["earnings-call"],
            start_date="2026-06-01",
            end_date="2026-12-31",
            limit=3,
        ))
        # 只看顶层结构 + 第一条，避免刷屏
        top_keys = list(cal.keys()) if isinstance(cal, dict) else type(cal).__name__
        _show("[2] 前瞻财报日历 events-calendar（REST）顶层 keys", top_keys)

    # ---- 3. SDK 缺失的前瞻一致预期（REST 逃生舱） ----
    if aapl_id:
        try:
            est = rc(lambda: rest.analyst_estimates(aapl_id, period="quarter", limit=3))
            top = list(est.keys()) if isinstance(est, dict) else type(est).__name__
            _show("[3] 前瞻一致预期 analyst-estimates（REST）顶层 keys", top)
        except Exception as e:  # endpoint 半文档化，schema 可能漂移
            print(f"\n[3] analyst-estimates 调用异常（半文档化 endpoint）: "
                  f"{type(e).__name__}: {str(e)[:160]}")

    # ---- 4. A 股结构化数据可用性验证（茅台 surprise） ----
    if moutai_id:
        try:
            surp = rc(lambda: rest.latest_surprise(moutai_id))
            top = list(surp.keys()) if isinstance(surp, dict) else type(surp).__name__
            _show("[4] 茅台 latest-surprise（REST，验证 A 股结构面可用）顶层 keys", top)
        except Exception as e:
            print(f"\n[4] 茅台 surprise 异常: {type(e).__name__}: {str(e)[:160]}")

    # ---- 5. 成本外推模型（纯计算，演示冷启动预算否决判断） ----
    model = CostModel(chunk_limit_per_query=500, tier="fast")
    est_poc = model.estimate(n_entities=20, n_windows=1)
    est_full = model.estimate(n_entities=100, n_windows=12)  # 100 标的 × 3 年季度
    _show("[5a] 冷启动成本外推 · PoC（20 标的单快照）", est_poc)
    _show("[5b] 冷启动成本外推 · 全量（100 标的 × 3 年季度）", est_full)
    if est_full["pct_of_trial_quota"] > 100:
        print(f"  ⚠️ 全量 backfill 需 {est_full['pct_of_trial_quota']}% trial 配额 "
              f"→ trial 做不了，需要更大的付费配额")

    # ---- 6.（可选）带标注 chunk 抽取（按 chunk 计费，显式开启） ----
    if args.with_search and aapl_id:
        rc(lambda: tracker.snapshot())
        searcher = AnnotatedSearcher(client)
        docs = rc(lambda: searcher.search_entity(aapl_id, keyword="revenue", chunk_limit=10))
        n_chunks = sum(len(d.get("chunks", [])) for d in docs)
        print(f"\n[6] search 返回 {len(docs)} docs / {n_chunks} chunks")
        if docs and docs[0].get("chunks"):
            ch = docs[0]["chunks"][0]
            text = ch.get("text") or ""
            print(f"    首 chunk sentiment={ch.get('sentiment')} "
                  f"entities={len(ch.get('entities') or [])} text[:80]={text[:80]!r}")
        _show("[6] search 实际消耗 delta", rc(lambda: tracker.delta()))
    else:
        print("\n[6] search 已跳过（加 --with-search 开启，按 chunk 计费）")

    quota1 = rc(lambda: tracker.quota())
    print(f"\n[*] 结束配额 used_chunks={quota1['used_chunks']} "
          f"(起始 {quota0['used_chunks']}, 本次净增 "
          f"{quota1['used_chunks'] - quota0['used_chunks']} chunks)")
    return 0


if __name__ == "__main__":
    sys.exit(main())