diff --git a/.env.example b/.env.example index b7470e61f..f1653df44 100644 --- a/.env.example +++ b/.env.example @@ -826,6 +826,8 @@ SCHEDULE_TIME=18:00 SCHEDULE_TIMES= # Web/API runtime scheduler 单次分析硬超时（秒，最小 60 秒） DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS=2700 +# 硬超时后是否推送已落库个股的部分完成通知（true/false；默认 true，未设置即开启，此前超时不推送） +DSA_TIMEOUT_PARTIAL_NOTIFY=true # 定时模式启动时是否立即执行一次分析（true/false） # 若未显式设置，定时模式会沿用 RUN_IMMEDIATELY 的运行时覆盖语义以兼容旧配置 SCHEDULE_RUN_IMMEDIATELY=true diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 1fda984c5..0141e06cc 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +- [新功能] Web/API runtime scheduler 硬超时后扫描已落库分析历史，**默认发送**部分完成通知（`DSA_TIMEOUT_PARTIAL_NOTIFY` 未设置或为 true；此前超时不推送已落库个股），并在 `last_error` 中记录 `completed/pending` 摘要；可用 `DSA_TIMEOUT_PARTIAL_NOTIFY=false` 关闭推送（Refs #2328）。 - [测试] 修复股票名称解析冷启动超时并发测试的同步竞态：在放行后台抓取前确认两个等待者均已结束并返回空结果，避免 Docker 发布门禁偶发失败。 - [文档] 将仓库内所有 SerpApi 链接统一更新为新的赞助转化追踪地址。 - [修复] 智能导入兼容带 UTF-8 BOM 的 CSV 与剪贴板文本，避免 `code` 表头被误当成数据并丢失有效股票代码。 diff --git a/docs/full-guide.md b/docs/full-guide.md index 071c01ac4..d5a0795a1 100644 --- a/docs/full-guide.md +++ b/docs/full-guide.md @@ -496,6 +496,7 @@ daily_stock_analysis/ | `SCHEDULE_TIME` | 定时执行时间 | `18:00` | | `SCHEDULE_TIMES` | 多个定时执行时间，逗号分隔；为空时使用 `SCHEDULE_TIME` | 空 | | `DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS` | Web/API runtime scheduler 单次分析硬超时（秒，最小 60 秒）；超时后终止独立分析进程，不阻塞后续任务 | `2700` | +| `DSA_TIMEOUT_PARTIAL_NOTIFY` | **默认开启（行为变化）**：硬超时后对已落库个股发送「部分完成」通知；此前超时不推送。`false` 时仍写入结构化 `last_error`，但不推送 | `true` | | `LOG_DIR` | 日志目录 | `./logs` | | `SAVE_CONTEXT_SNAPSHOT` | 保存分析历史 `context_snapshot`；设为 `false` 时新历史不保存 enhanced_context、market_phase_summary、AnalysisContextPack overview 或诊断快照，但不关闭当次 Prompt 低敏摘要 | `true` | @@ -890,10 +891,13 @@ python main.py --schedule --no-run-immediately | `SCHEDULE_TIME` | 每日执行时间 (HH:MM) | `18:00` | `09:30` | | `SCHEDULE_TIMES` | 多个每日执行时间，逗号分隔；为空时使用 `SCHEDULE_TIME` | 空 | `09:20,12:30,15:10,18:00` | | `DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS` | Web/API runtime scheduler 单次分析硬超时（秒，最小 60 秒） | `2700` | `3600` | +| `DSA_TIMEOUT_PARTIAL_NOTIFY` | **默认开启（行为变化）**：硬超时后是否推送已落库个股的部分完成通知；此前超时不推送 | `true` | `false` | | `SCHEDULE_RUN_IMMEDIATELY` | 定时模式启动时是否立即运行一次；未显式设置时沿用 `RUN_IMMEDIATELY` 的运行时覆盖语义 | `true` | `false` | | `RUN_IMMEDIATELY` | 非定时模式启动时是否立即运行一次；同时作为未显式设置 `SCHEDULE_RUN_IMMEDIATELY` 时的 legacy 回退 | `true` | `false` | | `TRADING_DAY_CHECK_ENABLED` | 交易日检查：非交易日跳过执行；设为 `false` 可强制执行 | `true` | `false` | +> 超时部分完成排障：通知渠道异常只记 warning（`Partial timeout notification failed`）并跳过推送，**不抛出、不占用** `status().running`（notify 在分析锁释放后的后台线程）。扫库 / `src.storage` 导入失败同样 fail-open：API 上可能仍是 baseline `timed out after Ns`，或被 enrich 成 `completed=0`（与「确实没有落库」无法区分）。运维请搜 `Failed to collect completed analyses after timeout`、`Timeout partial delivery failed open`、`Partial timeout notification failed`。合入后首次发布请关注通知量，避免默认开启造成骚扰；可设 `DSA_TIMEOUT_PARTIAL_NOTIFY=false` 关闭推送。 + 例如在 Docker 中配置： ```bash diff --git a/docs/full-guide_EN.md b/docs/full-guide_EN.md index 43e5c93c0..12e2e0ea7 100644 --- a/docs/full-guide_EN.md +++ b/docs/full-guide_EN.md @@ -417,11 +417,14 @@ For the notification baseline, diagnostics, and deployment notes, see [Notificat | `SCHEDULE_TIME` | Scheduled execution time | `18:00` | | `SCHEDULE_TIMES` | Multiple scheduled execution times, comma-separated; falls back to `SCHEDULE_TIME` when empty | empty | | `DSA_RUNTIME_SCHEDULER_TIMEOUT_SECONDS` | Hard timeout in seconds for each Web/API runtime scheduler analysis (minimum 60); the isolated analysis process is terminated on timeout so later runs can continue | `2700` | +| `DSA_TIMEOUT_PARTIAL_NOTIFY` | **Default on (behavior change):** after a hard timeout, send a partial notification for analyses already saved to history; previously timeout sent no per-stock push. When `false`, structured `last_error` is still recorded but no push is sent | `true` | | `SCHEDULE_RUN_IMMEDIATELY` | Run once immediately when scheduler mode starts; when unset it keeps following the legacy `RUN_IMMEDIATELY` runtime override | `true` | | `RUN_IMMEDIATELY` | Run once immediately for non-scheduler startup; also acts as the legacy fallback when `SCHEDULE_RUN_IMMEDIATELY` is unset | `true` | | `LOG_DIR` | Log directory | `./logs` | | `SAVE_CONTEXT_SNAPSHOT` | Persist analysis-history `context_snapshot`. When false, new history records do not save enhanced_context, market_phase_summary, AnalysisContextPack overview, or diagnostic snapshots, but current-run prompt summaries remain enabled | `true` | +> Timeout partial-delivery ops: a notification-channel exception is logged (`Partial timeout notification failed`) and skipped; it is not re-raised and cannot keep `status().running` true (notify runs on a daemon thread after the analysis lock is released). Storage import / collect failures also fail open: `status().last_error` may stay at the baseline `timed out after Ns` string, or be enriched to `completed=0`, which is indistinguishable from “nothing was saved”. Grep `Failed to collect completed analyses after timeout`, `Timeout partial delivery failed open`, and `Partial timeout notification failed`. After the first release, watch notification volume; set `DSA_TIMEOUT_PARTIAL_NOTIFY=false` to disable pushes. + > Behavior notes: > - When `TICKFLOW_API_KEY` is configured, TickFlow is instantiated as an optional A-share daily K-line data source and CN market-review enhancer. `TICKFLOW_PRIORITY` only affects the generic A-share daily K-line/provider fallback chain. Realtime quote priority is controlled separately by `REALTIME_SOURCE_PRIORITY`; TickFlow realtime quotes are used only when that list explicitly includes `tickflow`, and any source listed before `tickflow` is tried first. > - All SH/SZ indices currently registered in `IndexRegistry` (the `sh`/`sz`/`csi` prefixes; see the [Index watchlist configuration](#index-watchlist-configuration) registry notes and `scripts/stock_index_seeds/index_registry.csv` for the full list) bypass generic priority sorting when given as explicit-market inputs (exchange-suffix forms such as `000016.SH` are also accepted) and use the fixed Tencent → AkShare → TickFlow → YFinance fallback chain; unconfigured or unavailable providers are skipped. Bare `000016`-style inputs remain stocks and do not enter the index chain. This fixed chain ignores `EFINANCE_PRIORITY`, `AKSHARE_PRIORITY`, `TUSHARE_PRIORITY`, `TICKFLOW_PRIORITY`, `PYTDX_PRIORITY`, `BAOSTOCK_PRIORITY`, `YFINANCE_PRIORITY`, and `TENCENT_PRIORITY`. Existing stock and realtime-quote ordering is unchanged. diff --git a/src/core/config_registry.py b/src/core/config_registry.py index fdf803a48..d651371fb 100644 --- a/src/core/config_registry.py +++ b/src/core/config_registry.py @@ -3119,6 +3119,35 @@ _FIELD_DEFINITIONS: Dict[str, Dict[str, Any]] = { ], "warning_codes": [], }, + "DSA_TIMEOUT_PARTIAL_NOTIFY": { + "title": "Timeout Partial Notification", + "description": ( + "After a Web/API runtime scheduler hard timeout, send a partial " + "notification for analyses already saved to history." + ), + "category": "system", + "data_type": "boolean", + "ui_control": "switch", + "is_sensitive": False, + "is_required": False, + "is_editable": True, + "default_value": "true", + "options": [], + "validation": {}, + "display_order": 13, + "help_key": "settings.system.schedule", + "examples": [ + "DSA_TIMEOUT_PARTIAL_NOTIFY=true", + "DSA_TIMEOUT_PARTIAL_NOTIFY=false", + ], + "docs": [ + { + "label": "Full guide: configuration", + "href": "https://github.com/ZhuLinsen/daily_stock_analysis/blob/main/docs/full-guide.md#其他配置", + }, + ], + "warning_codes": [], + }, "HTTP_PROXY": { "title": "HTTP Proxy", "description": "Optional HTTP proxy endpoint.", diff --git a/src/services/analysis_timeout_partial.py b/src/services/analysis_timeout_partial.py new file mode 100644 index 000000000..250456b5a --- /dev/null +++ b/src/services/analysis_timeout_partial.py @@ -0,0 +1,361 @@ +# -*- coding: utf-8 -*- +"""Timeout partial delivery helpers for RuntimeSchedulerService. + +When the Web/API runtime scheduler hard-timeout kills a worker, successful +per-stock analyses may already be persisted. This module collects those rows +and optionally sends a partial notification with richer diagnostics. +""" + +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Dict, List, Optional, Sequence + +from sqlalchemy import and_, desc, select + +logger = logging.getLogger(__name__) + +TIMEOUT_PARTIAL_NOTIFY_ENV = "DSA_TIMEOUT_PARTIAL_NOTIFY" +_FALSEY = {"0", "false", "no", "off"} +_TRUTHY = {"1", "true", "yes", "on"} + + +@dataclass(frozen=True) +class CompletedAnalysisSummary: + """One stock analysis already saved before the timeout kill.""" + + code: str + name: Optional[str] = None + operation_advice: Optional[str] = None + sentiment_score: Optional[int] = None + created_at: Optional[datetime] = None + history_id: Optional[int] = None + + +@dataclass(frozen=True) +class TimeoutPartialOutcome: + """Result of timeout partial collection + optional notify.""" + + completed: List[CompletedAnalysisSummary] + pending_codes: List[str] + notified: bool + notify_skipped_reason: Optional[str] + error_message: str + + +def is_timeout_partial_notify_enabled() -> bool: + """Return whether ``DSA_TIMEOUT_PARTIAL_NOTIFY`` allows timeout notify. + + Default: enabled (true) when unset. Accepts common truthy/falsey strings. + """ + raw = os.getenv(TIMEOUT_PARTIAL_NOTIFY_ENV) + if raw is None or not str(raw).strip(): + return True + normalized = str(raw).strip().lower() + if normalized in _FALSEY: + return False + if normalized in _TRUTHY: + return True + logger.warning( + "Invalid %s=%r; treating as enabled", + TIMEOUT_PARTIAL_NOTIFY_ENV, + raw, + ) + return True + + +def resolve_expected_stock_codes( + stock_codes: Optional[Sequence[str]], + *, + config: Optional[Any] = None, +) -> List[str]: + """Resolve the stock universe for this run. + + When ``stock_codes`` is provided, normalize and return it. Otherwise fall + back to ``config.stock_list`` (caller may pass a loaded Config). + """ + if stock_codes is not None: + return _normalize_codes(stock_codes) + + if config is not None: + stock_list = getattr(config, "stock_list", None) or [] + return _normalize_codes(stock_list) + + return [] + + +def collect_completed_analyses_since( + *, + run_started_at: datetime, + expected_codes: Sequence[str], + db: Optional[Any] = None, +) -> List[CompletedAnalysisSummary]: + """Load successful analysis_history rows created at/after run start. + + Only codes in ``expected_codes`` are considered when that list is non-empty. + When the same code has multiple rows in the window, keep the latest. + Fail-open: DB/import errors return an empty list after logging a warning. + Callers cannot tell "no rows yet" from "collect failed"; operators must + grep ``Failed to collect completed analyses after timeout``. ``status()`` + may then show ``completed=0`` or keep the baseline timeout ``last_error``. + """ + try: + rows = _query_history_rows_since( + run_started_at=run_started_at, + expected_codes=expected_codes, + db=db, + ) + return _summaries_from_history_rows(rows, expected_codes) + except Exception as exc: # noqa: BLE001 - timeout path must stay fail-open + logger.warning("Failed to collect completed analyses after timeout: %s", exc) + return [] + + +def _query_history_rows_since( + *, + run_started_at: datetime, + expected_codes: Sequence[str], + db: Optional[Any] = None, +) -> List[Any]: + """Fetch raw analysis_history ORM rows for the timeout window.""" + storage_mod = _resolve_storage_module() + database = db if db is not None else storage_mod.get_db() + AnalysisHistory = storage_mod.AnalysisHistory + codes = _normalize_codes(expected_codes) + conditions = [ + AnalysisHistory.created_at >= run_started_at, + AnalysisHistory.report_type != "market_review", + ] + if codes: + conditions.append(AnalysisHistory.code.in_(codes)) + + with database.get_session() as session: + return list( + session.execute( + select(AnalysisHistory) + .where(and_(*conditions)) + .order_by(desc(AnalysisHistory.created_at), desc(AnalysisHistory.id)) + .limit(500) + ).scalars().all() + ) + + +def _resolve_storage_module() -> Any: + """Load ``src.storage``, cleaning up broken partial imports on failure. + + On import failure, drop ``src.storage`` / ``src.storage.*`` from + ``sys.modules`` and raise. The timeout collect path fail-opens to an + empty list; this is not surfaced on ``status().last_error`` beyond a + possible ``completed=0`` enrich. Watch the warning log, not the API. + """ + import importlib + import sys + + existing = sys.modules.get("src.storage") + if existing is not None: + return existing + try: + return importlib.import_module("src.storage") + except Exception as exc: + for name in list(sys.modules): + if name == "src.storage" or name.startswith("src.storage."): + sys.modules.pop(name, None) + raise RuntimeError("storage unavailable for timeout partial collect") from exc + + +def _summaries_from_history_rows( + rows: Sequence[Any], + expected_codes: Sequence[str], +) -> List[CompletedAnalysisSummary]: + """Convert ORM/history-like rows into latest-per-code summaries.""" + codes = _normalize_codes(expected_codes) + latest_by_code: Dict[str, CompletedAnalysisSummary] = {} + for row in rows: + code = str(getattr(row, "code", "") or "").strip() + if not code or code in latest_by_code: + continue + if codes and code not in set(codes): + continue + latest_by_code[code] = CompletedAnalysisSummary( + code=code, + name=getattr(row, "name", None), + operation_advice=getattr(row, "operation_advice", None), + sentiment_score=getattr(row, "sentiment_score", None), + created_at=getattr(row, "created_at", None), + history_id=getattr(row, "id", None), + ) + + if codes: + return [latest_by_code[code] for code in codes if code in latest_by_code] + return list(latest_by_code.values()) + + +def format_timeout_error_message( + *, + timeout_seconds: int, + completed: Sequence[CompletedAnalysisSummary], + pending_codes: Sequence[str], +) -> str: + """Build structured ``last_error`` text for RuntimeSchedulerService.status().""" + completed_codes = [item.code for item in completed] + pending = [str(code).strip() for code in pending_codes if str(code).strip()] + parts = [ + f"runtime scheduled analysis timed out after {timeout_seconds}s", + f"completed={len(completed_codes)}", + f"pending={len(pending)}", + ] + if completed_codes: + parts.append(f"completed_codes={','.join(completed_codes)}") + if pending: + parts.append(f"pending_codes={','.join(pending)}") + return "; ".join(parts) + + +def build_partial_timeout_report( + *, + timeout_seconds: int, + completed: Sequence[CompletedAnalysisSummary], + pending_codes: Sequence[str], +) -> str: + """Build Markdown body for the partial-success timeout notification.""" + lines = [ + "## 定时分析超时（部分完成）", + "", + f"- 硬超时：`{timeout_seconds}` 秒", + f"- 已完成：`{len(completed)}` 只", + f"- 未完成：`{len(list(pending_codes))}` 只", + "", + ] + if completed: + lines.append("### 已落库结果") + for item in completed: + name = item.name or "-" + advice = item.operation_advice or "-" + score = item.sentiment_score if item.sentiment_score is not None else "-" + lines.append(f"- `{item.code}` {name}｜建议：{advice}｜评分：{score}") + lines.append("") + if pending_codes: + lines.append("### 未完成") + lines.append(", ".join(f"`{code}`" for code in pending_codes)) + lines.append("") + lines.append( + "_本轮因 runtime scheduler 硬超时被终止；已完成个股结果已写入分析历史，" + "未完成部分可稍后手动重跑。_" + ) + return "\n".join(lines) + + +def send_partial_timeout_notification( + report: str, + *, + completed_codes: Sequence[str], + no_notify: bool = False, +) -> bool: + """Send ``report`` via NotificationService unless ``no_notify`` is set. + + Returns True when a send was attempted and succeeded. + + Channel/import exceptions are swallowed: this function logs a warning + (``Partial timeout notification failed``) and returns False. It does not + re-raise, so it cannot keep ``RuntimeSchedulerService`` ``_run_lock`` + held or flip ``status().running`` — notify runs on a daemon thread after + that lock is already released. Structured ``last_error`` is still returned + by ``handle_runtime_analysis_timeout`` with ``notify_skipped_reason=send_failed``. + """ + if no_notify: + return False + if not is_timeout_partial_notify_enabled(): + return False + if not report or not str(report).strip(): + return False + + try: + from src.notification import NotificationService + + service = NotificationService() + return bool( + service.send( + report, + email_stock_codes=list(completed_codes) or None, + route_type="system_error", + severity="warning", + dedup_key=f"runtime_timeout_partial:{','.join(completed_codes)}", + ) + ) + except Exception as exc: # noqa: BLE001 - never break scheduler on notify failure + logger.warning("Partial timeout notification failed: %s", exc) + return False + + +def handle_runtime_analysis_timeout( + *, + timeout_seconds: int, + run_started_at: datetime, + stock_codes: Optional[Sequence[str]] = None, + no_notify: bool = False, + config: Optional[Any] = None, + db: Optional[Any] = None, +) -> TimeoutPartialOutcome: + """Orchestrate collect → format error → optional partial notify. + + Intended call site: ``RuntimeSchedulerService._run_analysis_with_watchdog`` + timeout branch, after the worker process tree is terminated. + """ + expected = resolve_expected_stock_codes(stock_codes, config=config) + completed = collect_completed_analyses_since( + run_started_at=run_started_at, + expected_codes=expected, + db=db, + ) + completed_code_set = {item.code for item in completed} + pending_codes = [code for code in expected if code not in completed_code_set] + error_message = format_timeout_error_message( + timeout_seconds=timeout_seconds, + completed=completed, + pending_codes=pending_codes, + ) + + notified = False + notify_skipped_reason: Optional[str] = None + if no_notify: + notify_skipped_reason = "no_notify" + elif not is_timeout_partial_notify_enabled(): + notify_skipped_reason = "env_disabled" + elif not completed: + notify_skipped_reason = "no_completed_results" + else: + report = build_partial_timeout_report( + timeout_seconds=timeout_seconds, + completed=completed, + pending_codes=pending_codes, + ) + notified = send_partial_timeout_notification( + report, + completed_codes=[item.code for item in completed], + no_notify=False, + ) + if not notified: + notify_skipped_reason = "send_failed" + + return TimeoutPartialOutcome( + completed=list(completed), + pending_codes=pending_codes, + notified=notified, + notify_skipped_reason=notify_skipped_reason, + error_message=error_message, + ) + + +def _normalize_codes(codes: Sequence[Any]) -> List[str]: + seen = set() + ordered: List[str] = [] + for raw in codes: + code = str(raw or "").strip() + if not code or code in seen: + continue + seen.add(code) + ordered.append(code) + return ordered diff --git a/src/services/runtime_scheduler.py b/src/services/runtime_scheduler.py index d7207bc06..847b4ea2d 100644 --- a/src/services/runtime_scheduler.py +++ b/src/services/runtime_scheduler.py @@ -359,6 +359,107 @@ class RuntimeSchedulerService: ) return DEFAULT_RUNTIME_SCHEDULER_TIMEOUT_SECONDS + def _build_timeout_last_error( + self, + *, + timeout_seconds: int, + run_started_at: datetime, + stock_codes: Optional[List[str]], + ) -> str: + """Collect partial DB results and build a structured timeout error. + + Notify-channel failures are fail-open and do not change this string. + Collect/import failures are also fail-open: this may return a + ``completed=0`` structured message (indistinguishable from no saved + rows) or, if the helper itself raises, the baseline timeout fallback. + ``status().last_error`` therefore cannot be used as a collect-failure + signal; grep ``Failed to collect completed analyses after timeout`` + or ``Timeout partial delivery failed open``. + """ + fallback = f"runtime scheduled analysis timed out after {timeout_seconds}s" + try: + from src.services.analysis_timeout_partial import handle_runtime_analysis_timeout + + config = None + try: + config = self._config_provider() + except Exception as exc: # noqa: BLE001 - config is optional for diagnostics + logger.warning("Timeout partial notify could not load config: %s", exc) + + no_notify = bool(getattr(self._make_schedule_args(), "no_notify", False)) + outcome = handle_runtime_analysis_timeout( + timeout_seconds=timeout_seconds, + run_started_at=run_started_at, + stock_codes=stock_codes, + no_notify=no_notify, + config=config, + ) + if outcome.notified: + logger.info( + "Timeout partial notification sent: completed=%s pending=%s", + len(outcome.completed), + len(outcome.pending_codes), + ) + elif outcome.notify_skipped_reason: + logger.info( + "Timeout partial notification skipped (%s): completed=%s pending=%s", + outcome.notify_skipped_reason, + len(outcome.completed), + len(outcome.pending_codes), + ) + return outcome.error_message or fallback + except Exception as exc: # noqa: BLE001 - never lose the original timeout signal + logger.warning("Timeout partial delivery failed open: %s", exc) + return fallback + + def _apply_timeout_partial_outcome(self, context: Dict[str, Any]) -> None: + """Enrich timeout diagnostics after the run lock is released. + + Runs in a daemon thread so DB/import work cannot delay the next run or + keep the watchdog finally block occupied. + + Consistency with ``status()``: + - The watchdog first writes a baseline ``timed out after Ns`` string + under ``_analysis_process_lock``, then releases ``_run_lock``. + - This thread later replaces ``_last_error`` with the structured + completed/pending message, still under ``_analysis_process_lock``. + - The replace is skipped if ``generation`` no longer matches (a newer + run started) or if ``_last_error`` no longer contains + ``timed out after`` (another outcome already replaced it). + - ``status()`` reads ``_last_error`` without that lock. CPython + pointer assignment is atomic, so callers observe either the baseline + or the fully replaced string—never a torn mix. They may briefly see + the baseline until this thread finishes. + - Notify-channel exceptions stay inside this thread and cannot + re-acquire ``_run_lock``. Collect/import fail-open is only visible + in warning logs, not as a distinct ``last_error`` code. + """ + generation = context.get("generation") + + def _enrich() -> None: + try: + enriched = self._build_timeout_last_error( + timeout_seconds=int(context["timeout_seconds"]), + run_started_at=context["run_started_at"], + stock_codes=context.get("stock_codes"), + ) + except Exception as exc: # noqa: BLE001 - baseline timeout error remains + logger.warning("Timeout partial enrichment failed open: %s", exc) + return + with self._analysis_process_lock: + if generation is not None and generation != self._analysis_generation: + return + current = self._last_error or "" + if "timed out after" not in current: + return + self._last_error = enriched + + threading.Thread( + target=_enrich, + daemon=True, + name="runtime-timeout-partial", + ).start() + def _run_analysis_with_watchdog( self, stock_codes: Optional[List[str]] = None, @@ -374,6 +475,7 @@ class RuntimeSchedulerService: generation = self._analysis_generation result_queue = None + timeout_partial_context: Optional[Dict[str, Any]] = None try: context = multiprocessing.get_context("spawn") result_queue = context.Queue() @@ -383,12 +485,13 @@ class RuntimeSchedulerService: name="runtime-scheduled-analysis", ) timeout = self._analysis_timeout_seconds() + run_started_at = datetime.now() with self._analysis_process_lock: if generation != self._analysis_generation: return process.start() self._analysis_process = process - self._last_run_at = datetime.now().isoformat() + self._last_run_at = run_started_at.isoformat() result = None deadline = time.monotonic() + timeout @@ -411,7 +514,16 @@ class RuntimeSchedulerService: with self._analysis_process_lock: if generation != self._analysis_generation: return - self._last_error = f"runtime scheduled analysis timed out after {timeout}s" + # Baseline error first so status.running can clear quickly. + self._last_error = ( + f"runtime scheduled analysis timed out after {timeout}s" + ) + timeout_partial_context = { + "timeout_seconds": timeout, + "run_started_at": run_started_at, + "stock_codes": stock_codes, + "generation": generation, + } return if result is None: @@ -457,6 +569,8 @@ class RuntimeSchedulerService: if result_queue is not None: result_queue.cancel_join_thread() result_queue.close() + if timeout_partial_context is not None: + self._apply_timeout_partial_outcome(timeout_partial_context) def _start_analysis_watchdog( self, @@ -682,6 +796,8 @@ class RuntimeSchedulerService: "next_run_at": next_run, "last_run_at": self._last_run_at, "last_success_at": self._last_success_at, + # Unlocked read: may briefly show the baseline timeout string + # until _apply_timeout_partial_outcome finishes. See that method. "last_error": self._last_error, "last_skipped_at": self._last_skipped_at, "last_skip_reason": self._last_skip_reason, diff --git a/tests/test_analysis_timeout_partial_notify.py b/tests/test_analysis_timeout_partial_notify.py new file mode 100644 index 000000000..3bf62e577 --- /dev/null +++ b/tests/test_analysis_timeout_partial_notify.py @@ -0,0 +1,387 @@ +# -*- coding: utf-8 -*- +"""Unit tests for timeout partial delivery helpers.""" + +from __future__ import annotations + +import os +import time +import unittest +from datetime import datetime +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from src.services.analysis_timeout_partial import ( + CompletedAnalysisSummary, + TIMEOUT_PARTIAL_NOTIFY_ENV, + build_partial_timeout_report, + collect_completed_analyses_since, + format_timeout_error_message, + handle_runtime_analysis_timeout, + is_timeout_partial_notify_enabled, + resolve_expected_stock_codes, + send_partial_timeout_notification, +) + + +class TimeoutPartialHelpersTests(unittest.TestCase): + def tearDown(self) -> None: + os.environ.pop(TIMEOUT_PARTIAL_NOTIFY_ENV, None) + + def test_env_defaults_enabled_and_parses_false(self) -> None: + os.environ.pop(TIMEOUT_PARTIAL_NOTIFY_ENV, None) + self.assertTrue(is_timeout_partial_notify_enabled()) + os.environ[TIMEOUT_PARTIAL_NOTIFY_ENV] = "false" + self.assertFalse(is_timeout_partial_notify_enabled()) + + def test_resolve_expected_stock_codes_prefers_explicit_list(self) -> None: + config = SimpleNamespace(stock_list=["000001", "000002"]) + self.assertEqual( + resolve_expected_stock_codes(["600519", "600519", ""], config=config), + ["600519"], + ) + self.assertEqual( + resolve_expected_stock_codes(None, config=config), + ["000001", "000002"], + ) + + def test_format_timeout_error_message_includes_counts(self) -> None: + completed = [ + CompletedAnalysisSummary(code="600519", name="贵州茅台"), + CompletedAnalysisSummary(code="000001", name="平安银行"), + ] + message = format_timeout_error_message( + timeout_seconds=90, + completed=completed, + pending_codes=["300750"], + ) + self.assertIn("timed out after 90s", message) + self.assertIn("completed=2", message) + self.assertIn("pending=1", message) + self.assertIn("completed_codes=600519,000001", message) + self.assertIn("pending_codes=300750", message) + + def test_build_partial_timeout_report_lists_completed_and_pending(self) -> None: + report = build_partial_timeout_report( + timeout_seconds=120, + completed=[ + CompletedAnalysisSummary( + code="600519", + name="贵州茅台", + operation_advice="买入", + sentiment_score=80, + ) + ], + pending_codes=["300750"], + ) + self.assertIn("部分完成", report) + self.assertIn("600519", report) + self.assertIn("300750", report) + self.assertIn("120", report) + + def test_summaries_from_history_rows_keeps_latest_per_code(self) -> None: + from src.services.analysis_timeout_partial import _summaries_from_history_rows + + older = SimpleNamespace( + id=1, + code="600519", + name="old", + operation_advice="持有", + sentiment_score=50, + created_at=datetime(2026, 1, 1, 10, 0, 0), + ) + newer = SimpleNamespace( + id=2, + code="600519", + name="new", + operation_advice="买入", + sentiment_score=70, + created_at=datetime(2026, 1, 1, 11, 0, 0), + ) + rows = _summaries_from_history_rows( + [newer, older], + ["600519", "300750"], + ) + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0].code, "600519") + self.assertEqual(rows[0].name, "new") + self.assertEqual(rows[0].history_id, 2) + + def test_timeout_partial_notify_registry_uses_switch_ui_control(self) -> None: + from api.v1.schemas.system_config import SystemConfigFieldSchema + from src.core.config_registry import get_field_definition + + field = get_field_definition("DSA_TIMEOUT_PARTIAL_NOTIFY") + self.assertEqual(field["data_type"], "boolean") + self.assertEqual(field["ui_control"], "switch") + self.assertEqual(field["default_value"], "true") + SystemConfigFieldSchema.model_validate(field) + + def test_collect_completed_analyses_since_fail_open(self) -> None: + with patch( + "src.services.analysis_timeout_partial._query_history_rows_since", + side_effect=RuntimeError("db down"), + ): + rows = collect_completed_analyses_since( + run_started_at=datetime.now(), + expected_codes=["600519"], + ) + self.assertEqual(rows, []) + + def test_resolve_storage_module_cleans_partial_import_on_failure(self) -> None: + import sys + + from src.services.analysis_timeout_partial import _resolve_storage_module + + original = { + name: module + for name, module in sys.modules.items() + if name == "src.storage" or name.startswith("src.storage.") + } + for name in list(original): + sys.modules.pop(name, None) + + def _boom(name: str, *args, **kwargs): + if name == "src.storage": + sys.modules["src.storage"] = object() + sys.modules["src.storage.models"] = object() + raise ImportError("broken storage") + raise AssertionError(f"unexpected import: {name}") + + try: + with patch("importlib.import_module", side_effect=_boom): + with self.assertRaises(RuntimeError) as ctx: + _resolve_storage_module() + self.assertIn("storage unavailable", str(ctx.exception)) + self.assertNotIn("src.storage", sys.modules) + self.assertFalse( + any(name.startswith("src.storage.") for name in sys.modules) + ) + finally: + for name in list(sys.modules): + if name == "src.storage" or name.startswith("src.storage."): + sys.modules.pop(name, None) + sys.modules.update(original) + + def test_collect_fail_open_when_storage_import_fails(self) -> None: + with patch( + "src.services.analysis_timeout_partial._resolve_storage_module", + side_effect=RuntimeError("storage unavailable for timeout partial collect"), + ): + rows = collect_completed_analyses_since( + run_started_at=datetime.now(), + expected_codes=["600519"], + ) + self.assertEqual(rows, []) + + def test_send_respects_no_notify_and_env(self) -> None: + self.assertFalse( + send_partial_timeout_notification( + "hello", + completed_codes=["600519"], + no_notify=True, + ) + ) + os.environ[TIMEOUT_PARTIAL_NOTIFY_ENV] = "0" + self.assertFalse( + send_partial_timeout_notification( + "hello", + completed_codes=["600519"], + no_notify=False, + ) + ) + + def test_send_swallows_channel_exception_and_returns_false(self) -> None: + notification_mod = MagicMock() + notification_mod.NotificationService.return_value.send.side_effect = RuntimeError( + "webhook 500" + ) + with patch.dict("sys.modules", {"src.notification": notification_mod}): + self.assertFalse( + send_partial_timeout_notification( + "hello", + completed_codes=["600519"], + no_notify=False, + ) + ) + notification_mod.NotificationService.return_value.send.assert_called_once() + + @patch("src.services.analysis_timeout_partial.collect_completed_analyses_since") + def test_handle_keeps_error_message_when_channel_raises( + self, + collect_mock: MagicMock, + ) -> None: + collect_mock.return_value = [ + CompletedAnalysisSummary(code="600519", name="贵州茅台"), + ] + notification_mod = MagicMock() + notification_mod.NotificationService.return_value.send.side_effect = RuntimeError( + "channel down" + ) + + with patch.dict("sys.modules", {"src.notification": notification_mod}): + outcome = handle_runtime_analysis_timeout( + timeout_seconds=60, + run_started_at=datetime(2026, 1, 1, 12, 0, 0), + stock_codes=["600519", "300750"], + no_notify=False, + ) + + self.assertFalse(outcome.notified) + self.assertEqual(outcome.notify_skipped_reason, "send_failed") + self.assertIn("completed=1", outcome.error_message) + self.assertIn("pending=1", outcome.error_message) + + @patch("src.services.analysis_timeout_partial.send_partial_timeout_notification") + @patch("src.services.analysis_timeout_partial.collect_completed_analyses_since") + def test_handle_runtime_analysis_timeout_notifies_when_completed( + self, + collect_mock: MagicMock, + send_mock: MagicMock, + ) -> None: + collect_mock.return_value = [ + CompletedAnalysisSummary(code="600519", name="贵州茅台"), + ] + send_mock.return_value = True + + outcome = handle_runtime_analysis_timeout( + timeout_seconds=60, + run_started_at=datetime(2026, 1, 1, 12, 0, 0), + stock_codes=["600519", "300750"], + no_notify=False, + config=SimpleNamespace(stock_list=[]), + db=MagicMock(), + ) + + self.assertTrue(outcome.notified) + self.assertEqual(outcome.pending_codes, ["300750"]) + self.assertIn("completed=1", outcome.error_message) + self.assertIn("pending=1", outcome.error_message) + send_mock.assert_called_once() + + @patch("src.services.analysis_timeout_partial.send_partial_timeout_notification") + @patch("src.services.analysis_timeout_partial.collect_completed_analyses_since") + def test_handle_skips_notify_without_completed( + self, + collect_mock: MagicMock, + send_mock: MagicMock, + ) -> None: + collect_mock.return_value = [] + outcome = handle_runtime_analysis_timeout( + timeout_seconds=60, + run_started_at=datetime.now(), + stock_codes=["600519"], + no_notify=False, + ) + self.assertFalse(outcome.notified) + self.assertEqual(outcome.notify_skipped_reason, "no_completed_results") + send_mock.assert_not_called() + + +class RuntimeSchedulerTimeoutPartialIntegrationTests(unittest.TestCase): + def test_build_timeout_last_error_uses_partial_helper(self) -> None: + from src.services.runtime_scheduler import RuntimeSchedulerService + + config = SimpleNamespace(stock_list=["600519", "300750"]) + service = RuntimeSchedulerService(config_provider=lambda: config) + completed = [CompletedAnalysisSummary(code="600519", name="贵州茅台")] + + with patch( + "src.services.analysis_timeout_partial.collect_completed_analyses_since", + return_value=completed, + ), patch( + "src.services.analysis_timeout_partial.send_partial_timeout_notification", + return_value=True, + ): + message = service._build_timeout_last_error( + timeout_seconds=90, + run_started_at=datetime(2026, 1, 1, 12, 0, 0), + stock_codes=["600519", "300750"], + ) + + self.assertIn("timed out after 90s", message) + self.assertIn("completed=1", message) + self.assertIn("pending=1", message) + self.assertIn("completed_codes=600519", message) + self.assertIn("pending_codes=300750", message) + + def test_timeout_branch_sets_structured_last_error(self) -> None: + from src.services.runtime_scheduler import RuntimeSchedulerService + + config = SimpleNamespace(stock_list=["600519", "300750"]) + service = RuntimeSchedulerService(config_provider=lambda: config) + service._analysis_timeout_seconds = lambda: 1 + + class _AliveProcess: + def __init__(self, *args, **kwargs): + self.pid = 424242 + self.exitcode = None + self._alive = True + + def start(self) -> None: + return None + + def is_alive(self) -> bool: + return self._alive + + def join(self, timeout=None) -> None: + return None + + def terminate(self) -> None: + self._alive = False + + def kill(self) -> None: + self._alive = False + + class _EmptyQueue: + def get(self, timeout=None): + from queue import Empty + + raise Empty + + def cancel_join_thread(self) -> None: + return None + + def close(self) -> None: + return None + + fake_context = SimpleNamespace( + Queue=lambda: _EmptyQueue(), + Process=_AliveProcess, + ) + + with patch( + "src.services.runtime_scheduler.multiprocessing.get_context", + return_value=fake_context, + ), patch( + "src.services.runtime_scheduler._terminate_analysis_process_tree", + ), patch.object( + service, + "_build_timeout_last_error", + return_value=( + "runtime scheduled analysis timed out after 1s; " + "completed=1; pending=1; completed_codes=600519; pending_codes=300750" + ), + ): + # Watchdog finally always releases the shared lock. + self.assertTrue(service._run_lock.acquire(blocking=False)) + service._run_analysis_with_watchdog( + ["600519", "300750"], + lock_held=True, + ) + + deadline = time.time() + 2 + last_error = service.status()["last_error"] + while ( + last_error is None or "completed=1" not in last_error + ) and time.time() < deadline: + time.sleep(0.02) + last_error = service.status()["last_error"] + + self.assertIsNotNone(last_error) + self.assertIn("timed out after 1s", last_error) + self.assertIn("completed=1", last_error) + self.assertIn("pending_codes=300750", last_error) + + +if __name__ == "__main__": + unittest.main()