diff --git a/bot/commands/analyze.py b/bot/commands/analyze.py index 2ad7cec0c..982935e3c 100644 --- a/bot/commands/analyze.py +++ b/bot/commands/analyze.py @@ -4,16 +4,24 @@ 股票分析命令 =================================== -分析指定股票，调用 AI 生成分析报告。 +分析指定股票或已登记指数，调用 AI 生成分析报告。 """ -import re import logging -from typing import List, Optional +import re +import unicodedata +from dataclasses import replace +from typing import List, Optional, Tuple, Union from bot.commands.base import BotCommand from bot.models import BotMessage, BotResponse -from src.services.stock_code_utils import resolve_index_stock_code_for_analysis +from src.services.stock_code_utils import is_code_like, resolve_index_stock_code_for_analysis +from src.services.stock_list_parser import ( + AnalysisTarget, + ParseStatus, + default_index_registry, + parse_analysis_target, +) logger = logging.getLogger(__name__) @@ -22,11 +30,14 @@ class AnalyzeCommand(BotCommand): """ 股票分析命令 - 分析指定股票代码，生成 AI 分析报告并推送。 + 分析指定股票代码或已登记指数，生成 AI 分析报告并推送。 用法： /analyze 600519 - 分析贵州茅台（精简报告） /analyze 600519 full - 分析并生成完整报告 + /analyze sh000016 - 分析上证50指数 + /analyze 930955.CSI - 分析红利低波100指数（alias 收敛） + /analyze 上证50 - 按注册名称分析上证50指数 """ @property @@ -39,69 +50,195 @@ class AnalyzeCommand(BotCommand): @property def description(self) -> str: - return "分析指定股票" + return "分析指定股票或指数" @property def usage(self) -> str: - return "/analyze <股票代码> [full]" + return "/analyze <股票代码/指数代码/指数名称> [full]" def validate_args(self, args: List[str]) -> Optional[str]: - """验证参数""" + """验证参数（仅结构检查；语义校验在 execute 中完成）""" if not args: - return "请输入股票代码" - - code = args[0].upper() - - # 验证股票代码格式 - # A股：6位数字 - # 港股：HK+5位数字 - # 美股：1-5个大写字母+.+2个后缀字母 - is_a_stock = re.match(r'^\d{6}$', code) - is_hk_stock = re.match(r'^HK\d{5}$', code) - is_us_stock = re.match(r'^[A-Z]{1,5}(\.[A-Z]{1,2})?$', code) - - if not (is_a_stock or is_hk_stock or is_us_stock): - return f"无效的股票代码: {code}（A股6位数字 / 港股HK+5位数字 / 美股1-5个字母）" - + return "请输入股票代码或指数名称" return None def execute(self, message: BotMessage, args: List[str]) -> BotResponse: """执行分析命令""" - code = resolve_index_stock_code_for_analysis(args[0]) - + raw = (args[0] or "").strip() + if not raw: + return BotResponse.error_response("请输入股票代码或指数名称") + # 检查是否需要完整报告（默认精简，传 full/完整/详细 切换） report_type = "simple" if len(args) > 1 and args[1].lower() in ["full", "完整", "详细"]: report_type = "full" - logger.info(f"[AnalyzeCommand] 分析股票: {code}, 报告类型: {report_type}") - + try: + code, analysis_target = self._resolve_analysis_input(raw) + if code is None: + error_msg = analysis_target + if not isinstance(error_msg, str): + error_msg = "无法识别标的" + return BotResponse.error_response(error_msg) + if isinstance(analysis_target, str): + return BotResponse.error_response(analysis_target) + + logger.info(f"[AnalyzeCommand] 分析标的: {code}, 报告类型: {report_type}") + # 调用分析服务 from src.services.task_service import get_task_service from src.enums import ReportType - + service = get_task_service() - + # 提交异步分析任务 result = service.submit_analysis( code=code, report_type=ReportType.from_str(report_type), - source_message=message + source_message=message, + analysis_target=analysis_target, ) - + if result.get("success"): task_id = result.get("task_id", "") - return BotResponse.markdown_response( + # ``extra`` 仅供内部任务 identity 透传（transport-independent + # 消费者，如 E2E smoke），文本保持既有契约不变，平台适配器可忽略。 + response = BotResponse.markdown_response( f"✅ **分析任务已提交**\n\n" - f"• 股票代码: `{code}`\n" + f"• 标的: `{code}`\n" f"• 报告类型: {ReportType.from_str(report_type).display_name}\n" f"• 任务 ID: `{task_id[:20]}...`\n\n" f"分析完成后将自动推送结果。" ) - else: - error = result.get("error", "未知错误") - return BotResponse.error_response(f"提交分析任务失败: {error}") - + response.extra = { + "task_id": task_id, + "stock_code": code, + } + return response + error = result.get("error", "未知错误") + return BotResponse.error_response(f"提交分析任务失败: {error}") + except Exception as e: logger.error(f"[AnalyzeCommand] 执行失败: {e}") return BotResponse.error_response(f"分析失败: {str(e)[:100]}") + + def _resolve_analysis_input( + self, raw: str + ) -> Tuple[Optional[str], Union[AnalysisTarget, str, None]]: + """Resolve one ``/analyze`` argument into ``(code, analysis_target)``. + + Registered index first, existing stock-name resolution as fallback. + + 1. Explicit index identity/alias (``sh000016`` / ``930955.CSI`` / + ``csi930955``) via + :meth:`IndexRegistry.find_by_explicit_key`. + 2. Exact registered display name (``上证50``) via + :meth:`IndexRegistry.find_by_display_name` — before any stock-name + fallback; ambiguous names fail with an explicit error. + 3. Parser INDEX acceptance: when :func:`parse_analysis_target` + classifies the raw input as INDEX (e.g. dotted-prefix aliases such + as ``SH.000016`` the registry key lookup misses), submit the + registry canonical with the matching structured target. Parser + STOCK results are deliberately ignored here so the legacy stock + gate stays authoritative for stock shapes. + 4. Explicit CSI forms (``csi`` + one or more digits, or one or more + digits + ``.csi``) that no registry entry claimed — they surface + the parser's ``unsupported`` details (never a stock-name fallback + or a US-ticker guess). + 5. Legacy stock-code gate, preserved exactly as before this change + (case-insensitive): A-share six digits, ``HK`` + five digits, and + US 1-5 letters with optional ``.XX`` suffix. Matches keep the + legacy code path (no structured target) so lowercase real tickers + like ``usfd`` still resolve to ``USFD``. + 6. Any remaining parser-UNSUPPORTED input (e.g. ``us1``, + ``600519.BJ``, ``1234567.SH``) is an explicit error — never sent + into stock-name resolution. + 7. Anything code-like the legacy gate rejected (``12345``, bare + ``00700``, ``600519.SH``, unregistered ``sh999999``) is an + explicit error — never silently submitted. + 8. Non-code names through :func:`resolve_name_to_code` (stock only). + + Only INDEX targets are carried downstream; stock inputs keep the + legacy code path so ``600519`` is never rewritten into a parser + canonical. On failure returns ``(None, error_message)``. + """ + from src.services.name_to_code_resolver import resolve_name_to_code + + registry = default_index_registry() + + # 1) Explicit index identity/alias — registered index wins over any + # stock-name fallback and is submitted with its lowercase + # canonical_id verbatim. + if registry.find_by_explicit_key(raw) is not None: + target = parse_analysis_target(raw, registry=registry) + return target.canonical_id, target + + # 2) Exact registered display name — independent of identity aliases; + # a Chinese name must never become a parser identity alias. + if registry.is_ambiguous_display_name(raw): + return None, ( + f"指数名称 `{raw}` 存在歧义，请改用显式代码（如 `sh000016`）" + ) + entry = registry.find_by_display_name(raw) + if entry is not None: + target = replace( + parse_analysis_target(entry.canonical_id, registry=registry), + raw_input=raw, + ) + return target.canonical_id, target + + # 3) Parser INDEX acceptance — the shared parser recognizes forms the + # registry key lookup above misses (e.g. dotted-prefix aliases + # ``SH.000016`` / ``SZ.399001``). Only an INDEX result short-circuits + # here; STOCK and UNSUPPORTED results continue to the legacy gates + # below so the legacy stock contract stays authoritative. + target = parse_analysis_target(raw, registry=registry) + if target.asset_type == ParseStatus.INDEX: + return target.canonical_id, target + + # 4) Explicit CSI forms — registered ones were claimed in step 1, so + # anything reaching here is unregistered and must surface the + # parser's UNSUPPORTED details. The full numeric explicit CSI + # family (``csi`` + one or more digits, or one or more digits + + # ``.csi``) is covered, not only six-digit forms. + normalized = unicodedata.normalize("NFKC", raw).strip().casefold() + if re.fullmatch(r"(?:csi\d+|\d+\.csi)", normalized): + reason = target.unsupported_reason or f"无法识别标的: {raw}" + return None, f"无法分析 `{raw}`：{reason}" + + # 5) Legacy stock-code gate — case-insensitive, exactly the shapes the + # old ``validate_args`` accepted. Matches keep the legacy code + # path (no structured target), including lowercase real tickers + # like ``usfd`` -> ``USFD``. + upper = raw.upper() + if ( + re.fullmatch(r"\d{6}", upper) + or re.fullmatch(r"HK\d{5}", upper) + or re.fullmatch(r"[A-Z]{1,5}(?:\.[A-Z]{1,2})?", upper) + ): + return resolve_index_stock_code_for_analysis(raw), None + + # 6) Any remaining parser-UNSUPPORTED input is an explicit error — + # never sent into stock-name resolution. This covers malformed + # code shapes the parser rejected (``us1``, ``600519.BJ``, + # ``1234567.SH``) that the legacy gate above did not claim. + if target.asset_type == ParseStatus.UNSUPPORTED: + reason = target.unsupported_reason or f"无法识别标的: {raw}" + return None, f"无法分析 `{raw}`：{reason}" + + # 7) Code-like inputs the legacy gate rejected are explicit errors — + # never submitted, never routed into stock-name resolution. This + # includes ``sh``/``sz`` prefixed six-digit forms (``sh999999``) + # that ``is_code_like`` misses because the bare digits classify as + # a different exchange. + if is_code_like(raw) or re.fullmatch(r"(?:sh|sz)\d{6}", normalized): + return None, ( + f"无效的标的代码: `{raw}`" + f"（A股6位数字 / HK+5位数字 / 美股1-5个字母 / 已登记指数代码或名称）" + ) + + # 8) Name input → stock-name fallback. + code = resolve_name_to_code(raw) + if not code: + return None, f"无法识别标的: {raw}" + return code, None diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 4b1183d19..e8fb2ceaa 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -8,6 +8,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/). > For user-friendly release highlights, see the [GitHub Releases](https://github.com/ZhuLinsen/daily_stock_analysis/releases) page. ## [Unreleased] +- [测试] 新增 Bot 指数入口 transport-independent 在线 E2E smoke（`scripts/smoke_bot_index_entry.py`）：worker 子进程经真实 `CommandDispatcher.dispatch_async` 提交并在同进程轮询 `TaskService` 贯穿到 `StockAnalysisPipeline`，父进程只负责 deadline、进程树清理（Windows `taskkill /T /F`、POSIX 杀进程组）与退出码（0=成功/1=失败/124=超时），输出单行 `E2E_EVENT {json}` 事件（`phase=submitted|completed|failed|timeout`）；smoke 覆盖 `SH.000016`/`上证50`/`930955.CSI` 矩阵，期望 code/name 取脚本内置权威映射（不信任响应/结果自报身份，提交 code mismatch 输出含期望值与实际值的显式错误并携带结构化实际 `stock_code`，dispatcher 路由错误也失败），矩阵外 target 与非正 `--timeout` 在提交与 spawn 之前即被拒绝（退出码 2），completed 须 exact canonical code、exact 注册名称且 `analysis_summary`/`operation_advice`/`trend_prediction` 非空（仅空白视为空），失败或结果不完整即非零退出；超时清理进程树（Windows `taskkill /T /F`、POSIX 杀进程组），清理成功输出 `timeout` 事件并退出 124、清理失败输出含清理错误的 `failed` 事件并退出 1，不回滚 DB/报告/通知副作用；用户 Ctrl-C 中止时父进程同样先清理进程树，清理成功透传中断、清理失败输出含清理错误的 `failed` 事件并退出 1，绝不静默吞掉清理失败；worker 意外异常输出 `failed` 事件并退出 1（stderr 保留异常证据，`KeyboardInterrupt` 不按普通失败处理），父进程将 worker 任意其他退出码归一化为 1（运行时契约只暴露 0/1/124）；父进程以内部 `--worker` flag 显式拉起子进程（不依赖环境变量，防外部预置绕过硬超时）；不 mock 在线依赖、不 dry-run、不修 transport，不加入离线 gate。 +- [新功能] Bot `/analyze` 支持已登记指数入口：显式代码（`sh000016`）、CSI alias（`930955.CSI` 收敛为 `csi930955`）与注册中文名（`上证50`）均可提交，指数以结构化 `AnalysisTarget` 经 `TaskService` 贯穿到 Pipeline `process_single_stock`（`sh000016` 不再被改写为 `SH000016`）；注册名称查询独立于 parser identity alias（中文名不进入 `find_by_explicit_key`/`parse_analysis_target`），同名歧义返回明确错误并要求显式代码，未登记 CSI 与未知名称返回明确错误且不提交任务；个股代码（A/HK/US）保持既有 legacy code 路径不变，股票名称输入（如 `贵州茅台`）由本次 Bot 入口新暴露——复用既有名称解析器（`resolve_name_to_code`）解析后提交 legacy code，不携带结构化 target；提交成功响应在 `BotResponse.extra` 暴露内部任务 identity（`task_id`/`stock_code`）供在线验收等内部流程使用，文本与错误路径不变、平台适配器可忽略 `extra`。 - [新功能] 新增最小 Agent 轨迹评估入口 `evals/agent_trajectory/`(Refs #1956):纯函数指标层只消费真实 `tool_calls_log + AgentResult`,冻结最小指标契约(工具命中、冗余/缓存、失败/重试、总步数/max_steps),`run_eval.py` 经 `build_agent_executor` 真实执行并输出文本摘要 + 结构化 JSON 报告;评估为 reporter 非 gate,零 `src/` 改动 - [修复] 将 litellm 依赖窗口上界收敛到 `<1.99.0`：1.99.0 起把 `prompt_cache_key` 透传给 OpenAI provider，破坏 provider 缓存测试对不透传行为的既有断言（CI backend-tests 3/3 与 backend-gate 失败）；保留历史最低版本与 `!=1.82.7`/`!=1.82.8` 事故排除，同时同步更新各 LLM 兼容文档中写死的依赖约束表述，避免文档与 requirements.txt 漂移 diff --git a/docs/bot-command.md b/docs/bot-command.md index d8558aa75..5a12bfd60 100644 --- a/docs/bot-command.md +++ b/docs/bot-command.md @@ -52,7 +52,7 @@ bot/ ├── commands/ # 命令处理器 │ ├── __init__.py │ ├── base.py # 命令抽象基类 -│ ├── analyze.py # /analyze 股票分析 +│ ├── analyze.py # /analyze 股票/指数分析 │ ├── market.py # /market 大盘复盘 │ ├── help.py # /help 帮助信息 │ └── status.py # /status 系统状态 @@ -185,7 +185,7 @@ class CommandDispatcher: |------|------|------|------| -| /analyze | /a, 分析 | 分析指定股票 | `/analyze 600519` | +| /analyze | /a, 分析 | 分析指定股票或已登记指数 | `/analyze 600519`、`/analyze sh000016`、`/analyze 上证50` | | /market | /m, 大盘 | 大盘复盘 | `/market` | @@ -195,6 +195,31 @@ class CommandDispatcher: | /status | /s, 状态 | 系统状态 | `/status` | +### `/analyze` 支持边界 + +- 股票：A 股 6 位数字（`600519`）、港股 `HK+5 位数字`（`hk00700`）、美股 1-5 个字母（`AAPL`、`usfd`→`USFD`）。股票名称（`贵州茅台`）由本次 Bot 入口新暴露：复用既有名称解析器（`resolve_name_to_code`）解析后提交 legacy code，不携带结构化 target。 +- 已登记指数：显式代码（`sh000016`）、CSI alias（`930955.CSI` 收敛为 `csi930955`）、注册中文名（`上证50`）均可提交；指数以结构化 `AnalysisTarget` 贯穿到分析 Pipeline，`sh000016` 不会被改写为 `SH000016`。 +- 未登记 CSI（如 `930956.CSI`）、旧闸门拒绝的代码形态（如 `12345`、裸 `00700`、`600519.SH`、未登记 `sh999999`）或无法识别的名称：返回明确错误，不提交任务。 +- 注册表内存在等价名称歧义时：要求改用显式代码，不猜测、不进入股票名称兜底。 + +### Bot 指数入口在线 E2E smoke（`scripts/smoke_bot_index_entry.py`） + +用于从真实 `CommandDispatcher.dispatch_async -> AnalyzeCommand -> TaskService -> StockAnalysisPipeline` 在线贯穿单标的分析，覆盖 `SH.000016` / `上证50` / `930955.CSI` 三条矩阵场景，不经过任何 webhook transport（不 mock 在线依赖、不 dry-run）。 + +运行（单 target 参数，`--timeout` 默认 900 秒）： + +```powershell +.venv\Scripts\python.exe scripts/smoke_bot_index_entry.py "SH.000016" +.venv\Scripts\python.exe scripts/smoke_bot_index_entry.py "上证50" +.venv\Scripts\python.exe scripts/smoke_bot_index_entry.py "930955.CSI" +``` + +进程职责：worker 独立子进程提交并轮询（同进程 `TaskService`），输出单行 `E2E_EVENT {json}` 事件（`phase=submitted|completed|failed|timeout`，含 `target`，按阶段附加 `task_id`/`stock_code`/`result`/`error`）；父进程只负责 deadline、进程树清理和退出码（0=成功 / 1=失败 / 124=超时）。 + +失败语义：任务失败、completed 但结果不完整（canonical code 或注册名称与矩阵期望不符，或 `analysis_summary`/`operation_advice`/`trend_prediction` 任一为空或仅空白）即输出 `failed` 事件并不为零退出；超时清理进程树（Windows `taskkill /T /F`，POSIX 杀进程组），清理成功输出 `timeout` 事件并退出 124，清理失败则输出含清理错误的 `failed` 事件并退出 1，不回滚 DB / 报告 / 通知等既有副作用。用户 Ctrl-C 中止时父进程同样先清理进程树：清理成功透传中断，清理失败输出含清理错误的 `failed` 事件并退出 1，绝不静默吞掉清理失败。worker 的期望 code/name 来自脚本内置的矩阵权威映射（`SH.000016`/`上证50` → `sh000016`/`上证50`，`930955.CSI` → `csi930955`/`红利低波100`），不信任响应或任务结果自报身份：提交响应的 `stock_code` 与期望不符时输出含期望值与实际值的显式 mismatch 错误（failed 事件同时携带结构化 `stock_code` 实际值）；矩阵之外的 target 与非正 `--timeout` 在提交与子进程 spawn 之前即被拒绝（参数/输入错误，退出码 2）。worker 意外异常（dispatch/轮询/事件序列化）输出 `failed` 事件并退出 1（异常证据保留在 stderr），`KeyboardInterrupt` 不按普通失败处理；父进程将 worker 的任意其他退出码归一化为 1，运行时契约只暴露 0 / 1 / 124。父进程以内部 `--worker` flag 显式拉起子进程，不依赖环境变量。 + +前置条件：需要网络与数据源/AI 凭据配置齐全（在线链路）。本脚本不属于离线 gate，不进入 CI。 + ## 五、`/status` 与模型配置诊断说明 ### 可配置层级与可用性判断依据 diff --git a/docs/bot-command_EN.md b/docs/bot-command_EN.md index e4b80fe45..6cb4216f5 100644 --- a/docs/bot-command_EN.md +++ b/docs/bot-command_EN.md @@ -57,7 +57,7 @@ bot/ ├── commands/ # Command handlers │ ├── __init__.py │ ├── base.py # Abstract base class for commands -│ ├── analyze.py # /analyze — stock analysis +│ ├── analyze.py # /analyze — stock/index analysis │ ├── ask.py # /ask — single-turn question │ ├── batch.py # /batch — batch watchlist analysis │ ├── chat.py # /chat — multi-turn strategy chat @@ -152,7 +152,7 @@ class BotCommand(ABC): | Command | Description | Example | |---------|-------------|---------| -| `/analyze` | Analyze a specific stock | `/analyze AAPL` or `/analyze 600519` | +| `/analyze` | Analyze a specific stock or a registered index | `/analyze AAPL`, `/analyze 600519`, `/analyze sh000016`, `/analyze 上证50` | | `/ask` | Single-turn question about a stock or the market | `/ask what is RSI for AAPL` | | `/batch` | Batch-analyze your configured watchlist | `/batch` | | `/chat` | Multi-turn strategy chat (maintains conversation context) | `/chat` | @@ -162,6 +162,26 @@ class BotCommand(ABC): > **Stock code formats:** A-shares use 6-digit codes (e.g. `600519`); HK stocks prefix `hk` (e.g. `hk00700`); US stocks use ticker symbols (e.g. `AAPL`, `TSLA`). +> **Registered index inputs:** explicit codes (`sh000016`), CSI aliases (`930955.CSI` converges to `csi930955`) and registered Chinese names (`上证50`) are all accepted. Index inputs are submitted as a structured `AnalysisTarget` that flows through to the analysis pipeline, so `sh000016` is never rewritten to `SH000016`. Unregistered CSI forms (e.g. `930956.CSI`), code shapes the legacy gate rejected (`12345`, bare `00700`, `600519.SH`, unregistered `sh999999`) and unrecognized names return an explicit error without submitting a task; ambiguous registered names require an explicit code instead of guessing. Stock inputs (A-share 6-digit, `HK`+5-digit, US 1-5 letters, e.g. `usfd`→`USFD`) keep the legacy code path. Stock-name inputs (e.g. `贵州茅台`) are newly exposed by this Bot entry: they reuse the existing name resolver (`resolve_name_to_code`) and then submit the legacy code without a structured target. + +### Bot index entry online E2E smoke (`scripts/smoke_bot_index_entry.py`) + +Walks the real `CommandDispatcher.dispatch_async -> AnalyzeCommand -> TaskService -> StockAnalysisPipeline` online for a single target, covering the `SH.000016` / `上证50` / `930955.CSI` matrix scenarios without any webhook transport (no mocked online dependencies, no dry-run). + +Run with a single target argument (`--timeout` defaults to 900 seconds): + +```powershell +.venv\Scripts\python.exe scripts/smoke_bot_index_entry.py "SH.000016" +.venv\Scripts\python.exe scripts/smoke_bot_index_entry.py "上证50" +.venv\Scripts\python.exe scripts/smoke_bot_index_entry.py "930955.CSI" +``` + +Process responsibilities: a worker subprocess submits and polls the in-process `TaskService` and prints single-line `E2E_EVENT {json}` events (`phase=submitted|completed|failed|timeout`, with `target`, plus `task_id`/`stock_code`/`result`/`error` per phase); the parent only enforces the deadline, cleans up the process tree and maps exit codes (0=success / 1=failure / 124=timeout). + +Failure semantics: a failed task or a completed-but-incomplete result (canonical code or registered name not matching the matrix expectation, or any of `analysis_summary`/`operation_advice`/`trend_prediction` empty or whitespace-only) emits a `failed` event and exits non-zero; on timeout the process tree is killed (Windows `taskkill /T /F`, POSIX process group) — a successful cleanup emits a `timeout` event and exits 124, a failed cleanup emits a `failed` event carrying the cleanup error and exits 1 — never rolling back DB / reports / notifications side effects. On a user Ctrl-C the parent also cleans up the process tree first: a successful cleanup propagates the interrupt, a failed cleanup emits a `failed` event carrying the cleanup error and exits 1 — a cleanup failure is never silently swallowed. The worker's expected code/name come from the script's built-in authoritative matrix map (`SH.000016`/`上证50` → `sh000016`/`上证50`, `930955.CSI` → `csi930955`/`红利低波100`) — never from the response or the result's self-reported identity: a submission whose `extra.stock_code` does not match the expectation fails with an explicit mismatch error carrying the expected and actual values (the failed event also carries the structured actual `stock_code`), and targets outside the matrix as well as non-positive `--timeout` are rejected before any submission or subprocess spawn (argument/input error, exit code 2). Unexpected worker exceptions (dispatch / poll / event serialization) emit a `failed` event and exit 1 with the exception evidence kept on stderr; `KeyboardInterrupt` is not treated as an ordinary failure; the parent normalizes any other worker exit code to 1 so the runtime contract only exposes 0 / 1 / 124. The parent spawns the worker with an internal `--worker` flag, not via environment variables. + +Prerequisites: network plus configured data-source and AI credentials (online chain). This script is not part of the offline gate and does not run in CI. + --- ## 5. `/status` and LLM configuration diagnostics diff --git a/docs/bot/discord-bot-config.md b/docs/bot/discord-bot-config.md index e432c2d7f..d35317744 100644 --- a/docs/bot/discord-bot-config.md +++ b/docs/bot/discord-bot-config.md @@ -73,19 +73,21 @@ DISCORD_BOT_STATUS=A股智能分析 | /help Discord机器人支持以下Slash命令： -1. `/analyze [full_report]` - 分析指定股票代码 - - `stock_code`: 股票代码，如 600519 +1. `/analyze [full_report]` - 分析指定股票、已登记指数或股票名称 + - `target`: 分析标的，可为股票代码（如 `600519`）、已登记指数（如 `sh000016`、`930955.CSI`、`上证50`）或股票名称（如 `贵州茅台`） - `full_report`: 可选，是否生成完整报告（包含大盘） 2. `/market_review` - 获取大盘复盘报告 3. `/help` - 查看帮助信息 +> 边界：`/analyze` 支持已登记指数入口与股票名称输入（股票名称复用既有名称解析器并提交 legacy code）；`/ask` 与 `/batch` 暂不支持指数输入。 + ## 测试机器人 1. 确保机器人已成功添加到你的服务器 2. 在频道中输入`/help`，机器人会返回帮助信息 -3. 输入`/analyze 600519`测试股票分析功能 +3. 输入`/analyze 600519`测试股票分析功能（已登记指数如 `/analyze sh000016`、股票名称如 `/analyze 贵州茅台` 同样可用） 4. 输入`/market_review`测试大盘复盘功能 ## 注意事项 diff --git a/docs/full-guide.md b/docs/full-guide.md index e24903a92..62bbd1c8c 100644 --- a/docs/full-guide.md +++ b/docs/full-guide.md @@ -744,7 +744,9 @@ Web 自动补全与搜索已放行已登记指数：搜索注册中文名（如 API `/analyze` 对显式指数输入构造结构化 `AnalysisTarget`：`sh000016` 以 `asset_type=INDEX` 且 `canonical_id=sh000016` 入队，`930955.CSI`/`csi930955` 收敛为 `csi930955`。指数与同码个股（如 `sh000016` 与 `000016`）独立去重调度、互不折叠；未登记的 CSI 输入（如 `930956.CSI`）在异步单股或同步模式返回明确的 4xx，在异步批量中仅该目标进入响应 `rejected` 列表、同批其他目标正常入队。中文名称输入（如 `贵州茅台`）仍走既有股票名解析，不进入指数判型。 -> **Phase 2 边界**：默认 `STOCK_LIST`、`--schedule`、Bot 与 GitHub Actions 每日工作流暂不开放指数入口；Web/API 与一次性 `--stocks` 已支持指数，Bot/定时/每日工作流入口留待 Phase 2 后续 PR。 +Bot `/analyze` 已支持已登记指数的显式代码（`sh000016`）、CSI alias（`930955.CSI`）和注册中文名（`上证50`）。指数以 registry canonical 和结构化 `AnalysisTarget` 进入与 CLI/API 相同的 Pipeline；未登记 CSI、未知名称或歧义注册名称会明确报错且不提交任务。普通 A/HK/US 代码与股票名称继续沿用 legacy code 路径。 + +> **Phase 2 边界**：默认 `STOCK_LIST`、`--schedule`、Bot `/ask`、Bot `/batch` 与 GitHub Actions 每日工作流仍未开放指数入口；Web/API、Bot `/analyze` 与一次性 `--stocks` 已支持指数，其余定时/每日工作流入口留待后续 PR。 ### 指数与个股 Dashboard canonical 隔离（PR #2312） diff --git a/docs/full-guide_EN.md b/docs/full-guide_EN.md index 47253fc76..625929751 100644 --- a/docs/full-guide_EN.md +++ b/docs/full-guide_EN.md @@ -683,7 +683,9 @@ Web autocomplete and search now expose registered indices: searching a registry The API `/analyze` endpoint builds a structured `AnalysisTarget` for explicit index inputs: `sh000016` is enqueued as `asset_type=INDEX` with `canonical_id=sh000016`, and `930955.CSI`/`csi930955` converge to `csi930955`. Indices and same-digit stocks (e.g. `sh000016` vs `000016`) are deduplicated independently and never collapse. An unregistered CSI input (e.g. `930956.CSI`) returns an explicit 4xx for a single async or sync request, and in an async batch only that target enters the response `rejected` list while the rest of the batch is enqueued normally. Chinese-name inputs (e.g. `贵州茅台`) keep the existing stock-name resolution path and never enter index classification. -> **Phase 2 boundary**: default `STOCK_LIST`, `--schedule`, Bot, and the GitHub Actions daily workflow do not yet expose index entrypoints; Web/API and the one-shot `--stocks` entry support indices, with Bot/scheduled/daily-workflow entries landing in later Phase 2 PRs. +Bot `/analyze` now accepts registered-index explicit codes (`sh000016`), CSI aliases (`930955.CSI`), and registered Chinese names (`上证50`). The registry canonical and structured `AnalysisTarget` flow into the same Pipeline used by CLI/API; unregistered CSI forms, unknown names, and ambiguous registered names return an explicit error without submitting a task. Existing A/HK/US codes and stock names keep the legacy-code path. + +> **Phase 2 boundary**: default `STOCK_LIST`, `--schedule`, Bot `/ask`, Bot `/batch`, and the GitHub Actions daily workflow still do not expose index entrypoints. Web/API, Bot `/analyze`, and the one-shot `--stocks` entry support indices; the remaining scheduled/daily-workflow entry lands in a later PR. ### Index vs stock Dashboard canonical isolation (PR #2312) diff --git a/scripts/smoke_bot_index_entry.py b/scripts/smoke_bot_index_entry.py new file mode 100644 index 000000000..95a3c78e3 --- /dev/null +++ b/scripts/smoke_bot_index_entry.py @@ -0,0 +1,506 @@ +# -*- coding: utf-8 -*- +""" +============================================= +Bot 指数入口 transport-independent 在线 E2E smoke +============================================= + +从真实 ``CommandDispatcher.dispatch_async -> AnalyzeCommand -> +TaskService -> StockAnalysisPipeline`` 在线贯穿单标的分析，覆盖 +SH_ALIAS / REGISTERED_NAME / CSI_ALIAS 三条矩阵场景，不经过任何 webhook +transport（不修 transport，不 mock 在线依赖，不 dry-run）。 + +运行方式（单 target 参数，--timeout 默认 900 秒）:: + + .venv\\Scripts\\python.exe scripts/smoke_bot_index_entry.py "SH.000016" + .venv\\Scripts\\python.exe scripts/smoke_bot_index_entry.py "上证50" + .venv\\Scripts\\python.exe scripts/smoke_bot_index_entry.py "930955.CSI" + .venv\\Scripts\\python.exe scripts/smoke_bot_index_entry.py "SH.000016" --timeout 1800 + +进程职责: +- 父进程: 只负责 deadline、进程树清理和退出码；worker 继承控制台。 +- worker: 独立子进程（由父进程以内部 ``--worker`` flag 显式拉起，不依赖 + 任何环境变量），经真实 dispatcher 提交并在同一进程内轮询 + ``TaskService``，输出单行 ``E2E_EVENT {json}`` 事件。 + +事件契约（单行 ``E2E_EVENT {json}``，phase=submitted|completed|failed|timeout， +含 target，按阶段附加 task_id / stock_code / result / error）: +- submitted : 任务已提交，附带 ``task_id`` / ``stock_code``。 +- completed : 任务完成且结构断言通过，附带 ``result``。 +- failed : 任务失败或 completed 但结果不完整，附带 ``error``。 +- timeout : worker 超过 deadline 被清理（父进程输出）。 + +退出码: 0 = 成功；1 = 失败（结构断言或 failed 事件）；124 = 超时。 + +超时清理: Windows 使用 ``taskkill /T /F`` 结束整个进程树；POSIX 向进程组 +发送 SIGKILL。超时只清理进程树，不回滚 DB / 报告 / 通知等既有副作用。 + +Ctrl-C 中止: 父进程在轮询窗口收到 Ctrl-C 时同样先清理进程树——清理成功 +透传中断，清理失败输出含清理错误的 ``failed`` 事件并退出 1，绝不静默 +吞掉清理失败。 +""" + +from __future__ import annotations + +import argparse +import json +import os +import signal +import subprocess +import sys +import time +from pathlib import Path +from typing import Optional + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from scripts.check_env import _reconfigure_output_stream # noqa: E402 + +EVENT_PREFIX = "E2E_EVENT " + +# 矩阵输入 -> (canonical code, registered display name) 的权威映射。 +# worker 的期望 code/name 只从这里取值，绝不信任响应或任务结果自报的身份， +# 防止 dispatcher 路由错误时仍假通过。 +EXPECTED_OUTCOMES = { + "SH.000016": ("sh000016", "上证50"), + "上证50": ("sh000016", "上证50"), + "930955.CSI": ("csi930955", "红利低波100"), +} + + +def _resolve_expected(target: str): + """Return ``(expected_code, expected_name)`` for a matrix target. + + Raises ``ValueError`` for targets outside the required matrix. + """ + try: + return EXPECTED_OUTCOMES[target] + except KeyError: + raise ValueError( + f"unsupported smoke target: {target!r} (must be one of " + f"{sorted(EXPECTED_OUTCOMES)!r})" + ) + + +def _emit_event(phase: str, target: str, **fields) -> None: + """Print one single-line ``E2E_EVENT {json}`` event to stdout.""" + payload = {"phase": phase, "target": target} + payload.update(fields) + print(f"{EVENT_PREFIX}{json.dumps(payload, ensure_ascii=False)}", flush=True) + + +def _spawn_kwargs(platform_name: str) -> dict: + """Worker spawn kwargs: inherit the console (never ``PIPE``) and detach + the child into its own session / new process group so the parent can + kill the whole tree on timeout.""" + if platform_name == "nt": + creationflags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) + return {"creationflags": creationflags} if creationflags else {} + return {"start_new_session": True} + + +def _terminate_process_tree(proc, platform_name: str) -> Optional[str]: + """Kill the whole worker process tree. + + Returns ``None`` only when tree cleanup is confirmed: a successful + ``taskkill /T /F`` (Windows) or process-group kill (POSIX), or a + ``ProcessLookupError`` for the tree/group (already gone). A failed tree + cleanup returns an error string even when the direct ``proc.kill()`` + best-effort fallback succeeds — direct kill only proves the worker + process itself was killed, not its descendants, so it never erases the + original tree-cleanup error. + """ + if platform_name == "nt": + try: + result = subprocess.run( + ["taskkill", "/T", "/F", "/PID", str(proc.pid)], + timeout=30, + ) + if result.returncode == 0: + return None + taskkill_error = f"taskkill exited {result.returncode}" + except Exception as exc: + taskkill_error = f"taskkill failed: {exc}" + # Best-effort damage control: kill the worker process itself, but + # never report a confirmed tree cleanup. + try: + proc.kill() + return ( + f"{taskkill_error}; direct worker kill succeeded but " + f"tree cleanup is unconfirmed" + ) + except ProcessLookupError: + return ( + f"{taskkill_error}; worker process already gone but " + f"tree cleanup is unconfirmed" + ) + except Exception as exc: + return f"{taskkill_error}; direct worker kill also failed: {exc}" + kill_signal = getattr(signal, "SIGKILL", signal.SIGTERM) + killpg = getattr(os, "killpg", None) + try: + if killpg is not None: + killpg(proc.pid, kill_signal) + else: + # 无进程组原语：直接 kill 只能证明 worker 本身被杀，不能证明 + # 后代已清理，不得返回 None 声称树清理已确认。 + proc.kill() + return ( + "process-group kill unavailable; direct worker kill " + "succeeded but tree cleanup is unconfirmed" + ) + return None + except ProcessLookupError: + # The process group is already gone — confirmed success. + return None + except OSError as exc: + try: + proc.kill() + return ( + f"process-group kill failed: {exc}; " + f"direct worker kill succeeded but tree cleanup is unconfirmed" + ) + except ProcessLookupError: + return ( + f"process-group kill failed: {exc}; " + f"worker process already gone but tree cleanup is unconfirmed" + ) + except Exception as kill_exc: + return ( + f"process-group kill failed: {exc}; " + f"direct worker kill also failed: {kill_exc}" + ) + + +def _cleanup_process_tree(proc, platform_name: str) -> Optional[str]: + """Terminate the worker tree and wait for the worker to exit. + + Returns ``None`` only when tree cleanup is confirmed; any terminate / + post-cleanup wait / kill failure is merged into the returned error + string so callers never silently claim a confirmed cleanup. + """ + cleanup_error = _terminate_process_tree(proc, platform_name) + # Post-cleanup handling must never escape as an uncaught exception: + # merge any failure into ``cleanup_error``. + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + try: + proc.kill() + except Exception as exc: + post_error = f"post-cleanup kill failed: {exc}" + cleanup_error = ( + f"{cleanup_error}; {post_error}" + if cleanup_error is not None + else post_error + ) + except Exception as exc: + post_error = f"post-cleanup wait failed: {exc}" + cleanup_error = ( + f"{cleanup_error}; {post_error}" + if cleanup_error is not None + else post_error + ) + return cleanup_error + + +def _assert_complete_result(result: dict, expected_code: str, expected_name: str): + """Validate a completed task result. + + Returns ``None`` on success, or a human-readable failure reason string. + A completed task is only a success when the canonical code equals the + authoritative expected code, the name equals the registered display + name, and the three non-empty result fields (analysis_summary / + operation_advice / trend_prediction) are present. + """ + if not isinstance(result, dict): + return "completed result is not a dict" + if result.get("code") != expected_code: + return ( + f"result code mismatch: expected {expected_code!r}, " + f"got {result.get('code')!r}" + ) + if result.get("name") != expected_name: + return ( + f"result name mismatch: expected {expected_name!r}, " + f"got {result.get('name')!r}" + ) + missing = [ + key + for key in ("analysis_summary", "operation_advice", "trend_prediction") + if not (result.get(key) or "").strip() + ] + if missing: + return f"completed result missing non-empty fields: {missing}" + return None + + +def _poll_once(service, task_id: str, target: str, stock_code: str): + """One poll of the in-process ``TaskService``. + + Empty initial state (``None`` / ``running``) returns ``None`` so the + caller keeps waiting. Terminal states return exactly one event dict: + ``completed`` (structure assertions passed against the authoritative + matrix expectation) or ``failed`` (task failed or completed but + incomplete). + """ + expected_code, expected_name = _resolve_expected(target) + status = service.get_task_status(task_id) + if status is None or status.get("status") in ("running", "pending"): + return None + if status.get("status") == "completed": + result = status.get("result") + reason = _assert_complete_result(result, expected_code, expected_name) + if reason is not None: + return { + "phase": "failed", + "target": target, + "task_id": task_id, + "stock_code": stock_code, + "error": reason, + } + return { + "phase": "completed", + "target": target, + "task_id": task_id, + "stock_code": stock_code, + "result": result, + } + return { + "phase": "failed", + "target": target, + "task_id": task_id, + "stock_code": stock_code, + "error": status.get("error") or f"task status is {status.get('status')!r}", + } + + +# --------------------------------------------------------------------------- +# Worker +# --------------------------------------------------------------------------- +def _run_worker(target: str) -> int: + """Worker entrypoint: submit via the real dispatcher in this process and + poll the same process's TaskService until a terminal state. + + Only exits 0 when a completed event with passing structure assertions + was produced. Authoritative target validation runs BEFORE any dispatcher + construction or submission, so an unsupported target can never submit. + Unexpected exceptions from dispatch / poll / event serialization are + converted into a ``failed`` E2E event with the exception evidence on + stderr and exit 1 — never an uncaught traceback. ``KeyboardInterrupt`` + is not treated as an ordinary failure. + """ + expected_code = _resolve_expected(target)[0] + + try: + return _run_worker_inner(target, expected_code) + except KeyboardInterrupt: + raise + except Exception as exc: + _emit_event("failed", target, error=f"worker exception: {exc}") + print(f"ERROR: worker exception: {exc}", file=sys.stderr) + return 1 + + +def _run_worker_inner(target: str, expected_code: str) -> int: + """Dispatch + poll loop of the worker (see :func:`_run_worker`).""" + from datetime import datetime + + from bot.commands.analyze import AnalyzeCommand + from bot.dispatcher import CommandDispatcher + from bot.models import BotMessage, ChatType + from src.services.task_service import get_task_service + + dispatcher = CommandDispatcher() + dispatcher.register(AnalyzeCommand()) + + message = BotMessage( + platform="smoke", + message_id="smoke-e2e", + user_id="smoke", + user_name="smoke", + chat_id="smoke", + chat_type=ChatType.PRIVATE, + content=f"/analyze {target}", + raw_content=f"/analyze {target}", + mentioned=False, + timestamp=datetime.now(), + ) + + import asyncio + + response = asyncio.run(dispatcher.dispatch_async(message)) + + # 权威期望只来自矩阵映射，绝不信任 ``extra.stock_code`` 或任务结果自报 + # 身份：dispatcher 路由错误（提交了错误 code/name）时仍会失败。 + response_code = response.extra.get("stock_code") + response_task_id = response.extra.get("task_id") + if response_task_id and response_code == expected_code: + task_id = response_task_id + _emit_event( + "submitted", + target, + task_id=task_id, + stock_code=expected_code, + ) + elif response_task_id: + # 有任务 identity 但 code 与矩阵期望不符：提交了错误的标的，输出 + # 明确的 mismatch 错误（含期望值与实际值）与结构化实际 code， + # 而不是成功响应文本。 + _emit_event( + "failed", + target, + task_id=response_task_id, + stock_code=response_code, + error=( + f"submitted code mismatch: expected {expected_code!r}, " + f"got {response_code!r}" + ), + ) + return 1 + else: + _emit_event("failed", target, error=response.text) + return 1 + + service = get_task_service() + while True: + event = _poll_once(service, task_id, target, expected_code) + if event is not None: + _emit_event(event["phase"], target, **{ + k: v + for k, v in event.items() + if k not in ("phase", "target") + }) + return 0 if event["phase"] == "completed" else 1 + time.sleep(5) + + +# --------------------------------------------------------------------------- +# Parent +# --------------------------------------------------------------------------- +def _run_parent(target: str, timeout: int) -> int: + """Parent entrypoint: spawn the worker (inheriting the console), enforce + the deadline, clean up the process tree on timeout and on Ctrl-C, and + map to exit codes 0 / 1 / 124. Authoritative target validation runs + before the subprocess spawn, so an unsupported target never reaches + ``Popen``; a non-positive timeout is rejected the same way. On Ctrl-C + the parent cleans up the worker tree first: a confirmed cleanup + propagates the interrupt, a failed cleanup emits a structured ``failed`` + event and returns 1.""" + _resolve_expected(target) # raises ValueError for unsupported targets + if timeout <= 0: + raise ValueError(f"timeout must be positive, got {timeout}") + + # ``proc.wait`` without ``timeout`` would block SIGINT handling on + # Windows, so poll with the deadline instead. + child_args = [sys.executable, str(Path(__file__).resolve()), "--worker", target] + if os.name == "nt": + proc = subprocess.Popen( + child_args, + cwd=str(REPO_ROOT), + **_spawn_kwargs("nt"), + ) + else: + proc = subprocess.Popen( + child_args, + cwd=str(REPO_ROOT), + **_spawn_kwargs("posix"), + ) + deadline = time.monotonic() + timeout + while True: + try: + exit_code = proc.wait(timeout=1) + except subprocess.TimeoutExpired: + exit_code = None + except KeyboardInterrupt: + # 用户主动中止：清理 worker 进程树。清理失败不得静默吞掉—— + # 与 timeout 分支一致，输出结构化 failed 事件并返回 1，避免 + # 用户误以为进程树已清理而 worker 后代仍在跑真实副作用。 + if proc.poll() is not None: + # worker 已自行退出：无需清理，直接透传中断。 + raise + cleanup_error = _cleanup_process_tree( + proc, "nt" if os.name == "nt" else "posix" + ) + if cleanup_error is not None: + _emit_event( + "failed", + target, + error=f"interrupt cleanup failed: {cleanup_error}", + ) + return 1 + raise + if exit_code is not None: + # 文档化运行时契约只暴露 0 / 1 / 124：worker 的任意其他退出码 + # 归一化为 1，避免泄漏任意崩溃码。 + if exit_code in (0, 1, 124): + return exit_code + return 1 + if time.monotonic() >= deadline: + cleanup_error = _cleanup_process_tree( + proc, "nt" if os.name == "nt" else "posix" + ) + if cleanup_error is not None: + # 清理失败不得静默声称成功：输出结构化 failed 事件并返回 1。 + _emit_event( + "failed", + target, + error=f"timeout cleanup failed: {cleanup_error}", + ) + return 1 + _emit_event("timeout", target) + return 124 + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser( + description=( + "Bot 指数入口 transport-independent 在线 E2E smoke：" + "从真实 dispatcher 贯穿 Pipeline，单标的在线分析。" + ) + ) + parser.add_argument("target", help="矩阵输入：SH.000016 / 上证50 / 930955.CSI") + parser.add_argument( + "--timeout", + type=int, + default=900, + help="worker 超时秒数（默认 900 秒；超时清理进程树，不回滚副作用）", + ) + parser.add_argument( + "--worker", + action="store_true", + help=argparse.SUPPRESS, + ) + args = parser.parse_args(argv) + + target = (args.target or "").strip() + if not target: + print("ERROR: target 不能为空", file=sys.stderr) + return 2 + + # 权威 target 校验必须先于任何角色分支：不支持的 target 不得构造 + # dispatcher、不得提交、不得 spawn 子进程（参数/输入错误，非零返回）。 + try: + _resolve_expected(target) + except ValueError as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 2 + + # ``--timeout`` 必须为正：非正超时在 spawn/提交之前即拒绝（输入错误）。 + if args.timeout <= 0: + print(f"ERROR: --timeout 必须为正整数，got {args.timeout}", file=sys.stderr) + return 2 + + _reconfigure_output_stream(sys.stdout) + _reconfigure_output_stream(sys.stderr) + + # 角色选择只由显式 ``--worker`` flag 决定（父进程拉起子进程时显式传入）， + # 不依赖任何环境变量，避免外部预置环境变量绕过父进程的硬超时。 + if args.worker: + return _run_worker(target) + return _run_parent(target, args.timeout) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/services/stock_list_parser.py b/src/services/stock_list_parser.py index 9e3e0f7fc..65105c877 100644 --- a/src/services/stock_list_parser.py +++ b/src/services/stock_list_parser.py @@ -87,6 +87,20 @@ def _normalize_index_key(value: str) -> str: return normalized +def _normalize_display_name(value: str) -> str: + """Normalize a display name for exact name lookup only. + + Exactly NFKC + trim + casefold. Deliberately **not** the identity + normalization of :func:`_normalize_index_key`, which rewrites exchange + variants (``000016.SH`` collapses to ``sh000016``): code-shaped display + names that are distinct under exact text equality must stay distinct here, + otherwise a name lookup would falsely report them as one key or as + ambiguous. Display-name equality is exact-text equality after this + normalization — no fuzzy search. + """ + return unicodedata.normalize("NFKC", str(value or "")).strip().casefold() + + # --------------------------------------------------------------------------- # Legacy helpers (unchanged). # --------------------------------------------------------------------------- @@ -392,6 +406,47 @@ class IndexRegistry: """ return self._by_resolver_key.get(_normalize_index_key(key)) + def find_by_display_name(self, name: str) -> Optional[IndexEntry]: + """Look up an index by its exact registered display name. + + Only NFKC + trim + casefold exact matches are considered — no fuzzy + search. Display names are deliberately NOT identity aliases: this + lookup is independent of :meth:`find_by_explicit_key` and never feeds + ``_by_resolver_key`` / :func:`parse_analysis_target`, so a Chinese + name can never become a parser identity alias. When two registered + entries have NFKC/casefold-equivalent display names the lookup is + ambiguous and returns ``None``; callers should detect that via + :meth:`is_ambiguous_display_name` and ask the user for an explicit + code instead of falling back to stock-name resolution. + """ + key = _normalize_display_name(name) + if not key: + return None + matches = [ + entry + for entry in self._entries + if _normalize_display_name(entry.display_name) == key + ] + if len(matches) == 1: + return matches[0] + return None + + def is_ambiguous_display_name(self, name: str) -> bool: + """True when two or more registered entries share an NFKC/casefold- + equivalent display name, so :meth:`find_by_display_name` cannot pick + a single entry.""" + key = _normalize_display_name(name) + if not key: + return False + return ( + sum( + 1 + for entry in self._entries + if _normalize_display_name(entry.display_name) == key + ) + > 1 + ) + def find_by_bare_conflict(self, bare_code: str) -> Optional[IndexEntry]: """Return the index whose explicit alias base equals ``bare_code``.""" return self._by_bare_conflict.get(_normalize_index_key(bare_code)) diff --git a/src/services/task_service.py b/src/services/task_service.py index 4606864e4..1411bc709 100644 --- a/src/services/task_service.py +++ b/src/services/task_service.py @@ -24,6 +24,7 @@ from src.enums import ReportType from src.storage import get_db from bot.models import BotMessage from src.services.stock_code_utils import resolve_index_stock_code_for_analysis +from src.services.stock_list_parser import AnalysisTarget, ParseStatus logger = logging.getLogger(__name__) @@ -72,7 +73,8 @@ class TaskService: report_type: Union[ReportType, str] = ReportType.SIMPLE, source_message: Optional[BotMessage] = None, save_context_snapshot: Optional[bool] = None, - query_source: str = "bot" + query_source: str = "bot", + analysis_target: Optional[AnalysisTarget] = None ) -> Dict[str, Any]: """ 提交异步分析任务 @@ -83,6 +85,9 @@ class TaskService: source_message: 来源消息（用于回复） save_context_snapshot: 是否保存上下文快照 query_source: 任务来源标识（bot/api/cli/system） + analysis_target: 可选结构化分析目标（INDEX 时原样使用 + ``target.canonical_id`` 并跳过股票代码规范化，防止 + ``sh000016`` 被改写为 ``SH000016``） Returns: 任务信息字典 @@ -91,7 +96,17 @@ class TaskService: if isinstance(report_type, str): report_type = ReportType.from_str(report_type) - normalized_code = resolve_index_stock_code_for_analysis(code) + if ( + analysis_target is not None + and analysis_target.asset_type != ParseStatus.INDEX + ): + raise ValueError("analysis_target must be an INDEX target") + if analysis_target is not None: + # INDEX target 携带时原样使用 canonical_id，不再走股票代码 + # 规范化路径（避免 ``sh000016`` 被改写为 ``SH000016``）。 + normalized_code = analysis_target.canonical_id + else: + normalized_code = resolve_index_stock_code_for_analysis(code) if not normalized_code: raise ValueError("股票代码不能为空或仅包含空白字符") @@ -105,7 +120,8 @@ class TaskService: report_type, source_message, save_context_snapshot, - query_source + query_source, + analysis_target ) logger.info( @@ -153,7 +169,8 @@ class TaskService: report_type: ReportType = ReportType.SIMPLE, source_message: Optional[BotMessage] = None, save_context_snapshot: Optional[bool] = None, - query_source: str = "bot" + query_source: str = "bot", + analysis_target: Optional[AnalysisTarget] = None ) -> Dict[str, Any]: """ 执行单只股票分析 @@ -191,12 +208,15 @@ class TaskService: ) # 执行单只股票分析（启用单股推送） - result = pipeline.process_single_stock( - code=code, - skip_analysis=False, - single_stock_notify=True, - report_type=report_type - ) + process_kwargs = { + "code": code, + "skip_analysis": False, + "single_stock_notify": True, + "report_type": report_type, + } + if analysis_target is not None: + process_kwargs["analysis_target"] = analysis_target + result = pipeline.process_single_stock(**process_kwargs) if result and result.success: result_data = { diff --git a/tests/test_bot_analyze_command.py b/tests/test_bot_analyze_command.py new file mode 100644 index 000000000..9f69180c3 --- /dev/null +++ b/tests/test_bot_analyze_command.py @@ -0,0 +1,363 @@ +# -*- coding: utf-8 -*- +"""Tests for the shared ``/analyze`` command (Bot 指数入口, PR2). + +Covers the real ``CommandDispatcher`` gate (``validate_args`` before +``execute``) with a stubbed ``TaskService`` so the tests assert the +code/target handed to ``submit_analysis`` without touching the network or +the real pipeline. +""" + +import unittest +from datetime import datetime +from unittest.mock import patch + +# Keep tests runnable when optional deps are missing. +try: + import litellm # noqa: F401 +except ModuleNotFoundError: + from tests.litellm_stub import ensure_litellm_stub + ensure_litellm_stub() + +from bot.commands.analyze import AnalyzeCommand +from bot.dispatcher import CommandDispatcher +from bot.models import BotMessage, BotResponse, ChatType +from src.services.stock_list_parser import ParseStatus + + +class _StubTaskService: + """Records the last ``submit_analysis`` call; never touches the network. + + Supports two canned outcomes (success by default, or failure) so tests + can assert both the success ``extra`` contract and the no-identity error + contract without touching the network or the real pipeline. + """ + + def __init__(self, fail: bool = False): + self.calls = [] + self._fail = fail + + def submit_analysis(self, **kwargs): + self.calls.append(kwargs) + if self._fail: + return {"success": False, "task_id": "", "error": "boom"} + return { + "success": True, + "task_id": "task-1234567890abcdef", + "code": kwargs.get("code", ""), + } + + +def _make_message(content: str) -> BotMessage: + return BotMessage( + platform="feishu", + message_id="m1", + user_id="u1", + user_name="tester", + chat_id="c1", + chat_type=ChatType.PRIVATE, + content=content, + raw_content=content, + mentioned=False, + timestamp=datetime.now(), + ) + + +class TestAnalyzeCommandDispatcherGate(unittest.TestCase): + """Real dispatcher gate: ``validate_args`` runs before ``execute`` and the + command submits through the shared ``AnalyzeCommand``.""" + + def _dispatch(self, content: str): + dispatcher = CommandDispatcher() + command = AnalyzeCommand() + dispatcher.register(command) + stub = _StubTaskService() + with patch("src.services.task_service.get_task_service", return_value=stub): + response = dispatcher.dispatch(_make_message(content)) + return response, stub + + def test_registered_code_submits_index_target(self): + response, stub = self._dispatch("/analyze sh000016") + self.assertIsInstance(response, BotResponse) + self.assertIn("分析任务已提交", response.text) + self.assertEqual(len(stub.calls), 1) + call = stub.calls[0] + self.assertEqual(call["code"], "sh000016") + target = call["analysis_target"] + self.assertIsNotNone(target) + self.assertEqual(target.asset_type, ParseStatus.INDEX) + self.assertEqual(target.canonical_id, "sh000016") + + def test_csi_alias_converges_to_canonical_index_target(self): + response, stub = self._dispatch("/analyze 930955.CSI") + self.assertIn("分析任务已提交", response.text) + call = stub.calls[0] + self.assertEqual(call["code"], "csi930955") + target = call["analysis_target"] + self.assertEqual(target.asset_type, ParseStatus.INDEX) + self.assertEqual(target.canonical_id, "csi930955") + + def test_registered_prefix_alias_submits_index_target(self): + """``csi930955`` is the registered canonical CSI prefix form; it must + submit with a matching INDEX target.""" + response, stub = self._dispatch("/analyze csi930955") + self.assertIn("分析任务已提交", response.text) + call = stub.calls[0] + self.assertEqual(call["code"], "csi930955") + target = call["analysis_target"] + self.assertIsNotNone(target) + self.assertEqual(target.asset_type, ParseStatus.INDEX) + self.assertEqual(target.canonical_id, "csi930955") + + def test_registered_suffix_alias_submits_index_target(self): + """``000016.SH`` is the registered suffix alias of sh000016; it must + submit the registry canonical with a matching INDEX target.""" + response, stub = self._dispatch("/analyze 000016.SH") + self.assertIn("分析任务已提交", response.text) + call = stub.calls[0] + self.assertEqual(call["code"], "sh000016") + target = call["analysis_target"] + self.assertIsNotNone(target) + self.assertEqual(target.asset_type, ParseStatus.INDEX) + self.assertEqual(target.canonical_id, "sh000016") + + def test_dotted_us_ticker_keeps_legacy_path_without_target(self): + """``BRK.B`` is a dotted US ticker the legacy gate accepts; it must + resolve to ``BRK.B`` with no structured target.""" + response, stub = self._dispatch("/analyze BRK.B") + self.assertIn("分析任务已提交", response.text) + call = stub.calls[0] + self.assertEqual(call["code"], "BRK.B") + self.assertIsNone(call["analysis_target"]) + + def test_parser_index_alias_ss_dotted_prefix_submits_index_target(self): + """``SS.000300`` is a parser-recognized dotted-prefix alias of the + registered sh000300 index (SS = Shanghai alias); the Bot must submit + the registry canonical with a matching INDEX target.""" + response, stub = self._dispatch("/analyze SS.000300") + self.assertIn("分析任务已提交", response.text) + call = stub.calls[0] + self.assertEqual(call["code"], "sh000300") + target = call["analysis_target"] + self.assertIsNotNone(target) + self.assertEqual(target.asset_type, ParseStatus.INDEX) + self.assertEqual(target.canonical_id, "sh000300") + + def test_dotted_prefix_sh_alias_submits_index_target(self): + """``SH.000016`` is a parser-recognized dotted-prefix alias of the + registered sh000016 index; the Bot must submit the registry canonical + with a matching INDEX target.""" + response, stub = self._dispatch("/analyze SH.000016") + self.assertIn("分析任务已提交", response.text) + call = stub.calls[0] + self.assertEqual(call["code"], "sh000016") + target = call["analysis_target"] + self.assertIsNotNone(target) + self.assertEqual(target.asset_type, ParseStatus.INDEX) + self.assertEqual(target.canonical_id, "sh000016") + + def test_dotted_prefix_sz_alias_submits_index_target(self): + """``SZ.399001`` is a parser-recognized dotted-prefix alias of the + registered sz399001 index; the Bot must submit the registry canonical + with a matching INDEX target.""" + response, stub = self._dispatch("/analyze SZ.399001") + self.assertIn("分析任务已提交", response.text) + call = stub.calls[0] + self.assertEqual(call["code"], "sz399001") + target = call["analysis_target"] + self.assertIsNotNone(target) + self.assertEqual(target.asset_type, ParseStatus.INDEX) + self.assertEqual(target.canonical_id, "sz399001") + + def test_registered_name_wins_over_stock_name_fallback(self): + with patch( + "src.services.name_to_code_resolver.resolve_name_to_code" + ) as mock_resolve: + response, stub = self._dispatch("/analyze 上证50") + self.assertIn("分析任务已提交", response.text) + call = stub.calls[0] + self.assertEqual(call["code"], "sh000016") + target = call["analysis_target"] + self.assertEqual(target.asset_type, ParseStatus.INDEX) + self.assertEqual(target.canonical_id, "sh000016") + self.assertEqual(target.raw_input, "上证50") + mock_resolve.assert_not_called() + + def test_stock_code_keeps_legacy_code_and_no_target(self): + response, stub = self._dispatch("/analyze 600519") + self.assertIn("分析任务已提交", response.text) + call = stub.calls[0] + self.assertEqual(call["code"], "600519") + self.assertIsNone(call["analysis_target"]) + + def test_stock_name_fallback_keeps_legacy_code(self): + """Deterministic: the AkShare/network online map is stubbed empty so + the real top-level ``resolve_name_to_code`` runs its local tables + only and still resolves 贵州茅台 to its legacy code.""" + response, stub = None, None + with patch( + "src.services.name_to_code_resolver._get_akshare_name_to_code", + return_value={}, + ): + response, stub = self._dispatch("/analyze 贵州茅台") + self.assertIn("分析任务已提交", response.text) + call = stub.calls[0] + self.assertEqual(call["code"], "600519") + self.assertIsNone(call["analysis_target"]) + + def test_hk_and_us_stock_keep_legacy_code(self): + # ``hk00700`` is uppercased to ``HK00700`` by the legacy resolver — + # the same behavior the pre-PR command already had. + for content, expected in (("/analyze hk00700", "HK00700"), ("/analyze AAPL", "AAPL")): + response, stub = self._dispatch(content) + self.assertIn("分析任务已提交", response.text) + call = stub.calls[0] + self.assertEqual(call["code"], expected) + self.assertIsNone(call["analysis_target"]) + + def test_lowercase_us_ticker_keeps_legacy_uppercase_path(self): + """``usfd`` is a real US ticker the old gate accepted (case-insensitive + 1-5 letters); it must resolve to ``USFD`` with no target, exactly as + before this change.""" + response, stub = self._dispatch("/analyze usfd") + self.assertIn("分析任务已提交", response.text) + call = stub.calls[0] + self.assertEqual(call["code"], "USFD") + self.assertIsNone(call["analysis_target"]) + + def test_legacy_invalid_shapes_return_error_without_submission(self): + """Shapes the old validate_args gate rejected must stay rejected — + explicit error and no TaskService call.""" + for code in ("12345", "00700", "600519.SH", "sh999999"): + with self.subTest(code=code): + response, stub = self._dispatch(f"/analyze {code}") + self.assertIn("无效的标的代码", response.text) + self.assertEqual(len(stub.calls), 0) + + def test_unregistered_csi_returns_error_and_does_not_submit(self): + response, stub = self._dispatch("/analyze 930956.CSI") + self.assertIn("无法分析", response.text) + self.assertIn("CSI", response.text) + self.assertEqual(len(stub.calls), 0) + + def test_malformed_csi_numeric_forms_return_csi_error_without_submission( + self, + ): + """The full numeric explicit CSI family (``csi`` + digits, or digits + + ``.csi``) is rejected with the CSI-specific error — never submitted, + never routed into stock-name resolution.""" + for code in ("csi930956", "csi123", "12345.csi"): + with self.subTest(code=code): + response, stub = self._dispatch(f"/analyze {code}") + self.assertIn("无法分析", response.text) + self.assertIn("CSI", response.text) + self.assertEqual(len(stub.calls), 0) + + def test_parser_unsupported_malformed_code_returns_error_without_submission( + self, + ): + """A non-CSI parser-UNSUPPORTED malformed code shape (``us1``) must + return an explicit error and never submit a task.""" + response, stub = self._dispatch("/analyze us1") + self.assertIn("无法分析", response.text) + self.assertEqual(len(stub.calls), 0) + + def test_unknown_name_returns_error_and_does_not_submit(self): + """Deterministic: the AkShare/network online map is stubbed empty so + the real ``resolve_name_to_code`` runs its local tables only.""" + response, stub = None, None + with patch( + "src.services.name_to_code_resolver._get_akshare_name_to_code", + return_value={}, + ): + response, stub = self._dispatch("/analyze 不存在标的") + self.assertIn("无法识别标的", response.text) + self.assertEqual(len(stub.calls), 0) + + def test_empty_args_rejected_by_validate_args_gate(self): + response, stub = self._dispatch("/analyze") + self.assertIn("请输入", response.text) + self.assertEqual(len(stub.calls), 0) + + def test_full_report_flag_still_works(self): + response, stub = self._dispatch("/analyze sh000016 full") + self.assertIn("分析任务已提交", response.text) + call = stub.calls[0] + self.assertEqual(call["code"], "sh000016") + self.assertEqual(call["report_type"].value, "full") + + def test_success_text_is_unchanged_and_carries_task_identity_extra(self): + """On a successful submission the user-visible text must stay exactly + as before, while ``extra`` carries the internal task identity + (``task_id`` + ``stock_code``) for transport-independent consumers.""" + response, stub = self._dispatch("/analyze sh000016") + expected_text = ( + "✅ **分析任务已提交**\n\n" + "• 标的: `sh000016`\n" + "• 报告类型: 精简报告\n" + "• 任务 ID: `task-1234567890abcde...`\n\n" + "分析完成后将自动推送结果。" + ) + self.assertEqual(response.text, expected_text) + self.assertEqual(response.extra, { + "task_id": "task-1234567890abcdef", + "stock_code": "sh000016", + }) + + def test_success_extra_uses_normalized_code(self): + """``extra.stock_code`` carries the normalized code the task was + submitted under (the registry canonical for an index alias).""" + response, stub = self._dispatch("/analyze 930955.CSI") + self.assertIn("分析任务已提交", response.text) + self.assertEqual(response.extra["task_id"], "task-1234567890abcdef") + self.assertEqual(response.extra["stock_code"], "csi930955") + + def test_error_response_has_no_task_identity_extra(self): + """On a submission failure there is no task identity: no task was + created, so ``extra`` must stay empty.""" + dispatcher = CommandDispatcher() + dispatcher.register(AnalyzeCommand()) + stub = _StubTaskService(fail=True) + with patch("src.services.task_service.get_task_service", return_value=stub): + response = dispatcher.dispatch(_make_message("/analyze sh000016")) + self.assertIn("提交分析任务失败", response.text) + self.assertEqual(response.extra, {}) + + +class TestAnalyzeCommandAmbiguousName(unittest.TestCase): + """Ambiguous registered display names must fail with an explicit error and + never fall back to stock-name resolution.""" + + def test_ambiguous_display_name_returns_error(self): + from src.services.stock_list_parser import IndexEntry, IndexRegistry + + registry = IndexRegistry(( + IndexEntry( + bare_code="000300", + exchange="SH", + canonical_id="sh000300", + display_name="沪深300", + ), + IndexEntry( + bare_code="000999", + exchange="SH", + canonical_id="sh000999", + display_name="沪深300", + ), + )) + dispatcher = CommandDispatcher() + dispatcher.register(AnalyzeCommand()) + stub = _StubTaskService() + with patch( + "bot.commands.analyze.default_index_registry", return_value=registry + ), patch( + "src.services.name_to_code_resolver.resolve_name_to_code" + ) as mock_resolve, patch( + "src.services.task_service.get_task_service", return_value=stub + ): + response = dispatcher.dispatch(_make_message("/analyze 沪深300")) + self.assertIn("歧义", response.text) + self.assertEqual(stub.calls, []) + mock_resolve.assert_not_called() + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_smoke_bot_index_entry.py b/tests/test_smoke_bot_index_entry.py new file mode 100644 index 000000000..8c6e42315 --- /dev/null +++ b/tests/test_smoke_bot_index_entry.py @@ -0,0 +1,1170 @@ +# -*- coding: utf-8 -*- +"""Offline contract tests for the Bot 指数入口 online E2E smoke runner +(``scripts/smoke_bot_index_entry.py``). + +Covers, without touching the network or spawning real children: + +* the I/O & edge-case matrix resolution (SH_ALIAS / REGISTERED_NAME / + CSI_ALIAS -> canonical id + registered display name), +* transport independence: worker spawn kwargs never use ``PIPE``, + POSIX uses ``start_new_session`` and Windows uses a new process group, +* failure/timeout semantics: failed tasks and incomplete completed results + produce a ``failed`` phase event; timeout cleanup kills the process tree, +* the single-line ``E2E_EVENT {json}`` wire format. +""" + +import io +import json +import os +import signal +import subprocess +import unittest +from contextlib import redirect_stdout +from unittest.mock import patch + +from scripts import smoke_bot_index_entry as smoke # noqa: E402 + + +class TestExpectedOutcomes(unittest.TestCase): + """The authoritative matrix map must cover exactly the three required + inputs and reject everything else.""" + + def test_matrix_covers_three_required_scenarios(self): + self.assertEqual( + smoke.EXPECTED_OUTCOMES, + { + "SH.000016": ("sh000016", "上证50"), + "上证50": ("sh000016", "上证50"), + "930955.CSI": ("csi930955", "红利低波100"), + }, + ) + + def test_resolve_expected_sh_alias(self): + self.assertEqual( + smoke._resolve_expected("SH.000016"), ("sh000016", "上证50") + ) + + def test_resolve_expected_registered_name(self): + self.assertEqual( + smoke._resolve_expected("上证50"), ("sh000016", "上证50") + ) + + def test_resolve_expected_csi_alias(self): + self.assertEqual( + smoke._resolve_expected("930955.CSI"), ("csi930955", "红利低波100") + ) + + def test_unknown_target_is_rejected(self): + with self.assertRaises(ValueError): + smoke._resolve_expected("AAPL") + with self.assertRaises(ValueError): + smoke._resolve_expected("600519") + + +class TestSpawnKwargs(unittest.TestCase): + """Worker spawn must inherit the console (no ``PIPE``) and detach into + its own session/process group so the parent can kill the whole tree.""" + + def test_posix_spawn_kwargs_use_start_new_session(self): + kwargs = smoke._spawn_kwargs("posix") + self.assertEqual(kwargs, {"start_new_session": True}) + + def test_windows_spawn_kwargs_use_new_process_group(self): + with patch.object( + smoke.subprocess, "CREATE_NEW_PROCESS_GROUP", 0x00000200, create=True + ): + kwargs = smoke._spawn_kwargs("nt") + self.assertEqual(kwargs, {"creationflags": 0x00000200}) + + def test_spawn_kwargs_never_use_pipe(self): + with patch.object( + smoke.subprocess, "CREATE_NEW_PROCESS_GROUP", 0x00000200, create=True + ): + for platform in ("posix", "nt"): + kwargs = smoke._spawn_kwargs(platform) + self.assertNotIn("stdout", kwargs) + self.assertNotIn("stderr", kwargs) + self.assertNotIn("stdin", kwargs) + + +class TestTerminateProcessTree(unittest.TestCase): + """Timeout cleanup must kill the whole tree: ``taskkill /T /F`` on + Windows, the process group on POSIX. Failures must be surfaced as an + error string — never silently swallowed as success — while an + already-gone process counts as success.""" + + @patch("subprocess.run") + def test_windows_uses_taskkill_tree_force(self, mock_run): + mock_run.return_value = type("R", (), {"returncode": 0})() + proc = type("Proc", (), {"pid": 4242})() + self.assertIsNone(smoke._terminate_process_tree(proc, "nt")) + mock_run.assert_called_once_with( + ["taskkill", "/T", "/F", "/PID", "4242"], timeout=30 + ) + + @patch.object(os, "killpg", create=True) + def test_posix_kills_process_group(self, mock_killpg): + proc = type("Proc", (), {"pid": 4242})() + self.assertIsNone(smoke._terminate_process_tree(proc, "posix")) + # ``SIGKILL`` does not exist on Windows; the script resolves it via + # ``getattr`` so this expectation stays portable. + kill_signal = getattr(signal, "SIGKILL", signal.SIGTERM) + mock_killpg.assert_called_once_with(4242, kill_signal) + + @patch("subprocess.run") + def test_windows_taskkill_failure_is_surfaced(self, mock_run): + """A non-zero taskkill must not be reported as success: the error is + returned (direct kill also failing keeps the error).""" + mock_run.return_value = type("R", (), {"returncode": 1})() + proc = type("Proc", (), {"pid": 4242, "kill": _raise_oserror})() + error = smoke._terminate_process_tree(proc, "nt") + self.assertIsNotNone(error) + self.assertIn("taskkill exited 1", error or "") + + @patch("subprocess.run") + def test_windows_taskkill_exception_is_surfaced(self, mock_run): + mock_run.side_effect = RuntimeError("taskkill missing") + proc = type("Proc", (), {"pid": 4242, "kill": _raise_oserror})() + error = smoke._terminate_process_tree(proc, "nt") + self.assertIsNotNone(error) + self.assertIn("taskkill failed", error or "") + + @patch("subprocess.run") + def test_windows_taskkill_failure_with_direct_kill_success(self, mock_run): + """When taskkill fails, a successful direct kill must NOT erase the + tree-cleanup error: direct kill only proves the worker process was + killed, not its descendants.""" + mock_run.return_value = type("R", (), {"returncode": 1})() + proc = type("Proc", (), {"pid": 4242, "kill": lambda self: None})() + error = smoke._terminate_process_tree(proc, "nt") + self.assertIsNotNone(error) + self.assertIn("taskkill exited 1", error or "") + self.assertIn("tree cleanup is unconfirmed", error or "") + + @patch.object(os, "killpg", create=True) + def test_posix_killpg_failure_is_surfaced(self, mock_killpg): + mock_killpg.side_effect = OSError("permission denied") + proc = type("Proc", (), {"pid": 4242, "kill": _raise_oserror})() + error = smoke._terminate_process_tree(proc, "posix") + self.assertIsNotNone(error) + self.assertIn("process-group kill failed", error or "") + + @patch.object(os, "killpg", create=True) + def test_posix_killpg_failure_with_direct_kill_success_still_fails(self, mock_killpg): + """A successful direct kill after a killpg failure must still surface + the tree-cleanup error (descendants are unconfirmed).""" + mock_killpg.side_effect = OSError("permission denied") + proc = type("Proc", (), {"pid": 4242, "kill": lambda self: None})() + error = smoke._terminate_process_tree(proc, "posix") + self.assertIsNotNone(error) + self.assertIn("process-group kill failed", error or "") + self.assertIn("tree cleanup is unconfirmed", error or "") + + @patch.object(os, "killpg", create=True) + def test_posix_already_gone_counts_as_success(self, mock_killpg): + mock_killpg.side_effect = ProcessLookupError() + proc = type("Proc", (), {"pid": 4242})() + self.assertIsNone(smoke._terminate_process_tree(proc, "posix")) + + @patch.object(os, "killpg", None, create=True) + def test_posix_without_killpg_never_claims_confirmed_tree_cleanup(self): + """Without a process-group primitive, a direct kill only proves the + worker itself was killed, not its descendants: the function must + return an unconfirmed-tree-cleanup error instead of None.""" + proc = type("Proc", (), {"pid": 4242, "kill": lambda self: None})() + error = smoke._terminate_process_tree(proc, "posix") + self.assertIsNotNone(error) + self.assertIn("tree cleanup is unconfirmed", error or "") + + +def _raise_oserror(): + raise OSError("direct kill failed") + + +class TestParentSpawn(unittest.TestCase): + """The parent must spawn the worker with the explicit hidden ``--worker`` + flag (never via environment), with no ``PIPE``, and the child must not + re-enter the parent branch when an environment role variable is preset.""" + + def _patch_popen(self, target, env=None, exit_code=0): + class _FakeProc: + def __init__(self, args, **kwargs): + self.args = args + self.kwargs = kwargs + + def wait(self, timeout=None): + return exit_code + + with patch.object(subprocess, "Popen", side_effect=_FakeProc) as mock_popen: + try: + result = smoke._run_parent(target, timeout=60) + finally: + pass + return result, mock_popen + + def test_parent_spawns_worker_with_explicit_flag_and_no_pipe(self): + result, mock_popen = self._patch_popen("SH.000016") + self.assertEqual(result, 0) + self.assertEqual(mock_popen.call_count, 1) + args, kwargs = mock_popen.call_args + self.assertIn("--worker", args[0]) + self.assertEqual(args[0][-1], "SH.000016") + self.assertNotIn("stdout", kwargs) + self.assertNotIn("stderr", kwargs) + self.assertNotIn("stdin", kwargs) + + def test_preset_env_role_cannot_force_worker_mode(self): + """An externally preset role variable must not switch the entrypoint + into worker mode: without ``--worker`` the parent branch still runs.""" + with patch.dict( + os.environ, {"DSA_SMOKE_ROLE": "worker"}, clear=False + ): + result, mock_popen = self._patch_popen("上证50") + self.assertEqual(result, 0) + self.assertEqual(mock_popen.call_count, 1) + args, kwargs = mock_popen.call_args + self.assertIn("--worker", args[0]) + self.assertEqual(args[0][-1], "上证50") + + def test_worker_flag_dispatches_to_worker_and_parent_flag_never_spawns(self): + """``main(["--worker", target])`` runs the worker directly; the + parent path (which is the only spawner) never sees ``--worker``.""" + with patch.object(smoke, "_run_worker", return_value=3) as mock_worker, patch.object( + smoke, "_run_parent", return_value=4 + ) as mock_parent: + self.assertEqual(smoke.main(["--worker", "SH.000016"]), 3) + mock_worker.assert_called_once_with("SH.000016") + mock_parent.assert_not_called() + + def test_run_parent_unsupported_target_never_spawns(self): + """Unsupported targets are rejected before the subprocess spawn: the + parent must raise and never reach ``Popen``.""" + def _fail_popen(*args, **kwargs): + raise AssertionError("Popen must not be called for unsupported targets") + + with patch.object(subprocess, "Popen", side_effect=_fail_popen) as mock_popen: + with self.assertRaises(ValueError): + smoke._run_parent("AAPL", timeout=60) + mock_popen.assert_not_called() + + def test_run_worker_unsupported_target_raises_before_dispatch(self): + """``_run_worker`` defensively validates before any dispatcher + construction: an unsupported target raises and never dispatches.""" + def _fail_dispatcher(*args, **kwargs): + raise AssertionError("dispatcher must not be constructed") + + with patch("bot.dispatcher.CommandDispatcher", side_effect=_fail_dispatcher) as mock_disp: + with self.assertRaises(ValueError): + smoke._run_worker("600519") + mock_disp.assert_not_called() + + def test_main_unsupported_target_returns_error_without_running_branches(self): + """``main()`` must reject unsupported targets with a non-zero + argument/input error without calling ``_run_parent`` or + ``_run_worker`` (so nothing can submit or spawn).""" + def _fail_worker(target): + raise AssertionError("_run_worker must not be called") + + def _fail_parent(target, timeout): + raise AssertionError("_run_parent must not be called") + + with patch.object(smoke, "_run_worker", side_effect=_fail_worker) as mock_worker, patch.object( + smoke, "_run_parent", side_effect=_fail_parent + ) as mock_parent: + result = smoke.main(["AAPL"]) + self.assertEqual(result, 2) + mock_worker.assert_not_called() + mock_parent.assert_not_called() + + def test_run_parent_timeout_kills_tree_emits_event_and_exits_124(self): + """TIMEOUT matrix row: when the worker outlives the deadline the + parent cleans up the process tree, emits a timeout ``E2E_EVENT`` with + the target and exits 124.""" + import io + from contextlib import redirect_stdout + + class _NeverEndingProc: + def __init__(self, args, **kwargs): + self.args = args + self.kwargs = kwargs + + def wait(self, timeout=None): + raise subprocess.TimeoutExpired("worker", timeout or 1) + + def kill(self): + pass + + def _kill(proc, platform_name): + killed.append((proc, platform_name)) + + killed = [] + buf = io.StringIO() + # 第一次 monotonic 调用计算 deadline；后续调用（轮询循环内）必须越过 + # deadline。用单调递增序列而非固定两项列表，容忍实现中任何无害的 + # 额外 ``time.monotonic()`` 调用。 + monotonic_values = iter([1000.0, 1061.0, 1062.0, 1063.0]) + with patch.object( + subprocess, "Popen", side_effect=_NeverEndingProc + ), patch.object(smoke, "_terminate_process_tree", side_effect=_kill), patch( + "time.monotonic", side_effect=lambda: next(monotonic_values) + ), redirect_stdout(buf): + result = smoke._run_parent("SH.000016", timeout=60) + + self.assertEqual(result, 124) + self.assertEqual(len(killed), 1) + event_line = next( + line for line in buf.getvalue().splitlines() + if line.startswith("E2E_EVENT ") + ) + payload = json.loads(event_line[len("E2E_EVENT "):]) + self.assertEqual(payload["phase"], "timeout") + self.assertEqual(payload["target"], "SH.000016") + + def test_run_parent_timeout_cleanup_failure_emits_failed_and_exits_1(self): + """When tree cleanup fails on timeout, the parent must NOT claim + success: it emits a structured failed event with the cleanup error + and returns 1 (not 124).""" + import io + from contextlib import redirect_stdout + + class _NeverEndingProc: + def __init__(self, args, **kwargs): + self.args = args + self.kwargs = kwargs + + def wait(self, timeout=None): + raise subprocess.TimeoutExpired("worker", timeout or 1) + + def kill(self): + pass + + monotonic_values = iter([1000.0, 1061.0, 1062.0]) + buf = io.StringIO() + with patch.object( + subprocess, "Popen", side_effect=_NeverEndingProc + ), patch.object( + smoke, + "_terminate_process_tree", + return_value="taskkill exited 128", + ), patch( + "time.monotonic", side_effect=lambda: next(monotonic_values) + ), redirect_stdout(buf): + result = smoke._run_parent("SH.000016", timeout=60) + + self.assertEqual(result, 1) + event_line = next( + line for line in buf.getvalue().splitlines() + if line.startswith("E2E_EVENT ") + ) + payload = json.loads(event_line[len("E2E_EVENT "):]) + self.assertEqual(payload["phase"], "failed") + self.assertEqual(payload["target"], "SH.000016") + self.assertIn("timeout cleanup failed", payload["error"]) + self.assertIn("taskkill exited 128", payload["error"]) + + def test_run_parent_post_cleanup_kill_failure_emits_failed_and_exits_1(self): + """A post-cleanup ``proc.kill()`` failure must not escape as an + uncaught exception: it merges into the cleanup error, emits a + structured failed event and returns 1 (never 124).""" + import io + from contextlib import redirect_stdout + + class _StuckProc: + def __init__(self, args, **kwargs): + self.args = args + self.kwargs = kwargs + + def wait(self, timeout=None): + raise subprocess.TimeoutExpired("worker", timeout or 1) + + def kill(self): + raise OSError("post-cleanup kill exploded") + + monotonic_values = iter([1000.0, 1061.0, 1062.0]) + buf = io.StringIO() + with patch.object( + subprocess, "Popen", side_effect=_StuckProc + ), patch.object( + smoke, "_terminate_process_tree", return_value=None + ), patch( + "time.monotonic", side_effect=lambda: next(monotonic_values) + ), redirect_stdout(buf): + result = smoke._run_parent("SH.000016", timeout=60) + + self.assertEqual(result, 1) + event_line = next( + line for line in buf.getvalue().splitlines() + if line.startswith("E2E_EVENT ") + ) + payload = json.loads(event_line[len("E2E_EVENT "):]) + self.assertEqual(payload["phase"], "failed") + self.assertEqual(payload["target"], "SH.000016") + self.assertIn("timeout cleanup failed", payload["error"]) + self.assertIn("post-cleanup kill failed", payload["error"]) + + def test_run_parent_normalizes_unknown_worker_exit_code_to_1(self): + """The documented runtime contract only exposes 0 / 1 / 124: any + other worker exit code is normalized to 1.""" + result, mock_popen = self._patch_popen("SH.000016", exit_code=42) + self.assertEqual(result, 1) + self.assertEqual(mock_popen.call_count, 1) + + def test_run_parent_keyboard_interrupt_cleans_tree_then_reraises(self): + """A Ctrl-C in the parent poll window must clean up the worker + process tree before propagating the interrupt, so no orphan keeps + running the real data/LLM/report/notification chain.""" + + class _InterruptingProc: + def __init__(self, args, **kwargs): + self.args = args + self.kwargs = kwargs + self._interrupted = False + + def poll(self): + return None + + def wait(self, timeout=None): + if not self._interrupted: + self._interrupted = True + raise KeyboardInterrupt() + return 0 + + def kill(self): + pass + + killed = [] + with patch.object( + subprocess, "Popen", side_effect=_InterruptingProc + ), patch.object( + smoke, "_terminate_process_tree", side_effect=lambda p, n: killed.append((p, n)) + ): + with self.assertRaises(KeyboardInterrupt): + smoke._run_parent("SH.000016", timeout=60) + self.assertEqual(len(killed), 1) + self.assertEqual(killed[0][1], "nt" if os.name == "nt" else "posix") + + def test_run_parent_keyboard_interrupt_cleanup_failure_emits_failed_and_exits_1(self): + """When tree cleanup fails on Ctrl-C, the parent must NOT silently + swallow the failure: it emits a structured failed event with the + cleanup error and returns 1 instead of re-raising the interrupt.""" + import io + from contextlib import redirect_stdout + + class _InterruptingProc: + def __init__(self, args, **kwargs): + self.args = args + self.kwargs = kwargs + self._interrupted = False + + def poll(self): + return None + + def wait(self, timeout=None): + if not self._interrupted: + self._interrupted = True + raise KeyboardInterrupt() + return 0 + + def kill(self): + pass + + buf = io.StringIO() + killed = [] + with patch.object( + subprocess, "Popen", side_effect=_InterruptingProc + ), patch.object( + smoke, + "_terminate_process_tree", + side_effect=lambda p, n: killed.append((p, n)) or "taskkill exited 128", + ), redirect_stdout(buf): + result = smoke._run_parent("SH.000016", timeout=60) + + self.assertEqual(result, 1) + self.assertEqual(len(killed), 1) + self.assertEqual(killed[0][1], "nt" if os.name == "nt" else "posix") + event_line = next( + line for line in buf.getvalue().splitlines() + if line.startswith("E2E_EVENT ") + ) + payload = json.loads(event_line[len("E2E_EVENT "):]) + self.assertEqual(payload["phase"], "failed") + self.assertEqual(payload["target"], "SH.000016") + self.assertIn("interrupt cleanup failed", payload["error"]) + self.assertIn("taskkill exited 128", payload["error"]) + + def test_run_parent_keyboard_interrupt_post_cleanup_kill_failure_emits_failed(self): + """A post-cleanup ``proc.kill()`` failure on the Ctrl-C path must + merge into the cleanup error and emit a structured failed event + (return 1), never escape as an uncaught exception.""" + import io + from contextlib import redirect_stdout + + class _StuckInterruptingProc: + def __init__(self, args, **kwargs): + self.args = args + self.kwargs = kwargs + self._interrupted = False + + def poll(self): + return None + + def wait(self, timeout=None): + if not self._interrupted: + self._interrupted = True + raise KeyboardInterrupt() + raise subprocess.TimeoutExpired("worker", timeout or 1) + + def kill(self): + raise OSError("post-cleanup kill exploded") + + buf = io.StringIO() + with patch.object( + subprocess, "Popen", side_effect=_StuckInterruptingProc + ), patch.object( + smoke, "_terminate_process_tree", return_value=None + ), redirect_stdout(buf): + result = smoke._run_parent("SH.000016", timeout=60) + + self.assertEqual(result, 1) + event_line = next( + line for line in buf.getvalue().splitlines() + if line.startswith("E2E_EVENT ") + ) + payload = json.loads(event_line[len("E2E_EVENT "):]) + self.assertEqual(payload["phase"], "failed") + self.assertEqual(payload["target"], "SH.000016") + self.assertIn("interrupt cleanup failed", payload["error"]) + self.assertIn("post-cleanup kill failed", payload["error"]) + + def test_run_parent_keyboard_interrupt_cleanup_confirmed_via_kill_reraises(self): + """When tree cleanup succeeds but the post-cleanup wait times out and + the follow-up kill succeeds, cleanup is still confirmed: the parent + re-raises the interrupt instead of emitting a failed event.""" + import io + from contextlib import redirect_stdout + + class _KillableInterruptingProc: + def __init__(self, args, **kwargs): + self.args = args + self.kwargs = kwargs + self._interrupted = False + + def poll(self): + return None + + def wait(self, timeout=None): + if not self._interrupted: + self._interrupted = True + raise KeyboardInterrupt() + raise subprocess.TimeoutExpired("worker", timeout or 1) + + def kill(self): + pass + + buf = io.StringIO() + with patch.object( + subprocess, "Popen", side_effect=_KillableInterruptingProc + ), patch.object( + smoke, "_terminate_process_tree", return_value=None + ), redirect_stdout(buf): + with self.assertRaises(KeyboardInterrupt): + smoke._run_parent("SH.000016", timeout=60) + self.assertEqual(buf.getvalue(), "") + + def test_run_parent_keyboard_interrupt_skips_cleanup_when_worker_exited(self): + """A Ctrl-C arriving after the worker already exited must NOT run + tree cleanup (which would misreport a spurious failure on Windows + taskkill against a dead PID): the interrupt is propagated directly.""" + import io + from contextlib import redirect_stdout + + class _ExitedInterruptingProc: + def __init__(self, args, **kwargs): + self.args = args + self.kwargs = kwargs + self._interrupted = False + + def poll(self): + return 0 + + def wait(self, timeout=None): + if not self._interrupted: + self._interrupted = True + raise KeyboardInterrupt() + return 0 + + def kill(self): + pass + + buf = io.StringIO() + with patch.object( + subprocess, "Popen", side_effect=_ExitedInterruptingProc + ), patch.object( + smoke, "_terminate_process_tree", side_effect=AssertionError( + "cleanup must not run when the worker already exited" + ) + ), redirect_stdout(buf): + with self.assertRaises(KeyboardInterrupt): + smoke._run_parent("SH.000016", timeout=60) + self.assertEqual(buf.getvalue(), "") + + def test_run_parent_preserves_known_exit_codes(self): + for code in (0, 1, 124): + result, _ = self._patch_popen("SH.000016", exit_code=code) + self.assertEqual(result, code) + + def test_run_parent_rejects_non_positive_timeout_before_spawn(self): + """A non-positive timeout is a defensive input error: ``_run_parent`` + raises before any subprocess spawn.""" + def _fail_popen(*args, **kwargs): + raise AssertionError("Popen must not be called for non-positive timeout") + + with patch.object(subprocess, "Popen", side_effect=_fail_popen) as mock_popen: + with self.assertRaises(ValueError): + smoke._run_parent("SH.000016", timeout=0) + with self.assertRaises(ValueError): + smoke._run_parent("SH.000016", timeout=-5) + mock_popen.assert_not_called() + + def test_main_rejects_non_positive_timeout_without_running_branches(self): + """``main()`` rejects ``--timeout <= 0`` as an input error (exit 2) + without calling ``_run_parent`` or ``_run_worker`` (no spawn).""" + def _fail_worker(target): + raise AssertionError("_run_worker must not be called") + + def _fail_parent(target, timeout): + raise AssertionError("_run_parent must not be called") + + with patch.object(smoke, "_run_worker", side_effect=_fail_worker) as mock_worker, patch.object( + smoke, "_run_parent", side_effect=_fail_parent + ) as mock_parent: + result = smoke.main(["SH.000016", "--timeout", "0"]) + self.assertEqual(result, 2) + mock_worker.assert_not_called() + mock_parent.assert_not_called() + + +class TestWorkerCodeMismatch(unittest.TestCase): + """When the response carries a task identity but the extra stock code + does not match the authoritative matrix expectation, the worker must + emit an explicit mismatch error with expected and actual values — not + the successful response text.""" + + def _run_worker_with_response(self, response): + class _FakeDispatcher: + def __init__(self, *args, **kwargs): + pass + + def register(self, command): + pass + + async def dispatch_async(self, message): + return response + + with patch("bot.dispatcher.CommandDispatcher", side_effect=_FakeDispatcher), patch( + "src.services.task_service.get_task_service" + ) as mock_service: + buf = io.StringIO() + with redirect_stdout(buf): + exit_code = smoke._run_worker("SH.000016") + mock_service.assert_not_called() + return exit_code, buf.getvalue() + + def test_code_mismatch_emits_explicit_error_with_expected_and_actual(self): + from bot.models import BotResponse + + exit_code, output = self._run_worker_with_response( + BotResponse(text="ok", extra={"task_id": "t1", "stock_code": "csi930955"}) + ) + self.assertEqual(exit_code, 1) + self.assertNotIn("E2E_EVENT {\"phase\": \"completed\"", output) + event_line = next( + line for line in output.splitlines() + if line.startswith("E2E_EVENT ") + ) + payload = json.loads(event_line[len("E2E_EVENT "):]) + self.assertEqual(payload["phase"], "failed") + self.assertEqual(payload["target"], "SH.000016") + self.assertIn("submitted code mismatch", payload["error"]) + self.assertIn("expected 'sh000016'", payload["error"]) + self.assertIn("got 'csi930955'", payload["error"]) + # 结构化 ``stock_code`` 携带实际提交的 code（与 error 中的实际值一致）。 + self.assertEqual(payload["stock_code"], "csi930955") + # 不是成功响应文本。 + self.assertNotIn("分析任务已提交", payload["error"]) + + def test_code_mismatch_with_missing_stock_code_emits_null_structured_code(self): + """When the response has a task id but no ``extra.stock_code``, the + failed event still carries a structured ``stock_code`` (null) plus + the explicit mismatch error.""" + from bot.models import BotResponse + + exit_code, output = self._run_worker_with_response( + BotResponse(text="ok", extra={"task_id": "t1"}) + ) + self.assertEqual(exit_code, 1) + event_line = next( + line for line in output.splitlines() + if line.startswith("E2E_EVENT ") + ) + payload = json.loads(event_line[len("E2E_EVENT "):]) + self.assertEqual(payload["phase"], "failed") + self.assertIn("submitted code mismatch", payload["error"]) + self.assertIn("got None", payload["error"]) + self.assertIsNone(payload["stock_code"]) + + +class TestWorkerSuccessPath(unittest.TestCase): + """Offline integration-style success path: a matching ``BotResponse.extra`` + plus a fake in-process TaskService that reaches completed must produce + ordered ``submitted`` then ``completed`` events, poll the service, and + exit 0.""" + + def _run_worker_with_service(self, service, response=None): + from bot.models import BotResponse + + if response is None: + response = BotResponse( + text="ok", extra={"task_id": "t1", "stock_code": "sh000016"} + ) + + class _FakeDispatcher: + def __init__(self, *args, **kwargs): + pass + + def register(self, command): + pass + + async def dispatch_async(self, message): + return response + + with patch("bot.dispatcher.CommandDispatcher", side_effect=_FakeDispatcher), patch( + "src.services.task_service.get_task_service", return_value=service + ), patch("time.sleep"): + buf = io.StringIO() + with redirect_stdout(buf): + exit_code = smoke._run_worker("SH.000016") + return exit_code, buf.getvalue() + + def test_success_path_polls_service_and_emits_ordered_events(self): + class _FakeService: + def __init__(self): + self.polls = 0 + + def get_task_status(self, task_id): + self.polls += 1 + if self.polls == 1: + return None # 空初始状态：等待 + return { + "status": "completed", + "code": "sh000016", + "result": { + "code": "sh000016", + "name": "上证50", + "analysis_summary": "摘要", + "operation_advice": "持有", + "trend_prediction": "震荡", + }, + } + + service = _FakeService() + exit_code, output = self._run_worker_with_service(service) + self.assertEqual(exit_code, 0) + self.assertEqual(service.polls, 2) + events = [ + json.loads(line[len("E2E_EVENT "):]) + for line in output.splitlines() + if line.startswith("E2E_EVENT ") + ] + self.assertEqual([e["phase"] for e in events], ["submitted", "completed"]) + self.assertEqual(events[0]["task_id"], "t1") + self.assertEqual(events[0]["stock_code"], "sh000016") + self.assertEqual(events[1]["result"]["code"], "sh000016") + self.assertEqual(events[1]["result"]["name"], "上证50") + + def test_no_task_id_response_returns_1_with_failed_event(self): + """A response without a task identity (e.g. submission failure) must + return 1 with a failed event and never poll the service.""" + class _FakeService: + def get_task_status(self, task_id): + raise AssertionError("service must not be polled") + + from bot.models import BotResponse + + exit_code, output = self._run_worker_with_service( + _FakeService(), + response=BotResponse(text="❌ 错误：提交分析任务失败: boom"), + ) + self.assertEqual(exit_code, 1) + events = [ + json.loads(line[len("E2E_EVENT "):]) + for line in output.splitlines() + if line.startswith("E2E_EVENT ") + ] + self.assertEqual([e["phase"] for e in events], ["failed"]) + self.assertIn("提交分析任务失败", events[0]["error"]) + + +class TestWorkerExceptionContract(unittest.TestCase): + """Unexpected exceptions from dispatch / poll / event serialization must + produce a ``failed`` E2E event and exit 1 (with evidence on stderr), + never an uncaught traceback. ``KeyboardInterrupt`` is not treated as an + ordinary failure.""" + + def _run_worker_with_dispatcher(self, dispatcher): + with patch("bot.dispatcher.CommandDispatcher", side_effect=dispatcher): + buf = io.StringIO() + err = io.StringIO() + with redirect_stdout(buf), patch("sys.stderr", err): + exit_code = smoke._run_worker("SH.000016") + return exit_code, buf.getvalue(), err.getvalue() + + def test_dispatch_exception_emits_failed_event_and_exit_1(self): + class _BoomDispatcher: + def __init__(self, *args, **kwargs): + pass + + def register(self, command): + pass + + async def dispatch_async(self, message): + raise RuntimeError("dispatch exploded") + + exit_code, output, err = self._run_worker_with_dispatcher(_BoomDispatcher) + self.assertEqual(exit_code, 1) + event_line = next( + line for line in output.splitlines() + if line.startswith("E2E_EVENT ") + ) + payload = json.loads(event_line[len("E2E_EVENT "):]) + self.assertEqual(payload["phase"], "failed") + self.assertEqual(payload["target"], "SH.000016") + self.assertIn("worker exception", payload["error"]) + self.assertIn("dispatch exploded", payload["error"]) + self.assertIn("dispatch exploded", err) + + def test_poll_exception_emits_failed_event_and_exit_1(self): + from bot.models import BotResponse + + class _BoomService: + def get_task_status(self, task_id): + raise RuntimeError("poll exploded") + + class _FakeDispatcher: + def __init__(self, *args, **kwargs): + pass + + def register(self, command): + pass + + async def dispatch_async(self, message): + return BotResponse( + text="ok", extra={"task_id": "t1", "stock_code": "sh000016"} + ) + + with patch("bot.dispatcher.CommandDispatcher", side_effect=_FakeDispatcher), patch( + "src.services.task_service.get_task_service", return_value=_BoomService() + ), patch("time.sleep"): + buf = io.StringIO() + err = io.StringIO() + with redirect_stdout(buf), patch("sys.stderr", err): + exit_code = smoke._run_worker("SH.000016") + self.assertEqual(exit_code, 1) + events = [ + json.loads(line[len("E2E_EVENT "):]) + for line in buf.getvalue().splitlines() + if line.startswith("E2E_EVENT ") + ] + self.assertEqual([e["phase"] for e in events], ["submitted", "failed"]) + self.assertIn("poll exploded", events[1]["error"]) + self.assertIn("poll exploded", err.getvalue()) + + def test_keyboard_interrupt_is_not_swallowed(self): + class _KillDispatcher: + def __init__(self, *args, **kwargs): + pass + + def register(self, command): + pass + + async def dispatch_async(self, message): + raise KeyboardInterrupt() + + with patch("bot.dispatcher.CommandDispatcher", side_effect=_KillDispatcher): + with self.assertRaises(KeyboardInterrupt): + smoke._run_worker("SH.000016") + + +class TestResultCompleteness(unittest.TestCase): + """A completed task is only a success when the canonical code equals the + authoritative expected code, the name equals the registered display name, + and the three non-empty result fields (analysis_summary / + operation_advice / trend_prediction) are present.""" + + def _complete(self): + return { + "code": "sh000016", + "name": "上证50", + "analysis_summary": "摘要", + "operation_advice": "持有", + "trend_prediction": "震荡", + } + + def test_complete_result_has_no_error(self): + self.assertIsNone( + smoke._assert_complete_result(self._complete(), "sh000016", "上证50") + ) + + def test_missing_code_is_failure(self): + result = self._complete() + del result["code"] + self.assertIn("code", smoke._assert_complete_result(result, "sh000016", "上证50") or "") + + def test_wrong_code_is_failure(self): + result = self._complete() + result["code"] = "csi930955" + self.assertIn("code", smoke._assert_complete_result(result, "sh000016", "上证50") or "") + + def test_missing_name_is_failure(self): + result = self._complete() + del result["name"] + self.assertIn("name", smoke._assert_complete_result(result, "sh000016", "上证50") or "") + + def test_wrong_name_is_failure(self): + result = self._complete() + result["name"] = "红利低波100" + self.assertIn("name", smoke._assert_complete_result(result, "sh000016", "上证50") or "") + + def test_missing_analysis_summary_is_failure(self): + result = self._complete() + del result["analysis_summary"] + self.assertIn( + "analysis_summary", + smoke._assert_complete_result(result, "sh000016", "上证50") or "", + ) + + def test_missing_operation_advice_is_failure(self): + result = self._complete() + del result["operation_advice"] + self.assertIn( + "operation_advice", + smoke._assert_complete_result(result, "sh000016", "上证50") or "", + ) + + def test_missing_trend_prediction_is_failure(self): + result = self._complete() + del result["trend_prediction"] + self.assertIn( + "trend_prediction", + smoke._assert_complete_result(result, "sh000016", "上证50") or "", + ) + + def test_empty_operation_advice_is_failure(self): + result = self._complete() + result["operation_advice"] = "" + self.assertIn( + "operation_advice", + smoke._assert_complete_result(result, "sh000016", "上证50") or "", + ) + + def test_whitespace_only_analysis_summary_is_failure(self): + result = self._complete() + result["analysis_summary"] = " \t " + self.assertIn( + "analysis_summary", + smoke._assert_complete_result(result, "sh000016", "上证50") or "", + ) + + def test_whitespace_only_operation_advice_is_failure(self): + result = self._complete() + result["operation_advice"] = " \n " + self.assertIn( + "operation_advice", + smoke._assert_complete_result(result, "sh000016", "上证50") or "", + ) + + def test_whitespace_only_trend_prediction_is_failure(self): + result = self._complete() + result["trend_prediction"] = " " + self.assertIn( + "trend_prediction", + smoke._assert_complete_result(result, "sh000016", "上证50") or "", + ) + + def test_sentiment_score_is_not_required(self): + """``sentiment_score`` is not part of the spec's required fields.""" + result = self._complete() + self.assertNotIn("sentiment_score", result) + self.assertIsNone( + smoke._assert_complete_result(result, "sh000016", "上证50") + ) + + +class TestPollOnce(unittest.TestCase): + """Empty initial state must wait (None), then surface exactly one final + phase event: completed / failed.""" + + def _service(self, status_payload): + class _FakeService: + def get_task_status(self, task_id): + return status_payload + + return _FakeService() + + def test_empty_initial_state_returns_none(self): + self.assertIsNone( + smoke._poll_once( + self._service(None), "t1", "SH.000016", "sh000016" + ) + ) + + def test_running_state_returns_none(self): + self.assertIsNone( + smoke._poll_once( + self._service({"status": "running"}), + "t1", + "SH.000016", + "sh000016", + ) + ) + + def test_completed_state_emits_completed_event(self): + event = smoke._poll_once( + self._service( + { + "status": "completed", + "code": "sh000016", + "result": { + "code": "sh000016", + "name": "上证50", + "analysis_summary": "摘要", + "operation_advice": "持有", + "trend_prediction": "震荡", + }, + } + ), + "t1", + "SH.000016", + "sh000016", + ) + self.assertEqual(event["phase"], "completed") + self.assertEqual(event["target"], "SH.000016") + self.assertEqual(event["task_id"], "t1") + self.assertEqual(event["stock_code"], "sh000016") + self.assertEqual(event["result"]["code"], "sh000016") + + def test_completed_with_wrong_code_emits_failed_event(self): + """Dispatcher routed the wrong code: the completed result carries a + different canonical, which must fail against the matrix expectation.""" + event = smoke._poll_once( + self._service( + { + "status": "completed", + "code": "csi930955", + "result": { + "code": "csi930955", + "name": "红利低波100", + "analysis_summary": "摘要", + "operation_advice": "持有", + "trend_prediction": "震荡", + }, + } + ), + "t1", + "SH.000016", + "sh000016", + ) + self.assertEqual(event["phase"], "failed") + self.assertIn("code", event.get("error", "")) + + def test_completed_with_wrong_name_emits_failed_event(self): + """The pipeline reported the wrong display name: must fail.""" + event = smoke._poll_once( + self._service( + { + "status": "completed", + "code": "sh000016", + "result": { + "code": "sh000016", + "name": "红利低波100", + "analysis_summary": "摘要", + "operation_advice": "持有", + "trend_prediction": "震荡", + }, + } + ), + "t1", + "SH.000016", + "sh000016", + ) + self.assertEqual(event["phase"], "failed") + self.assertIn("name", event.get("error", "")) + + def test_completed_without_analysis_summary_emits_failed_event(self): + event = smoke._poll_once( + self._service( + { + "status": "completed", + "code": "sh000016", + "result": { + "code": "sh000016", + "name": "上证50", + "operation_advice": "持有", + "trend_prediction": "震荡", + }, + } + ), + "t1", + "SH.000016", + "sh000016", + ) + self.assertEqual(event["phase"], "failed") + self.assertIn("analysis_summary", event.get("error", "")) + + def test_failed_state_emits_failed_event(self): + event = smoke._poll_once( + self._service({"status": "failed", "error": "数据源不可用"}), + "t1", + "SH.000016", + "sh000016", + ) + self.assertEqual(event["phase"], "failed") + self.assertIn("数据源不可用", event.get("error", "")) + + def test_completed_but_incomplete_result_emits_failed_event(self): + event = smoke._poll_once( + self._service( + { + "status": "completed", + "code": "sh000016", + "result": {"code": "sh000016", "name": "上证50"}, + } + ), + "t1", + "SH.000016", + "sh000016", + ) + self.assertEqual(event["phase"], "failed") + self.assertIn("analysis_summary", event.get("error", "")) + + +class TestEventEmit(unittest.TestCase): + """Events are single-line ``E2E_EVENT {json}`` on stdout.""" + + def test_emit_event_is_single_line_json(self): + import io + from contextlib import redirect_stdout + + buf = io.StringIO() + with redirect_stdout(buf): + smoke._emit_event("submitted", "SH.000016", task_id="t1") + line = buf.getvalue().strip() + self.assertTrue(line.startswith("E2E_EVENT ")) + payload = json.loads(line[len("E2E_EVENT "):]) + self.assertEqual(payload["phase"], "submitted") + self.assertEqual(payload["target"], "SH.000016") + self.assertEqual(payload["task_id"], "t1") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_stock_list_parser.py b/tests/test_stock_list_parser.py index 03dfaf73e..1cfb214a9 100644 --- a/tests/test_stock_list_parser.py +++ b/tests/test_stock_list_parser.py @@ -429,6 +429,60 @@ class TestDefaultIndexRegistry: target = parse_analysis_target("沪深300") assert target.asset_type != ParseStatus.INDEX + def test_find_by_display_name_resolves_registered_name(self) -> None: + """PR2: exact registered display names resolve through the dedicated + name lookup — independent of ``find_by_explicit_key``.""" + registry = default_index_registry() + entry = registry.find_by_display_name("上证50") + assert entry is not None + assert entry.canonical_id == "sh000016" + entry2 = registry.find_by_display_name("沪深300") + assert entry2 is not None + assert entry2.canonical_id == "sh000300" + + def test_find_by_display_name_is_nfkc_trim_casefold_exact(self) -> None: + """PR2: name lookup normalizes NFKC + trim + casefold before the exact + match, but never falls back to fuzzy matching.""" + registry = default_index_registry() + # Full-width / whitespace variants of a registered name still match. + assert registry.find_by_display_name("　上证50　") is not None + assert registry.find_by_display_name("上证５０") is not None + # Unknown names return None (no fuzzy fallback). + assert registry.find_by_display_name("不存在标的") is None + assert registry.find_by_display_name("上证5") is None + + def test_find_by_display_name_does_not_leak_into_identity_alias(self) -> None: + """PR2: a display-name hit must not make the name resolvable as an + identity alias — the parser contract stays untouched.""" + registry = default_index_registry() + assert registry.find_by_display_name("上证50") is not None + assert registry.find_by_explicit_key("上证50") is None + target = parse_analysis_target("上证50") + assert target.asset_type != ParseStatus.INDEX + + def test_ambiguous_display_name_is_reported(self) -> None: + """PR2: two entries with NFKC/casefold-equivalent display names make + the name lookup ambiguous — no guessing, no single winner.""" + registry = IndexRegistry(( + IndexEntry( + bare_code="000300", + exchange="SH", + canonical_id="sh000300", + display_name="沪深300", + ), + IndexEntry( + bare_code="000999", + exchange="SH", + canonical_id="sh000999", + display_name="沪深300", + ), + )) + assert registry.is_ambiguous_display_name("沪深300") is True + assert registry.find_by_display_name("沪深300") is None + # Explicit identity still resolves deterministically. + assert registry.find_by_explicit_key("sh000300") is not None + assert registry.find_by_explicit_key("sh000999") is not None + def test_custom_registry_rejects_text_identity_alias(self) -> None: with pytest.raises(ValueError, match="explicit code form"): IndexRegistry(( @@ -441,6 +495,40 @@ class TestDefaultIndexRegistry: ), )) + def test_code_shaped_display_names_stay_distinct_under_exact_name_lookup( + self, + ) -> None: + """PR2 review fix: display-name equality is exactly NFKC + trim + + casefold. Code-shaped names (``000016.SH`` vs ``sh000016``) that + identity normalization would falsely collapse (``_normalize_index_key`` + treats them as the same resolver key ``sh000016``) must stay distinct + for the name lookup, so each exact query resolves its own entry and + neither is reported ambiguous.""" + registry = IndexRegistry(( + IndexEntry( + bare_code="000016", + exchange="SH", + canonical_id="sh000016", + display_name="000016.SH", + ), + IndexEntry( + bare_code="000999", + exchange="SH", + canonical_id="sh000999", + display_name="sh000016", + ), + )) + # Identity normalization collapses both names to ``sh000016``, but the + # name lookup must NOT reuse it — each display name resolves exactly. + entry = registry.find_by_display_name("000016.SH") + assert entry is not None + assert entry.display_name == "000016.SH" + entry = registry.find_by_display_name("sh000016") + assert entry is not None + assert entry.display_name == "sh000016" + assert registry.is_ambiguous_display_name("000016.SH") is False + assert registry.is_ambiguous_display_name("sh000016") is False + # --------------------------------------------------------------------------- # Batch parsing helper. diff --git a/tests/test_task_service.py b/tests/test_task_service.py index 7053394e3..f33d93df5 100644 --- a/tests/test_task_service.py +++ b/tests/test_task_service.py @@ -43,6 +43,19 @@ class _FakePipeline: return _make_failed_result(kwargs["code"]) +class _CapturingPipeline: + """Records the final ``process_single_stock`` kwargs for target assertions.""" + + calls = [] # class-level so tests can assert via the fake module attribute + + def __init__(self, *args, **kwargs): + pass + + def process_single_stock(self, *args, **kwargs): + type(self).calls.append(kwargs) + return _make_failed_result(kwargs["code"]) + + class TestTaskService(unittest.TestCase): def test_run_analysis_marks_failed_for_unsuccessful_result(self): service = TaskService() @@ -116,6 +129,94 @@ class TestTaskService(unittest.TestCase): # executor.submit(self._run_analysis, code, task_id, ...) — code is arg[1] self.assertEqual(captured["args"][1], "csi930955") + def test_submit_analysis_with_index_target_skips_resolver_and_keeps_canonical(self): + """PR2: when an INDEX ``AnalysisTarget`` is supplied, ``submit_analysis`` + must use ``target.canonical_id`` verbatim and skip the stock-code + resolver — otherwise ``sh000016`` would be rewritten to ``SH000016``.""" + from src.services.stock_list_parser import AnalysisTarget, ParseStatus + + service = TaskService() + service._tasks = {} + service._tasks_lock = threading.Lock() + captured = {} + + executor = MagicMock() + + def capture_submit(*args, **kwargs): + captured["args"] = args + return "future" + + executor.submit.side_effect = capture_submit + service._executor = executor + + target = AnalysisTarget( + raw_input="sh000016", + asset_type=ParseStatus.INDEX, + canonical_id="sh000016", + display_code="上证50", + exchange="SH", + ) + + with patch("src.services.task_service.resolve_index_stock_code_for_analysis") as mock_resolve: + result = service.submit_analysis( + "sh000016", report_type="simple", query_source="cli", + analysis_target=target, + ) + + mock_resolve.assert_not_called() + self.assertEqual(result["code"], "sh000016") + self.assertIn("args", captured) + # executor.submit(self._run_analysis, code, task_id, ...) — code is arg[1] + self.assertEqual(captured["args"][1], "sh000016") + # analysis_target is the last positional arg handed to _run_analysis + self.assertIs(captured["args"][-1], target) + + def test_submit_analysis_rejects_non_index_target(self): + from src.services.stock_list_parser import parse_analysis_target + + service = TaskService() + service._executor = MagicMock() + target = parse_analysis_target("600519") + + with self.assertRaisesRegex(ValueError, "must be an INDEX target"): + service.submit_analysis("600519", analysis_target=target) + service._executor.submit.assert_not_called() + + def test_run_analysis_passes_index_target_to_pipeline(self): + """PR2: the background task must forward the same ``analysis_target`` + to ``process_single_stock`` so the pipeline receives canonical code and + the structured INDEX target together.""" + from src.services.stock_list_parser import AnalysisTarget, ParseStatus + + service = TaskService() + service._tasks = {} + service._tasks_lock = threading.Lock() + + fake_main = ModuleType("main") + fake_main.StockAnalysisPipeline = _CapturingPipeline + + target = AnalysisTarget( + raw_input="930955.CSI", + asset_type=ParseStatus.INDEX, + canonical_id="csi930955", + display_code="红利低波100", + exchange="CSI", + ) + + _CapturingPipeline.calls = [] + with patch.dict("sys.modules", {"main": fake_main}), patch( + "src.config.get_config", return_value=SimpleNamespace() + ): + result = service._run_analysis( + code="csi930955", task_id="task-index-1", analysis_target=target + ) + + self.assertFalse(result["success"]) # _CapturingPipeline returns failure + self.assertEqual(len(_CapturingPipeline.calls), 1) + call = _CapturingPipeline.calls[0] + self.assertEqual(call["code"], "csi930955") + self.assertIs(call["analysis_target"], target) + if __name__ == "__main__": import unittest