mirror of
https://github.com/ZhuLinsen/daily_stock_analysis
synced 2026-09-20 10:53:33 +08:00
972c314656
* feat: Web/API 指数入口与共享 canonical 去重基础 为 Issue #2303 Phase 2 PR1 落地 Web/API 指数入口适配： - API 使用 parse_analysis_target 构造结构化 AnalysisTarget 并贯通到 pipeline - API 与 TaskQueue 去重按 asset_type 分支（指数用 canonical_id，个股用 legacy code） - BatchTaskAcceptedResponse 追加可选 rejected 字段，未登记 CSI 单股 400、批量仅该目标失败 - TaskInfo 固化 dedupe_key，避免指数与同码个股折叠及 _analyzing_stocks 残留 - Web 移除 assetType=index 全局过滤，Chat 名称识别保护指数 canonical - 补齐 Pipeline 指数 DecisionSignal market_override=cn 真实分支测试 * fix: 收敛指数 canonical 身份与批量响应契约 PR #2312 review 修复：报告 meta 补充 asset_type 隐藏指数自选；/analyze 在解析前限制非空原始 token；is_single 统一驱动 metadata/409/单任务 202；HomePage 三元计数继续后续 chunk。 * fix: avoid double space in index news search query * fix: preserve canonical index identity * fix: validate legacy task asset type * fix: preserve canonical index identity in chat * fix: preserve index identity across chat backends
1659 lines
64 KiB
Python
1659 lines
64 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
===================================
|
||
股票分析接口
|
||
===================================
|
||
|
||
职责：
|
||
1. 提供 POST /api/v1/analysis/analyze 触发分析接口
|
||
2. 提供 GET /api/v1/analysis/status/{task_id} 查询任务状态接口
|
||
3. 提供 GET /api/v1/analysis/tasks 获取任务列表接口
|
||
4. 提供 GET /api/v1/analysis/tasks/stream SSE 实时推送接口
|
||
|
||
特性：
|
||
- 异步任务队列：分析任务异步执行，不阻塞请求
|
||
- 防重复提交：相同股票代码正在分析时返回 409
|
||
- SSE 实时推送：任务状态变化实时通知前端
|
||
"""
|
||
|
||
import asyncio
|
||
import copy
|
||
import json
|
||
import logging
|
||
import re
|
||
import unicodedata
|
||
import uuid
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Optional, Union, Dict, Any
|
||
|
||
from fastapi import APIRouter, HTTPException, Depends, Query, Body
|
||
from fastapi.responses import JSONResponse, StreamingResponse
|
||
|
||
from api.deps import get_config_dep
|
||
from api.v1.errors import api_error
|
||
from api.v1.schemas.analysis import (
|
||
AnalyzeRequest,
|
||
AnalysisResultResponse,
|
||
TaskAccepted,
|
||
BatchTaskAcceptedResponse,
|
||
BatchTaskAcceptedItem,
|
||
BatchDuplicateTaskItem,
|
||
RejectedTaskItem,
|
||
TaskStatus,
|
||
TaskInfo,
|
||
TaskListResponse,
|
||
DuplicateTaskErrorResponse,
|
||
MarketReviewRequest,
|
||
MarketReviewAccepted,
|
||
)
|
||
from api.v1.schemas.common import ErrorResponse
|
||
from api.v1.schemas.history import (
|
||
AnalysisReport,
|
||
ReportMeta,
|
||
ReportSummary,
|
||
ReportStrategy,
|
||
ReportDetails,
|
||
)
|
||
from api.v1.schemas.run_flow import RunFlowSnapshot
|
||
from data_provider.base import canonical_stock_code, normalize_stock_code
|
||
from src.data.stock_index_loader import resolve_index_stock_code
|
||
from src.config import Config
|
||
from src.core.market_review_lock import (
|
||
MarketReviewExecutionLock as _MarketReviewExecutionLock,
|
||
market_review_lock_path,
|
||
release_market_review_lock as _release_market_review_lock,
|
||
try_acquire_market_review_lock as _try_acquire_market_review_lock,
|
||
)
|
||
from src.core.market_review_runtime import (
|
||
build_market_review_runtime as _runtime_build_market_review_runtime,
|
||
)
|
||
from src.analysis_context_pack_overview import (
|
||
extract_analysis_context_pack_overview,
|
||
sanitize_context_snapshot_for_api,
|
||
)
|
||
from src.market_phase_summary import (
|
||
extract_market_phase_summary,
|
||
rebuild_market_phase_summary_for_stock_code,
|
||
)
|
||
from src.services.stock_code_utils import is_code_like, resolve_index_stock_code_for_analysis
|
||
from src.services.stock_list_parser import ParseStatus, parse_analysis_target
|
||
from src.report_language import get_localized_stock_name, normalize_report_language
|
||
from src.schemas.decision_action import build_action_fields
|
||
from src.services.name_to_code_resolver import resolve_name_to_code
|
||
from src.services.task_queue import (
|
||
get_task_queue,
|
||
DuplicateTaskError,
|
||
TaskStatus as TaskStatusEnum,
|
||
)
|
||
from src.services.analysis_service import asset_type_from_canonical_code
|
||
from src.services.run_diagnostics import build_run_diagnostic_summary
|
||
from src.services.run_flow import build_task_run_flow_snapshot
|
||
from src.services.empty_news import empty_news_disclosure_from_stored
|
||
from src.utils.data_processing import (
|
||
normalize_model_used,
|
||
parse_json_field,
|
||
extract_fundamental_detail_fields,
|
||
extract_board_detail_fields,
|
||
extract_market_structure_detail_field,
|
||
extract_realtime_detail_fields,
|
||
)
|
||
from src.utils.market_review_region import normalize_market_review_region_lenient
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter()
|
||
|
||
_SUPPORTED_FREE_TEXT_RE = re.compile(r"^[A-Za-z0-9.*\-+\u3400-\u9fff\s]+$")
|
||
|
||
|
||
def _get_task_trace_id(task: Any) -> Optional[str]:
|
||
trace_id = getattr(task, "trace_id", None)
|
||
if isinstance(trace_id, str) and trace_id.strip():
|
||
return trace_id
|
||
task_id = getattr(task, "task_id", None)
|
||
if isinstance(task_id, str) and task_id.strip():
|
||
return task_id
|
||
return None
|
||
|
||
|
||
def _task_asset_type(task: Any) -> Optional[str]:
|
||
"""Return the task's optional asset type only when it is a real literal.
|
||
|
||
The Pydantic literal domain (``stock``/``index``) is the only allowed set:
|
||
real in-domain strings pass through verbatim. Legacy mock/proxy tasks whose
|
||
``getattr`` yields a ``MagicMock`` child, and missing values, degrade to
|
||
``None`` so schema defaults keep legacy responses unchanged. Genuine
|
||
out-of-domain strings are logged and degraded instead of widening the enum
|
||
or silently masking drift.
|
||
"""
|
||
raw = getattr(task, "asset_type", None)
|
||
if not isinstance(raw, str):
|
||
return None
|
||
if raw in {"stock", "index"}:
|
||
return raw
|
||
# 任何真实字符串只要不精确等于字面量域（含空串、纯空白、大小写或带
|
||
# 空格形态）都必须记录 warning 后降级，避免静默掩盖漂移；仅非字符串
|
||
# 代理（如 MagicMock 子对象）保持静默 None。
|
||
logger.warning(
|
||
"task asset_type 超出字面量域，降级为 None: task_id=%s asset_type=%r",
|
||
getattr(task, "task_id", None),
|
||
raw,
|
||
)
|
||
return None
|
||
|
||
|
||
def _market_review_lock_path(config: Config) -> Path:
|
||
return market_review_lock_path(config)
|
||
|
||
|
||
def _build_market_review_runtime(config: Config, source_message: Optional[Any] = None) -> tuple[Any, Any, Any]:
|
||
return _runtime_build_market_review_runtime(config, source_message)
|
||
|
||
|
||
def _with_request_report_language(config: Config, report_language: Optional[str]) -> Config:
|
||
"""Return a request-scoped config copy when the caller overrides report language."""
|
||
normalized = normalize_report_language(report_language, default="")
|
||
if not normalized:
|
||
return config
|
||
|
||
scoped_config = copy.copy(config)
|
||
scoped_config.report_language = normalized
|
||
return scoped_config
|
||
|
||
|
||
def _run_market_review_background(
|
||
send_notification: bool,
|
||
effective_region: str,
|
||
lock_token: Optional[_MarketReviewExecutionLock] = None,
|
||
config: Optional[Config] = None,
|
||
query_id: Optional[str] = None,
|
||
) -> Dict[str, Any]:
|
||
"""Run market review after the API response has been accepted."""
|
||
from src.core.market_review import run_market_review
|
||
|
||
runtime_config = config or get_config_dep()
|
||
try:
|
||
notifier, analyzer, search_service = _build_market_review_runtime(runtime_config)
|
||
review_kwargs = {
|
||
"notifier": notifier,
|
||
"analyzer": analyzer,
|
||
"search_service": search_service,
|
||
"config": runtime_config,
|
||
"send_notification": send_notification,
|
||
"override_region": effective_region,
|
||
"return_structured": True,
|
||
"trigger_source": "api",
|
||
}
|
||
if query_id:
|
||
review_kwargs["query_id"] = query_id
|
||
logger.info(
|
||
"[MarketReview] component=market_review action=background_start "
|
||
"trigger_source=api task_id=%s region=%s",
|
||
query_id or "-",
|
||
effective_region,
|
||
)
|
||
report = run_market_review(**review_kwargs)
|
||
if not report:
|
||
raise RuntimeError("大盘复盘未返回可持久化报告")
|
||
if hasattr(report, "report"):
|
||
return {
|
||
"result": report.report,
|
||
"market_review_payload": getattr(report, "market_review_payload", None),
|
||
"region": effective_region,
|
||
}
|
||
return {"result": report, "region": effective_region}
|
||
finally:
|
||
_release_market_review_lock(lock_token)
|
||
|
||
|
||
def _coalesce_text(*values: Any) -> Optional[str]:
|
||
for value in values:
|
||
if value is None:
|
||
continue
|
||
text = str(value).strip()
|
||
if text:
|
||
return text
|
||
return None
|
||
|
||
|
||
def _extract_guardrail_reason(raw_result: Any) -> Optional[str]:
|
||
if not isinstance(raw_result, dict):
|
||
return None
|
||
for reason in (
|
||
raw_result.get("guardrail_reason"),
|
||
raw_result.get("downgrade_reason"),
|
||
raw_result.get("decision_score_guardrail_reason"),
|
||
):
|
||
if reason is not None:
|
||
text = str(reason).strip()
|
||
if text:
|
||
return text
|
||
metadata = raw_result.get("metadata")
|
||
if isinstance(metadata, dict):
|
||
metadata_reason = metadata.get("guardrail_reason") or metadata.get("downgrade_reason")
|
||
if metadata_reason is not None:
|
||
text = str(metadata_reason).strip()
|
||
if text:
|
||
return text
|
||
return None
|
||
|
||
|
||
def _invalid_analysis_input_error() -> HTTPException:
|
||
return api_error(400, "validation_error", "请输入有效的股票代码或股票名称")
|
||
|
||
|
||
def _is_obviously_invalid_analysis_input(text: str) -> bool:
|
||
"""Reject mixed alphanumeric noise and unsupported symbols early."""
|
||
if not text or is_code_like(text):
|
||
return False
|
||
|
||
if not _SUPPORTED_FREE_TEXT_RE.fullmatch(text):
|
||
return True
|
||
|
||
has_letters = any(ch.isalpha() and ch.isascii() for ch in text)
|
||
has_digits = any(ch.isdigit() for ch in text)
|
||
return has_letters and has_digits
|
||
|
||
|
||
def _resolve_analysis_input(raw_value: str):
|
||
"""
|
||
Resolve one analysis request input into ``(code, analysis_target)``.
|
||
|
||
Code-like tokens go through :func:`parse_analysis_target` (the single
|
||
asset-type authority): registered indices keep their structured
|
||
``AnalysisTarget`` (canonical id + index semantics), unsupported targets
|
||
(e.g. unregistered ``930956.CSI``) are surfaced with their reason so the
|
||
caller can reject them explicitly, and stock tokens keep the legacy
|
||
resolution path. Non-code names (e.g. ``贵州茅台``) keep
|
||
:func:`resolve_name_to_code` and never enter index classification.
|
||
"""
|
||
text = (raw_value or "").strip()
|
||
if not text:
|
||
return None
|
||
|
||
# CSI explicit forms (``csi930955`` / ``930955.CSI`` / ``CSI930955``) are
|
||
# explicit code forms but ``is_code_like`` does not recognise the ``.CSI``
|
||
# suffix; route them through parse_analysis_target so registered CSI
|
||
# converges to its canonical index identity and unregistered CSI surfaces
|
||
# as an explicit unsupported error (never a name-resolution fallback).
|
||
normalized_csi = unicodedata.normalize("NFKC", text).strip().casefold()
|
||
if re.fullmatch(r"(?:csi\d{6}|\d{6}\.csi)", normalized_csi):
|
||
target = parse_analysis_target(text)
|
||
if target.asset_type == ParseStatus.INDEX:
|
||
return (target.canonical_id, target)
|
||
return (text, target)
|
||
|
||
# SH/SZ prefixed six-digit forms (``sh000016`` / ``SH000016``) are explicit
|
||
# index/stock code shapes, but ``is_code_like`` rejects them when the bare
|
||
# digits do not match the exchange's *stock* digit rules (``000016`` is an
|
||
# SH index yet classifies as an SZ stock digit shape). Route them through
|
||
# parse_analysis_target so registered SH/SZ indices keep their structured
|
||
# target; unknown prefixed forms degrade to the stock path as usual.
|
||
normalized_sh_sz = re.fullmatch(r"(sh|sz)(\d{6})", normalized_csi)
|
||
if normalized_sh_sz is not None:
|
||
target = parse_analysis_target(text)
|
||
if target.asset_type == ParseStatus.INDEX:
|
||
return (target.canonical_id, target)
|
||
return (resolve_index_stock_code_for_analysis(text), None)
|
||
|
||
if is_code_like(text):
|
||
target = parse_analysis_target(text)
|
||
if target.asset_type == ParseStatus.INDEX:
|
||
return (target.canonical_id, target)
|
||
if target.asset_type == ParseStatus.UNSUPPORTED:
|
||
return (text, target)
|
||
# Stock target: keep the existing canonical resolution path and do not
|
||
# carry a stock target downstream (stock semantics must not change).
|
||
return (resolve_index_stock_code_for_analysis(text), None)
|
||
|
||
if text.isdigit() and len(text) == 4:
|
||
resolved_index_code = resolve_index_stock_code_for_analysis(text)
|
||
if resolved_index_code != canonical_stock_code(text):
|
||
return (resolved_index_code, None)
|
||
|
||
if _is_obviously_invalid_analysis_input(text):
|
||
raise _invalid_analysis_input_error()
|
||
|
||
resolved = resolve_name_to_code(text)
|
||
if resolved:
|
||
return (canonical_stock_code(resolved), None)
|
||
|
||
raise _invalid_analysis_input_error()
|
||
|
||
|
||
# ============================================================
|
||
# POST /analyze - 触发股票分析
|
||
# ============================================================
|
||
|
||
@router.post(
|
||
"/analyze",
|
||
response_model=AnalysisResultResponse,
|
||
responses={
|
||
200: {"description": "分析完成（同步模式）", "model": AnalysisResultResponse},
|
||
202: {
|
||
"description": "分析任务已接受（异步模式）",
|
||
"model": Union[TaskAccepted, BatchTaskAcceptedResponse],
|
||
},
|
||
400: {"description": "请求参数错误", "model": ErrorResponse},
|
||
409: {"description": "股票正在分析中，拒绝重复提交", "model": DuplicateTaskErrorResponse},
|
||
500: {"description": "分析失败", "model": ErrorResponse},
|
||
},
|
||
summary="触发股票分析",
|
||
description="启动 AI 智能分析任务，支持同步和异步模式。异步模式下相同股票代码不允许重复提交。"
|
||
)
|
||
def trigger_analysis(
|
||
request: AnalyzeRequest,
|
||
config: Config = Depends(get_config_dep)
|
||
) -> Union[AnalysisResultResponse, JSONResponse]:
|
||
"""
|
||
触发股票分析
|
||
|
||
启动 AI 智能分析任务，支持单只或多只股票批量分析
|
||
|
||
流程：
|
||
1. 校验请求参数
|
||
2. 异步模式：检查重复 -> 提交任务队列 -> 返回 202
|
||
3. 同步模式：直接执行分析 -> 返回 200
|
||
|
||
Args:
|
||
request: 分析请求参数
|
||
config: 配置依赖
|
||
|
||
Returns:
|
||
AnalysisResultResponse: 分析结果（同步模式）
|
||
TaskAccepted | BatchTaskAcceptedResponse: 任务已接受（异步模式，返回 202）
|
||
|
||
Raises:
|
||
HTTPException: 400 - 请求参数错误
|
||
HTTPException: 409 - 股票正在分析中
|
||
HTTPException: 500 - 分析失败
|
||
"""
|
||
# 校验请求参数
|
||
stock_codes = []
|
||
if request.stock_code:
|
||
stock_codes.append(request.stock_code)
|
||
if request.stock_codes:
|
||
stock_codes.extend(request.stock_codes)
|
||
|
||
if not stock_codes:
|
||
raise api_error(400, "validation_error", "必须提供 stock_code 或 stock_codes 参数")
|
||
|
||
# Limit the number of non-blank raw tokens BEFORE resolution. Rejected and
|
||
# duplicate tokens must also count toward the cap, otherwise a request that
|
||
# mixes one valid token with many rejected/duplicate tokens could bypass the
|
||
# DoS limit via the post-dedup check below.
|
||
MAX_BATCH_SIZE = 50
|
||
non_empty_raw_tokens = [c for c in stock_codes if str(c or "").strip()]
|
||
if len(non_empty_raw_tokens) > MAX_BATCH_SIZE:
|
||
raise api_error(400, "validation_error", f"单次分析请求最多支持 {MAX_BATCH_SIZE} 只股票")
|
||
|
||
# Normalize and de-duplicate inputs while preserving compatibility.
|
||
# Code-like tokens go through parse_analysis_target (the single asset-type
|
||
# authority) so registered indices keep their structured target; non-code
|
||
# names keep the legacy stock-name resolution path.
|
||
resolved_entries = [_resolve_analysis_input(c) for c in stock_codes]
|
||
|
||
seen = set()
|
||
unique_codes = []
|
||
unique_targets = []
|
||
rejected_entries = []
|
||
for entry in resolved_entries:
|
||
if entry is None:
|
||
continue
|
||
code, target = entry
|
||
if not code:
|
||
continue
|
||
if target is not None and target.asset_type == ParseStatus.UNSUPPORTED:
|
||
rejected_entries.append((code, target))
|
||
continue
|
||
# 去重键按 asset_type 分支：INDEX 用 canonical_id（指数与同码股票不折叠、
|
||
# CSI alias 收敛），STOCK 保持 normalize_stock_code 既有语义
|
||
# （'600519' 与 '600519.SH' 合并）。
|
||
if target is not None and target.asset_type == ParseStatus.INDEX:
|
||
norm = target.canonical_id
|
||
else:
|
||
norm = normalize_stock_code(code)
|
||
if norm not in seen:
|
||
seen.add(norm)
|
||
unique_codes.append(code)
|
||
unique_targets.append(target)
|
||
|
||
stock_codes = unique_codes
|
||
|
||
if not stock_codes:
|
||
if not rejected_entries:
|
||
raise api_error(400, "validation_error", "股票代码不能为空或仅包含空白字符")
|
||
# 全部目标都被拒绝（单请求或批量全拒）：无任务被接受时返回 202 会误导
|
||
# 客户端，一律明确 400；部分被拒则走下方 rejected 语义。
|
||
code, target = rejected_entries[0]
|
||
reason = target.unsupported_reason or f"不支持的目标: {code}"
|
||
raise api_error(400, "validation_error", reason)
|
||
|
||
# Sync mode only supports single-stock analysis.
|
||
if not request.async_mode:
|
||
if len(stock_codes) > 1:
|
||
raise api_error(
|
||
400,
|
||
"validation_error",
|
||
"同步模式仅支持单只股票分析，请使用 async_mode=true 进行批量分析",
|
||
)
|
||
if rejected_entries:
|
||
# 同步模式不支持 rejected 语义：只要存在被拒绝目标（含全部被拒绝）
|
||
# 就以第一个未登记目标的明确校验错误返回 4xx。
|
||
code, target = rejected_entries[0]
|
||
reason = target.unsupported_reason or f"不支持的目标: {code}"
|
||
raise api_error(400, "validation_error", reason)
|
||
return _handle_sync_analysis(stock_codes[0], request, analysis_target=unique_targets[0] if unique_targets else None)
|
||
|
||
# Async mode submits one task per stock.
|
||
return _handle_async_analysis_batch(stock_codes, request, analysis_targets=unique_targets, rejected_entries=rejected_entries)
|
||
|
||
|
||
def _handle_async_analysis_batch(
|
||
stock_codes: list,
|
||
request: AnalyzeRequest,
|
||
analysis_targets: Optional[list] = None,
|
||
rejected_entries: Optional[list] = None,
|
||
) -> JSONResponse:
|
||
"""
|
||
Handle asynchronous analysis requests, including batch submission.
|
||
|
||
Args:
|
||
stock_codes: canonical codes to submit
|
||
request: the analysis request
|
||
analysis_targets: optional per-code structured targets (index targets
|
||
flow through to the pipeline)
|
||
rejected_entries: optional list of ``(code, target)`` pairs that were
|
||
explicitly rejected (e.g. unregistered CSI); returned in the
|
||
``rejected`` field for batch requests only.
|
||
"""
|
||
task_queue = get_task_queue()
|
||
|
||
# Preserve metadata for single-stock requests. For batch requests,
|
||
# only carry through metadata that semantically applies to the whole
|
||
# batch, such as import/image source tracking.
|
||
# A single "accepted" code alongside any rejected entries is a batch:
|
||
# rejected entries mean the server has not fully disposed of a single-stock
|
||
# request, so single-stock metadata/409/single-202 semantics must not apply.
|
||
is_single = len(stock_codes) == 1 and not rejected_entries
|
||
preserve_batch_metadata = request.selection_source in {"import", "image"}
|
||
|
||
stock_name = request.stock_name if is_single else None
|
||
original_query = request.original_query if (is_single or preserve_batch_metadata) else None
|
||
selection_source = request.selection_source if (is_single or preserve_batch_metadata) else None
|
||
notify = getattr(request, "notify", True)
|
||
skills = getattr(request, "skills", None)
|
||
analysis_phase = request.analysis_phase
|
||
report_language = normalize_report_language(getattr(request, "report_language", None), default="")
|
||
|
||
submit_kwargs = dict(
|
||
stock_codes=stock_codes,
|
||
stock_name=stock_name,
|
||
original_query=original_query,
|
||
selection_source=selection_source,
|
||
report_type=request.report_type,
|
||
analysis_phase=analysis_phase,
|
||
force_refresh=request.force_refresh,
|
||
notify=notify,
|
||
)
|
||
if report_language:
|
||
submit_kwargs["report_language"] = report_language
|
||
if skills is not None:
|
||
submit_kwargs["skills"] = skills
|
||
# 仅当存在非 None 的结构化 target（如指数）时才传递，保持纯股票请求
|
||
# 的既有 kwargs 契约不变。
|
||
if analysis_targets is not None and any(t is not None for t in analysis_targets):
|
||
submit_kwargs["analysis_targets"] = analysis_targets
|
||
|
||
accepted_tasks, duplicate_errors = task_queue.submit_tasks_batch(**submit_kwargs)
|
||
|
||
accepted = [
|
||
BatchTaskAcceptedItem(
|
||
task_id=task.task_id,
|
||
trace_id=_get_task_trace_id(task),
|
||
stock_code=task.stock_code,
|
||
status="pending",
|
||
message=f"分析任务已加入队列: {task.stock_code}",
|
||
analysis_phase=task.analysis_phase,
|
||
asset_type=_task_asset_type(task),
|
||
)
|
||
for task in accepted_tasks
|
||
]
|
||
duplicates = [
|
||
BatchDuplicateTaskItem(
|
||
stock_code=dup.stock_code,
|
||
existing_task_id=dup.existing_task_id,
|
||
message=str(dup),
|
||
)
|
||
for dup in duplicate_errors
|
||
]
|
||
rejected = [
|
||
RejectedTaskItem(
|
||
stock_code=code,
|
||
message=(target.unsupported_reason or f"不支持的目标: {code}"),
|
||
)
|
||
for code, target in (rejected_entries or [])
|
||
]
|
||
|
||
# 单只股票且被拒绝：保持 409 兼容性
|
||
if is_single and duplicates:
|
||
dup = duplicates[0]
|
||
error_response = DuplicateTaskErrorResponse(
|
||
error="duplicate_task",
|
||
message=dup.message,
|
||
stock_code=dup.stock_code,
|
||
existing_task_id=dup.existing_task_id,
|
||
)
|
||
return JSONResponse(
|
||
status_code=409,
|
||
content=error_response.model_dump()
|
||
)
|
||
|
||
# 单只股票成功（且无 rejected）：保持原有响应格式兼容性
|
||
if is_single and accepted and not rejected:
|
||
task_accepted = TaskAccepted(
|
||
task_id=accepted[0].task_id,
|
||
trace_id=accepted[0].trace_id,
|
||
status="pending",
|
||
message=accepted[0].message,
|
||
analysis_phase=accepted[0].analysis_phase,
|
||
)
|
||
return JSONResponse(
|
||
status_code=202,
|
||
content=task_accepted.model_dump()
|
||
)
|
||
|
||
# 批量：返回汇总结果（rejected 仅 async 批量返回）
|
||
rejected_count = len(rejected)
|
||
if rejected_count:
|
||
message = f"已提交 {len(accepted)} 个任务，{len(duplicates)} 个重复跳过，{rejected_count} 个被拒绝"
|
||
else:
|
||
message = f"已提交 {len(accepted)} 个任务，{len(duplicates)} 个重复跳过"
|
||
batch_response = BatchTaskAcceptedResponse(
|
||
accepted=accepted,
|
||
duplicates=duplicates,
|
||
rejected=rejected if rejected else None,
|
||
message=message,
|
||
)
|
||
return JSONResponse(
|
||
status_code=202,
|
||
content=batch_response.model_dump()
|
||
)
|
||
|
||
|
||
def _handle_sync_analysis(
|
||
stock_code: str,
|
||
request: AnalyzeRequest,
|
||
analysis_target: Optional[Any] = None,
|
||
) -> AnalysisResultResponse:
|
||
"""
|
||
处理同步分析请求
|
||
|
||
直接执行分析，等待完成后返回结果
|
||
"""
|
||
import uuid
|
||
from src.services.analysis_service import AnalysisService
|
||
|
||
query_id = uuid.uuid4().hex
|
||
|
||
try:
|
||
service = AnalysisService()
|
||
result = service.analyze_stock(
|
||
stock_code=stock_code,
|
||
report_type=request.report_type,
|
||
force_refresh=request.force_refresh,
|
||
query_id=query_id,
|
||
send_notification=getattr(request, "notify", True),
|
||
skills=getattr(request, "skills", None),
|
||
analysis_phase=request.analysis_phase,
|
||
report_language=getattr(request, "report_language", None),
|
||
analysis_target=analysis_target,
|
||
)
|
||
|
||
if result is None:
|
||
error_message = service.last_error or f"分析股票 {stock_code} 失败"
|
||
raise api_error(500, "analysis_failed", error_message)
|
||
|
||
# 构建报告结构
|
||
report_data = result.get("report", {})
|
||
context_snapshot, fundamental_snapshot, raw_result_snapshot = _load_sync_fundamental_sources(
|
||
query_id=query_id,
|
||
stock_code=result.get("stock_code", stock_code),
|
||
)
|
||
report = _build_analysis_report(
|
||
report_data,
|
||
query_id,
|
||
stock_code,
|
||
result.get("stock_name"),
|
||
context_snapshot=context_snapshot,
|
||
fallback_fundamental_payload=fundamental_snapshot,
|
||
fallback_raw_result_payload=raw_result_snapshot or result,
|
||
)
|
||
|
||
return AnalysisResultResponse(
|
||
query_id=query_id,
|
||
trace_id=result.get("trace_id") or query_id,
|
||
stock_code=result.get("stock_code", stock_code),
|
||
stock_name=result.get("stock_name"),
|
||
report=report.model_dump() if report else None,
|
||
diagnostic_summary=result.get("diagnostic_summary"),
|
||
created_at=datetime.now().isoformat()
|
||
)
|
||
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
logger.error(f"分析失败: {e}", exc_info=True)
|
||
raise api_error(500, "internal_error", f"分析过程发生错误: {str(e)}")
|
||
|
||
|
||
# ============================================================
|
||
# POST /market-review - 触发大盘复盘
|
||
# ============================================================
|
||
|
||
@router.post(
|
||
"/market-review",
|
||
response_model=MarketReviewAccepted,
|
||
status_code=202,
|
||
responses={
|
||
202: {"description": "大盘复盘任务已接受", "model": MarketReviewAccepted},
|
||
409: {"description": "大盘复盘正在执行", "model": ErrorResponse},
|
||
500: {"description": "提交失败", "model": ErrorResponse},
|
||
},
|
||
summary="触发大盘复盘",
|
||
description="提交一个后台大盘复盘任务，复用 CLI 的大盘复盘运行时装配并保存报告。该人工触发入口不按交易日检查跳过；接口内部仅提供进程内/单机防重，如多实例（多 Worker/多容器）部署，需结合外部幂等机制避免重复触发。",
|
||
)
|
||
def trigger_market_review(
|
||
request: Optional[MarketReviewRequest] = Body(None),
|
||
config: Config = Depends(get_config_dep),
|
||
) -> MarketReviewAccepted:
|
||
"""Trigger market review from Web/API without blocking the request."""
|
||
request = request or MarketReviewRequest()
|
||
|
||
runtime_config = _with_request_report_language(config, request.report_language)
|
||
effective_region = request.region or (
|
||
normalize_market_review_region_lenient(runtime_config.market_review_region) or "cn"
|
||
)
|
||
|
||
lock_token = _try_acquire_market_review_lock(runtime_config)
|
||
if lock_token is None:
|
||
raise api_error(409, "duplicate_market_review", "大盘复盘正在执行中，请稍后再试")
|
||
|
||
try:
|
||
task_id = uuid.uuid4().hex
|
||
logger.info(
|
||
"[MarketReview] component=market_review action=submit trigger_source=api "
|
||
"task_id=%s region=%s send_notification=%s",
|
||
task_id,
|
||
effective_region,
|
||
request.send_notification,
|
||
)
|
||
task = get_task_queue().submit_background_task(
|
||
lambda: _run_market_review_background(
|
||
request.send_notification,
|
||
effective_region=effective_region,
|
||
lock_token=lock_token,
|
||
config=runtime_config,
|
||
query_id=task_id,
|
||
),
|
||
stock_code="market_review",
|
||
stock_name="大盘复盘",
|
||
message="大盘复盘任务已提交",
|
||
task_id=task_id,
|
||
region=effective_region,
|
||
)
|
||
except Exception:
|
||
_release_market_review_lock(lock_token)
|
||
raise
|
||
|
||
return MarketReviewAccepted(
|
||
status="accepted",
|
||
message="大盘复盘任务已提交，完成后会保存报告并按配置推送通知",
|
||
send_notification=request.send_notification,
|
||
region=effective_region,
|
||
task_id=task.task_id,
|
||
trace_id=_get_task_trace_id(task),
|
||
)
|
||
|
||
|
||
# ============================================================
|
||
# GET /tasks - 获取任务列表
|
||
# ============================================================
|
||
|
||
@router.get(
|
||
"/tasks",
|
||
response_model=TaskListResponse,
|
||
responses={
|
||
200: {"description": "任务列表"},
|
||
},
|
||
summary="获取分析任务列表",
|
||
description="获取当前所有分析任务，可按状态筛选"
|
||
)
|
||
def get_task_list(
|
||
status: Optional[str] = Query(
|
||
None,
|
||
description="筛选状态：pending, processing, completed, failed, cancel_requested, cancelled（支持逗号分隔多个）"
|
||
),
|
||
limit: int = Query(20, description="返回数量限制", ge=1, le=100),
|
||
) -> TaskListResponse:
|
||
"""
|
||
获取分析任务列表
|
||
|
||
Args:
|
||
status: 状态筛选（可选）
|
||
limit: 返回数量限制
|
||
|
||
Returns:
|
||
TaskListResponse: 任务列表响应
|
||
"""
|
||
task_queue = get_task_queue()
|
||
|
||
# 获取所有任务
|
||
all_tasks = task_queue.list_all_tasks(limit=limit)
|
||
|
||
# 状态筛选
|
||
if status:
|
||
status_list = [s.strip().lower() for s in status.split(",")]
|
||
all_tasks = [t for t in all_tasks if t.status.value in status_list]
|
||
|
||
# 统计信息
|
||
stats = task_queue.get_task_stats()
|
||
|
||
# 转换为 Schema
|
||
task_infos = [
|
||
TaskInfo(
|
||
task_id=t.task_id,
|
||
trace_id=_get_task_trace_id(t),
|
||
stock_code=t.stock_code,
|
||
stock_name=t.stock_name,
|
||
status=t.status.value,
|
||
progress=t.progress,
|
||
message=t.message,
|
||
report_type=t.report_type,
|
||
created_at=t.created_at.isoformat(),
|
||
started_at=t.started_at.isoformat() if t.started_at else None,
|
||
completed_at=t.completed_at.isoformat() if t.completed_at else None,
|
||
error=t.error,
|
||
original_query=t.original_query,
|
||
selection_source=t.selection_source,
|
||
analysis_phase=t.analysis_phase,
|
||
skills=getattr(t, "skills", None),
|
||
region=t.region,
|
||
asset_type=_task_asset_type(t),
|
||
)
|
||
for t in all_tasks
|
||
]
|
||
|
||
return TaskListResponse(
|
||
total=stats["total"],
|
||
pending=stats["pending"],
|
||
processing=stats["processing"],
|
||
tasks=task_infos,
|
||
)
|
||
|
||
|
||
# ============================================================
|
||
# GET /tasks/stream - SSE 实时推送
|
||
# ============================================================
|
||
|
||
@router.get(
|
||
"/tasks/stream",
|
||
responses={
|
||
200: {"description": "SSE 事件流", "content": {"text/event-stream": {}}},
|
||
},
|
||
summary="任务状态 SSE 流",
|
||
description="通过 Server-Sent Events 实时推送任务状态变化"
|
||
)
|
||
async def task_stream():
|
||
"""
|
||
SSE 任务状态流
|
||
|
||
事件类型：
|
||
- connected: 连接成功
|
||
- task_created: 新任务创建
|
||
- task_started: 任务开始执行
|
||
- task_progress: 任务阶段进度更新
|
||
- task_completed: 任务完成
|
||
- task_failed: 任务失败
|
||
- heartbeat: 心跳（每 30 秒）
|
||
|
||
Returns:
|
||
StreamingResponse: SSE 事件流
|
||
"""
|
||
async def event_generator():
|
||
task_queue = get_task_queue()
|
||
event_queue: asyncio.Queue = asyncio.Queue()
|
||
|
||
# 发送连接成功事件
|
||
yield _format_sse_event("connected", {"message": "Connected to task stream"})
|
||
|
||
# 发送当前进行中的任务
|
||
pending_tasks = task_queue.list_pending_tasks()
|
||
for task in pending_tasks:
|
||
yield _format_sse_event("task_created", task.to_dict())
|
||
|
||
# 订阅任务事件
|
||
task_queue.subscribe(event_queue)
|
||
|
||
try:
|
||
while True:
|
||
try:
|
||
# 等待事件，超时发送心跳
|
||
event = await asyncio.wait_for(event_queue.get(), timeout=30)
|
||
yield _format_sse_event(event["type"], event["data"])
|
||
except asyncio.TimeoutError:
|
||
# 心跳
|
||
yield _format_sse_event("heartbeat", {
|
||
"timestamp": datetime.now().isoformat()
|
||
})
|
||
except asyncio.CancelledError:
|
||
logger.debug("SSE client disconnected, cancelling event generator")
|
||
raise
|
||
finally:
|
||
task_queue.unsubscribe(event_queue)
|
||
|
||
return StreamingResponse(
|
||
event_generator(),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache",
|
||
"Connection": "keep-alive",
|
||
"X-Accel-Buffering": "no", # 禁用 Nginx 缓冲
|
||
}
|
||
)
|
||
|
||
|
||
def _format_sse_event(event_type: str, data: Dict[str, Any]) -> str:
|
||
"""
|
||
格式化 SSE 事件
|
||
|
||
Args:
|
||
event_type: 事件类型
|
||
data: 事件数据
|
||
|
||
Returns:
|
||
SSE 格式字符串
|
||
"""
|
||
return f"event: {event_type}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
|
||
|
||
|
||
def _load_history_run_flow_by_query_id(
|
||
query_id: str,
|
||
*,
|
||
code: Optional[str] = None,
|
||
report_type: Optional[str] = None,
|
||
fail_open: bool = False,
|
||
) -> Optional[RunFlowSnapshot]:
|
||
try:
|
||
from src.storage import DatabaseManager
|
||
from src.services.history_service import HistoryService
|
||
|
||
service = HistoryService(DatabaseManager.get_instance())
|
||
return service.resolve_and_get_run_flow(
|
||
query_id,
|
||
code=code,
|
||
report_type=report_type,
|
||
)
|
||
except Exception as e:
|
||
if fail_open:
|
||
logger.debug(
|
||
"load history run-flow failed, falling back to task skeleton: query_id=%s err=%s",
|
||
query_id,
|
||
e,
|
||
)
|
||
return None
|
||
raise
|
||
|
||
|
||
@router.get(
|
||
"/tasks/{task_id}/flow",
|
||
response_model=RunFlowSnapshot,
|
||
responses={
|
||
200: {"description": "任务运行流快照"},
|
||
404: {"description": "任务不存在", "model": ErrorResponse},
|
||
500: {"description": "服务器错误", "model": ErrorResponse},
|
||
},
|
||
summary="获取分析任务运行流",
|
||
description="根据 task_id 查询任务数据流/信息流快照；活跃任务缺少诊断时返回骨架流。",
|
||
)
|
||
def get_task_run_flow(task_id: str) -> RunFlowSnapshot:
|
||
"""
|
||
查询分析任务运行流。
|
||
|
||
Active tasks are served from the in-memory task queue. Completed tasks try
|
||
to hydrate from persisted history diagnostics using the same task_id/query_id.
|
||
"""
|
||
task_queue = get_task_queue()
|
||
task = task_queue.get_task(task_id)
|
||
|
||
if task:
|
||
if task.status == TaskStatusEnum.COMPLETED:
|
||
task_report_type = _history_report_type_for_task_flow(
|
||
getattr(task, "report_type", None)
|
||
)
|
||
task_stock_code = _safe_task_flow_text(getattr(task, "stock_code", None), max_length=32)
|
||
if task_report_type == "market_review":
|
||
task_stock_code = "MARKET"
|
||
history_snapshot = _load_history_run_flow_by_query_id(
|
||
task_id,
|
||
code=task_stock_code,
|
||
report_type=task_report_type,
|
||
fail_open=True,
|
||
)
|
||
if history_snapshot is not None:
|
||
return history_snapshot
|
||
return build_task_run_flow_snapshot(task)
|
||
|
||
try:
|
||
history_snapshot = _load_history_run_flow_by_query_id(task_id)
|
||
if history_snapshot is not None:
|
||
return history_snapshot
|
||
except Exception as e:
|
||
logger.error(f"查询任务运行流失败: {e}", exc_info=True)
|
||
raise api_error(500, "internal_error", f"查询任务运行流失败: {str(e)}")
|
||
|
||
raise api_error(404, "not_found", f"任务 {task_id} 不存在或已过期")
|
||
|
||
|
||
def _safe_task_flow_text(value: Any, *, max_length: int) -> Optional[str]:
|
||
if value is None:
|
||
return None
|
||
text = str(value).strip()
|
||
if not text:
|
||
return None
|
||
return text[:max_length]
|
||
|
||
|
||
def _history_report_type_for_task_flow(value: Any) -> Optional[str]:
|
||
text = _safe_task_flow_text(value, max_length=64)
|
||
if text is None:
|
||
return None
|
||
normalized = text.lower().strip().replace("-", "_")
|
||
aliases = {
|
||
"detailed": "full",
|
||
"simple": "simple",
|
||
"full": "full",
|
||
"brief": "brief",
|
||
"market": "market_review",
|
||
"market_review": "market_review",
|
||
}
|
||
return aliases.get(normalized, normalized)
|
||
|
||
|
||
def _datetime_to_iso(value: Any) -> Optional[str]:
|
||
if isinstance(value, datetime):
|
||
return value.isoformat()
|
||
if isinstance(value, str) and value.strip():
|
||
return value
|
||
return None
|
||
|
||
|
||
def _extract_report_created_at(payload: Dict[str, Any]) -> Optional[str]:
|
||
report = payload.get("report")
|
||
if not isinstance(report, dict):
|
||
return None
|
||
|
||
meta = report.get("meta")
|
||
if not isinstance(meta, dict):
|
||
return None
|
||
|
||
return _datetime_to_iso(meta.get("created_at"))
|
||
|
||
|
||
def _display_stock_code_from_index(stock_code: Any) -> str:
|
||
code = str(stock_code or "").strip()
|
||
if not code:
|
||
return code
|
||
return resolve_index_stock_code(code) or code
|
||
|
||
|
||
def _display_market_phase_summary(stock_code: Any, context_snapshot: Any) -> Any:
|
||
return rebuild_market_phase_summary_for_stock_code(
|
||
_display_stock_code_from_index(stock_code),
|
||
context_snapshot,
|
||
)
|
||
|
||
|
||
def _prepare_report_for_task_enrichment(
|
||
report_data: Dict[str, Any],
|
||
created_at: Optional[str],
|
||
) -> Dict[str, Any]:
|
||
enriched_report = dict(report_data)
|
||
meta = dict(enriched_report.get("meta") or {})
|
||
if created_at and not _datetime_to_iso(meta.get("created_at")):
|
||
meta["created_at"] = created_at
|
||
enriched_report["meta"] = meta
|
||
return enriched_report
|
||
|
||
|
||
def _first_non_empty_report_value(*values: Any) -> Any:
|
||
for value in values:
|
||
if value is None:
|
||
continue
|
||
if isinstance(value, str) and not value.strip():
|
||
continue
|
||
return value
|
||
return None
|
||
|
||
|
||
def _ensure_report_action_fields(report_data: Dict[str, Any]) -> Dict[str, Any]:
|
||
enriched_report = dict(report_data)
|
||
meta = dict(enriched_report.get("meta") or {})
|
||
summary = dict(enriched_report.get("summary") or {})
|
||
details = enriched_report.get("details") if isinstance(enriched_report.get("details"), dict) else {}
|
||
raw_result = details.get("raw_result") if isinstance(details.get("raw_result"), dict) else {}
|
||
report_language = normalize_report_language(
|
||
meta.get("report_language") or raw_result.get("report_language")
|
||
)
|
||
action_fields = build_action_fields(
|
||
operation_advice=raw_result.get("operation_advice") or summary.get("operation_advice"),
|
||
explicit_action=raw_result.get("action") or summary.get("action"),
|
||
report_type=meta.get("report_type"),
|
||
report_language=report_language,
|
||
sentiment_score=_first_non_empty_report_value(
|
||
summary.get("sentiment_score"),
|
||
raw_result.get("sentiment_score"),
|
||
),
|
||
guardrail_reason=_extract_guardrail_reason(raw_result),
|
||
align_with_score=True,
|
||
)
|
||
summary["action"] = action_fields["action"]
|
||
summary["action_label"] = action_fields["action_label"]
|
||
enriched_report["summary"] = summary
|
||
return enriched_report
|
||
|
||
|
||
def _build_task_analysis_result(task: Any) -> AnalysisResultResponse:
|
||
"""
|
||
Normalize an in-memory completed task result to the public API contract.
|
||
|
||
Older AnalysisService payloads contain stock_code/stock_name/report only.
|
||
The status endpoint owns the task metadata, so it can supply the missing
|
||
response fields without waiting for the database fallback path.
|
||
"""
|
||
payload = dict(task.result)
|
||
if not payload.get("query_id"):
|
||
payload["query_id"] = task.task_id
|
||
if not payload.get("trace_id"):
|
||
payload["trace_id"] = _get_task_trace_id(task) or task.task_id
|
||
if not payload.get("stock_code"):
|
||
payload["stock_code"] = task.stock_code
|
||
display_stock_code = _display_stock_code_from_index(payload.get("stock_code"))
|
||
if display_stock_code:
|
||
payload["stock_code"] = display_stock_code
|
||
|
||
if not payload.get("stock_name") and getattr(task, "stock_name", None):
|
||
payload["stock_name"] = task.stock_name
|
||
|
||
if not payload.get("created_at"):
|
||
payload["created_at"] = (
|
||
_extract_report_created_at(payload)
|
||
or _datetime_to_iso(getattr(task, "created_at", None))
|
||
or _datetime_to_iso(getattr(task, "completed_at", None))
|
||
or datetime.now().isoformat()
|
||
)
|
||
|
||
report_data = payload.get("report")
|
||
stock_code = payload.get("stock_code")
|
||
query_id = payload.get("query_id")
|
||
report_enriched = False
|
||
|
||
if isinstance(report_data, dict) and stock_code and query_id:
|
||
context_snapshot, fundamental_snapshot, raw_result_snapshot = _load_sync_fundamental_sources(
|
||
query_id=query_id,
|
||
stock_code=stock_code,
|
||
)
|
||
report_task_details = report_data.get("details")
|
||
report_task_raw_result = (
|
||
report_task_details.get("raw_result")
|
||
if isinstance(report_task_details, dict)
|
||
else None
|
||
)
|
||
should_rebuild_report = (
|
||
context_snapshot is not None
|
||
or fundamental_snapshot is not None
|
||
or raw_result_snapshot is not None
|
||
or report_task_raw_result is not None
|
||
)
|
||
if should_rebuild_report:
|
||
try:
|
||
report = _build_analysis_report(
|
||
_prepare_report_for_task_enrichment(
|
||
report_data,
|
||
payload.get("created_at"),
|
||
),
|
||
query_id,
|
||
stock_code,
|
||
payload.get("stock_name") or getattr(task, "stock_name", None),
|
||
context_snapshot=context_snapshot,
|
||
fallback_fundamental_payload=fundamental_snapshot,
|
||
fallback_raw_result_payload=raw_result_snapshot or payload,
|
||
)
|
||
payload["report"] = report.model_dump()
|
||
report_enriched = True
|
||
except Exception as e:
|
||
logger.debug(
|
||
"enrich in-memory task report failed (fail-open): task_id=%s err=%s",
|
||
getattr(task, "task_id", None),
|
||
e,
|
||
)
|
||
|
||
if not report_enriched and isinstance(report_data, dict):
|
||
meta = report_data.get("meta")
|
||
if isinstance(meta, dict) and display_stock_code:
|
||
raw_meta_code = meta.get("stock_code") or getattr(task, "stock_code", None)
|
||
meta["stock_code"] = display_stock_code
|
||
meta["market_phase_summary"] = _display_market_phase_summary(
|
||
raw_meta_code,
|
||
{"market_phase_summary": meta.get("market_phase_summary")},
|
||
)
|
||
payload["report"] = _ensure_report_action_fields(report_data)
|
||
|
||
return AnalysisResultResponse.model_validate(payload)
|
||
|
||
|
||
# ============================================================
|
||
# GET /status/{task_id} - 查询单个任务状态
|
||
# ============================================================
|
||
|
||
@router.get(
|
||
"/status/{task_id}",
|
||
response_model=TaskStatus,
|
||
responses={
|
||
200: {"description": "任务状态"},
|
||
404: {"description": "任务不存在", "model": ErrorResponse},
|
||
},
|
||
summary="查询分析任务状态",
|
||
description="根据 task_id 查询单个任务的状态"
|
||
)
|
||
def get_analysis_status(task_id: str) -> TaskStatus:
|
||
"""
|
||
查询分析任务状态
|
||
|
||
优先从任务队列查询，如果不存在则从数据库查询历史记录
|
||
|
||
Args:
|
||
task_id: 任务 ID
|
||
|
||
Returns:
|
||
TaskStatus: 任务状态信息
|
||
|
||
Raises:
|
||
HTTPException: 404 - 任务不存在
|
||
"""
|
||
# 1. 先从任务队列查询
|
||
task_queue = get_task_queue()
|
||
task = task_queue.get_task(task_id)
|
||
|
||
if task:
|
||
result: Optional[AnalysisResultResponse] = None
|
||
market_review_report = None
|
||
market_review_payload = None
|
||
|
||
if task.status == TaskStatusEnum.COMPLETED and isinstance(task.result, dict):
|
||
if task.stock_code == "market_review":
|
||
report_text = task.result.get("result")
|
||
if isinstance(report_text, str) and report_text.strip():
|
||
market_review_report = report_text
|
||
payload = task.result.get("market_review_payload")
|
||
if isinstance(payload, dict):
|
||
market_review_payload = payload
|
||
else:
|
||
try:
|
||
result = _build_task_analysis_result(task)
|
||
except Exception:
|
||
logger.warning(
|
||
"解析任务结果失败，回退为空返回: task_id=%s",
|
||
task.task_id,
|
||
)
|
||
|
||
return TaskStatus(
|
||
task_id=task.task_id,
|
||
trace_id=_get_task_trace_id(task),
|
||
status=task.status.value,
|
||
progress=task.progress,
|
||
result=result,
|
||
market_review_report=market_review_report,
|
||
market_review_payload=market_review_payload,
|
||
region=task.region,
|
||
error=task.error,
|
||
stock_name=task.stock_name,
|
||
original_query=task.original_query,
|
||
selection_source=task.selection_source,
|
||
analysis_phase=task.analysis_phase,
|
||
skills=getattr(task, "skills", None),
|
||
)
|
||
|
||
# 2. 从数据库查询已完成的记录
|
||
try:
|
||
from src.storage import DatabaseManager
|
||
db = DatabaseManager.get_instance()
|
||
records = db.get_analysis_history(query_id=task_id, limit=1)
|
||
|
||
if records:
|
||
record = records[0]
|
||
raw_result = parse_json_field(record.raw_result)
|
||
if getattr(record, "report_type", None) == "market_review":
|
||
market_review_report = None
|
||
context_snapshot = parse_json_field(getattr(record, "context_snapshot", None))
|
||
market_review_payload = None
|
||
region = None
|
||
if isinstance(context_snapshot, dict):
|
||
raw_region = context_snapshot.get("market_review_region")
|
||
if isinstance(raw_region, str) and raw_region.strip():
|
||
region = raw_region.strip()
|
||
payload = context_snapshot.get("market_review_payload")
|
||
if isinstance(payload, dict):
|
||
market_review_payload = payload
|
||
if region is None:
|
||
payload_region = payload.get("region")
|
||
if isinstance(payload_region, str) and payload_region.strip():
|
||
region = payload_region.strip()
|
||
if isinstance(raw_result, dict):
|
||
report_text = raw_result.get("raw_response") or raw_result.get("market_review_report")
|
||
if isinstance(report_text, str) and report_text.strip():
|
||
market_review_report = report_text
|
||
if not market_review_report and record.news_content:
|
||
market_review_report = record.news_content
|
||
|
||
return TaskStatus(
|
||
task_id=task_id,
|
||
trace_id=task_id,
|
||
status="completed",
|
||
progress=100,
|
||
result=None,
|
||
market_review_report=market_review_report,
|
||
market_review_payload=market_review_payload,
|
||
region=region,
|
||
error=None,
|
||
stock_name=record.name,
|
||
)
|
||
|
||
model_used = normalize_model_used(
|
||
(raw_result or {}).get("model_used") if isinstance(raw_result, dict) else None
|
||
)
|
||
report_language = normalize_report_language(
|
||
(raw_result or {}).get("report_language") if isinstance(raw_result, dict) else None
|
||
)
|
||
stock_name = get_localized_stock_name(record.name, record.code, report_language)
|
||
display_stock_code = _display_stock_code_from_index(record.code)
|
||
|
||
# Extract current_price / change_pct from context_snapshot
|
||
skills = None
|
||
context_snapshot = parse_json_field(getattr(record, 'context_snapshot', None))
|
||
analysis_context_pack_overview = extract_analysis_context_pack_overview(context_snapshot)
|
||
market_phase_summary = _display_market_phase_summary(record.code, context_snapshot)
|
||
api_context_snapshot = sanitize_context_snapshot_for_api(context_snapshot)
|
||
if context_snapshot and isinstance(context_snapshot, dict):
|
||
raw_skills = context_snapshot.get("skills")
|
||
if isinstance(raw_skills, list):
|
||
skills = [str(skill) for skill in raw_skills]
|
||
realtime_fields = extract_realtime_detail_fields(context_snapshot)
|
||
current_price = realtime_fields.get("current_price")
|
||
change_pct = realtime_fields.get("change_pct")
|
||
fallback_fundamental = db.get_latest_fundamental_snapshot(
|
||
query_id=task_id,
|
||
code=record.code,
|
||
)
|
||
extracted_fundamental = extract_fundamental_detail_fields(
|
||
context_snapshot=context_snapshot,
|
||
fallback_fundamental_payload=fallback_fundamental,
|
||
)
|
||
extracted_boards = extract_board_detail_fields(
|
||
context_snapshot=context_snapshot,
|
||
fallback_fundamental_payload=fallback_fundamental,
|
||
)
|
||
market_structure = extract_market_structure_detail_field(
|
||
context_snapshot,
|
||
raw_result,
|
||
)
|
||
news_disclosure = empty_news_disclosure_from_stored(
|
||
raw_result,
|
||
context_snapshot,
|
||
report_language,
|
||
)
|
||
has_board_details = (
|
||
bool(extracted_boards.get("belong_boards"))
|
||
or extracted_boards.get("sector_rankings") is not None
|
||
or extracted_boards.get("concept_rankings") is not None
|
||
)
|
||
details = None
|
||
if (
|
||
any(extracted_fundamental.values())
|
||
or has_board_details
|
||
or market_structure is not None
|
||
or context_snapshot is not None
|
||
or analysis_context_pack_overview is not None
|
||
or news_disclosure is not None
|
||
):
|
||
details = ReportDetails(
|
||
news_content=getattr(record, "news_content", None),
|
||
empty_news_disclosure=news_disclosure,
|
||
raw_result=raw_result,
|
||
context_snapshot=api_context_snapshot,
|
||
analysis_context_pack_overview=analysis_context_pack_overview,
|
||
financial_report=extracted_fundamental.get("financial_report"),
|
||
dividend_metrics=extracted_fundamental.get("dividend_metrics"),
|
||
belong_boards=extracted_boards.get("belong_boards"),
|
||
sector_rankings=extracted_boards.get("sector_rankings"),
|
||
concept_rankings=extracted_boards.get("concept_rankings"),
|
||
market_structure=market_structure,
|
||
)
|
||
|
||
raw_dict = raw_result if isinstance(raw_result, dict) else {}
|
||
action_fields = build_action_fields(
|
||
operation_advice=raw_dict.get("operation_advice") or record.operation_advice,
|
||
explicit_action=raw_dict.get("action"),
|
||
report_type=getattr(record, 'report_type', None),
|
||
report_language=report_language,
|
||
sentiment_score=record.sentiment_score if record.sentiment_score is not None else raw_dict.get("sentiment_score"),
|
||
guardrail_reason=_extract_guardrail_reason(raw_dict),
|
||
align_with_score=True,
|
||
)
|
||
|
||
# Build report from DB record so completed tasks return real data
|
||
report_dict = AnalysisReport(
|
||
meta=ReportMeta(
|
||
id=record.id,
|
||
query_id=task_id,
|
||
stock_code=display_stock_code,
|
||
stock_name=stock_name,
|
||
report_type=getattr(record, 'report_type', None),
|
||
report_language=report_language,
|
||
created_at=record.created_at.isoformat() if record.created_at else None,
|
||
model_used=model_used,
|
||
current_price=current_price,
|
||
change_pct=change_pct,
|
||
market_phase_summary=market_phase_summary,
|
||
asset_type=asset_type_from_canonical_code(record.code),
|
||
),
|
||
summary=ReportSummary(
|
||
sentiment_score=record.sentiment_score,
|
||
operation_advice=record.operation_advice,
|
||
action=action_fields["action"],
|
||
action_label=action_fields["action_label"],
|
||
trend_prediction=record.trend_prediction,
|
||
analysis_summary=record.analysis_summary,
|
||
),
|
||
strategy=ReportStrategy(
|
||
ideal_buy=_stringify_report_strategy_value(getattr(record, 'ideal_buy', None)),
|
||
secondary_buy=_stringify_report_strategy_value(getattr(record, 'secondary_buy', None)),
|
||
stop_loss=_stringify_report_strategy_value(getattr(record, 'stop_loss', None)),
|
||
take_profit=_stringify_report_strategy_value(getattr(record, 'take_profit', None)),
|
||
),
|
||
details=details,
|
||
).model_dump()
|
||
return TaskStatus(
|
||
task_id=task_id,
|
||
trace_id=task_id,
|
||
status="completed",
|
||
progress=100,
|
||
result=AnalysisResultResponse(
|
||
query_id=task_id,
|
||
trace_id=task_id,
|
||
stock_code=display_stock_code,
|
||
stock_name=stock_name,
|
||
report=report_dict,
|
||
diagnostic_summary=build_run_diagnostic_summary(
|
||
context_snapshot=context_snapshot,
|
||
raw_result=raw_result,
|
||
report_saved=True,
|
||
query_id=task_id,
|
||
stock_code=display_stock_code,
|
||
),
|
||
created_at=record.created_at.isoformat() if record.created_at else datetime.now().isoformat()
|
||
),
|
||
error=None,
|
||
skills=skills,
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"查询任务状态失败: {e}", exc_info=True)
|
||
raise api_error(500, "internal_error", f"查询任务状态失败: {str(e)}")
|
||
|
||
# 3. 任务不存在
|
||
raise api_error(404, "not_found", f"任务 {task_id} 不存在或已过期")
|
||
|
||
|
||
# ============================================================
|
||
# 辅助函数
|
||
# ============================================================
|
||
|
||
def _load_sync_fundamental_sources(
|
||
query_id: str,
|
||
stock_code: str,
|
||
) -> tuple[Optional[Any], Optional[Dict[str, Any]], Optional[Any]]:
|
||
"""
|
||
Load report enrichment payloads for sync analyze response.
|
||
"""
|
||
try:
|
||
from src.storage import DatabaseManager
|
||
|
||
db = DatabaseManager.get_instance()
|
||
records = db.get_analysis_history(query_id=query_id, code=stock_code, limit=1)
|
||
context_snapshot = None
|
||
raw_result_snapshot = None
|
||
if records:
|
||
latest_record = records[0]
|
||
context_snapshot = parse_json_field(getattr(latest_record, "context_snapshot", None))
|
||
raw_result_snapshot = parse_json_field(getattr(latest_record, "raw_result", None))
|
||
|
||
fallback_fundamental = db.get_latest_fundamental_snapshot(
|
||
query_id=query_id,
|
||
code=stock_code,
|
||
)
|
||
return context_snapshot, fallback_fundamental, raw_result_snapshot
|
||
except Exception as e:
|
||
logger.debug(
|
||
"load sync fundamental sources failed (fail-open): query_id=%s stock_code=%s err=%s",
|
||
query_id,
|
||
stock_code,
|
||
e,
|
||
)
|
||
return None, None, None
|
||
|
||
|
||
def _stringify_report_strategy_value(value: Any) -> Optional[str]:
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, str):
|
||
return value
|
||
return str(value)
|
||
|
||
|
||
def _build_analysis_report(
|
||
report_data: Dict[str, Any],
|
||
query_id: str,
|
||
stock_code: str,
|
||
stock_name: Optional[str] = None,
|
||
context_snapshot: Optional[Any] = None,
|
||
fallback_fundamental_payload: Optional[Dict[str, Any]] = None,
|
||
fallback_raw_result_payload: Optional[Any] = None,
|
||
) -> AnalysisReport:
|
||
"""
|
||
构建符合 API 规范的分析报告
|
||
|
||
Args:
|
||
report_data: 原始报告数据
|
||
query_id: 查询 ID
|
||
stock_code: 股票代码
|
||
stock_name: 股票名称
|
||
context_snapshot: 上下文快照（可选）
|
||
fallback_fundamental_payload: 基本面快照 payload（可选）
|
||
fallback_raw_result_payload: 原始分析结果 payload（可选）
|
||
|
||
Returns:
|
||
AnalysisReport: 结构化的分析报告
|
||
"""
|
||
meta_data = report_data.get("meta", {})
|
||
summary_data = report_data.get("summary", {})
|
||
strategy_data = report_data.get("strategy", {})
|
||
details_data = report_data.get("details", {})
|
||
report_language = normalize_report_language(
|
||
meta_data.get("report_language")
|
||
or (context_snapshot or {}).get("report_language")
|
||
or getattr(Config.get_instance(), "report_language", "zh")
|
||
)
|
||
display_stock_code = _display_stock_code_from_index(meta_data.get("stock_code", stock_code))
|
||
localized_stock_name = get_localized_stock_name(
|
||
meta_data.get("stock_name", stock_name),
|
||
display_stock_code,
|
||
report_language,
|
||
)
|
||
realtime_fields = extract_realtime_detail_fields(context_snapshot)
|
||
current_price = meta_data.get("current_price")
|
||
if current_price is None:
|
||
current_price = realtime_fields.get("current_price")
|
||
change_pct = meta_data.get("change_pct")
|
||
if change_pct is None:
|
||
change_pct = realtime_fields.get("change_pct")
|
||
raw_stock_code = meta_data.get("stock_code", stock_code)
|
||
market_phase_summary = _display_market_phase_summary(raw_stock_code, context_snapshot)
|
||
if market_phase_summary is None:
|
||
meta_phase_summary = meta_data.get("market_phase_summary")
|
||
if meta_phase_summary is not None:
|
||
market_phase_summary = _display_market_phase_summary(
|
||
raw_stock_code,
|
||
{"market_phase_summary": meta_phase_summary},
|
||
)
|
||
|
||
meta = ReportMeta(
|
||
query_id=meta_data.get("query_id", query_id),
|
||
stock_code=display_stock_code,
|
||
stock_name=localized_stock_name,
|
||
report_type=meta_data.get("report_type", "detailed"),
|
||
report_language=report_language,
|
||
created_at=meta_data.get("created_at", datetime.now().isoformat()),
|
||
current_price=current_price,
|
||
change_pct=change_pct,
|
||
model_used=normalize_model_used(meta_data.get("model_used")),
|
||
market_phase_summary=market_phase_summary,
|
||
asset_type=asset_type_from_canonical_code(raw_stock_code),
|
||
)
|
||
|
||
def _looks_like_raw_result_payload(candidate: Any) -> bool:
|
||
return (
|
||
isinstance(candidate, dict)
|
||
and (
|
||
"analysis_summary" in candidate
|
||
or "operation_advice" in candidate
|
||
or "trend_prediction" in candidate
|
||
or "sentiment_score" in candidate
|
||
or "market_structure_context" in candidate
|
||
or "model_used" in candidate
|
||
or "dashboard" in candidate
|
||
or "action" in candidate
|
||
)
|
||
)
|
||
|
||
raw_result_data = details_data.get("raw_result")
|
||
if not isinstance(raw_result_data, dict):
|
||
raw_result_data = {}
|
||
if isinstance(fallback_raw_result_payload, dict):
|
||
if isinstance(fallback_raw_result_payload.get("raw_result"), dict):
|
||
raw_result_data = fallback_raw_result_payload["raw_result"]
|
||
elif _looks_like_raw_result_payload(fallback_raw_result_payload):
|
||
raw_result_data = fallback_raw_result_payload
|
||
if not raw_result_data and isinstance(details_data, dict):
|
||
raw_result_data = details_data
|
||
action_fields = build_action_fields(
|
||
operation_advice=(
|
||
raw_result_data.get("operation_advice")
|
||
or details_data.get("operation_advice")
|
||
or summary_data.get("operation_advice")
|
||
),
|
||
explicit_action=raw_result_data.get("action") or details_data.get("action") or summary_data.get("action"),
|
||
report_type=meta.report_type,
|
||
report_language=report_language,
|
||
sentiment_score=_first_non_empty_report_value(
|
||
summary_data.get("sentiment_score"),
|
||
raw_result_data.get("sentiment_score"),
|
||
details_data.get("sentiment_score"),
|
||
),
|
||
guardrail_reason=_extract_guardrail_reason(raw_result_data),
|
||
align_with_score=True,
|
||
)
|
||
|
||
summary = ReportSummary(
|
||
analysis_summary=summary_data.get("analysis_summary"),
|
||
operation_advice=summary_data.get("operation_advice"),
|
||
action=action_fields["action"],
|
||
action_label=action_fields["action_label"],
|
||
trend_prediction=summary_data.get("trend_prediction"),
|
||
sentiment_score=summary_data.get("sentiment_score"),
|
||
sentiment_label=summary_data.get("sentiment_label")
|
||
)
|
||
|
||
strategy = None
|
||
if strategy_data:
|
||
strategy = ReportStrategy(
|
||
ideal_buy=_stringify_report_strategy_value(strategy_data.get("ideal_buy")),
|
||
secondary_buy=_stringify_report_strategy_value(strategy_data.get("secondary_buy")),
|
||
stop_loss=_stringify_report_strategy_value(strategy_data.get("stop_loss")),
|
||
take_profit=_stringify_report_strategy_value(strategy_data.get("take_profit"))
|
||
)
|
||
|
||
extracted_fundamental = extract_fundamental_detail_fields(
|
||
context_snapshot=context_snapshot,
|
||
fallback_fundamental_payload=fallback_fundamental_payload,
|
||
)
|
||
extracted_boards = extract_board_detail_fields(
|
||
context_snapshot=context_snapshot,
|
||
fallback_fundamental_payload=fallback_fundamental_payload,
|
||
)
|
||
market_structure = None
|
||
for raw_candidate in (fallback_raw_result_payload, raw_result_data, details_data):
|
||
if raw_candidate is None:
|
||
continue
|
||
market_structure = extract_market_structure_detail_field(
|
||
context_snapshot,
|
||
raw_candidate,
|
||
)
|
||
if market_structure is not None:
|
||
break
|
||
analysis_context_pack_overview = extract_analysis_context_pack_overview(context_snapshot)
|
||
api_context_snapshot = sanitize_context_snapshot_for_api(context_snapshot)
|
||
news_disclosure = empty_news_disclosure_from_stored(
|
||
raw_result_data,
|
||
context_snapshot,
|
||
report_language,
|
||
)
|
||
if news_disclosure is None and isinstance(details_data, dict):
|
||
news_disclosure = details_data.get("empty_news_disclosure")
|
||
details = None
|
||
has_board_details = (
|
||
bool(extracted_boards.get("belong_boards"))
|
||
or extracted_boards.get("sector_rankings") is not None
|
||
or extracted_boards.get("concept_rankings") is not None
|
||
)
|
||
if (
|
||
details_data
|
||
or any(extracted_fundamental.values())
|
||
or has_board_details
|
||
or market_structure is not None
|
||
or context_snapshot is not None
|
||
or analysis_context_pack_overview is not None
|
||
or news_disclosure is not None
|
||
):
|
||
details = ReportDetails(
|
||
news_content=details_data.get("news_summary") or details_data.get("news_content"),
|
||
empty_news_disclosure=news_disclosure,
|
||
raw_result=raw_result_data,
|
||
context_snapshot=api_context_snapshot,
|
||
analysis_context_pack_overview=analysis_context_pack_overview,
|
||
financial_report=extracted_fundamental.get("financial_report"),
|
||
dividend_metrics=extracted_fundamental.get("dividend_metrics"),
|
||
belong_boards=extracted_boards.get("belong_boards"),
|
||
sector_rankings=extracted_boards.get("sector_rankings"),
|
||
concept_rankings=extracted_boards.get("concept_rankings"),
|
||
market_structure=market_structure,
|
||
)
|
||
|
||
return AnalysisReport(
|
||
meta=meta,
|
||
summary=summary,
|
||
strategy=strategy,
|
||
details=details
|
||
)
|