diff --git a/README.md b/README.md index 62ffe2e..a477de7 100644 --- a/README.md +++ b/README.md @@ -132,6 +132,16 @@ easy-tdx server-info --table easy-tdx symbol-info SZ 000001 --table ``` +### 公告检索(巨潮资讯网) + +```bash +easy-tdx announcement 688017 # 默认 30 条,JSON 输出 +easy-tdx announcement 600519 --count 50 --page 2 # 翻页 +easy-tdx announcement 000001 --table # 表格输出 +``` + +> 独立数据源(巨潮资讯网),无需连接 TDX 行情服务器即可使用。 + ### 技术指标 ```bash @@ -855,6 +865,9 @@ curl "http://localhost:8000/api/v1/mac/symbol-info?market=SZ&code=000001" # 服务器交易时段信息 curl "http://localhost:8000/api/v1/mac/server-info" +# ── 公告检索(巨潮资讯网,独立数据源)── +curl "http://localhost:8000/api/v1/announcements?code=688017&count=30&page=1" + # ── 排行 / 竞价 / 异动 ── # 全 A 涨幅排行前 20 curl "http://localhost:8000/api/v1/mac/quote-list?category=A&count=20&sort_type=CHANGE_PCT" @@ -1515,6 +1528,16 @@ ruff format --check src/ tests/ # format check ## Changelog +### 1.13.0 (2026-06-14) + +**新增巨潮公告检索** — 三层接入(编程 API / CLI / Web API),独立数据源,无需连接 TDX 行情服务器。 + +- 新模块 `easy_tdx.cninfo`:`CninfoClient().get_announcements(code, count=, page=)` 返回 `DataFrame[title, type, date, url]` +- CLI:`easy-tdx announcement 688017 [--count N --page N --table]` +- Web:`GET /api/v1/announcements?code=&count=&page=` +- 标准库 urllib 实现,零新依赖 +- 沿用 #19 修复的 orgId 动态映射 + 三段硬编码 fallback(保证 601xxx 段可查) + ### 1.12.0 (2026-06-14) **新增 4 个技术指标(30 → 34)** — 按"语义空白"补齐三类现有指标库缺失的维度:止损位、机构成本价、趋势启动时机。均为纯 numpy 实现,零新依赖。 diff --git a/pyproject.toml b/pyproject.toml index b57f17c..176e881 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "easy-tdx" -version = "1.12.0" +version = "1.13.0" description = "通达信 TCP 协议行情数据客户端,支持在线行情、离线数据读取与写入同步" readme = "README.md" requires-python = ">=3.10" diff --git a/src/easy_tdx/cli/__init__.py b/src/easy_tdx/cli/__init__.py index 9eb96f3..bd833eb 100644 --- a/src/easy_tdx/cli/__init__.py +++ b/src/easy_tdx/cli/__init__.py @@ -7,6 +7,7 @@ import click from ..backtest.cli import backtest, portfolio from ..screen.cli import screen from .cmd_admin import ping, version +from .cmd_announcement import announcement from .cmd_auction import auction from .cmd_board import ( belong_board, @@ -59,6 +60,7 @@ def cli() -> None: cli.add_command(ping) cli.add_command(version) +cli.add_command(announcement) cli.add_command(kline) cli.add_command(quote) cli.add_command(quote_list) diff --git a/src/easy_tdx/cli/cmd_announcement.py b/src/easy_tdx/cli/cmd_announcement.py new file mode 100644 index 0000000..bd80d58 --- /dev/null +++ b/src/easy_tdx/cli/cmd_announcement.py @@ -0,0 +1,36 @@ +"""公告检索命令(巨潮资讯网数据源,无需 TDX 服务器)。""" + +from __future__ import annotations + +import click + + +@click.command("announcement") +@click.argument("code") +@click.option("--count", default=30, type=int, help="每页数量") +@click.option("--page", default=1, type=int, help="页码(1 起始)") +@click.option("--table", "use_table", is_flag=True, help="表格输出") +@click.option("--output", "output_fmt", type=click.Choice(["json", "table", "csv"]), default="json") +def announcement(code: str, count: int, page: int, use_table: bool, output_fmt: str) -> None: + """检索公司公告(巨潮资讯网,独立数据源,无需连接 TDX)。 + + \b + 示例: + + easy-tdx announcement 688017 + + easy-tdx announcement 600519 --count 50 --page 2 + + easy-tdx announcement 000001 --table + """ + from ..cninfo import CninfoClient, CninfoError + from .output import print_error, print_output + + fmt = "table" if use_table else output_fmt + client = CninfoClient() + try: + df = client.get_announcements(code, count=count, page=page) + except CninfoError as e: + print_error(str(e)) + raise SystemExit(1) from e + print_output(df, fmt) diff --git a/src/easy_tdx/cninfo/__init__.py b/src/easy_tdx/cninfo/__init__.py new file mode 100644 index 0000000..36ad6f0 --- /dev/null +++ b/src/easy_tdx/cninfo/__init__.py @@ -0,0 +1,18 @@ +"""巨潮资讯网(cninfo)公告检索 —— 独立于 TDX 协议的 HTTP 数据源。 + +零额外依赖(标准库 urllib),无需连接 TDX 服务器即可使用。 + +用法:: + + from easy_tdx.cninfo import CninfoClient + + df = CninfoClient().get_announcements("688017", count=30) + # → DataFrame[title, type, date, url] +""" + +from __future__ import annotations + +from .client import CninfoClient +from .models import Announcement, CninfoError + +__all__ = ["CninfoClient", "Announcement", "CninfoError"] diff --git a/src/easy_tdx/cninfo/client.py b/src/easy_tdx/cninfo/client.py new file mode 100644 index 0000000..7e44df3 --- /dev/null +++ b/src/easy_tdx/cninfo/client.py @@ -0,0 +1,193 @@ +"""巨潮资讯网(cninfo)公告检索客户端。 + +独立于 TDX 协议的 HTTP 数据源(标准库 urllib,零额外依赖)。 +公开方法返回 ``pd.DataFrame``,遵循项目 ``get_*`` 约定。 + +参考实现说明(来自 #19 修复): + +- 巨潮 ``orgId`` 并非统一的 ``gssx0{code}`` 格式(如 601318→9900002221、 + 601398→jjxt0000019、688017→9900041602),硬编码会导致大量股票 + (尤其 601xxx 段)返回 ``totalAnnouncement=0``、查不到公告。 +- 优先动态查官方映射表,查不到再回退硬编码规则。 +""" + +from __future__ import annotations + +import json +import logging +from datetime import datetime +from typing import Any +from urllib import parse +from urllib import request as urlrequest + +import pandas as pd + +from .models import Announcement, CninfoError + +logger = logging.getLogger(__name__) + +# 巨潮公告检索请求固定头(Referer/Origin 必填,否则被反爬拦截) +_UA = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" +) +_STOCK_MAP_URL = "http://www.cninfo.com.cn/new/data/szse_stock.json" +_QUERY_URL = "https://www.cninfo.com.cn/new/hisAnnouncement/query" +_DETAIL_URL = "https://www.cninfo.com.cn/new/disclosure/detail?annoId=" + +# 模块级 orgId 映射缓存:首次拉取后全程复用(Cpython dict 读写原子, +# 并发下最坏多发一次请求,可接受) +_ORGID_MAP: dict[str, str] = {} + + +def _ts_to_date(ts: Any) -> str: + """巨潮 ``announcementTime`` 返回 Unix 毫秒整数,转 ``YYYY-MM-DD``。""" + if isinstance(ts, (int, float)): + return datetime.fromtimestamp(ts / 1000).strftime("%Y-%m-%d") + return str(ts)[:10] if ts else "" + + +def _http_get_json(url: str, timeout: float = 15.0) -> Any: + """GET JSON(stdlib urllib,monkeypatch 点)。""" + req = urlrequest.Request(url, headers={"User-Agent": _UA}) + with urlrequest.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read().decode("utf-8")) + + +def _http_post_form(url: str, payload: dict[str, str], timeout: float = 15.0) -> Any: + """POST x-www-form-urlencoded,返回 JSON(stdlib urllib,monkeypatch 点)。""" + data = parse.urlencode(payload).encode("utf-8") + req = urlrequest.Request( + url, + data=data, + headers={ + "User-Agent": _UA, + "Content-Type": "application/x-www-form-urlencoded", + "Referer": "https://www.cninfo.com.cn/new/disclosure", + "Origin": "https://www.cninfo.com.cn", + }, + method="POST", + ) + with urlrequest.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read().decode("utf-8")) + + +class CninfoClient: + """巨潮公告检索客户端(无状态 HTTP,无需 connect/close)。 + + 用法:: + + from easy_tdx.cninfo import CninfoClient + + df = CninfoClient().get_announcements("688017", count=30) + # → DataFrame[title, type, date, url] + """ + + def __init__(self, *, timeout: float = 15.0) -> None: + self.timeout = timeout + + # ------------------------------------------------------------------ + # orgId 解析(#19 修复逻辑:动态表 + 三段硬编码 fallback) + # ------------------------------------------------------------------ + + def _fetch_stock_map(self) -> dict[str, str]: + """拉取官方 szse_stock.json,构建 code→orgId 映射。""" + try: + d = _http_get_json(_STOCK_MAP_URL, timeout=self.timeout) + stock_list = d.get("stockList", []) if isinstance(d, dict) else [] + return {s["code"]: s["orgId"] for s in stock_list} + except Exception as e: # noqa: BLE001 — 拉取失败需 graceful fallback + logger.warning("巨潮 orgId 映射表拉取失败,回退硬编码规则: %s", e) + return {} + + def _resolve_orgid(self, code: str) -> str: + """查股票真实 orgId,动态表优先,查不到回退硬编码规则。""" + global _ORGID_MAP + if not _ORGID_MAP: + fetched = self._fetch_stock_map() + # 只在确实取到数据时写入缓存;空结果保留以便下次重试 + if fetched: + _ORGID_MAP.update(fetched) + org = _ORGID_MAP.get(code) + if org: + return org + # fallback:老格式(仅部分老股票如 600519/600036 适用) + if code.startswith("6"): + return f"gssh0{code}" + if code.startswith("8") or code.startswith("4"): + return f"gsbj0{code}" + return f"gssz0{code}" + + # ------------------------------------------------------------------ + # 公开 API + # ------------------------------------------------------------------ + + def get_announcements( + self, + code: str, + *, + count: int = 30, + page: int = 1, + ) -> pd.DataFrame: + """检索指定股票的公告列表。 + + Args: + code: 6 位股票代码(不含市场前缀),如 ``688017``。 + count: 每页数量(即 pageSize)。 + page: 页码(1 起始)。 + + Returns: + ``DataFrame[title, type, date, url]``,按服务器返回顺序(最新在前)。 + 无结果时返回空 DataFrame(含列名)。 + """ + rows = self._query_announcements(code, count=count, page=page) + if not rows: + return pd.DataFrame(columns=["title", "type", "date", "url"]) + return pd.DataFrame([r.__dict__ for r in rows]) + + def _query_announcements(self, code: str, *, count: int, page: int) -> list[Announcement]: + """POST 公告检索接口,解析为 Announcement 列表。 + + 整个 HTTP + 解析过程统一捕获异常并转为 ``CninfoError``,避免 + ``announcementTime`` 等字段畸形时 ``_ts_to_date`` 抛出未捕获异常。 + """ + org_id = self._resolve_orgid(code) + payload = { + "stock": f"{code},{org_id}", + "tabName": "fulltext", + "pageSize": str(count), + "pageNum": str(page), + "column": "", + "category": "", + "plate": "", + "seDate": "", + "searchkey": "", + "secid": "", + "sortName": "", + "sortType": "", + "isHLtitle": "true", + } + try: + d = _http_post_form(_QUERY_URL, payload, timeout=self.timeout) + items = d.get("announcements", []) if isinstance(d, dict) else None + if not items: + return [] + + result: list[Announcement] = [] + for item in items: + if not isinstance(item, dict): + continue + anno_id = item.get("announcementId", "") + result.append( + Announcement( + title=item.get("announcementTitle", ""), + type=item.get("announcementTypeName", ""), + date=_ts_to_date(item.get("announcementTime")), + url=f"{_DETAIL_URL}{anno_id}", + ) + ) + return result + except CninfoError: + raise + except Exception as e: # noqa: BLE001 — HTTP 失败 / JSON 解析 / 日期转换统一转领域异常 + raise CninfoError(f"巨潮公告检索失败: {e}") from e diff --git a/src/easy_tdx/cninfo/models.py b/src/easy_tdx/cninfo/models.py new file mode 100644 index 0000000..1e40dbe --- /dev/null +++ b/src/easy_tdx/cninfo/models.py @@ -0,0 +1,24 @@ +"""巨潮资讯网(cninfo)数据模型。""" + +from __future__ import annotations + +from dataclasses import dataclass + +from easy_tdx.exceptions import TdxError + + +@dataclass(frozen=True) +class Announcement: + """单条公告记录。 + + 巨潮公告检索接口返回的标准化结构。 + """ + + title: str + type: str + date: str # YYYY-MM-DD + url: str + + +class CninfoError(TdxError): + """巨潮数据请求或解析失败。""" diff --git a/src/easy_tdx/web/app.py b/src/easy_tdx/web/app.py index 002a753..442d295 100644 --- a/src/easy_tdx/web/app.py +++ b/src/easy_tdx/web/app.py @@ -140,6 +140,7 @@ def _create_app( register_exception_handlers(app) # Mount routers + from easy_tdx.web.routers.announcement import router as announcement_router from easy_tdx.web.routers.bars import router as bars_router from easy_tdx.web.routers.block import router as block_router from easy_tdx.web.routers.board_mac import router as board_mac_router @@ -166,5 +167,7 @@ def _create_app( app.include_router(ex_market_router, prefix="/api/v1") # 技术指标路由 app.include_router(indicator_router, prefix="/api/v1") + # 公告检索路由(巨潮资讯网,独立数据源) + app.include_router(announcement_router, prefix="/api/v1") return app diff --git a/src/easy_tdx/web/routers/announcement.py b/src/easy_tdx/web/routers/announcement.py new file mode 100644 index 0000000..9f24c56 --- /dev/null +++ b/src/easy_tdx/web/routers/announcement.py @@ -0,0 +1,37 @@ +"""公告检索路由(巨潮资讯网,独立数据源,不依赖 TDX 服务器)。""" + +from __future__ import annotations + +import asyncio + +from fastapi import APIRouter, Query + +from easy_tdx.web.schemas import DataFrameResponse + +router = APIRouter(tags=["announcement"]) + + +@router.get("/announcements", response_model=DataFrameResponse) +async def announcements( + code: str = Query(..., min_length=6, max_length=6, description="6位股票代码"), + count: int = Query(30, ge=1, le=100, description="每页数量"), + page: int = Query(1, ge=1, description="页码(1 起始)"), +) -> DataFrameResponse: + """检索公司公告列表(巨潮资讯网,独立于 TDX 行情服务器)。 + + TDX 行情不可用时本接口仍可调用。 + """ + from easy_tdx.cninfo import CninfoClient, CninfoError + + client = CninfoClient() + + def _fetch() -> DataFrameResponse: + df = client.get_announcements(code, count=count, page=page) + return DataFrameResponse.from_dataframe(df) + + try: + return await asyncio.to_thread(_fetch) + except CninfoError as e: + from fastapi import HTTPException + + raise HTTPException(status_code=503, detail=str(e)) from e diff --git a/tests/unit/test_cninfo.py b/tests/unit/test_cninfo.py new file mode 100644 index 0000000..dbd3257 --- /dev/null +++ b/tests/unit/test_cninfo.py @@ -0,0 +1,456 @@ +"""巨潮(cninfo)模块离线测试 —— mock HTTP,零网络依赖。 + +覆盖:日期转换、orgId 解析(动态表/三段 fallback)、公告解析、分页、 +错误转换、模块导出。 +""" + +from __future__ import annotations + +import json +from typing import Any + +import pandas as pd +import pytest + +# --------------------------------------------------------------------------- +# 导出 +# --------------------------------------------------------------------------- + + +def test_public_exports() -> None: + """模块应导出 CninfoClient / Announcement / CninfoError。""" + from easy_tdx import cninfo + + assert hasattr(cninfo, "CninfoClient") + assert hasattr(cninfo, "Announcement") + assert hasattr(cninfo, "CninfoError") + + +def test_announcement_is_frozen_dataclass() -> None: + """Announcement 应为 frozen dataclass,含 title/type/date/url。""" + from easy_tdx.cninfo import Announcement + + a = Announcement(title="t", type="ty", date="2026-06-14", url="http://x") + assert a.title == "t" + assert a.type == "ty" + assert a.date == "2026-06-14" + assert a.url == "http://x" + # frozen + with pytest.raises(Exception): + a.title = "mutated" # type: ignore[misc] + + +def test_cninfo_error_is_exception() -> None: + from easy_tdx.cninfo import CninfoError + from easy_tdx.exceptions import TdxError + + assert issubclass(CninfoError, Exception) + # 回归 #1:CninfoError 必须继承 TdxError,保证全局 except TdxError 覆盖 + assert issubclass(CninfoError, TdxError) + + +# --------------------------------------------------------------------------- +# 日期转换 +# --------------------------------------------------------------------------- + + +def test_ts_to_date_from_millis() -> None: + """Unix 毫秒整数应转为 YYYY-MM-DD。""" + from easy_tdx.cninfo.client import _ts_to_date + + # 1718323200000 ms = 2024-06-14 00:00:00 UTC ≈ 当地日期 + assert _ts_to_date(1718323200000) # 非空字符串,长度 10 + assert len(_ts_to_date(1718323200000)) == 10 + + +def test_ts_to_date_from_string() -> None: + """字符串输入应取前 10 字符。""" + from easy_tdx.cninfo.client import _ts_to_date + + assert _ts_to_date("2026-06-14T08:00:00") == "2026-06-14" + + +def test_ts_to_date_empty() -> None: + from easy_tdx.cninfo.client import _ts_to_date + + assert _ts_to_date("") == "" + assert _ts_to_date(None) == "" + + +# --------------------------------------------------------------------------- +# orgId 解析(动态表 + 三段 fallback) +# --------------------------------------------------------------------------- + + +@pytest.fixture +def reset_orgid_cache() -> Any: + """每个测试前后清空 orgId 缓存,保证隔离。""" + import easy_tdx.cninfo.client as mod + + mod._ORGID_MAP.clear() + yield + mod._ORGID_MAP.clear() + + +def _patch_stock_map(monkeypatch: pytest.MonkeyPatch, mapping: dict[str, str]) -> None: + """让 _fetch_stock_map 返回给定映射(不触网)。""" + monkeypatch.setattr( + "easy_tdx.cninfo.client._http_get_json", + lambda url, timeout=15.0: { + "stockList": [{"code": c, "orgId": o} for c, o in mapping.items()] + }, + ) + + +def test_resolve_orgid_from_dynamic_map( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """动态表命中应返回表中 orgId。""" + from easy_tdx.cninfo import CninfoClient + + _patch_stock_map(monkeypatch, {"688017": "9900041602", "601318": "9900002221"}) + client = CninfoClient() + assert client._resolve_orgid("688017") == "9900041602" + assert client._resolve_orgid("601318") == "9900002221" + + +def test_resolve_orgid_fallback_6_prefix( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """动态表无此 code 且 6 开头 → gssh0{code}。""" + from easy_tdx.cninfo import CninfoClient + + _patch_stock_map(monkeypatch, {}) + client = CninfoClient() + assert client._resolve_orgid("600519") == "gssh0600519" + + +def test_resolve_orgid_fallback_8_prefix( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """北交所 8/4 开头 → gsbj0{code}。""" + from easy_tdx.cninfo import CninfoClient + + _patch_stock_map(monkeypatch, {}) + client = CninfoClient() + assert client._resolve_orgid("830799") == "gsbj0830799" + assert client._resolve_orgid("430047") == "gsbj0430047" + + +def test_resolve_orgid_fallback_sz_default( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """其他前缀(深圳)→ gssz0{code}。""" + from easy_tdx.cninfo import CninfoClient + + _patch_stock_map(monkeypatch, {}) + client = CninfoClient() + assert client._resolve_orgid("000001") == "gssz0000001" + + +def test_resolve_orgid_empty_map_does_not_cache( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """映射表为空时不写入缓存,下次仍会重试(避免永久 fallback)。""" + import easy_tdx.cninfo.client as mod + from easy_tdx.cninfo import CninfoClient + + _patch_stock_map(monkeypatch, {}) + client = CninfoClient() + client._resolve_orgid("600519") + assert mod._ORGID_MAP == {} + + +def test_resolve_orgid_fetch_failure_fallback( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """映射表拉取异常应 graceful fallback 到硬编码规则,不抛错。""" + from easy_tdx.cninfo import CninfoClient + + def _boom(url: str, timeout: float = 15.0) -> Any: + raise OSError("network down") + + monkeypatch.setattr("easy_tdx.cninfo.client._http_get_json", _boom) + client = CninfoClient() + # 不抛错,回退 SH 规则 + assert client._resolve_orgid("600519") == "gssh0600519" + + +def test_resolve_orgid_cache_reused( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """第二次调用不应再次拉取映射表(缓存命中)。""" + call_count = {"n": 0} + + def _fake(url: str, timeout: float = 15.0) -> Any: + call_count["n"] += 1 + return {"stockList": [{"code": "688017", "orgId": "9900041602"}]} + + monkeypatch.setattr("easy_tdx.cninfo.client._http_get_json", _fake) + from easy_tdx.cninfo import CninfoClient + + client = CninfoClient() + client._resolve_orgid("688017") + client._resolve_orgid("688017") + assert call_count["n"] == 1 + + +# --------------------------------------------------------------------------- +# 公告查询与解析 +# --------------------------------------------------------------------------- + + +_QUERY_RESPONSE: dict[str, Any] = { + "announcements": [ + { + "announcementTitle": "关于召开2025年年度股东大会的通知", + "announcementTypeName": "股东大会", + "announcementTime": 1749859200000, + "announcementId": "abc123", + }, + { + "announcementTitle": "2024年年度报告", + "announcementTypeName": "定期报告", + "announcementTime": 1740614400000, + "announcementId": "def456", + }, + ], + "totalAnnouncement": 2, +} + + +def test_get_announcements_returns_dataframe( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """应返回 DataFrame[title, type, date, url]。""" + _patch_stock_map(monkeypatch, {"688017": "9900041602"}) + monkeypatch.setattr( + "easy_tdx.cninfo.client._http_post_form", + lambda url, payload, timeout=15.0: _QUERY_RESPONSE, + ) + from easy_tdx.cninfo import CninfoClient + + df = CninfoClient().get_announcements("688017", count=30, page=1) + assert isinstance(df, pd.DataFrame) + assert list(df.columns) == ["title", "type", "date", "url"] + assert len(df) == 2 + assert df.iloc[0]["title"] == "关于召开2025年年度股东大会的通知" + assert df.iloc[0]["type"] == "股东大会" + assert len(df.iloc[0]["date"]) == 10 # YYYY-MM-DD + assert df.iloc[0]["url"].endswith("abc123") + assert "cninfo.com.cn" in df.iloc[0]["url"] + + +def test_get_announcements_empty(monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any) -> None: + """无公告应返回带列名的空 DataFrame。""" + _patch_stock_map(monkeypatch, {"688017": "9900041602"}) + monkeypatch.setattr( + "easy_tdx.cninfo.client._http_post_form", + lambda url, payload, timeout=15.0: {"announcements": []}, + ) + from easy_tdx.cninfo import CninfoClient + + df = CninfoClient().get_announcements("688017") + assert isinstance(df, pd.DataFrame) + assert df.empty + assert list(df.columns) == ["title", "type", "date", "url"] + + +def test_get_announcements_missing_key( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """响应缺少 announcements 键应视为空结果。""" + _patch_stock_map(monkeypatch, {"688017": "9900041602"}) + monkeypatch.setattr( + "easy_tdx.cninfo.client._http_post_form", + lambda url, payload, timeout=15.0: {"totalAnnouncement": 0}, + ) + from easy_tdx.cninfo import CninfoClient + + df = CninfoClient().get_announcements("688017") + assert df.empty + + +def test_get_announcements_request_failure_raises_cninfo_error( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """HTTP 异常应转为 CninfoError。""" + _patch_stock_map(monkeypatch, {"688017": "9900041602"}) + + def _boom(url: str, payload: dict[str, str], timeout: float = 15.0) -> Any: + raise OSError("connection refused") + + monkeypatch.setattr("easy_tdx.cninfo.client._http_post_form", _boom) + from easy_tdx.cninfo import CninfoClient, CninfoError + + with pytest.raises(CninfoError): + CninfoClient().get_announcements("688017") + + +def test_get_announcements_pagination( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """count/page 应正确传入 payload。""" + _patch_stock_map(monkeypatch, {"688017": "9900041602"}) + captured: dict[str, Any] = {} + + def _capture(url: str, payload: dict[str, str], timeout: float = 15.0) -> Any: + captured.update(payload) + return {"announcements": []} + + monkeypatch.setattr("easy_tdx.cninfo.client._http_post_form", _capture) + from easy_tdx.cninfo import CninfoClient + + CninfoClient().get_announcements("688017", count=50, page=3) + assert captured["pageSize"] == "50" + assert captured["pageNum"] == "3" + assert captured["stock"] == "688017,9900041602" + assert captured["tabName"] == "fulltext" + + +def test_get_announcements_uses_fallback_orgid( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """601xxx 段动态表未命中时用 gssh0 fallback,stock 字段含该 orgId。""" + _patch_stock_map(monkeypatch, {}) # 空表 → fallback + captured: dict[str, Any] = {} + + def _capture(url: str, payload: dict[str, str], timeout: float = 15.0) -> Any: + captured.update(payload) + return {"announcements": []} + + monkeypatch.setattr("easy_tdx.cninfo.client._http_post_form", _capture) + from easy_tdx.cninfo import CninfoClient + + CninfoClient().get_announcements("601318") + assert captured["stock"] == "601318,gssh0601318" + + +def test_get_announcements_skips_non_dict_items( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """announcements 列表中混入非 dict 元素应被跳过。""" + _patch_stock_map(monkeypatch, {"688017": "9900041602"}) + monkeypatch.setattr( + "easy_tdx.cninfo.client._http_post_form", + lambda url, payload, timeout=15.0: { + "announcements": [ + "not a dict", + { + "announcementTitle": "ok", + "announcementTime": 1749859200000, + "announcementId": "x", + }, + ] + }, + ) + from easy_tdx.cninfo import CninfoClient + + df = CninfoClient().get_announcements("688017") + assert len(df) == 1 + assert df.iloc[0]["title"] == "ok" + + +def test_get_announcements_response_not_dict( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """响应非 dict(如 list)应视为空结果,不抛错。""" + _patch_stock_map(monkeypatch, {"688017": "9900041602"}) + monkeypatch.setattr( + "easy_tdx.cninfo.client._http_post_form", + lambda url, payload, timeout=15.0: ["unexpected", "list"], + ) + from easy_tdx.cninfo import CninfoClient + + df = CninfoClient().get_announcements("688017") + assert df.empty + + +def test_get_announcements_malformed_timestamp_wrapped_as_cninfo_error( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """announcementTime 畸形(导致 fromtimestamp 溢出)应转 CninfoError,不裸抛。 + + 回归 #2:修复前 _ts_to_date 在 try 块外,畸形时间戳会抛 OverflowError/ + ValueError 裸异常;修复后整个解析路径统一转 CninfoError。 + """ + _patch_stock_map(monkeypatch, {"688017": "9900041602"}) + monkeypatch.setattr( + "easy_tdx.cninfo.client._http_post_form", + lambda url, payload, timeout=15.0: { + "announcements": [ + {"announcementTitle": "x", "announcementTime": 10**30, "announcementId": "y"} + ] + }, + ) + from easy_tdx.cninfo import CninfoClient, CninfoError + + with pytest.raises(CninfoError): + CninfoClient().get_announcements("688017") + + +# --------------------------------------------------------------------------- +# urllib helper 烟雾测试(不触网,仅验证 JSON 解码路径) +# --------------------------------------------------------------------------- + + +def test_http_post_form_urlencoded_body(monkeypatch: pytest.MonkeyPatch) -> None: + """_http_post_form 应以 application/x-www-form-urlencoded 发送。""" + import easy_tdx.cninfo.client as mod + + captured: dict[str, Any] = {} + + class _FakeResp: + def __init__(self, body: bytes) -> None: + self._body = body + + def read(self) -> bytes: + return self._body + + def __enter__(self) -> _FakeResp: + return self + + def __exit__(self, *args: Any) -> None: + pass + + def _fake_urlopen(req: Any, timeout: float = 15.0) -> _FakeResp: + captured["data"] = req.data + captured["headers"] = {k: v for k, v in req.header_items()} + captured["method"] = req.get_method() + return _FakeResp(json.dumps({"ok": True}).encode("utf-8")) + + monkeypatch.setattr(mod.urlrequest, "urlopen", _fake_urlopen) + result = mod._http_post_form("https://example.com/api", {"pageNum": "2", "pageSize": "30"}) + assert result == {"ok": True} + assert b"pageNum=2" in captured["data"] + assert b"pageSize=30" in captured["data"] + assert captured["method"] == "POST" + headers = {k.lower(): v for k, v in captured["headers"].items()} + assert "cninfo.com.cn" in headers.get("referer", "") + assert "x-www-form-urlencoded" in headers.get("content-type", "") + + +def test_http_get_json_headers(monkeypatch: pytest.MonkeyPatch) -> None: + """_http_get_json 应携带 User-Agent。""" + import easy_tdx.cninfo.client as mod + + captured: dict[str, Any] = {} + + class _FakeResp: + def read(self) -> bytes: + return json.dumps({"ok": 1}).encode("utf-8") + + def __enter__(self) -> _FakeResp: + return self + + def __exit__(self, *args: Any) -> None: + pass + + def _fake_urlopen(req: Any, timeout: float = 15.0) -> _FakeResp: + captured["headers"] = {k: v for k, v in req.header_items()} + return _FakeResp() + + monkeypatch.setattr(mod.urlrequest, "urlopen", _fake_urlopen) + assert mod._http_get_json("https://example.com/x.json") == {"ok": 1} + headers = {k.lower(): v for k, v in captured["headers"].items()} + assert "user-agent" in headers diff --git a/tests/unit/test_web_api.py b/tests/unit/test_web_api.py index 02e1046..c328f6e 100644 --- a/tests/unit/test_web_api.py +++ b/tests/unit/test_web_api.py @@ -268,6 +268,7 @@ def test_full_app_routes_registered(): "/api/v1/xdxr", "/api/v1/block", "/api/v1/chanlun", + "/api/v1/announcements", "/ws/realtime", ] for prefix in expected_prefixes: