From 574b284dd50e01dd2c1101d3dd9b4aeb33a66471 Mon Sep 17 00:00:00 2001 From: Justin Gu <97915@qq.com> Date: Sun, 14 Jun 2026 23:58:24 +0800 Subject: [PATCH] =?UTF-8?q?fix(cninfo):=20URL=20404=20+=20type=20null=20+?= =?UTF-8?q?=20=E8=A1=A8=E6=A0=BC=E6=88=AA=E6=96=AD=20+=20PDF=20=E4=B8=8B?= =?UTF-8?q?=E8=BD=BD=EF=BC=88=E5=AE=9E=E6=B5=8B=20601088=20=E6=9A=B4?= =?UTF-8?q?=E9=9C=B2=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 58 +++-- src/easy_tdx/cli/cmd_announcement.py | 71 ++++++- src/easy_tdx/cli/output.py | 19 +- src/easy_tdx/cninfo/client.py | 99 ++++++++- src/easy_tdx/cninfo/models.py | 40 ++++ tests/unit/test_cninfo.py | 304 +++++++++++++++++++++++++-- 6 files changed, 547 insertions(+), 44 deletions(-) diff --git a/README.md b/README.md index 7c488fc..329d657 100644 --- a/README.md +++ b/README.md @@ -136,11 +136,15 @@ 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 # 表格输出 +easy-tdx announcement 601088 --count 10 --page 2 # 翻页 +easy-tdx announcement 000001 --table # 表格输出（不截断 url） + +# 下载最新 5 条公告的 PDF 到 ./pdfs 目录 +easy-tdx announcement 601088 --count 5 --download 5 --download-dir ./pdfs ``` > 独立数据源（巨潮资讯网），无需连接 TDX 行情服务器即可使用。 +> 返回的 ``url`` 含 4 参数可直接打开，``pdf_url`` 为 PDF 直链。 ### 技术指标 @@ -868,7 +872,8 @@ curl "http://localhost:8000/api/v1/mac/server-info" # ── 公告检索（巨潮资讯网，独立数据源）── # 检索公司公告（无需 TDX 行情服务器） curl "http://localhost:8000/api/v1/announcements?code=688017&count=30&page=1" -# 返回：{"data": [{"title": "...", "type": "...", "date": "...", "url": "..."}], "count": 30} +# 返回每条含 url（4 参数可直点打开）和 pdf_url（PDF 直链）： +# {"data": [{"title":"...","type":"...","date":"...","url":".../detail?stockCode=...","pdf_url":"http://static.cninfo.com.cn/.../xxx.PDF",...}], "count": 30} # ── 排行 / 竞价 / 异动 ── # 全 A 涨幅排行前 20 @@ -1338,19 +1343,39 @@ client = CninfoClient() # 检索公告（默认 30 条，最新在前） df = client.get_announcements("688017") -# → DataFrame[title, type, date, url] +# → DataFrame[title, type, date, url, code, org_id, announcement_id, announcement_time, pdf_url] # 翻页 + 自定义数量 -df = client.get_announcements("600519", count=50, page=2) +df = client.get_announcements("601088", count=10, page=2) -# 返回示例： -# title type date url -# 0 关于召开2025年年度股东大会的通知 股东大会 2025-06-14 https://www.cninfo.com.cn/new/disclosure/detail?annoId=abc123 -# 1 2024年年度报告 定期报告 2025-03-28 https://www.cninfo.com.cn/new/disclosure/detail?annoId=def456 +# 返回示例（url 含 4 参数可直点打开，pdf_url 为 PDF 直链）： +# title type date url pdf_url +# 0 关于召开2025年年度股东大会... 股东大会 2025-06-14 .../detail?stockCode=688017&announcementId=... http://static.cninfo.com.cn/.../xxx.PDF +# 1 2024年年度报告 PDF 2025-03-28 .../detail?stockCode=688017&announcementId=... http://static.cninfo.com.cn/.../yyy.PDF ``` -> orgId 解析沿用 #19 修复：动态拉取官方映射表，查不到回退硬编码规则， -> 保证 601xxx 等非标 orgId 段也能正常查询。 +> - ``type`` 优先取 cninfo 的 ``announcementTypeName``；该字段对很多公告为 null +> （数据源限制），此时回退到 ``adjunctType``（如 "PDF"），再为空给空字符串。 +> - ``url`` 必须含 4 参数（``stockCode``/``announcementId``/``orgId``/``announcementTime``） +> 才能打开，少参数会 404。 +> - orgId 解析沿用 #19 修复：动态拉取官方映射表，查不到回退硬编码规则， +> 保证 601xxx 等非标 orgId 段也能正常查询。 + +#### 下载公告 PDF + +```python +# 下载最新一条公告的 PDF 到当前目录 +df = client.get_announcements("601088", count=5) +path = client.download_pdf(df.iloc[0]) # 接受 Announcement 或 DataFrame 的一行 +print(path) # /abs/path/20260605_1225351400.PDF + +# 批量下载 +for _, row in df.iterrows(): + try: + path = client.download_pdf(row, dest_dir="./pdfs") + except Exception as e: + print(f"跳过（无附件或失败）: {e}") +``` ## 枚举参考 @@ -1560,12 +1585,19 @@ ruff format --check src/ tests/ # format check **新增巨潮公告检索** — 三层接入（编程 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]` +- 新模块 `easy_tdx.cninfo`：`CninfoClient().get_announcements(code, count=, page=)` 返回 `DataFrame[title, type, date, url, code, org_id, announcement_id, announcement_time, pdf_url]` +- CLI：`easy-tdx announcement 688017 [--count N --page N --table] [--download N --download-dir DIR]` - Web：`GET /api/v1/announcements?code=&count=&page=` +- `CninfoClient().download_pdf(announcement, dest_dir=)` 下载公告 PDF（接受 `Announcement` 或 DataFrame 一行） - 标准库 urllib 实现，零新依赖 - 沿用 #19 修复的 orgId 动态映射 + 三段硬编码 fallback（保证 601xxx 段可查） +**Bug 修复**（实测 601088 暴露）： + +- `type` 列全 null：cninfo 对很多公告不填 `announcementTypeName`，回退到 `adjunctType`（如 "PDF"） +- `url` 404：原仅拼 `announcementId` 一个参数，补全 4 参数 `stockCode`/`announcementId`/`orgId`/`announcementTime` +- 表格输出 url 被截断成 `https://www.cninfo.com.cn/new/`：`announcement --table` 改用不截断渲染 + ### 1.12.0 (2026-06-14) **新增 4 个技术指标（30 → 34）** — 按"语义空白"补齐三类现有指标库缺失的维度：止损位、机构成本价、趋势启动时机。均为纯 numpy 实现，零新依赖。 diff --git a/src/easy_tdx/cli/cmd_announcement.py b/src/easy_tdx/cli/cmd_announcement.py index bd80d58..5be2f15 100644 --- a/src/easy_tdx/cli/cmd_announcement.py +++ b/src/easy_tdx/cli/cmd_announcement.py @@ -11,7 +11,28 @@ import click @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: +@click.option( + "--download", + "download_n", + type=int, + default=0, + help="下载最新 N 条公告的 PDF（0=不下载），需配合 --download-dir", +) +@click.option( + "--download-dir", + "download_dir", + default=".", + help="PDF 保存目录（默认当前目录，自动创建）", +) +def announcement( + code: str, + count: int, + page: int, + use_table: bool, + output_fmt: str, + download_n: int, + download_dir: str, +) -> None: """检索公司公告（巨潮资讯网，独立数据源，无需连接 TDX）。 \b @@ -19,9 +40,12 @@ def announcement(code: str, count: int, page: int, use_table: bool, output_fmt: easy-tdx announcement 688017 - easy-tdx announcement 600519 --count 50 --page 2 + easy-tdx announcement 601088 --count 50 --page 2 easy-tdx announcement 000001 --table + + # 下载最新 5 条公告的 PDF 到 ./pdfs 目录 + easy-tdx announcement 601088 --count 5 --download 5 --download-dir ./pdfs """ from ..cninfo import CninfoClient, CninfoError from .output import print_error, print_output @@ -33,4 +57,45 @@ def announcement(code: str, count: int, page: int, use_table: bool, output_fmt: except CninfoError as e: print_error(str(e)) raise SystemExit(1) from e - print_output(df, fmt) + + # 表格模式下不截断长文本列（url/title 经常超 30 字符，默认 output 会切到不可读） + if fmt == "table": + from .output import _render_table_full + + click.echo(_render_table_full(df)) + else: + print_output(df, fmt) + + # PDF 下载 + if download_n > 0: + if df.empty: + print_error("无公告可下载") + raise SystemExit(1) + to_download = df.head(download_n) + click.echo(f"开始下载 {len(to_download)} 条公告 PDF 到 {download_dir} ...", err=True) + downloaded = 0 + skipped = 0 + # 复用 _query_announcements 已解析的 Announcement 对象需要重新查； + # 这里直接从 DataFrame 行构造 Announcement 以避免二次网络请求。 + from ..cninfo.models import Announcement + + for _, row in to_download.iterrows(): + anno = Announcement( + title=row["title"], + type=row["type"], + date=row["date"], + url=row["url"], + code=row["code"], + org_id=row["org_id"], + announcement_id=row["announcement_id"], + announcement_time=row["announcement_time"], + pdf_url=row["pdf_url"], + ) + try: + path = client.download_pdf(anno, dest_dir=download_dir) + click.echo(f" ✓ {path}", err=True) + downloaded += 1 + except CninfoError as e: + click.echo(f" ✗ 跳过（{row['title'][:30]}）: {e}", err=True) + skipped += 1 + click.echo(f"完成：{downloaded} 个下载，{skipped} 个跳过", err=True) diff --git a/src/easy_tdx/cli/output.py b/src/easy_tdx/cli/output.py index 36b897d..8170e15 100644 --- a/src/easy_tdx/cli/output.py +++ b/src/easy_tdx/cli/output.py @@ -35,13 +35,23 @@ def print_error(msg: str) -> None: def _render_table(df: pd.DataFrame) -> str: """将 DataFrame 渲染为人类可读的文本表格。""" + return _render_table_impl(df, truncate=30) + + +def _render_table_full(df: pd.DataFrame) -> str: + """渲染表格但**不截断**长文本列（适用于 url/title 等长字段）。""" + return _render_table_impl(df, truncate=None) + + +def _render_table_impl(df: pd.DataFrame, truncate: int | None) -> str: + """表格渲染实现。``truncate=None`` 时不截断 object 列。""" if df.empty: return "(无数据)" display_df = df.copy() for col in display_df.columns: - if display_df[col].dtype == object: - display_df[col] = display_df[col].astype(str).str.slice(0, 30) + if display_df[col].dtype == object and truncate is not None: + display_df[col] = display_df[col].astype(str).str.slice(0, truncate) try: import tabulate @@ -51,10 +61,11 @@ def _render_table(df: pd.DataFrame) -> str: lines: list[str] = [] cols = list(display_df.columns) header = " | ".join(str(c) for c in cols) - sep = "-+-".join("-" * min(len(str(c)), 30) for c in cols) + cap = truncate if truncate is not None else 100 + sep = "-+-".join("-" * min(len(str(c)), cap) for c in cols) lines.append(header) lines.append(sep) for _, row in display_df.iterrows(): - line = " | ".join(str(v)[:30] for v in row.values) + line = " | ".join(str(v)[:cap] for v in row.values) lines.append(line) return "\n".join(lines) diff --git a/src/easy_tdx/cninfo/client.py b/src/easy_tdx/cninfo/client.py index 7e44df3..da61ebe 100644 --- a/src/easy_tdx/cninfo/client.py +++ b/src/easy_tdx/cninfo/client.py @@ -15,6 +15,7 @@ from __future__ import annotations import json import logging +import os from datetime import datetime from typing import Any from urllib import parse @@ -22,7 +23,7 @@ from urllib import request as urlrequest import pandas as pd -from .models import Announcement, CninfoError +from .models import Announcement, CninfoError, build_detail_url, build_pdf_url logger = logging.getLogger(__name__) @@ -33,7 +34,6 @@ _UA = ( ) _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 读写原子， # 并发下最坏多发一次请求，可接受） @@ -137,14 +137,87 @@ class CninfoClient: page: 页码（1 起始）。 Returns: - ``DataFrame[title, type, date, url]``，按服务器返回顺序（最新在前）。 + ``DataFrame[title, type, date, url, code, org_id, announcement_id, + announcement_time, pdf_url]``，按服务器返回顺序（最新在前）。 无结果时返回空 DataFrame（含列名）。 + + Note: + ``type`` 列优先取 cninfo 的 ``announcementTypeName``；该字段对很多 + 公告为 null（数据源限制），此时回退到 ``adjunctType``（如 "PDF"）， + 再为空给空字符串。 """ rows = self._query_announcements(code, count=count, page=page) + cols = [ + "title", + "type", + "date", + "url", + "code", + "org_id", + "announcement_id", + "announcement_time", + "pdf_url", + ] if not rows: - return pd.DataFrame(columns=["title", "type", "date", "url"]) + return pd.DataFrame(columns=cols) return pd.DataFrame([r.__dict__ for r in rows]) + def download_pdf( + self, + announcement: Announcement | pd.Series[Any], + dest_dir: str | os.PathLike[str] = ".", + *, + filename: str | None = None, + ) -> str: + """下载公告 PDF 附件到本地。 + + Args: + announcement: ``get_announcements`` 返回的单条记录（需含 pdf_url）。 + 也接受 ``pd.Series``（DataFrame 的一行）。 + dest_dir: 目标目录，默认当前目录。不存在会自动创建。 + filename: 保存文件名（不含路径）。默认 ``{date}_{announcement_id}.PDF``。 + + Returns: + 下载后的本地文件绝对路径。 + + Raises: + CninfoError: 该公告无 PDF 附件（pdf_url 为空），或下载失败。 + """ + # 统一为字段访问：兼容 pd.Series（DataFrame.iloc[i]）和 Announcement + if isinstance(announcement, Announcement): + pdf_url = announcement.pdf_url + anno_time = announcement.announcement_time + anno_id = announcement.announcement_id + else: + # pd.Series 的 .get/__getitem__ 行为 + pdf_url = str(announcement.get("pdf_url", "") or "") + anno_time = int(announcement.get("announcement_time", 0) or 0) + anno_id = str(announcement.get("announcement_id", "x") or "") + + if not pdf_url: + raise CninfoError("该公告无 PDF 附件（pdf_url 为空）") + + dest_dir = os.fspath(dest_dir) + os.makedirs(dest_dir, exist_ok=True) + if filename is None: + # announcement_time 为毫秒时间戳，转 YYYYMMDD 更可读 + try: + date_str = datetime.fromtimestamp(anno_time / 1000).strftime("%Y%m%d") + except (OSError, ValueError, OverflowError): + date_str = "unknown" + filename = f"{date_str}_{anno_id}.PDF" + + filepath = os.path.join(dest_dir, filename) + try: + req = urlrequest.Request(pdf_url, headers={"User-Agent": _UA}) + with urlrequest.urlopen(req, timeout=self.timeout) as resp: + data = resp.read() + with open(filepath, "wb") as f: + f.write(data) + except Exception as e: # noqa: BLE001 — 下载失败统一转领域异常 + raise CninfoError(f"PDF 下载失败: {e}") from e + return os.path.abspath(filepath) + def _query_announcements(self, code: str, *, count: int, page: int) -> list[Announcement]: """POST 公告检索接口，解析为 Announcement 列表。 @@ -177,13 +250,21 @@ class CninfoClient: for item in items: if not isinstance(item, dict): continue - anno_id = item.get("announcementId", "") + anno_id = str(item.get("announcementId", "") or "") + anno_time = item.get("announcementTime", 0) or 0 + # type 回退：announcementTypeName 常为 null → adjunctType (如 "PDF") + type_name = item.get("announcementTypeName") or item.get("adjunctType") or "" result.append( Announcement( - title=item.get("announcementTitle", ""), - type=item.get("announcementTypeName", ""), - date=_ts_to_date(item.get("announcementTime")), - url=f"{_DETAIL_URL}{anno_id}", + title=item.get("announcementTitle", "") or "", + type=type_name, + date=_ts_to_date(anno_time), + url=build_detail_url(code, anno_id, org_id, anno_time), + code=code, + org_id=org_id, + announcement_id=anno_id, + announcement_time=anno_time, + pdf_url=build_pdf_url(item.get("adjunctUrl", "") or ""), ) ) return result diff --git a/src/easy_tdx/cninfo/models.py b/src/easy_tdx/cninfo/models.py index 1e40dbe..fcd8bf5 100644 --- a/src/easy_tdx/cninfo/models.py +++ b/src/easy_tdx/cninfo/models.py @@ -6,18 +6,58 @@ from dataclasses import dataclass from easy_tdx.exceptions import TdxError +# PDF 附件直链前缀（adjunctUrl 拼此 base 即真实 PDF 地址） +_PDF_BASE = "http://static.cninfo.com.cn/" + @dataclass(frozen=True) class Announcement: """单条公告记录。 巨潮公告检索接口返回的标准化结构。 + + Attributes: + title: 公告标题。 + type: 公告类型（优先 ``announcementTypeName``，缺失时回退 ``adjunctType`` + 如 "PDF"，再缺失给空字符串 — cninfo 对很多公告不填 typeName）。 + date: 公告日期 ``YYYY-MM-DD``。 + url: 公告详情页 URL（含 stockCode/announcementId/orgId/announcementTime 四参数）。 + code: 6 位股票代码。 + org_id: 巨潮 orgId（详情页 URL 参数）。 + announcement_id: 巨潮公告 ID（详情页 URL 参数 + PDF 文件名构成）。 + announcement_time: 原始 Unix 毫秒时间戳（详情页 URL 参数）。 + pdf_url: PDF 附件直链（``adjunctUrl`` 拼接 ``static.cninfo.com.cn``）， + 无附件时为空字符串。 """ title: str type: str date: str # YYYY-MM-DD url: str + code: str + org_id: str + announcement_id: str + announcement_time: int + pdf_url: str + + +def build_detail_url(code: str, announcement_id: str, org_id: str, announcement_time: int) -> str: + """构造公告详情页 URL（4 参数缺一不可，否则 404）。""" + return ( + "https://www.cninfo.com.cn/new/disclosure/detail?" + f"stockCode={code}&announcementId={announcement_id}" + f"&orgId={org_id}&announcementTime={announcement_time}" + ) + + +def build_pdf_url(adjunct_url: str) -> str: + """``adjunctUrl``（如 finalpage/2026-06-05/xxx.PDF）拼成完整 PDF 直链。 + + 无附件（adjunctUrl 为空）返回空字符串。 + """ + if not adjunct_url: + return "" + return f"{_PDF_BASE}{adjunct_url}" class CninfoError(TdxError): diff --git a/tests/unit/test_cninfo.py b/tests/unit/test_cninfo.py index dbd3257..ca7e0eb 100644 --- a/tests/unit/test_cninfo.py +++ b/tests/unit/test_cninfo.py @@ -1,12 +1,13 @@ """巨潮（cninfo）模块离线测试 —— mock HTTP，零网络依赖。 -覆盖：日期转换、orgId 解析（动态表/三段 fallback）、公告解析、分页、 -错误转换、模块导出。 +覆盖：日期转换、orgId 解析（动态表/三段 fallback）、公告解析（含 URL 4 参数、 +type 回退、pdf_url）、PDF 下载、分页、错误转换、模块导出。 """ from __future__ import annotations import json +from pathlib import Path from typing import Any import pandas as pd @@ -27,14 +28,28 @@ def test_public_exports() -> None: def test_announcement_is_frozen_dataclass() -> None: - """Announcement 应为 frozen dataclass，含 title/type/date/url。""" + """Announcement 应为 frozen dataclass，含全部字段。""" from easy_tdx.cninfo import Announcement - a = Announcement(title="t", type="ty", date="2026-06-14", url="http://x") + a = Announcement( + title="t", + type="ty", + date="2026-06-14", + url="http://x", + code="688017", + org_id="9900041602", + announcement_id="abc123", + announcement_time=1718323200000, + pdf_url="http://static.cninfo.com.cn/x.PDF", + ) assert a.title == "t" assert a.type == "ty" assert a.date == "2026-06-14" - assert a.url == "http://x" + assert a.code == "688017" + assert a.org_id == "9900041602" + assert a.announcement_id == "abc123" + assert a.announcement_time == 1718323200000 + assert a.pdf_url == "http://static.cninfo.com.cn/x.PDF" # frozen with pytest.raises(Exception): a.title = "mutated" # type: ignore[misc] @@ -49,6 +64,29 @@ def test_cninfo_error_is_exception() -> None: assert issubclass(CninfoError, TdxError) +def test_build_detail_url_has_four_params() -> None: + """回归 Bug2：详情页 URL 必须含 4 参数 stockCode/announcementId/orgId/announcementTime。""" + from easy_tdx.cninfo.models import build_detail_url + + url = build_detail_url("601088", "1225351323", "9900003701", 1780588800000) + assert "stockCode=601088" in url + assert "announcementId=1225351323" in url + assert "orgId=9900003701" in url + assert "announcementTime=1780588800000" in url + + +def test_build_pdf_url() -> None: + """adjunctUrl 应拼成 static.cninfo.com.cn 直链。""" + from easy_tdx.cninfo.models import build_pdf_url + + assert ( + build_pdf_url("finalpage/2026-06-05/1225351400.PDF") + == "http://static.cninfo.com.cn/finalpage/2026-06-05/1225351400.PDF" + ) + assert build_pdf_url("") == "" + assert build_pdf_url(None) == "" # type: ignore[arg-type] + + # --------------------------------------------------------------------------- # 日期转换 # --------------------------------------------------------------------------- @@ -58,7 +96,6 @@ 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 @@ -207,22 +244,34 @@ _QUERY_RESPONSE: dict[str, Any] = { "announcementTypeName": "股东大会", "announcementTime": 1749859200000, "announcementId": "abc123", + "adjunctUrl": "finalpage/2026-06-14/abc123.PDF", + "adjunctType": "PDF", }, { "announcementTitle": "2024年年度报告", - "announcementTypeName": "定期报告", + "announcementTypeName": None, # Bug1 场景：typeName 为 null "announcementTime": 1740614400000, "announcementId": "def456", + "adjunctUrl": "finalpage/2026-02-27/def456.PDF", + "adjunctType": "PDF", + }, + { + "announcementTitle": "无附件公告", + "announcementTypeName": None, + "announcementTime": 1740614400000, + "announcementId": "ghi789", + "adjunctUrl": "", # 无 PDF 附件 + "adjunctType": None, }, ], - "totalAnnouncement": 2, + "totalAnnouncement": 3, } def test_get_announcements_returns_dataframe( monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any ) -> None: - """应返回 DataFrame[title, type, date, url]。""" + """应返回 DataFrame，含全部新字段。""" _patch_stock_map(monkeypatch, {"688017": "9900041602"}) monkeypatch.setattr( "easy_tdx.cninfo.client._http_post_form", @@ -232,13 +281,75 @@ def test_get_announcements_returns_dataframe( 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 + expected_cols = [ + "title", + "type", + "date", + "url", + "code", + "org_id", + "announcement_id", + "announcement_time", + "pdf_url", + ] + assert list(df.columns) == expected_cols + assert len(df) == 3 + # 第一行：正常 typeName 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"] + assert len(df.iloc[0]["date"]) == 10 + assert df.iloc[0]["pdf_url"].endswith("abc123.PDF") + + +def test_get_announcements_type_fallback_to_adjunct_type( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """回归 Bug1：announcementTypeName 为 null 时回退 adjunctType。""" + _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") + # 第二行 typeName=null 但 adjunctType=PDF → type 应回退为 "PDF" + assert df.iloc[1]["type"] == "PDF" + # 第三行 typeName=null 且 adjunctType=null → type 为空字符串（非 nan） + assert df.iloc[2]["type"] == "" + + +def test_get_announcements_url_has_four_params( + monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any +) -> None: + """回归 Bug2：URL 必须含 4 参数才能打开（否则 404）。""" + _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") + url = df.iloc[0]["url"] + assert "stockCode=688017" in url + assert "announcementId=abc123" in url + assert "orgId=9900041602" in url + assert "announcementTime=" in url + + +def test_get_announcements_pdf_url(monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any) -> None: + """pdf_url 应为 static.cninfo.com.cn 直链，无附件时为空。""" + _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") + assert df.iloc[0]["pdf_url"] == "http://static.cninfo.com.cn/finalpage/2026-06-14/abc123.PDF" + assert df.iloc[2]["pdf_url"] == "" # 无附件 def test_get_announcements_empty(monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any) -> None: @@ -253,7 +364,7 @@ def test_get_announcements_empty(monkeypatch: pytest.MonkeyPatch, reset_orgid_ca df = CninfoClient().get_announcements("688017") assert isinstance(df, pd.DataFrame) assert df.empty - assert list(df.columns) == ["title", "type", "date", "url"] + assert len(df.columns) == 9 def test_get_announcements_missing_key( @@ -340,6 +451,7 @@ def test_get_announcements_skips_non_dict_items( "announcementTitle": "ok", "announcementTime": 1749859200000, "announcementId": "x", + "adjunctUrl": "", }, ] }, @@ -389,6 +501,168 @@ def test_get_announcements_malformed_timestamp_wrapped_as_cninfo_error( CninfoClient().get_announcements("688017") +# --------------------------------------------------------------------------- +# PDF 下载 +# --------------------------------------------------------------------------- + + +def _make_anno(**overrides: Any) -> Any: + """构造测试用 Announcement（默认有 pdf_url）。""" + from easy_tdx.cninfo import Announcement + + defaults: dict[str, Any] = { + "title": "测试公告", + "type": "PDF", + "date": "2026-06-14", + "url": "http://x", + "code": "688017", + "org_id": "9900041602", + "announcement_id": "abc123", + "announcement_time": 1718323200000, + "pdf_url": "http://static.cninfo.com.cn/x.PDF", + } + defaults.update(overrides) + return Announcement(**defaults) + + +def test_download_pdf_success(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """download_pdf 应写入文件并返回绝对路径。""" + from easy_tdx.cninfo import CninfoClient + + pdf_bytes = b"%PDF-1.4\nfake pdf content" + + 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: + return _FakeResp(pdf_bytes) + + import easy_tdx.cninfo.client as mod + + monkeypatch.setattr(mod.urlrequest, "urlopen", _fake_urlopen) + anno = _make_anno() + path = CninfoClient().download_pdf(anno, dest_dir=tmp_path) + assert Path(path).exists() + assert Path(path).read_bytes() == pdf_bytes + # 默认文件名格式：{date}_{announcement_id}.PDF + assert "abc123" in Path(path).name + assert Path(path).name.endswith(".PDF") + + +def test_download_pdf_custom_filename(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """自定义 filename 应被使用。""" + from easy_tdx.cninfo import CninfoClient + + class _FakeResp: + def read(self) -> bytes: + return b"%PDF-1.4" + + def __enter__(self) -> _FakeResp: + return self + + def __exit__(self, *args: Any) -> None: + pass + + import easy_tdx.cninfo.client as mod + + monkeypatch.setattr(mod.urlrequest, "urlopen", lambda req, timeout=15.0: _FakeResp()) + path = CninfoClient().download_pdf(_make_anno(), dest_dir=tmp_path, filename="custom.pdf") + assert Path(path).name == "custom.pdf" + + +def test_download_pdf_no_attachment_raises() -> None: + """pdf_url 为空应抛 CninfoError，不触网。""" + from easy_tdx.cninfo import CninfoClient, CninfoError + + anno = _make_anno(pdf_url="") + with pytest.raises(CninfoError): + CninfoClient().download_pdf(anno) + + +def test_download_pdf_creates_dest_dir(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """目标目录不存在应自动创建。""" + from easy_tdx.cninfo import CninfoClient + + class _FakeResp: + def read(self) -> bytes: + return b"%PDF-1.4" + + def __enter__(self) -> _FakeResp: + return self + + def __exit__(self, *args: Any) -> None: + pass + + import easy_tdx.cninfo.client as mod + + monkeypatch.setattr(mod.urlrequest, "urlopen", lambda req, timeout=15.0: _FakeResp()) + nested = tmp_path / "a" / "b" / "c" + path = CninfoClient().download_pdf(_make_anno(), dest_dir=nested) + assert Path(path).exists() + assert nested.is_dir() + + +def test_download_pdf_accepts_series(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """download_pdf 应兼容 pd.Series（DataFrame.iloc[i] 的返回类型）。""" + from easy_tdx.cninfo import CninfoClient + + class _FakeResp: + def read(self) -> bytes: + return b"%PDF-1.4" + + def __enter__(self) -> _FakeResp: + return self + + def __exit__(self, *args: Any) -> None: + pass + + import easy_tdx.cninfo.client as mod + + monkeypatch.setattr(mod.urlrequest, "urlopen", lambda req, timeout=15.0: _FakeResp()) + # 模拟 DataFrame 的一行 + row = pd.Series( + { + "title": "t", + "type": "PDF", + "date": "2026-06-14", + "url": "http://x", + "code": "688017", + "org_id": "9900041602", + "announcement_id": "abc", + "announcement_time": 1718323200000, + "pdf_url": "http://static.cninfo.com.cn/x.PDF", + } + ) + path = CninfoClient().download_pdf(row, dest_dir=tmp_path) + assert Path(path).exists() + + +def test_download_pdf_network_failure_wrapped( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """下载网络失败应转 CninfoError。""" + from easy_tdx.cninfo import CninfoClient, CninfoError + + def _boom(req: Any, timeout: float = 15.0) -> Any: + raise OSError("connection reset") + + import easy_tdx.cninfo.client as mod + + monkeypatch.setattr(mod.urlrequest, "urlopen", _boom) + with pytest.raises(CninfoError): + CninfoClient().download_pdf(_make_anno(), dest_dir=tmp_path) + + # --------------------------------------------------------------------------- # urllib helper 烟雾测试（不触网，仅验证 JSON 解码路径） # ---------------------------------------------------------------------------