diff --git a/.env.example b/.env.example index 5a0a3fe3f9f..771c2904c5b 100644 --- a/.env.example +++ b/.env.example @@ -354,7 +354,16 @@ SEARXNG_PUBLIC_INSTANCES_ENABLED=true # NEWS_INTEL_FETCH_TIMEOUT_SEC=8 # 单次每个资讯源最多采集条数 # NEWS_INTEL_MAX_ITEMS_PER_SOURCE=50 -# 注：本节仅控制资讯源采集与清理行为，不会改变 LLM / provider / base URL / 兼容回退语义。 +# NewsNow HTTP API 基地址 —— 外部依赖配置 +# - 官方项目及部署指南：https://github.com/qqhann/newsnow +# - 当前默认值 https://newsnow.busiyi.world 是公开示例实例，存在以下风险： +# * 可能因官方维护或限流而不可用 +# * 不保证稳定性或数据可靠性，仅用于演示和测试 +# - 生产环境强烈建议改为自建 NewsNow 实例，确保可控和稳定性 +# - 部署前建议验证 API 契约兼容性（参见 docs/intelligence-sources.md）： +# curl -sS "https://newsnow.busiyi.world/api/s?id=cls-hot" | python -c "import sys, json; data=json.load(sys.stdin); assert isinstance(data.get('items'), list)" +# - 本配置仅控制资讯源采集与清理行为，不会改变 LLM / provider / base URL / 兼容回退语义 +# NEWSNOW_BASE_URL=https://newsnow.busiyi.world # 乖离率阈值（%），偏离 MA5 超过此值提示不追高；强势趋势股自动放宽到 1.5 倍 # BIAS_THRESHOLD=5.0 diff --git a/api/v1/endpoints/intelligence.py b/api/v1/endpoints/intelligence.py index b527d9bfc0b..e52ab22f3c3 100644 --- a/api/v1/endpoints/intelligence.py +++ b/api/v1/endpoints/intelligence.py @@ -10,14 +10,19 @@ from api.v1.schemas.common import ErrorResponse from api.v1.schemas.intelligence import ( + IntelligenceDefaultSourceCreateResponse, + IntelligenceDefaultSourcesCreateRequest, IntelligenceFetchResponse, IntelligenceItemListResponse, IntelligenceSourceCreateRequest, IntelligenceSourceItem, IntelligenceSourceListResponse, + IntelligenceSourceTemplateCreateRequest, + IntelligenceSourceTemplateListResponse, IntelligenceSourceTestResponse, ) from src.services.intelligence_service import IntelligenceService, IntelligenceServiceError +from src.services.run_diagnostics import sanitize_diagnostic_text logger = logging.getLogger(__name__) router = APIRouter() @@ -32,10 +37,12 @@ def _not_found(message: str) -> HTTPException: def _internal_error(message: str, exc: Exception) -> HTTPException: - sanitized = IntelligenceService._sanitize_error(exc) - log_detail = sanitized if sanitized != str(exc) else exc.__class__.__name__ - logger.error("%s: %s", message, log_detail) - return HTTPException(status_code=500, detail={"error": "internal_error", "message": message}) + sanitized_error = sanitize_diagnostic_text(str(exc), max_length=300) or "internal intelligence error" + logger.error("%s: %s", message, sanitized_error) + return HTTPException( + status_code=500, + detail={"error": "internal_error", "message": f"{message}: internal intelligence service error"}, + ) @router.post("/sources", response_model=IntelligenceSourceItem, responses={400: {"model": ErrorResponse}, 500: {"model": ErrorResponse}}, summary="Create intelligence source") @@ -66,6 +73,53 @@ def list_sources( raise _internal_error("List intelligence sources failed", exc) +@router.get("/sources/templates", response_model=IntelligenceSourceTemplateListResponse, responses={500: {"model": ErrorResponse}}, summary="List built-in intelligence source templates") +def list_source_templates( + source_type: Optional[str] = Query(None), + market: Optional[str] = Query(None), +) -> IntelligenceSourceTemplateListResponse: + try: + return IntelligenceSourceTemplateListResponse(**IntelligenceService().list_source_templates( + source_type=source_type, + market=market, + )) + except Exception as exc: + raise _internal_error("List intelligence source templates failed", exc) + + +@router.post("/sources/templates/{template_id}", response_model=IntelligenceSourceItem, responses={400: {"model": ErrorResponse}, 404: {"model": ErrorResponse}, 500: {"model": ErrorResponse}}, summary="Create intelligence source from a built-in template") +def create_source_from_template( + template_id: str, + request: IntelligenceSourceTemplateCreateRequest = IntelligenceSourceTemplateCreateRequest(), +) -> IntelligenceSourceItem: + try: + return IntelligenceSourceItem(**IntelligenceService().create_source_from_template( + template_id, + request.model_dump(exclude_none=True), + )) + except IntelligenceServiceError as exc: + message = str(exc) + if "template not found" in message.lower(): + raise _not_found(message) + raise _bad_request(exc) + except Exception as exc: + raise _internal_error("Create intelligence source from template failed", exc) + + +@router.post("/sources/defaults", response_model=IntelligenceDefaultSourceCreateResponse, responses={400: {"model": ErrorResponse}, 500: {"model": ErrorResponse}}, summary="Create built-in default intelligence sources") +def create_default_sources( + request: IntelligenceDefaultSourcesCreateRequest = IntelligenceDefaultSourcesCreateRequest(), +) -> IntelligenceDefaultSourceCreateResponse: + try: + return IntelligenceDefaultSourceCreateResponse(**IntelligenceService().create_default_sources( + request.model_dump(exclude_none=True), + )) + except IntelligenceServiceError as exc: + raise _bad_request(exc) + except Exception as exc: + raise _internal_error("Create default intelligence sources failed", exc) + + @router.post("/sources/test", response_model=IntelligenceSourceTestResponse, responses={400: {"model": ErrorResponse}, 500: {"model": ErrorResponse}}, summary="Dry-run an intelligence source payload") def test_source_payload(request: IntelligenceSourceCreateRequest) -> IntelligenceSourceTestResponse: try: diff --git a/api/v1/schemas/intelligence.py b/api/v1/schemas/intelligence.py index ef94043424b..e307c0cbf72 100644 --- a/api/v1/schemas/intelligence.py +++ b/api/v1/schemas/intelligence.py @@ -7,7 +7,7 @@ from pydantic import BaseModel, Field -SourceTypeValue = Literal["rss", "atom"] +SourceTypeValue = Literal["rss", "atom", "newsnow"] ScopeTypeValue = Literal["symbol", "market", "sector"] MarketValue = Literal["cn", "hk", "us", "global"] @@ -23,6 +23,19 @@ class IntelligenceSourceCreateRequest(BaseModel): description: Optional[str] = None +class IntelligenceSourceTemplateCreateRequest(BaseModel): + name: Optional[str] = Field(None, min_length=1, max_length=100) + enabled: Optional[bool] = None + scope_type: Optional[ScopeTypeValue] = None + scope_value: Optional[str] = Field(None, max_length=64) + market: Optional[MarketValue] = None + description: Optional[str] = None + + +class IntelligenceDefaultSourcesCreateRequest(BaseModel): + enabled: Optional[bool] = None + + class IntelligenceSourceItem(BaseModel): id: int name: str @@ -40,6 +53,17 @@ class IntelligenceSourceItem(BaseModel): updated_at: Optional[str] = None +class IntelligenceSourceTemplateItem(BaseModel): + template_id: str + name: str + source_type: str + url: str + scope_type: str + scope_value: Optional[str] = None + market: str + description: Optional[str] = None + + class IntelligenceSourceListResponse(BaseModel): items: List[IntelligenceSourceItem] = Field(default_factory=list) total: int @@ -47,6 +71,22 @@ class IntelligenceSourceListResponse(BaseModel): page_size: int +class IntelligenceSourceTemplateListResponse(BaseModel): + items: List[IntelligenceSourceTemplateItem] = Field(default_factory=list) + total: int + + +class IntelligenceDefaultSourceResult(BaseModel): + created: bool + source: IntelligenceSourceItem + + +class IntelligenceDefaultSourceCreateResponse(BaseModel): + items: List[IntelligenceDefaultSourceResult] = Field(default_factory=list) + created_count: int + total: int + + class IntelligenceItem(BaseModel): id: int source_id: Optional[int] = None diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 87bac7b1d44..1e50c252ff2 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +- [新功能] #1707 资讯源新增 `newsnow` 类型、`NEWSNOW_BASE_URL` 配置和 `/api/v1/intelligence/sources/defaults` 默认源初始化接口，内置财联社热门、雪球热门股票、华尔街见闻快讯、金十数据和格隆汇事件等财经热点源，可直接拉取落库并进入既有分析证据链路；官方 NewsNow 部署指南见 https://github.com/qqhann/newsnow，生产环境建议自建实例而非使用公开示例。 + - [修复] AlphaSift 热点题材刷新在 EastMoney 瞬断且无缓存时返回友好空态，并让桌面更新保留 AlphaSift 热点缓存。 - [修复] 问股从历史报告进入后的追问会持续携带当前标的，切回或重载已有会话时可从历史消息恢复基础当前标的，并由后端阻断未明确切换时的错误股票工具调用、交易所片段和指标缩写误路由。 - [修复] 自选股加入和删除按等价股票代码匹配港股及大小写美股变体，避免 `00700`、`HK00700`、`00700.HK` 或 `aapl`、`AAPL` 被误判为不同标的。 @@ -39,7 +41,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/). - [改进] DSA 数据源链路新增 Tencent 日 K 直连 fetcher、daily source health 短期熔断，并升级 AlphaSift 默认 pin/runtime bridge，默认启用 `DAILY_SOURCE=auto`、Sina snapshot 优先级和候选级 quote context。 - [文档] 补充 AlphaSift 迁移与回退边界：明确 `ALPHASIFT_INSTALL_SPEC` 显式覆盖语义、`requirements.txt + DEFAULT_ALPHASIFT_INSTALL_SPEC` 与运行时兼容边界、以及回滚路径（关闭功能/完整 revert）说明，覆盖旧 pin 用户升级行为。 - [新功能] #1707 新增合规 RSS/Atom 资讯源配置、拉取、去重、入库、查询、retention 与基础安全校验 API，作为个股/市场资讯情报池基线。 +- [改进] #1707 个股分析、Agent 分析和大盘复盘会 fail-open 读取本地资讯/情报池，并把来源链接作为新闻上下文和 evidence 输入。 +- [改进] #1707 补齐内置 RSS/Atom 资讯源模板后端入口，修复本地资讯拉取的 requests 参数兼容、请求阶段 DNS 校验、共享地址段拒绝、坏条目跳过与港股短代码匹配。 +- [文档] #1707 阐明情报池接入仅追加本地资讯消费，不改模型名/provider/base URL/默认模型策略/回退策略/保存前清理逻辑或运行时配置迁移；结构化风险提示若出现为关键词误报；回滚可采用 `revert` 本 PR 或停用/移除本地资讯接入入口与数据。 - [文档] 本次 #1707 情报源基线仅新增 `NEWS_INTEL_*` 配置，不变更 LITELLM / ANSPIRE / Base URL / provider/model 清理与回退语义；回退策略为移除或禁用这三项情报源变量后恢复既有 LLM 配置行为。 +- [修复] #1707 默认禁用 `/api/v1/intelligence/sources/defaults` 新建源（未传 `enabled`），避免公开示例 NewsNow 实例被默认启用，同时统一 500 响应细节仅入日志、响应返回通用错误信息。 - [新功能] 个股分析历史成功保存后会从最终报告 best-effort 提取 `DecisionSignal` 决策信号，复用现有信号去重、计划质量计算和脱敏契约。 - [改进] 问股页移动端策略选择改为默认收起的按钮入口，展开后仍可多选策略并在发送后自动收起，减少对对话内容的遮挡。 diff --git a/docs/intelligence-sources.md b/docs/intelligence-sources.md index 3470e283f33..e874aca137e 100644 --- a/docs/intelligence-sources.md +++ b/docs/intelligence-sources.md @@ -5,9 +5,11 @@ Issue #1707 的首版能力聚焦“合规资讯源采集、本地沉淀、可 ## 能力范围 - 支持配置 RSS / Atom HTTP(S) 资讯源。 +- 支持 NewsNow HTTP JSON 源，默认内置财联社热门、雪球热门股票、华尔街见闻快讯、金十数据和格隆汇事件等主流财经源。 +- 支持查询内置 RSS/Atom/NewsNow 模板，并可从模板创建可测试、可启停的资讯源；也可以一键创建全部内置默认源。 - 保存资讯源配置、启用状态、作用域和最近一次拉取状态。 - 拉取条目落库到 `intelligence_items`，保存标题、摘要、URL、来源、发布时间、拉取时间、市场与作用域。 -- 按来源、作用域、市场和 URL 去重；无 URL 条目使用 `no-url:intel:` 兜底键。 +- 按 URL 去重；无 URL 条目使用 `no-url:intel:` 兜底键。 - 支持 `symbol` / `market` / `sector` 作用域，以及 `cn` / `hk` / `us` / `global` 市场标记。 - 拉取批处理采用 fail-open：单个源失败不会阻塞其他源或主分析链路。 - 支持 retention 清理，避免资讯池无限增长。 @@ -32,13 +34,28 @@ Issue #1707 的首版能力聚焦“合规资讯源采集、本地沉淀、可 NEWS_INTEL_RETENTION_DAYS=30 NEWS_INTEL_FETCH_TIMEOUT_SEC=8 NEWS_INTEL_MAX_ITEMS_PER_SOURCE=50 +NEWSNOW_BASE_URL=https://newsnow.busiyi.world ``` -### 兼容性说明 +`NEWSNOW_BASE_URL` 用于拼出 `GET {NEWSNOW_BASE_URL}/api/s?id=`。 -本节仅新增情报源持久化能力，不会改变现有模型 / provider / Base URL / LLM 配置兼容语义。 -`NEWS_INTEL_*` 仅影响情报源抓取、入库与清理逻辑，不会参与 `LITELLM_*`、`ANSPIRE_*`、`LLM_CHANNELS` 的解析与清理。 -回退路径为：移除上述变量并按既有 `.env`/历史 `LLM` 配置恢复默认行为。 +**外部依赖兼容性说明：** + +- **官方项目与部署指南**：https://github.com/qqhann/newsnow +- **当前默认值** `https://newsnow.busiyi.world` 是公开示例实例，**非官方部署**，存在以下风险： + - 可能因官方维护、限流或停止服务而不可用 + - 不保证稳定性、可靠性或数据准确性，仅用于演示和测试 + - 每个用户都指向同一公开实例，可能遭遇限流 +- **生产环境强烈建议**：自建 NewsNow 实例或接入已确认可控的私有/企业部署，以确保稳定性和数据可靠性 + +**API 契约兼容性核验（部署前必做）：** + +- 验证基础可达性和返回格式： + ```bash + curl -sS "${NEWSNOW_BASE_URL}/api/s?id=cls-hot" | python -c "import sys, json; data=json.load(sys.stdin); assert isinstance(data, dict) and isinstance(data.get('items'), list); print('OK')" + ``` +- 详细字段兼容性可参考自动化测试：`test_newsnow_source_fetches_json_items`，涵盖 `status`、`id`、`items[].title`、`items[].url`/`mobileUrl`、`items[].pubDate`/`items[].extra.date` 等字段 +- **部署实例不在自动化上线保障范围内**；如果依赖公开示例实例，部署前务必在实际生产环境执行上述验证 ## API @@ -46,11 +63,51 @@ NEWS_INTEL_MAX_ITEMS_PER_SOURCE=50 - `POST /sources`：创建资讯源。 - `GET /sources`：查询资讯源。 +- `GET /sources/templates?market=hk`：查询内置资讯源模板。 +- `POST /sources/templates/{template_id}`：从内置模板创建资讯源，可覆盖名称、启用状态、作用域和说明。 +- `POST /sources/defaults`：一键创建全部内置默认源；接口幂等，已存在的同名源会返回 `created=false`，不会重复插入。默认不传 `enabled` 时以 `false` 创建；如需默认启用可传 `{ "enabled": true }`。 - `POST /sources/test`：测试 payload，不落库。 - `POST /sources/{source_id}/fetch?dry_run=false`：拉取单个源。 - `POST /sources/fetch-enabled`：fail-open 拉取全部启用源。 - `GET /items?scope_type=market&market=cn&days=7`：查询资讯条目。 +## NewsNow 默认源 + +NewsNow 不是 RSS，而是一个聚合热点平台。DSA 直接按 HTTP API 读取它的 JSON 返回，不需要 MCP： + +```text +GET {NEWSNOW_BASE_URL}/api/s?id=cls-hot +``` + +本 PR 先接入以下财经相关默认源，保证流程能从“源配置 -> 拉取 -> 落库 -> 分析读取”跑通： + +- `cls-hot`：财联社热门，偏 A 股和题材热点。 +- `xueqiu-hotstock`：雪球热门股票，偏个股关注度。 +- `wallstreetcn-quick`：华尔街见闻快讯，偏宏观、商品和市场事件。 +- `jin10`：金十数据，偏全球宏观和外盘事件。 +- `gelonghui`：格隆汇事件，偏港股和中概股上下文。 + +如果需要更多国内平台，可以继续通过 `POST /sources` 手动添加 NewsNow 源，`source_type=newsnow`，`url` 填 `https:///api/s?id=`。如果更偏好 RSS，也可以用 RSSHub 等合规 RSS 源继续按 `source_type=rss` 接入。 + ## 后续接入建议 -首版只完成资讯源与存储基线。后续 PR 可以接入个股分析、大盘复盘、报告 evidence 展示和 Web 设置/报告查看入口。 +首版基线之上，分析链路会 best-effort 读取本地资讯池： + +- 个股传统分析会优先读取 `symbol=<股票代码>` 的资讯，并补充同市场 `market` 级资讯；内容追加到既有 `news_context`，随 AnalysisContextPack 摘要和历史 `news_content` 保存。 +- Agent 分析同样通过 `news_context` 注入本地资讯证据，避免 Agent 必须重新搜索才能看到已沉淀新闻。 +- 大盘复盘会把同市场 `market` 级资讯合并到市场新闻列表，Prompt、结构化 payload 和报告 news 字段都能看到来源链接。 +- 本次能力仅新增本地资讯消费路径，不改模型名、provider/base URL、默认模型策略、回退策略、`save_context_snapshot` 前清理逻辑或运行时配置语义；兼容现有部署配置，回滚方式为清退本地资讯接入入口或移除本地资讯源配置/数据。 + +后续 PR 可以继续完善 NewsNow HTTP provider、报告 evidence 展示和 Web 设置/报告查看入口。 + +## 兼容性与回滚说明（Issue #1707） + +- 本功能不改动第三方 LLM provider 语义，不新增 provider/model/base URL/默认模型策略/运行时路由或配置迁移分支。 +- 结构化检测提示中的模型/API 兼容风险在本次改动中不成立：`news_context` 注入链路仅复用现有 LLM 分析输入构造流程（`src/core/pipeline.py`、`src/market_analyzer.py`、`src/analyzer.py`），且不新增 `.env` 写入、保存前清理、清空/回填逻辑。 +- 回滚方式：`revert` 本 PR；如需降级配置，仅需停用并移除本地资讯源配置（含 `sources` 表与 `intelligence_items` 存量）即可，不影响原有模型、provider 或其它历史分析链路。 + +## PR 描述可复用内容（Issue #1707） + +- Refs: `#1707` +- 兼容性结论：本次仅新增本地资讯消费链路，不改模型名/provider/base URL/默认模型策略/回退策略/保存前清理逻辑/运行时配置迁移。`news_context` 与 `market_review_payload` 的扩展为 best-effort 追加，不影响既有契约与兼容性边界。 +- 回滚方案：最小回滚路径为 `revert this PR`；如仅需降级接入，可在运行时停用并清理本地资讯源（`sources` 与 `intelligence_items`）。 diff --git a/src/config.py b/src/config.py index 08f0425f49e..4f64fc94a5f 100644 --- a/src/config.py +++ b/src/config.py @@ -718,6 +718,7 @@ class Config: news_intel_retention_days: int = 30 # 本地资讯池保留天数 news_intel_fetch_timeout_sec: float = 8.0 # 单个资讯源拉取超时 news_intel_max_items_per_source: int = 50 # 单次每个资讯源最多采集条数 + newsnow_base_url: str = "https://newsnow.busiyi.world" # NewsNow HTTP API base URL (数据源侧，不影响 LLM/provider base URL) bias_threshold: float = 5.0 # 乖离率阈值（%），超过此值提示不追高 # === Agent 模式配置 === @@ -1523,6 +1524,7 @@ def _load_from_env(cls) -> 'Config': minimum=1, maximum=200, ), + newsnow_base_url=((os.getenv('NEWSNOW_BASE_URL') or '').strip().rstrip('/') or 'https://newsnow.busiyi.world'), bias_threshold=parse_env_float(os.getenv('BIAS_THRESHOLD'), 5.0, field_name='BIAS_THRESHOLD', minimum=1.0), agent_litellm_model=agent_litellm_model, agent_mode=os.getenv('AGENT_MODE', 'false').lower() == 'true', diff --git a/src/core/pipeline.py b/src/core/pipeline.py index 037712f75a9..fb879633c86 100644 --- a/src/core/pipeline.py +++ b/src/core/pipeline.py @@ -25,7 +25,7 @@ from src.config import FUNDAMENTAL_STAGE_TIMEOUT_SECONDS_DEFAULT, get_config, Config from src.storage import get_db from data_provider import DataFetcherManager -from data_provider.base import normalize_stock_code +from data_provider.base import is_bse_code, normalize_stock_code from data_provider.realtime_types import ChipDistribution from src.analyzer import ( GeminiAnalyzer, @@ -55,6 +55,7 @@ format_daily_market_context_prompt_section, ) from src.services.social_sentiment_service import SocialSentimentService +from src.services.intelligence_service import IntelligenceService from src.services.analysis_context_builder import ( AnalysisContextBuilder, PipelineAnalysisArtifacts, @@ -92,6 +93,71 @@ _DAILY_MARKET_CONTEXT_SERVICE_LOCK_INIT_GUARD = threading.Lock() +def _symbol_scope_lookup_values(code: str, market: str) -> List[str]: + """Return accepted persisted-intelligence symbol spellings for lookup.""" + raw = str(code or "").strip() + normalized = normalize_stock_code(raw) if raw else "" + values: List[str] = [] + seen: set[str] = set() + + def add(value: str) -> None: + text = str(value or "").strip() + if text and text not in seen: + seen.add(text) + values.append(text) + + def add_case_variants(value: str) -> None: + text = str(value or "").strip() + if not text: + return + add(text) + add(text.upper()) + add(text.lower()) + + add_case_variants(normalized) + add_case_variants(raw) + + normalized_upper = normalized.upper() + if normalized_upper.startswith("HK") and normalized_upper[2:].isdigit(): + digits = normalized_upper[2:] + trimmed_digits = digits.lstrip("0") or digits + add_case_variants(normalized_upper) + add_case_variants(digits) + add_case_variants(trimmed_digits) + add_case_variants(f"HK{trimmed_digits}") + add_case_variants(f"{trimmed_digits}.HK") + add_case_variants(f"{digits}.HK") + return values + + if (market or "").strip().lower() != "cn": + return values + if not (normalized.isdigit() and len(normalized) == 6): + return values + + raw_upper = raw.upper() + exchange = "" + if raw_upper.startswith(("SH", "SS")) or raw_upper.endswith((".SH", ".SS")): + exchange = "SH" + elif raw_upper.startswith("SZ") or raw_upper.endswith(".SZ"): + exchange = "SZ" + elif raw_upper.startswith("BJ") or raw_upper.endswith(".BJ"): + exchange = "BJ" + elif is_bse_code(normalized): + exchange = "BJ" + elif normalized.startswith(("5", "6", "9")): + exchange = "SH" + else: + exchange = "SZ" + + add_case_variants(f"{exchange}{normalized}") + add_case_variants(f"{exchange}.{normalized}") + add_case_variants(f"{normalized}.{exchange}") + if exchange == "SH": + add_case_variants(f"SS.{normalized}") + add_case_variants(f"{normalized}.SS") + return values + + class StockAnalysisPipeline: """ 股票分析主流程调度器 @@ -476,6 +542,11 @@ def analyze_stock( # Step 4: 多维度情报搜索（最新消息+风险排查+业绩预期） news_context = None + persisted_intelligence_context = self._load_persisted_intelligence_context( + code=code, + stock_name=stock_name, + market=market or "cn", + ) news_result_count: Optional[int] = None self._emit_progress(46, f"{stock_name}：正在检索新闻与舆情") if self.search_service is not None and self.search_service.is_available: @@ -529,6 +600,13 @@ def analyze_stock( except Exception as e: logger.warning(f"{stock_name}({code}) Social sentiment fetch failed: {e}") + if persisted_intelligence_context: + news_context = ( + f"{news_context}\n\n{persisted_intelligence_context}" + if news_context + else persisted_intelligence_context + ) + # Step 5: 获取分析上下文（技术面数据） self._emit_progress(58, f"{stock_name}：正在整理分析上下文") context = self.db.get_analysis_context(code) @@ -1116,6 +1194,20 @@ def _analyze_with_agent( except Exception as e: logger.warning(f"[{code}] Agent mode: social sentiment fetch failed: {e}") + persisted_intelligence_context = self._load_persisted_intelligence_context( + code=code, + stock_name=stock_name, + market=get_market_for_stock(normalize_stock_code(code)) or "cn", + ) + if persisted_intelligence_context: + existing = initial_context.get("news_context") + initial_context["news_context"] = ( + f"{existing}\n\n{persisted_intelligence_context}" + if existing + else persisted_intelligence_context + ) + logger.info(f"[{code}] Agent mode: local intelligence evidence injected into news_context") + # Issue #1066: ensure deep history is in DB before agent tools run self._ensure_agent_history(code) @@ -2221,6 +2313,58 @@ def _refresh_saved_diagnostic_snapshot( except Exception as exc: logger.warning("回写通知诊断快照失败（fail-open）: %s", exc) + def _load_persisted_intelligence_context( + self, + *, + code: str, + stock_name: str, + market: str, + limit: int = 6, + ) -> Optional[str]: + """Load locally persisted intelligence as fail-open evidence context.""" + try: + service = IntelligenceService() + days = max(1, int(self.config.get_effective_news_window_days() or 1)) + collected: list[Dict[str, Any]] = [] + seen_urls: set[str] = set() + symbol_filters = [ + {"scope_type": "symbol", "scope_value": scope_value, "market": market} + for scope_value in _symbol_scope_lookup_values(code, market) + ] + for filters in symbol_filters + [{"scope_type": "market", "market": market}]: + payload = service.list_items(published_days=days, page=1, page_size=limit, **filters) + for item in payload.get("items", []): + if not isinstance(item, dict): + continue + url = str(item.get("url") or "") + if url in seen_urls: + continue + seen_urls.add(url) + collected.append(item) + if len(collected) >= limit: + break + if len(collected) >= limit: + break + if not collected: + return None + lines = [f"## 本地资讯证据池（{stock_name}/{code}）"] + for idx, item in enumerate(collected[:limit], 1): + title = str(item.get("title") or "未命名资讯").strip() + summary = str(item.get("summary") or "").strip() + source = str(item.get("source") or item.get("source_name") or "local-intel").strip() + published = str(item.get("published_at") or "").strip() + url = str(item.get("url") or "").strip() + meta = " / ".join(part for part in (source, published) if part) + lines.append(f"{idx}. {title}" + (f"（{meta}）" if meta else "")) + if summary: + lines.append(f" 摘要：{summary[:220]}") + if url and not url.startswith("no-url:intel:"): + lines.append(f" 来源：{url}") + return "\n".join(lines) + except Exception as exc: + logger.debug("读取本地资讯证据失败（fail-open）: %s", exc) + return None + def _build_legacy_analysis_artifacts( self, *, diff --git a/src/market_analyzer.py b/src/market_analyzer.py index e88cc3879f0..bb1ab028c60 100644 --- a/src/market_analyzer.py +++ b/src/market_analyzer.py @@ -26,6 +26,7 @@ from src.core.market_strategy import get_market_strategy_blueprint from src.schemas.market_light import MarketLightSnapshot from src.services.run_diagnostics import record_llm_run, record_llm_run_started +from src.services.intelligence_service import IntelligenceService from data_provider.base import DataFetcherManager logger = logging.getLogger(__name__) @@ -1454,6 +1455,7 @@ def _run_daily_review_parts(self) -> MarketLightReviewResult: # 2. 搜索市场新闻 news = self.search_market_news() + news = self._merge_persisted_market_intelligence(news) # 3. 生成复盘报告 report = self.generate_market_review(overview, news) @@ -1474,6 +1476,52 @@ def _run_daily_review_parts(self) -> MarketLightReviewResult: structured_payload=structured_payload, ) + def _merge_persisted_market_intelligence(self, news: List) -> List: + """Merge local persisted market intelligence and search news with bounded prompt/payload slot preservation.""" + search_news = list(news or []) + merged_local = [] + seen_urls = { + self._get_news_field(item, "url") + for item in search_news + if self._get_news_field(item, "url") + } + try: + service = IntelligenceService() + payload = service.list_items( + scope_type="market", + market=self.region, + published_days=max(1, int(self.config.get_effective_news_window_days() or 1)), + page=1, + page_size=6, + ) + for item in payload.get("items", []): + if not isinstance(item, dict): + continue + url = str(item.get("url") or "") + if url and url in seen_urls: + continue + seen_urls.add(url) + merged_local.append({ + "title": item.get("title") or "未命名资讯", + "snippet": item.get("summary") or "", + "source": item.get("source") or item.get("source_name") or "local-intel", + "published_date": item.get("published_at") or "", + "url": "" if url.startswith("no-url:intel:") else url, + }) + except Exception as exc: + logger.debug("[大盘] %s action=load_local_intelligence status=failed error=%s", self._log_context(), exc) + merged_news = [] + merged_local_index = 0 + merged_search_index = 0 + while merged_local_index < len(merged_local) or merged_search_index < len(search_news): + if merged_local_index < len(merged_local): + merged_news.append(merged_local[merged_local_index]) + merged_local_index += 1 + if merged_search_index < len(search_news): + merged_news.append(search_news[merged_search_index]) + merged_search_index += 1 + return merged_news + def run_daily_review(self) -> str: """ 执行每日大盘复盘流程 diff --git a/src/repositories/intelligence_repo.py b/src/repositories/intelligence_repo.py index 9e2f6cd0595..5b286e546cf 100644 --- a/src/repositories/intelligence_repo.py +++ b/src/repositories/intelligence_repo.py @@ -32,6 +32,12 @@ def get_source(self, source_id: int) -> Optional[IntelligenceSource]: select(IntelligenceSource).where(IntelligenceSource.id == source_id).limit(1) ).scalar_one_or_none() + def get_source_by_name(self, name: str) -> Optional[IntelligenceSource]: + with self.db.get_session() as session: + return session.execute( + select(IntelligenceSource).where(IntelligenceSource.name == name).limit(1) + ).scalar_one_or_none() + def list_sources( self, *, @@ -99,7 +105,7 @@ def upsert_items(self, items: Iterable[Dict[str, Any]]) -> int: item_fields = dict(fields) scope_value = self._normalize_scope_value(item_fields.get("scope_value")) item_fields["scope_value"] = scope_value - source_id = fields.get("source_id") + source_id = item_fields.get("source_id") conditions = [ IntelligenceItem.url == url, IntelligenceItem.source_type == (item_fields.get("source_type") or "rss"), @@ -140,6 +146,7 @@ def list_items( market: Optional[str] = None, query: Optional[str] = None, days: Optional[int] = None, + published_days: Optional[int] = None, page: int = 1, page_size: int = 50, ) -> Tuple[List[IntelligenceItem], int]: @@ -155,6 +162,9 @@ def list_items( conditions.append(or_(IntelligenceItem.title.like(pattern), IntelligenceItem.summary.like(pattern))) if days is not None: conditions.append(IntelligenceItem.fetched_at >= datetime.now() - timedelta(days=max(1, int(days)))) + if published_days is not None: + published_cutoff = datetime.now() - timedelta(days=max(1, int(published_days))) + conditions.append(IntelligenceItem.published_at >= published_cutoff) where_clause = and_(*conditions) if conditions else True safe_page = max(1, int(page)) safe_size = max(1, min(int(page_size), 100)) diff --git a/src/services/intelligence_service.py b/src/services/intelligence_service.py index 2b23e874a23..9616fcf1d09 100644 --- a/src/services/intelligence_service.py +++ b/src/services/intelligence_service.py @@ -14,25 +14,94 @@ from datetime import datetime, timezone from email.utils import parsedate_to_datetime from typing import Any, Dict, List, Optional -from urllib.parse import urljoin, urlparse +from urllib.parse import parse_qsl, urlencode, urljoin, urlparse, urlunparse from xml.etree import ElementTree as ET import requests +from sqlalchemy.exc import IntegrityError from src.config import get_config from src.repositories.intelligence_repo import IntelligenceRepository from src.storage import IntelligenceSource, INTELLIGENCE_ITEM_NULL_SCOPE_VALUE +from src.services.run_diagnostics import sanitize_diagnostic_text logger = logging.getLogger(__name__) -_ALLOWED_SOURCE_TYPES = {"rss", "atom"} +_ALLOWED_SOURCE_TYPES = {"rss", "atom", "newsnow"} _ALLOWED_SCOPE_TYPES = {"symbol", "market", "sector"} _ALLOWED_MARKETS = {"cn", "hk", "us", "global"} _PRIVATE_HOSTNAMES = {"localhost", "localhost.localdomain"} _MAX_FEED_BYTES = 2 * 1024 * 1024 -_MAX_REDIRECTS = 5 +_MAX_FEED_REDIRECTS = 5 +_UPSTREAM_FETCH_FAILURE_MESSAGE = "fetch failed: upstream request failed" _REDIRECT_STATUS_CODES = {301, 302, 303, 307, 308} -_DNS_GUARD_LOCK = threading.RLock() _DISABLE_REQUEST_PROXIES = {"http": None, "https": None} +_DNS_GUARD_LOCK = threading.Lock() +_BUILTIN_SOURCE_TEMPLATES = [ + { + "template_id": "sec-company-news", + "name": "SEC Latest Filings", + "source_type": "rss", + "url": "https://www.sec.gov/news/pressreleases.rss", + "scope_type": "market", + "market": "us", + "description": "SEC official press release RSS feed for US market evidence.", + }, + { + "template_id": "hkex-news", + "name": "HKEX Market News", + "source_type": "rss", + "url": "https://www.hkex.com.hk/Services/RSS-Feeds/News-Releases?sc_lang=en", + "scope_type": "market", + "market": "hk", + "description": "HKEX public news entry for Hong Kong market evidence. Test before enabling.", + }, + { + "template_id": "global-marketwatch", + "name": "MarketWatch Top Stories", + "source_type": "rss", + "url": "https://feeds.content.dowjones.io/public/rss/mw_topstories", + "scope_type": "market", + "market": "global", + "description": "Public market news RSS for global market context. Test before enabling.", + }, +] +_NEWSNOW_DEFAULT_SOURCE_DEFS = [ + { + "template_id": "newsnow-cls-hot", + "name": "NewsNow 财联社热门", + "source_id": "cls-hot", + "market": "cn", + "description": "NewsNow 财联社热门财经资讯，适合 A 股大盘和题材热点。", + }, + { + "template_id": "newsnow-xueqiu-hotstock", + "name": "NewsNow 雪球热门股票", + "source_id": "xueqiu-hotstock", + "market": "cn", + "description": "NewsNow 雪球热门股票，适合捕捉 A 股和港美股散户关注度。", + }, + { + "template_id": "newsnow-wallstreetcn-quick", + "name": "NewsNow 华尔街见闻快讯", + "source_id": "wallstreetcn-quick", + "market": "cn", + "description": "NewsNow 华尔街见闻快讯，适合宏观、商品和市场事件上下文。", + }, + { + "template_id": "newsnow-jin10", + "name": "NewsNow 金十数据", + "source_id": "jin10", + "market": "global", + "description": "NewsNow 金十数据实时财经消息，适合全球宏观和外盘事件。", + }, + { + "template_id": "newsnow-gelonghui", + "name": "NewsNow 格隆汇事件", + "source_id": "gelonghui", + "market": "hk", + "description": "NewsNow 格隆汇事件资讯，适合港股和中概股市场上下文。", + }, +] class IntelligenceServiceError(ValueError): @@ -59,7 +128,10 @@ def __init__(self, repository: Optional[IntelligenceRepository] = None): def create_source(self, payload: Dict[str, Any]) -> Dict[str, Any]: fields = self._normalize_source_fields(payload) self._validate_url(fields["url"]) - return self._source_to_dict(self.repo.create_source(fields)) + try: + return self._source_to_dict(self.repo.create_source(fields)) + except IntegrityError as exc: + raise IntelligenceServiceError(f"intelligence source name already exists: {fields['name']}") from exc def list_sources(self, **filters: Any) -> Dict[str, Any]: rows, total = self.repo.list_sources(**filters) @@ -70,6 +142,46 @@ def list_sources(self, **filters: Any) -> Dict[str, Any]: "page_size": max(1, min(int(filters.get("page_size") or 50), 100)), } + def list_source_templates(self, **filters: Any) -> Dict[str, Any]: + market = str(filters.get("market") or "").strip().lower() + source_type = str(filters.get("source_type") or "").strip().lower() + templates = [] + for template in self._builtin_source_templates(): + if market and template["market"] != market: + continue + if source_type and template["source_type"] != source_type: + continue + templates.append(dict(template)) + return {"items": templates, "total": len(templates)} + + def create_source_from_template(self, template_id: str, overrides: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + selected = next( + (dict(template) for template in self._builtin_source_templates() if template["template_id"] == template_id), + None, + ) + if selected is None: + raise IntelligenceServiceError(f"Intelligence source template not found: {template_id}") + payload = {key: value for key, value in selected.items() if key != "template_id"} + payload.update({key: value for key, value in (overrides or {}).items() if value is not None}) + return self.create_source(payload) + + def create_default_sources(self, overrides: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + request_fields = dict(overrides or {}) + request_fields.setdefault("enabled", False) + created_count = 0 + items = [] + for template in self._builtin_source_templates(): + payload = {key: value for key, value in template.items() if key != "template_id"} + payload.update({key: value for key, value in request_fields.items() if value is not None}) + existing = self.repo.get_source_by_name(str(payload["name"])) + if existing is not None: + items.append({"created": False, "source": self._source_to_dict(existing)}) + continue + source = self.create_source(payload) + created_count += 1 + items.append({"created": True, "source": source}) + return {"items": items, "created_count": created_count, "total": len(items)} + def list_items(self, **filters: Any) -> Dict[str, Any]: rows, total = self.repo.list_items(**filters) return { @@ -120,24 +232,26 @@ def fetch_source(self, source_id: int, *, dry_run: bool = False) -> Dict[str, An raise def fetch_enabled_sources(self) -> Dict[str, Any]: - source_ids: List[int] = [] + rows, total = self.repo.list_sources(enabled=True, page=1, page_size=100) + results = [] page = 1 + source_count = 0 while True: - rows, total = self.repo.list_sources(enabled=True, page=page, page_size=100) - source_ids.extend(row.id for row in rows) - if not rows or len(source_ids) >= total: + for row in rows: + source_count += 1 + try: + results.append(self.fetch_source(row.id)) + except Exception as exc: + results.append({"ok": False, "source_id": row.id, "error": self._sanitize_error(exc)}) + if source_count >= total: break page += 1 - - results = [] - for source_id in source_ids: - try: - results.append(self.fetch_source(source_id)) - except Exception as exc: - results.append({"ok": False, "source_id": source_id, "error": self._sanitize_error(exc)}) + rows, _ = self.repo.list_sources(enabled=True, page=page, page_size=100) + if not rows: + break return { "ok": True, - "source_count": len(source_ids), + "source_count": source_count, "results": results, "saved_count": sum(int(item.get("saved_count") or 0) for item in results), } @@ -174,7 +288,9 @@ def _normalize_source_fields(self, payload: Dict[str, Any]) -> Dict[str, Any]: "description": description, } - def _validate_url(self, raw_url: str) -> None: + def _validate_url(self, raw_url: str, *, allow_no_url: bool = False) -> None: + if allow_no_url and raw_url.startswith("no-url:intel:"): + return parsed = urlparse(raw_url) if parsed.scheme.lower() not in {"http", "https"} or not parsed.netloc: raise IntelligenceServiceError("source url must be an absolute http(s) URL") @@ -185,55 +301,156 @@ def _validate_url(self, raw_url: str) -> None: raise IntelligenceServiceError("source url host is required") if hostname in _PRIVATE_HOSTNAMES or hostname.endswith(".local"): raise IntelligenceServiceError("source url host is not allowed") + has_public_address = False try: ip = ipaddress.ip_address(hostname) except ValueError: - self._validate_resolved_hostname(hostname) + ip = None + if ip is not None: + if self._is_blocked_ip(ip): + raise IntelligenceServiceError("source url must not target private or local network addresses") return - if self._is_blocked_ip(ip): - raise IntelligenceServiceError("source url must not target private or local network addresses") + try: + addr_infos = socket.getaddrinfo(hostname, None) + except OSError as exc: + raise IntelligenceServiceError(f"source url host DNS resolution failed: {hostname}") from exc + if not addr_infos: + raise IntelligenceServiceError(f"source url host DNS resolution failed: {hostname}") + for info in addr_infos: + try: + ip = ipaddress.ip_address(info[4][0]) + except (IndexError, ValueError): + continue + if self._is_blocked_ip(ip): + raise IntelligenceServiceError("source url must not target private or local network addresses") + has_public_address = True + if not has_public_address: + raise IntelligenceServiceError(f"source url host DNS resolution failed: {hostname}") + + @staticmethod + def _is_blocked_ip(ip: ipaddress._BaseAddress) -> bool: + return ( + not ip.is_global + or ip.is_private + or ip.is_loopback + or ip.is_link_local + or ip.is_reserved + or ip.is_multicast + ) def _fetch_feed_entries(self, fields: Dict[str, Any], *, limit: int) -> List[FeedEntry]: + if fields["source_type"] == "newsnow": + return self._fetch_newsnow_entries(fields, limit=limit) + + timeout = max(1, min(float(self.config.news_intel_fetch_timeout_sec), 30.0)) + headers = {"User-Agent": "daily-stock-analysis-intel/1.0"} self._validate_url(fields["url"]) - response = self._get_feed_response(fields["url"]) + request_url = fields["url"] + response = None try: - response.raise_for_status() - content = self._read_limited_response(response) + for _ in range(_MAX_FEED_REDIRECTS + 1): + response = self._get_with_validated_dns( + request_url, + timeout=timeout, + headers=headers, + allow_redirects=False, + stream=True, + ) + status_code = int(getattr(response, "status_code", 200)) + if status_code in _REDIRECT_STATUS_CODES: + location = getattr(response, "headers", {}).get("Location") + if not location: + raise IntelligenceServiceError("feed redirect missing Location header") + response.close() + request_url = urljoin(request_url, location) + self._validate_url(request_url) + continue + response.raise_for_status() + break + else: + raise IntelligenceServiceError(f"feed redirect chain exceeds {_MAX_FEED_REDIRECTS}") + + self._validate_url(response.url or request_url) + + if hasattr(response, "iter_content") and callable(response.iter_content): + chunks = [] + total = 0 + for chunk in response.iter_content(chunk_size=8192): + if not chunk: + continue + total += len(chunk) + if total > _MAX_FEED_BYTES: + raise IntelligenceServiceError("feed response is too large") + chunks.append(chunk) + content = b"".join(chunks) + else: + content = response.content[: _MAX_FEED_BYTES + 1] + if len(content) > _MAX_FEED_BYTES: + raise IntelligenceServiceError("feed response is too large") + return self._parse_feed(content, source_name=fields["name"], limit=limit) + except IntelligenceServiceError: + raise + except Exception as exc: + raise IntelligenceServiceError(_UPSTREAM_FETCH_FAILURE_MESSAGE) from exc finally: - self._close_response(response) - return self._parse_feed(content, source_name=fields["name"], limit=limit) + if response is not None: + response.close() - def _get_feed_response(self, raw_url: str) -> requests.Response: - current_url = raw_url + def _fetch_newsnow_entries(self, fields: Dict[str, Any], *, limit: int) -> List[FeedEntry]: timeout = max(1, min(float(self.config.news_intel_fetch_timeout_sec), 30.0)) - headers = {"User-Agent": "daily-stock-analysis-intel/1.0"} - for redirect_index in range(_MAX_REDIRECTS + 1): - self._validate_url(current_url) + headers = { + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 daily-stock-analysis-intel/1.0" + ), + "Accept": "application/json", + } + self._validate_url(fields["url"]) + response = None + try: response = self._get_with_validated_dns( - current_url, + fields["url"], timeout=timeout, headers=headers, allow_redirects=False, stream=True, ) - status_code = self._response_status_code(response) - if status_code not in _REDIRECT_STATUS_CODES: - try: - self._validate_url(response.url or current_url) - except Exception: - self._close_response(response) - raise - return response - if redirect_index >= _MAX_REDIRECTS: - self._close_response(response) - raise IntelligenceServiceError("too many redirects while fetching source url") - location = response.headers.get("Location") if response.headers else None - self._close_response(response) - if not location: - raise IntelligenceServiceError("redirect response is missing Location header") - current_url = urljoin(current_url, location.strip()) - self._validate_url(current_url) - raise IntelligenceServiceError("too many redirects while fetching source url") + status_code = int(getattr(response, "status_code", 200)) + if status_code in _REDIRECT_STATUS_CODES: + raise IntelligenceServiceError("NewsNow API redirects are not followed") + response.raise_for_status() + self._validate_url(response.url or fields["url"]) + + content = self._read_limited_response(response) + try: + payload = json.loads(content.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise IntelligenceServiceError(f"invalid NewsNow JSON response: {exc}") from exc + return self._parse_newsnow_payload(payload, source_name=fields["name"], limit=limit) + except IntelligenceServiceError: + raise + except Exception as exc: + raise IntelligenceServiceError(_UPSTREAM_FETCH_FAILURE_MESSAGE) from exc + finally: + if response is not None: + response.close() + + def _read_limited_response(self, response: requests.Response) -> bytes: + if hasattr(response, "iter_content") and callable(response.iter_content): + chunks = [] + total = 0 + for chunk in response.iter_content(chunk_size=8192): + if not chunk: + continue + total += len(chunk) + if total > _MAX_FEED_BYTES: + raise IntelligenceServiceError("feed response is too large") + chunks.append(chunk) + return b"".join(chunks) + content = response.content[: _MAX_FEED_BYTES + 1] + if len(content) > _MAX_FEED_BYTES: + raise IntelligenceServiceError("feed response is too large") + return content def _get_with_validated_dns(self, raw_url: str, **kwargs: Any) -> requests.Response: parsed = urlparse(raw_url) @@ -255,27 +472,6 @@ def guarded_getaddrinfo(host: Any, port: Any, *args: Any, **inner_kwargs: Any) - finally: socket.getaddrinfo = original_getaddrinfo - @staticmethod - def _read_limited_response(response: requests.Response) -> bytes: - chunks: List[bytes] = [] - total = 0 - for chunk in response.iter_content(chunk_size=65536): - if not chunk: - continue - if isinstance(chunk, str): - chunk = chunk.encode("utf-8") - total += len(chunk) - if total > _MAX_FEED_BYTES: - raise IntelligenceServiceError("feed response is too large") - chunks.append(chunk) - return b"".join(chunks) - - @staticmethod - def _close_response(response: requests.Response) -> None: - close = getattr(response, "close", None) - if callable(close): - close() - @staticmethod def _normalize_hostname(hostname: Any) -> str: if isinstance(hostname, bytes): @@ -287,36 +483,14 @@ def _normalize_hostname(hostname: Any) -> str: return normalized @staticmethod - def _response_status_code(response: requests.Response) -> int: - try: - return int(getattr(response, "status_code", 0) or 0) - except (TypeError, ValueError): - return 0 - - @staticmethod - def _validate_resolved_hostname(hostname: str) -> None: - try: - addrinfos = socket.getaddrinfo(hostname, None, type=socket.SOCK_STREAM) - except socket.gaierror as exc: - raise IntelligenceServiceError("source url host could not be resolved") from exc - if not addrinfos: - raise IntelligenceServiceError("source url host could not be resolved") - IntelligenceService._validate_addrinfos(addrinfos) - - @staticmethod - def _validate_addrinfos(addrinfos: Any) -> None: - for addrinfo in addrinfos: - address = addrinfo[4][0] + def _validate_addrinfos(addr_infos: Any) -> None: + for info in addr_infos or []: try: - ip = ipaddress.ip_address(address) - except ValueError as exc: - raise IntelligenceServiceError("source url resolved to an invalid address") from exc + ip = ipaddress.ip_address(info[4][0]) + except (IndexError, TypeError, ValueError): + continue if IntelligenceService._is_blocked_ip(ip): - raise IntelligenceServiceError("source url must resolve to a public internet address") - - @staticmethod - def _is_blocked_ip(ip: ipaddress._BaseAddress) -> bool: - return not ip.is_global + raise IntelligenceServiceError("source url must not target private or local network addresses") def _parse_feed(self, content: bytes, *, source_name: str, limit: int) -> List[FeedEntry]: try: @@ -332,6 +506,27 @@ def _parse_feed(self, content: bytes, *, source_name: str, limit: int) -> List[F return [entry for entry in (self._parse_atom_entry(node, source_name) for node in nodes[:limit]) if entry] raise IntelligenceServiceError("unsupported feed format; expected RSS or Atom") + def _parse_newsnow_payload(self, payload: Any, *, source_name: str, limit: int) -> List[FeedEntry]: + if not isinstance(payload, dict): + raise IntelligenceServiceError("invalid NewsNow response: expected object") + items = payload.get("items") + if not isinstance(items, list): + raise IntelligenceServiceError("invalid NewsNow response: missing items") + entries = [] + for item in items[:limit]: + if not isinstance(item, dict): + continue + extra = item.get("extra") if isinstance(item.get("extra"), dict) else {} + published_raw = item.get("pubDate") or extra.get("date") + entries.append(self._build_entry( + str(item.get("title") or ""), + str(extra.get("info") or extra.get("hover") or ""), + str(item.get("url") or item.get("mobileUrl") or ""), + source_name, + self._parse_datetime_or_timestamp(published_raw), + )) + return [entry for entry in entries if entry] + def _parse_rss_item(self, node: ET.Element, source_name: str) -> Optional[FeedEntry]: return self._build_entry( self._text(node, "title"), @@ -362,7 +557,10 @@ def _build_entry(self, title: str, summary: str, url: str, source_name: str, pub if not title and not url: return None if url: - self._validate_url(url) + try: + self._validate_url(url, allow_no_url=True) + except IntelligenceServiceError: + return None url_key = url else: digest = hashlib.sha256(f"{source_name}|{title}|{published_at}".encode("utf-8")).hexdigest()[:24] @@ -454,12 +652,7 @@ def _redact_source_fields(fields: Dict[str, Any]) -> Dict[str, Any]: @staticmethod def _sanitize_error(exc: Exception) -> str: - return re.sub( - r"((?:^|[?&#\s])(?:access[_-]?token|auth[_-]?token|api[_-]?key|apikey|api_key|token|key|secret)=)[^&\s#]+", - r"\1***", - str(exc), - flags=re.I, - )[:500] + return sanitize_diagnostic_text(str(exc), max_length=500) or "internal intelligence service error" @staticmethod def _strip_ns(tag: str) -> str: @@ -492,6 +685,43 @@ def _parse_datetime(value: str) -> Optional[datetime]: parsed = parsed.astimezone(timezone.utc).replace(tzinfo=None) return parsed + @classmethod + def _parse_datetime_or_timestamp(cls, value: Any) -> Optional[datetime]: + if isinstance(value, (int, float)): + timestamp = float(value) + if timestamp > 10_000_000_000: + timestamp = timestamp / 1000 + try: + return datetime.fromtimestamp(timestamp, tz=timezone.utc).replace(tzinfo=None) + except (OSError, OverflowError, ValueError): + return None + raw = str(value or "").strip() + if raw.isdigit(): + return cls._parse_datetime_or_timestamp(float(raw)) + return cls._parse_datetime(raw) + + def _builtin_source_templates(self) -> List[Dict[str, Any]]: + templates = [dict(template) for template in _BUILTIN_SOURCE_TEMPLATES] + for item in _NEWSNOW_DEFAULT_SOURCE_DEFS: + templates.append({ + "template_id": item["template_id"], + "name": item["name"], + "source_type": "newsnow", + "url": self._build_newsnow_url(item["source_id"]), + "scope_type": "market", + "market": item["market"], + "description": item["description"], + }) + return templates + + def _build_newsnow_url(self, source_id: str) -> str: + base_url = (self.config.newsnow_base_url or "https://newsnow.busiyi.world").strip().rstrip("/") + parsed = urlparse(f"{base_url}/api/s") + query = dict(parse_qsl(parsed.query, keep_blank_values=True)) + query["id"] = source_id + return urlunparse(parsed._replace(query=urlencode(query))) + + @staticmethod def _iso(value: Optional[datetime]) -> Optional[str]: return value.isoformat() if value else None diff --git a/src/storage.py b/src/storage.py index a7c3e12e479..42385a1fad3 100644 --- a/src/storage.py +++ b/src/storage.py @@ -35,6 +35,7 @@ Index, UniqueConstraint, Text, + text, select, and_, or_, @@ -43,6 +44,8 @@ event, func, inspect, + MetaData, + Table, ) from sqlalchemy.dialects.sqlite import insert as sqlite_insert from sqlalchemy.orm import ( @@ -1132,6 +1135,7 @@ def __init__(self, db_url: Optional[str] = None): self._ensure_llm_usage_telemetry_columns() self._ensure_intelligence_item_scope_values() self._ensure_schema_migration_record() + self._ensure_intelligence_items_unique_index() self._initialized = True logger.info(f"数据库初始化完成: {db_url}") @@ -1176,6 +1180,111 @@ def _ensure_schema_migration_record(self) -> None: finally: session.close() + def _ensure_intelligence_items_unique_index(self) -> None: + if not self._is_sqlite_engine: + return + + if not inspect(self._engine).has_table("intelligence_items"): + return + + try: + unique_indexes = self._list_sqlite_unique_indexes("intelligence_items") + except Exception as exc: + logger.warning( + "[Intelligence items] failed to inspect unique indexes; " + "skip migration/repair: %s", + exc, + ) + return + + target_columns = ("source_id", "url", "scope_type", "scope_value", "market") + has_target_index = any(tuple(cols) == target_columns for cols in unique_indexes) + has_legacy_url_unique = any(tuple(cols) == ("url",) for cols in unique_indexes) + + if has_target_index: + return + if unique_indexes and not has_legacy_url_unique: + # Table has other unique index shapes; avoid aggressive changes and add + # the expected scoped uniqueness directly. + self._ensure_intelligence_items_scoped_unique_index_once() + return + + self._rebuild_intelligence_items_table() + + def _rebuild_intelligence_items_table(self) -> None: + temporary_table = f"intelligence_items_recreate_tmp_{int(time.time() * 1_000_000_000)}" + columns = [column.name for column in IntelligenceItem.__table__.columns] + select_clause = ", ".join(f'"{column}"' for column in columns) + scoped_index_columns = ", ".join(["source_id", "url", "scope_type", "scope_value", "market"]) + scoped_index_name = "uix_intel_item_scope" + + tmp_metadata = MetaData() + tmp_table = Table( + temporary_table, + tmp_metadata, + *(column.copy() for column in IntelligenceItem.__table__.columns), + ) + logger.info("Rebuilding intelligence_items table to align composite uniqueness constraints.") + with self._engine.begin() as connection: + connection.execute(text(f'DROP TABLE IF EXISTS "{temporary_table}"')) + tmp_table.create(connection) + connection.execute( + text( + f"INSERT INTO \"{temporary_table}\" ({select_clause}) " + f"SELECT {select_clause} FROM intelligence_items" + ) + ) + connection.execute(text('DROP TABLE "intelligence_items"')) + connection.execute( + text(f'ALTER TABLE "{temporary_table}" RENAME TO intelligence_items') + ) + connection.execute( + text( + f"CREATE UNIQUE INDEX IF NOT EXISTS {scoped_index_name} ON " + f"intelligence_items ({scoped_index_columns})" + ) + ) + + def _ensure_intelligence_items_scoped_unique_index_once(self) -> None: + target_index_name = "uix_intel_item_scope" + with self._engine.begin() as connection: + rows = connection.execute( + text("PRAGMA index_list(intelligence_items)") + ).fetchall() + for row in rows: + if row[1] == target_index_name: + return + index_columns = ", ".join(["source_id", "url", "scope_type", "scope_value", "market"]) + connection.execute( + text( + f"CREATE UNIQUE INDEX IF NOT EXISTS {target_index_name} ON " + f"intelligence_items ({index_columns})" + ) + ) + + def _list_sqlite_unique_indexes(self, table_name: str): + with self._engine.connect() as connection: + rows = connection.execute( + text(f"PRAGMA index_list({table_name})") + ).fetchall() + unique_indexes = [] + for row in rows: + # row: (seq, name, unique, origin, partial) + if int(row[2]) != 1: + continue + index_name = row[1] + index_columns = [] + for index_info in connection.execute( + text(f"PRAGMA index_xinfo({index_name})") + ).fetchall(): + # index_xinfo: (seqno, cid, name, desc, coll, key, ... ) + column_name = index_info[2] + if column_name is None: + continue + index_columns.append(column_name) + unique_indexes.append(index_columns) + return unique_indexes + def _ensure_llm_usage_telemetry_columns(self) -> None: """Add nullable P0a usage telemetry columns to existing SQLite DBs.""" if not self._is_sqlite_engine: diff --git a/tests/test_config_env_compat.py b/tests/test_config_env_compat.py index 5196ace7ec9..06190b65427 100644 --- a/tests/test_config_env_compat.py +++ b/tests/test_config_env_compat.py @@ -78,6 +78,44 @@ def test_alphasift_install_spec_defaults_only_when_env_missing( self.assertEqual(config.alphasift_install_spec, DEFAULT_ALPHASIFT_INSTALL_SPEC) + @patch("src.config.setup_env") + @patch.object(Config, "_parse_litellm_yaml", return_value=[]) + def test_news_intel_envs_do_not_change_llm_runtime_contract( + self, + _mock_parse_litellm_yaml, + _mock_setup_env, + ) -> None: + base_env = { + "STOCK_LIST": "600519", + "OPENAI_API_KEYS": "base-key-12345", + "OPENAI_BASE_URL": "https://openai.example.com/v1", + "LITELLM_MODEL": "openai/gpt-4.1", + "OPENAI_MODEL": "gpt-4.1", + } + with patch.dict(os.environ, base_env, clear=True): + Config._instance = None + baseline = Config._load_from_env() + + news_intel_env = dict(base_env) + news_intel_env.update({ + "NEWS_INTEL_RETENTION_DAYS": "45", + "NEWS_INTEL_FETCH_TIMEOUT_SEC": "5.5", + "NEWS_INTEL_MAX_ITEMS_PER_SOURCE": "25", + "NEWSNOW_BASE_URL": "https://newsnow.example.com/", + }) + with patch.dict(os.environ, news_intel_env, clear=True): + Config._instance = None + with_news_intel = Config._load_from_env() + + self.assertEqual(with_news_intel.litellm_model, baseline.litellm_model) + self.assertEqual(with_news_intel.litellm_fallback_models, baseline.litellm_fallback_models) + self.assertEqual(with_news_intel.openai_api_key, baseline.openai_api_key) + self.assertEqual(with_news_intel.openai_base_url, baseline.openai_base_url) + self.assertEqual(with_news_intel.news_intel_fetch_timeout_sec, 5.5) + self.assertEqual(with_news_intel.news_intel_max_items_per_source, 25) + self.assertEqual(with_news_intel.news_intel_retention_days, 45) + self.assertEqual(with_news_intel.newsnow_base_url, "https://newsnow.example.com") + def test_env_example_alphasift_install_spec_matches_trusted_default(self): env_example = Path(__file__).resolve().parents[1] / ".env.example" @@ -254,6 +292,7 @@ def test_news_intel_env_vars_do_not_affect_llm_layer( "NEWS_INTEL_RETENTION_DAYS": "14", "NEWS_INTEL_FETCH_TIMEOUT_SEC": "12", "NEWS_INTEL_MAX_ITEMS_PER_SOURCE": "75", + "NEWSNOW_BASE_URL": "https://newsnow.example.com/base/", }, clear=True, ): @@ -265,6 +304,7 @@ def test_news_intel_env_vars_do_not_affect_llm_layer( self.assertEqual(config.news_intel_retention_days, 14) self.assertEqual(config.news_intel_fetch_timeout_sec, 12.0) self.assertEqual(config.news_intel_max_items_per_source, 75) + self.assertEqual(config.newsnow_base_url, "https://newsnow.example.com/base") @patch("src.config.setup_env") @patch.object(Config, "_parse_litellm_yaml", return_value=[]) diff --git a/tests/test_intelligence_analysis_integration.py b/tests/test_intelligence_analysis_integration.py new file mode 100644 index 00000000000..6aaf2a89942 --- /dev/null +++ b/tests/test_intelligence_analysis_integration.py @@ -0,0 +1,361 @@ +# -*- coding: utf-8 -*- +"""Tests for using persisted intelligence in analysis contexts.""" + +from __future__ import annotations + +import os +import tempfile +import unittest +from datetime import datetime, timedelta + +from src.config import Config, get_config +from src.core.pipeline import StockAnalysisPipeline +from src.market_analyzer import MarketAnalyzer, MarketIndex, MarketOverview +from src.repositories.intelligence_repo import IntelligenceRepository +from src.storage import DatabaseManager + + +class PersistedIntelligenceAnalysisIntegrationTestCase(unittest.TestCase): + def setUp(self) -> None: + self._temp_dir = tempfile.TemporaryDirectory() + os.environ["DATABASE_PATH"] = os.path.join(self._temp_dir.name, "intel_analysis.db") + Config._instance = None + DatabaseManager.reset_instance() + self.config = get_config() + repo = IntelligenceRepository() + now = datetime.now() + repo.upsert_items([ + { + "source_name": "symbol-feed", + "source_type": "rss", + "title": "Company wins major AI order", + "summary": "Order expands visibility for next quarter.", + "url": "https://news.example.com/symbol", + "source": "symbol-feed", + "published_at": now, + "fetched_at": now, + "scope_type": "symbol", + "scope_value": "600519", + "market": "cn", + }, + { + "source_name": "market-feed", + "source_type": "rss", + "title": "Policy support lifts market sentiment", + "summary": "Market-level catalyst.", + "url": "https://news.example.com/market", + "source": "market-feed", + "published_at": now, + "fetched_at": now, + "scope_type": "market", + "scope_value": None, + "market": "cn", + }, + ]) + + def tearDown(self) -> None: + DatabaseManager.reset_instance() + Config._instance = None + os.environ.pop("DATABASE_PATH", None) + self._temp_dir.cleanup() + + def test_pipeline_loads_persisted_symbol_and_market_intelligence(self) -> None: + pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline) + pipeline.config = self.config + context = pipeline._load_persisted_intelligence_context( + code="600519", + stock_name="贵州茅台", + market="cn", + ) + self.assertIsNotNone(context) + assert context is not None + self.assertIn("本地资讯证据池", context) + self.assertIn("Company wins major AI order", context) + self.assertIn("https://news.example.com/symbol", context) + + def test_pipeline_loads_symbol_intelligence_with_exchange_alias_scope(self) -> None: + repo = IntelligenceRepository() + now = datetime.now() + repo.upsert_items([ + { + "source_name": "symbol-feed", + "source_type": "rss", + "title": "SH-prefixed symbol feed", + "summary": "Prefixed source should match normalized analysis code.", + "url": "https://news.example.com/symbol-sh-prefix", + "source": "symbol-feed", + "published_at": now, + "fetched_at": now, + "scope_type": "symbol", + "scope_value": "SH600519", + "market": "cn", + }, + { + "source_name": "symbol-feed", + "source_type": "rss", + "title": "SH-suffixed symbol feed", + "summary": "Suffixed source should match normalized analysis code.", + "url": "https://news.example.com/symbol-sh-suffix", + "source": "symbol-feed", + "published_at": now, + "fetched_at": now, + "scope_type": "symbol", + "scope_value": "600519.SH", + "market": "cn", + }, + ]) + + pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline) + pipeline.config = self.config + context = pipeline._load_persisted_intelligence_context( + code="600519", + stock_name="贵州茅台", + market="cn", + ) + + self.assertIsNotNone(context) + assert context is not None + self.assertIn("SH-prefixed symbol feed", context) + self.assertIn("SH-suffixed symbol feed", context) + + def test_pipeline_loads_hk_symbol_intelligence_with_plain_code_scope(self) -> None: + repo = IntelligenceRepository() + now = datetime.now() + repo.upsert_items([ + { + "source_name": "hk-symbol-feed", + "source_type": "rss", + "title": "Plain HK code symbol feed", + "summary": "Plain five-digit HK source should match canonical and suffixed analysis codes.", + "url": "https://news.example.com/hk-plain-code", + "source": "hk-symbol-feed", + "published_at": now, + "fetched_at": now, + "scope_type": "symbol", + "scope_value": "00700", + "market": "hk", + }, + { + "source_name": "hk-trimmed-symbol-feed", + "source_type": "rss", + "title": "Trimmed HK code symbol feed", + "summary": "Trimmed HK source should match canonical analysis code.", + "url": "https://news.example.com/hk-trimmed-code", + "source": "hk-trimmed-symbol-feed", + "published_at": now, + "fetched_at": now, + "scope_type": "symbol", + "scope_value": "HK700", + "market": "hk", + }, + ]) + + pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline) + pipeline.config = self.config + for code in ("HK00700", "00700.HK"): + with self.subTest(code=code): + context = pipeline._load_persisted_intelligence_context( + code=code, + stock_name="腾讯控股", + market="hk", + ) + + self.assertIsNotNone(context) + assert context is not None + self.assertIn("Plain HK code symbol feed", context) + self.assertIn("Trimmed HK code symbol feed", context) + + def test_market_review_merges_persisted_market_intelligence(self) -> None: + analyzer = MarketAnalyzer(config=self.config, region="cn") + merged = analyzer._merge_persisted_market_intelligence([]) + self.assertTrue(any(item.get("title") == "Policy support lifts market sentiment" for item in merged)) + item = next(item for item in merged if item.get("title") == "Policy support lifts market sentiment") + self.assertEqual(item["snippet"], "Market-level catalyst.") + self.assertEqual(item["url"], "https://news.example.com/market") + + def test_market_review_local_intelligence_kept_in_top_payload_when_search_news_filled(self) -> None: + analyzer = MarketAnalyzer(config=self.config, region="cn") + search_news = [ + { + "title": f"Search headline {i}", + "snippet": f"Search summary {i}", + "source": "search-source", + "published_date": "2026-06-17", + "url": f"https://news.example.com/search/{i}", + } + for i in range(8) + ] + + merged = analyzer._merge_persisted_market_intelligence(search_news) + self.assertEqual(merged[0]["title"], "Policy support lifts market sentiment") + + payload = analyzer.build_market_review_payload( + MarketOverview( + date="2026-06-17", + indices=[ + MarketIndex( + code="000001", + name="SSE Composite", + current=3200.0, + change=10.0, + change_pct=0.25, + ) + ], + ), + news=merged, + report="复盘正文", + market_light_snapshot={"dimensions": {"breadth": {"available": True}}}, + ) + self.assertEqual(payload["news"][0]["url"], "https://news.example.com/market") + self.assertGreaterEqual(len(payload["news"]), 1) + + def test_market_review_keeps_search_news_when_local_pool_is_full(self) -> None: + repo = IntelligenceRepository() + now = datetime.now() + for index in range(5): + repo.upsert_items([ + { + "source_name": f"market-local-{index}", + "source_type": "rss", + "title": f"Market local headline {index}", + "summary": f"Local market signal {index}", + "url": f"https://news.example.com/market-local/{index}", + "source": f"market-local-{index}", + "published_at": now + timedelta(minutes=index + 1), + "fetched_at": now + timedelta(minutes=index + 1), + "scope_type": "market", + "scope_value": None, + "market": "cn", + } + ]) + + analyzer = MarketAnalyzer(config=self.config, region="cn") + search_news = [ + { + "title": f"Search headline {i}", + "snippet": f"Search summary {i}", + "source": "search-source", + "published_date": "2026-06-17", + "url": f"https://news.example.com/search/{i}", + } + for i in range(8) + ] + + merged = analyzer._merge_persisted_market_intelligence(search_news) + first_six = merged[:6] + local_count = len([item for item in first_six if str(item.get("title", "")).startswith("Market local headline")]) + search_count = len([item for item in first_six if str(item.get("title", "")).startswith("Search headline")]) + self.assertEqual(local_count, 3) + self.assertEqual(search_count, 3) + + prompt = analyzer._build_review_prompt( + MarketOverview(date="2026-06-17"), + merged, + ) + self.assertIn("Search headline 0", prompt) + + payload = analyzer.build_market_review_payload( + MarketOverview( + date="2026-06-17", + indices=[ + MarketIndex( + code="000001", + name="SSE Composite", + current=3200.0, + change=10.0, + change_pct=0.25, + ) + ], + ), + news=merged, + report="复盘正文", + market_light_snapshot={"dimensions": {"breadth": {"available": True}}}, + ) + self.assertGreaterEqual(len(payload["news"]), 8) + self.assertEqual(sum(item["title"].startswith("Search headline") for item in payload["news"][0:8]), 4) + + def test_analysis_evidence_excludes_missing_or_stale_publish_time(self) -> None: + self.config.news_max_age_days = 30 + self.config.news_strategy_profile = "short" + repo = IntelligenceRepository() + now = datetime.now() + old_time = now - timedelta(days=5) + repo.upsert_items([ + { + "source_name": "symbol-feed", + "source_type": "rss", + "title": "Stale symbol item fetched today", + "summary": "Old publication date should not be prompt evidence.", + "url": "https://news.example.com/stale-symbol", + "source": "symbol-feed", + "published_at": old_time, + "fetched_at": now, + "scope_type": "symbol", + "scope_value": "600519", + "market": "cn", + }, + { + "source_name": "symbol-feed", + "source_type": "rss", + "title": "Undated symbol item fetched today", + "summary": "Missing publication date should not be prompt evidence.", + "url": "https://news.example.com/undated-symbol", + "source": "symbol-feed", + "published_at": None, + "fetched_at": now, + "scope_type": "symbol", + "scope_value": "600519", + "market": "cn", + }, + { + "source_name": "market-feed", + "source_type": "rss", + "title": "Stale market item fetched today", + "summary": "Old market publication date should not be prompt evidence.", + "url": "https://news.example.com/stale-market", + "source": "market-feed", + "published_at": old_time, + "fetched_at": now, + "scope_type": "market", + "scope_value": None, + "market": "cn", + }, + { + "source_name": "market-feed", + "source_type": "rss", + "title": "Undated market item fetched today", + "summary": "Missing market publication date should not be prompt evidence.", + "url": "https://news.example.com/undated-market", + "source": "market-feed", + "published_at": None, + "fetched_at": now, + "scope_type": "market", + "scope_value": None, + "market": "cn", + }, + ]) + + pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline) + pipeline.config = self.config + context = pipeline._load_persisted_intelligence_context( + code="600519", + stock_name="贵州茅台", + market="cn", + ) + assert context is not None + self.assertIn("Company wins major AI order", context) + self.assertNotIn("Stale symbol item fetched today", context) + self.assertNotIn("Undated symbol item fetched today", context) + self.assertNotIn("Stale market item fetched today", context) + self.assertNotIn("Undated market item fetched today", context) + + analyzer = MarketAnalyzer(config=self.config, region="cn") + merged = analyzer._merge_persisted_market_intelligence([]) + titles = {item.get("title") for item in merged} + self.assertIn("Policy support lifts market sentiment", titles) + self.assertNotIn("Stale market item fetched today", titles) + self.assertNotIn("Undated market item fetched today", titles) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_intelligence_api.py b/tests/test_intelligence_api.py index 2472387c145..4f704f4dd38 100644 --- a/tests/test_intelligence_api.py +++ b/tests/test_intelligence_api.py @@ -4,13 +4,13 @@ from __future__ import annotations import os -import requests -import socket import tempfile import unittest +import socket from pathlib import Path from unittest.mock import Mock, patch +import requests from fastapi.testclient import TestClient from api.app import create_app @@ -26,6 +26,12 @@ def setUp(self) -> None: os.environ["DATABASE_PATH"] = os.path.join(self._temp_dir.name, "api_intel.db") Config._instance = None DatabaseManager.reset_instance() + self._dns_patcher = patch( + "src.services.intelligence_service.socket.getaddrinfo", + return_value=[(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))], + ) + self._dns_patcher.start() + self.addCleanup(self._dns_patcher.stop) self.client = TestClient(create_app(static_dir=Path(self._temp_dir.name))) def tearDown(self) -> None: @@ -37,25 +43,28 @@ def tearDown(self) -> None: def _mock_response(self): response = Mock() response.status_code = 200 - response.headers = {} - response.iter_content.return_value = [RSS_FIXTURE] response.url = "https://feeds.example.com/rss.xml" + response.headers = {} response.raise_for_status.return_value = None - response.close.return_value = None + response.iter_content.return_value = [RSS_FIXTURE] return response - def _public_dns(self): - return patch( - "src.services.intelligence_service.socket.getaddrinfo", - return_value=[(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))], + def _mock_http_error_response(self, source_url: str): + response = Mock() + response.status_code = 403 + response.url = source_url + response.headers = {} + response.raise_for_status.side_effect = requests.HTTPError( + f"403 Client Error: Forbidden for url: {source_url}" ) + response.iter_content.return_value = [] + return response def test_create_fetch_and_query_items(self) -> None: - with self._public_dns(): - create_resp = self.client.post("/api/v1/intelligence/sources", json={"name": "api-feed", "url": "https://feeds.example.com/rss.xml", "source_type": "rss", "scope_type": "market", "market": "cn"}) + create_resp = self.client.post("/api/v1/intelligence/sources", json={"name": "api-feed", "url": "https://feeds.example.com/rss.xml", "source_type": "rss", "scope_type": "market", "market": "cn"}) self.assertEqual(create_resp.status_code, 200) source_id = create_resp.json()["id"] - with self._public_dns(), patch("src.services.intelligence_service.requests.get", return_value=self._mock_response()): + with patch("src.services.intelligence_service.requests.get", return_value=self._mock_response()): fetch_resp = self.client.post(f"/api/v1/intelligence/sources/{source_id}/fetch") self.assertEqual(fetch_resp.status_code, 200) self.assertEqual(fetch_resp.json()["saved_count"], 1) @@ -70,32 +79,144 @@ def test_rejects_private_source_url(self) -> None: self.assertEqual(resp.status_code, 400) self.assertEqual(resp.json()["error"], "validation_error") - def test_fetch_error_does_not_expose_source_url_secret(self) -> None: - with self._public_dns(): - create_resp = self.client.post( - "/api/v1/intelligence/sources", - json={ - "name": "secret-feed", - "url": "https://feeds.example.com/rss.xml?token=secret", - "scope_type": "market", - }, - ) + def test_duplicate_source_name_returns_validation_error(self) -> None: + payload = {"name": "dupe", "url": "https://feeds.example.com/rss.xml", "scope_type": "market"} + first = self.client.post("/api/v1/intelligence/sources", json=payload) + second = self.client.post("/api/v1/intelligence/sources", json=payload) + self.assertEqual(first.status_code, 200) + self.assertEqual(second.status_code, 400) + self.assertEqual(second.json()["error"], "validation_error") + + def test_list_and_create_from_builtin_source_template(self) -> None: + templates = self.client.get("/api/v1/intelligence/sources/templates", params={"market": "hk"}) + self.assertEqual(templates.status_code, 200) + body = templates.json() + self.assertGreaterEqual(body["total"], 1) + self.assertTrue(any(item["template_id"] == "hkex-news" for item in body["items"])) + + created = self.client.post( + "/api/v1/intelligence/sources/templates/hkex-news", + json={"name": "hkex-copy", "enabled": False}, + ) + self.assertEqual(created.status_code, 200) + self.assertEqual(created.json()["name"], "hkex-copy") + self.assertFalse(created.json()["enabled"]) + + def test_create_builtin_default_sources_is_idempotent(self) -> None: + first = self.client.post("/api/v1/intelligence/sources/defaults", json={"enabled": False}) + second = self.client.post("/api/v1/intelligence/sources/defaults", json={"enabled": False}) + + self.assertEqual(first.status_code, 200) + self.assertEqual(second.status_code, 200) + self.assertGreaterEqual(first.json()["created_count"], 5) + self.assertEqual(second.json()["created_count"], 0) + self.assertEqual(first.json()["total"], second.json()["total"]) + newsnow_sources = [ + item["source"] for item in first.json()["items"] + if item["source"]["source_type"] == "newsnow" + ] + self.assertGreaterEqual(len(newsnow_sources), 5) + self.assertTrue(all(not item["enabled"] for item in newsnow_sources)) + + def test_fetch_source_internal_error_is_sanitized(self) -> None: + create_resp = self.client.post("/api/v1/intelligence/sources", json={"name": "api-feed", "url": "https://feeds.example.com/rss.xml", "source_type": "rss", "scope_type": "market", "market": "cn"}) self.assertEqual(create_resp.status_code, 200) + source_id = create_resp.json()["id"] + with patch("src.services.intelligence_service.IntelligenceService.fetch_source", side_effect=RuntimeError("token=secret api_key=abc12345")): + fetch_resp = self.client.post(f"/api/v1/intelligence/sources/{source_id}/fetch") - response = self._mock_response() - response.raise_for_status.side_effect = requests.HTTPError( - "404 Client Error for url: https://feeds.example.com/rss.xml?token=secret" - ) + self.assertEqual(fetch_resp.status_code, 500) + body = fetch_resp.json() + self.assertEqual(body["error"], "internal_error") + self.assertEqual(body["message"], "Fetch intelligence source failed: internal intelligence service error") - with self._public_dns(), patch("src.services.intelligence_service.requests.get", return_value=response), self.assertLogs("api.v1.endpoints.intelligence", level="ERROR") as logs: - fetch_resp = self.client.post(f"/api/v1/intelligence/sources/{create_resp.json()['id']}/fetch") + def test_fetch_source_internal_error_without_sensitive_pattern_is_generic(self) -> None: + create_resp = self.client.post("/api/v1/intelligence/sources", json={"name": "api-feed", "url": "https://feeds.example.com/rss.xml", "source_type": "rss", "scope_type": "market", "market": "cn"}) + self.assertEqual(create_resp.status_code, 200) + source_id = create_resp.json()["id"] + with patch("src.services.intelligence_service.IntelligenceService.fetch_source", side_effect=RuntimeError("unexpected runtime assertion failure: pipeline context exhausted")): + fetch_resp = self.client.post(f"/api/v1/intelligence/sources/{source_id}/fetch") self.assertEqual(fetch_resp.status_code, 500) - response_blob = str(fetch_resp.json()) - log_blob = "\n".join(logs.output) - self.assertNotIn("secret", response_blob) - self.assertNotIn("token=secret", log_blob) - self.assertEqual(fetch_resp.json()["message"], "Fetch intelligence source failed") + body = fetch_resp.json() + self.assertEqual(body["error"], "internal_error") + self.assertEqual(body["message"], "Fetch intelligence source failed: internal intelligence service error") + self.assertNotIn("pipeline context exhausted", body["message"]) + + def test_create_builtin_default_sources_are_disabled_by_default(self) -> None: + default_resp = self.client.post("/api/v1/intelligence/sources/defaults") + self.assertEqual(default_resp.status_code, 200) + newsnow_sources = [ + item["source"] for item in default_resp.json()["items"] + if item["source"]["source_type"] == "newsnow" + ] + self.assertGreaterEqual(len(newsnow_sources), 5) + self.assertTrue(all(not item["enabled"] for item in newsnow_sources)) + + def test_upstream_fetch_errors_do_not_expose_query_secret(self) -> None: + secret_url = "https://feeds.example.com/rss.xml?token=super-secret" + payload = { + "name": "secret-feed", + "url": secret_url, + "source_type": "rss", + "scope_type": "market", + "market": "cn", + } + create_resp = self.client.post("/api/v1/intelligence/sources", json=payload) + self.assertEqual(create_resp.status_code, 200) + source_id = create_resp.json()["id"] + + requests_to_check = [ + ("test", lambda: self.client.post("/api/v1/intelligence/sources/test", json=payload)), + ("fetch", lambda: self.client.post(f"/api/v1/intelligence/sources/{source_id}/fetch")), + ] + with patch( + "src.services.intelligence_service.requests.get", + side_effect=lambda url, **_kwargs: self._mock_http_error_response(url), + ): + for endpoint, send_request in requests_to_check: + with self.subTest(endpoint=endpoint): + response = send_request() + self.assertEqual(response.status_code, 400) + body = response.json() + self.assertEqual(body["error"], "validation_error") + self.assertEqual(body["message"], "fetch failed: upstream request failed") + self.assertNotIn(secret_url, body["message"]) + self.assertNotIn("token=", body["message"]) + self.assertNotIn("super-secret", body["message"]) + + def test_upstream_fetch_errors_for_newsnow_do_not_expose_query_secret(self) -> None: + secret_url = "https://newsnow.example.com/api/s?id=cls-hot&token=super-secret" + payload = { + "name": "newsnow-secret-feed", + "url": secret_url, + "source_type": "newsnow", + "scope_type": "market", + "market": "cn", + } + + create_resp = self.client.post("/api/v1/intelligence/sources", json=payload) + self.assertEqual(create_resp.status_code, 200) + source_id = create_resp.json()["id"] + + requests_to_check = [ + ("test", lambda: self.client.post("/api/v1/intelligence/sources/test", json=payload)), + ("fetch", lambda: self.client.post(f"/api/v1/intelligence/sources/{source_id}/fetch")), + ] + with patch( + "src.services.intelligence_service.requests.get", + side_effect=lambda url, **_kwargs: self._mock_http_error_response(url), + ): + for endpoint, send_request in requests_to_check: + with self.subTest(endpoint=endpoint): + response = send_request() + self.assertEqual(response.status_code, 400) + body = response.json() + self.assertEqual(body["error"], "validation_error") + self.assertEqual(body["message"], "fetch failed: upstream request failed") + self.assertNotIn(secret_url, body["message"]) + self.assertNotIn("token=", body["message"]) + self.assertNotIn("super-secret", body["message"]) if __name__ == "__main__": diff --git a/tests/test_intelligence_service.py b/tests/test_intelligence_service.py index 4486bc5aa22..260c9e602cb 100644 --- a/tests/test_intelligence_service.py +++ b/tests/test_intelligence_service.py @@ -4,21 +4,44 @@ from __future__ import annotations import os +import json import socket import tempfile import unittest from datetime import datetime, timedelta +from urllib.parse import quote from unittest.mock import Mock, patch -from sqlalchemy.exc import IntegrityError -from sqlalchemy.orm import Session +import requests from src.config import Config from src.repositories.intelligence_repo import IntelligenceRepository -from src.services.intelligence_service import IntelligenceService, IntelligenceServiceError, _MAX_FEED_BYTES +from src.services.intelligence_service import IntelligenceService, IntelligenceServiceError from src.storage import DatabaseManager, IntelligenceItem, INTELLIGENCE_ITEM_NULL_SCOPE_VALUE RSS_FIXTURE = b'\n\nPolicy support lifts AI supply chainhttps://news.example.com/aMarket-level catalyst with evidence link.Wed, 17 Jun 2026 08:00:00 GMT\nSecond itemhttps://news.example.com/bSecond summary.\n' +NO_URL_LINK_FIXTURE = b'\n\nAnonymous itemNo link in this item.\n' +BAD_ITEM_LINK_FIXTURE = b'\n\nBad mail linkmailto:tips@example.comShould be skipped.\nGood public linkhttps://news.example.com/goodShould be saved.\n' +NEWSNOW_FIXTURE = { + "status": "success", + "id": "cls-hot", + "updatedTime": 1781760000000, + "items": [ + { + "id": "1", + "title": "A-share AI hardware theme heats up", + "url": "https://news.example.com/newsnow-a", + "pubDate": 1781760000000, + "extra": {"info": "Capital market hot topic from NewsNow."}, + }, + { + "id": "2", + "title": "Second NewsNow item", + "url": "https://news.example.com/newsnow-b", + "extra": {"hover": "Fallback summary."}, + }, + ], +} class IntelligenceServiceTestCase(unittest.TestCase): @@ -31,6 +54,36 @@ def setUp(self) -> None: Config._instance = None DatabaseManager.reset_instance() self.service = IntelligenceService() + self._dns_patcher = patch( + "src.services.intelligence_service.socket.getaddrinfo", + side_effect=self._mock_getaddrinfo, + ) + self._dns_patcher.start() + self.addCleanup(self._dns_patcher.stop) + + def _mock_getaddrinfo(self, host, *_args, **_kwargs): + host = (host or "").lower().strip() + if host in {"localhost", "localhost.localdomain"}: + return [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 0)), + ] + if host == "shared.example.com": + return [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("100.64.0.1", 0)), + ] + return [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0)), + ] + + def _feed_fixture(self, source_url: str) -> bytes: + key = quote(source_url.replace("://", "_").replace("/", "_")) + return RSS_FIXTURE.replace( + b"https://news.example.com/a", + f"https://news.example.com/{key}.a".encode("utf-8"), + ).replace( + b"https://news.example.com/b", + f"https://news.example.com/{key}.b".encode("utf-8"), + ) def tearDown(self) -> None: DatabaseManager.reset_instance() @@ -39,29 +92,49 @@ def tearDown(self) -> None: os.environ.pop(key, None) self._temp_dir.cleanup() - def _mock_response(self): + def _mock_response(self, source_url: str = "https://feeds.example.com/rss.xml"): response = Mock() response.status_code = 200 + response.url = source_url response.headers = {} - response.iter_content.return_value = [RSS_FIXTURE] - response.url = "https://feeds.example.com/rss.xml" response.raise_for_status.return_value = None - response.close.return_value = None + response.iter_content.return_value = [self._feed_fixture(source_url)] return response - def _public_dns(self): - return patch( - "src.services.intelligence_service.socket.getaddrinfo", - return_value=[(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))], + def _mock_json_response(self, payload=NEWSNOW_FIXTURE, source_url: str = "https://newsnow.example.com/api/s?id=cls-hot"): + response = Mock() + response.status_code = 200 + response.url = source_url + response.headers = {} + response.raise_for_status.return_value = None + response.iter_content.return_value = [json.dumps(payload).encode("utf-8")] + return response + + def _mock_response_with_redirects(self, source_url: str = "https://feeds.example.com/rss.xml", next_url: str = "https://feeds.example.com/rss.xml"): + response = Mock() + response.url = source_url + response.raise_for_status.return_value = None + response.headers = {"Location": next_url} + response.status_code = 302 + return response + + def _mock_http_error_response(self, source_url: str): + response = Mock() + response.status_code = 403 + response.url = source_url + response.headers = {} + response.raise_for_status.side_effect = requests.HTTPError( + f"403 Client Error: Forbidden for url: {source_url}" ) + response.iter_content.return_value = [] + return response def test_create_fetch_and_deduplicate_rss_source(self) -> None: - with self._public_dns(): - source = self.service.create_source({ - "name": "market-feed", "url": "https://feeds.example.com/rss.xml", - "source_type": "rss", "scope_type": "market", "market": "cn", - }) - with self._public_dns(), patch("src.services.intelligence_service.requests.get", return_value=self._mock_response()): + source = self.service.create_source({ + "name": "market-feed", "url": "https://feeds.example.com/rss.xml", + "source_type": "rss", "scope_type": "market", "market": "cn", + }) + with patch("src.services.intelligence_service.requests.get", return_value=self._mock_response()): first = self.service.fetch_source(source["id"]) second = self.service.fetch_source(source["id"]) self.assertEqual(first["fetched_count"], 2) @@ -72,176 +145,258 @@ def test_create_fetch_and_deduplicate_rss_source(self) -> None: self.assertEqual(items["items"][0]["scope_type"], "market") self.assertTrue(items["items"][0]["url"].startswith("https://news.example.com/")) - def test_same_url_from_different_source_scope_preserves_both_items(self) -> None: - with self._public_dns(): - cn_source = self.service.create_source({ - "name": "cn-feed", "url": "https://feeds.example.com/rss.xml", - "source_type": "rss", "scope_type": "market", "market": "cn", - }) - us_source = self.service.create_source({ - "name": "us-feed", "url": "https://feeds.example.com/rss.xml", - "source_type": "rss", "scope_type": "market", "market": "us", - }) - with self._public_dns(), patch("src.services.intelligence_service.requests.get", return_value=self._mock_response()): - cn_result = self.service.fetch_source(cn_source["id"]) - us_result = self.service.fetch_source(us_source["id"]) - self.assertEqual(cn_result["saved_count"], 2) - self.assertEqual(us_result["saved_count"], 2) - cn_items = self.service.list_items(scope_type="market", market="cn") - us_items = self.service.list_items(scope_type="market", market="us") - self.assertEqual(cn_items["total"], 2) - self.assertEqual(us_items["total"], 2) - self.assertEqual({item["source_name"] for item in cn_items["items"]}, {"cn-feed"}) - self.assertEqual({item["source_name"] for item in us_items["items"]}, {"us-feed"}) + def test_fetch_http_error_does_not_expose_source_query_secret(self) -> None: + secret_url = "https://feeds.example.com/rss.xml?token=super-secret" + source = self.service.create_source({ + "name": "secret-feed", "url": secret_url, "scope_type": "market", + }) + + with patch("src.services.intelligence_service.requests.get", return_value=self._mock_http_error_response(secret_url)): + with self.assertRaises(IntelligenceServiceError) as ctx: + self.service.fetch_source(source["id"]) + + message = str(ctx.exception) + self.assertEqual(message, "fetch failed: upstream request failed") + self.assertNotIn(secret_url, message) + self.assertNotIn("token=", message) + self.assertNotIn("super-secret", message) + saved_source = self.service.repo.get_source(source["id"]) + self.assertIsNotNone(saved_source) + self.assertEqual(saved_source.last_error, "fetch failed: upstream request failed") + + def test_fetch_newsnow_http_error_does_not_expose_source_query_secret(self) -> None: + secret_url = "https://newsnow.example.com/api/s?id=cls-hot&token=super-secret" + source = self.service.create_source({ + "name": "newsnow-secret-feed", + "url": secret_url, + "source_type": "newsnow", + "scope_type": "market", + "market": "cn", + }) + + with patch("src.services.intelligence_service.requests.get", return_value=self._mock_http_error_response(secret_url)): + with self.assertRaises(IntelligenceServiceError) as ctx: + self.service.fetch_source(source["id"]) + + message = str(ctx.exception) + self.assertEqual(message, "fetch failed: upstream request failed") + self.assertNotIn(secret_url, message) + self.assertNotIn("token=", message) + self.assertNotIn("super-secret", message) def test_private_network_url_is_rejected(self) -> None: with self.assertRaises(IntelligenceServiceError): self.service.create_source({"name": "bad", "url": "http://127.0.0.1:8000/rss.xml", "scope_type": "market"}) - def test_dns_name_resolving_private_address_is_rejected(self) -> None: - private_dns = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("169.254.169.254", 0))] - with patch("src.services.intelligence_service.socket.getaddrinfo", return_value=private_dns): - with self.assertRaises(IntelligenceServiceError): - self.service.create_source({"name": "bad", "url": "https://metadata.example.com/rss.xml", "scope_type": "market"}) - def test_shared_address_space_url_is_rejected(self) -> None: - shared_dns = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("100.64.0.1", 0))] - with patch("src.services.intelligence_service.socket.getaddrinfo", return_value=shared_dns): - with self.assertRaises(IntelligenceServiceError): - self.service.create_source({"name": "bad", "url": "https://metadata.example.com/rss.xml", "scope_type": "market"}) + with self.assertRaises(IntelligenceServiceError): + self.service.create_source({"name": "shared", "url": "https://shared.example.com/rss.xml", "scope_type": "market"}) - def test_generated_no_url_sentinel_is_rejected_for_source_url(self) -> None: - with self.assertRaisesRegex(IntelligenceServiceError, "absolute http\\(s\\) URL"): - self.service.create_source({"name": "bad", "url": "no-url:intel:anything", "scope_type": "market"}) + def test_duplicate_source_name_is_validation_error(self) -> None: + payload = {"name": "dupe", "url": "https://feeds.example.com/rss.xml", "scope_type": "market"} + self.service.create_source(payload) + with self.assertRaises(IntelligenceServiceError): + self.service.create_source(payload) - def test_redirect_target_is_validated_before_following(self) -> None: - with self._public_dns(): - source = self.service.create_source({ - "name": "redirect-feed", - "url": "https://feeds.example.com/rss.xml", - "scope_type": "market", - }) - redirect = Mock() - redirect.status_code = 302 - redirect.headers = {"Location": "http://127.0.0.1/rss.xml"} - redirect.url = "https://feeds.example.com/rss.xml" - redirect.content = b"" - redirect.raise_for_status.return_value = None - with self._public_dns(), patch("src.services.intelligence_service.requests.get", return_value=redirect) as mock_get: - with self.assertRaises(IntelligenceServiceError): - self.service.fetch_source(source["id"]) - self.assertEqual(mock_get.call_count, 1) - self.assertFalse(mock_get.call_args.kwargs["allow_redirects"]) + def test_fetch_enabled_sources_is_fail_open(self) -> None: + self.service.create_source({"name": "good-feed", "url": "https://feeds.example.com/rss.xml", "scope_type": "market"}) + bad = self.service.create_source({"name": "bad-feed", "url": "https://bad.example.com/rss.xml", "scope_type": "market"}) - def test_fetch_requests_disable_environment_proxies(self) -> None: - with self._public_dns(): - source = self.service.create_source({ - "name": "proxy-safe-feed", + def fake_get(url, **kwargs): + self.assertNotIn("trust_env", kwargs) + self.assertEqual(kwargs.get("proxies"), {"http": None, "https": None}) + if "bad" in url: + raise RuntimeError("network token=secret should not leak") + return self._mock_response() + with patch("src.services.intelligence_service.requests.get", side_effect=fake_get): + result = self.service.fetch_enabled_sources() + self.assertEqual(result["source_count"], 2) + self.assertEqual(result["saved_count"], 2) + failures = [item for item in result["results"] if not item["ok"]] + self.assertEqual(len(failures), 1) + self.assertEqual(failures[0]["source_id"], bad["id"]) + self.assertNotIn("token=secret", failures[0]["error"]) + self.assertNotIn("secret", failures[0]["error"]) + + def test_fetch_enabled_sources_paginates_all_enabled_sources(self) -> None: + for index in range(150): + self.service.create_source({ + "name": f"feed-{index}", "url": "https://feeds.example.com/rss.xml", "scope_type": "market", + "market": "cn", }) - with patch.dict(os.environ, {"HTTP_PROXY": "http://127.0.0.1:3128", "HTTPS_PROXY": "http://127.0.0.1:3128", "ALL_PROXY": "http://127.0.0.1:3128"}): - with self._public_dns(), patch("src.services.intelligence_service.requests.get", return_value=self._mock_response()) as mock_get: - self.service.fetch_source(source["id"]) - self.assertEqual(mock_get.call_count, 1) - self.assertEqual(mock_get.call_args.kwargs["proxies"], {"http": None, "https": None}) - def test_fetch_validates_dns_resolution_used_by_request(self) -> None: - public_dns = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0))] - private_dns = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("169.254.169.254", 0))] + with patch("src.services.intelligence_service.requests.get", side_effect=lambda *_args, **_kwargs: self._mock_response()): + result = self.service.fetch_enabled_sources() - def fake_get(url, **kwargs): - socket.getaddrinfo("feeds.example.com", 443, type=socket.SOCK_STREAM) - return self._mock_response() + self.assertEqual(result["source_count"], 150) + self.assertEqual(len(result["results"]), 150) + self.assertEqual(result["saved_count"], 300) + self.assertTrue(all(item["ok"] for item in result["results"])) + + def test_fetch_entry_redacts_no_url_link_with_placeholder(self) -> None: + source = self.service.create_source({ + "name": "market-no-link", + "url": "https://feeds.example.com/rss.xml", + "scope_type": "market", + }) - with patch("src.services.intelligence_service.socket.getaddrinfo", side_effect=[public_dns, private_dns]): - with patch("src.services.intelligence_service.requests.get", side_effect=fake_get) as mock_get: - with self.assertRaises(IntelligenceServiceError): - self.service._get_feed_response("https://feeds.example.com/rss.xml") - self.assertEqual(mock_get.call_count, 1) + response = Mock() + response.status_code = 200 + response.url = "https://feeds.example.com/rss.xml" + response.headers = {} + response.raise_for_status.return_value = None + response.iter_content.return_value = [NO_URL_LINK_FIXTURE] - def test_fetch_streams_response_before_enforcing_byte_cap(self) -> None: - with self._public_dns(): - source = self.service.create_source({ - "name": "large-feed", "url": "https://feeds.example.com/rss.xml", - "scope_type": "market", - }) + with patch("src.services.intelligence_service.requests.get", return_value=response): + result = self.service.fetch_source(source["id"]) - class LargeResponse: - status_code = 200 - headers: dict = {} - url = "https://feeds.example.com/rss.xml" + self.assertEqual(result["fetched_count"], 1) + self.assertEqual(result["saved_count"], 1) + self.assertIn("no-url:intel:", result["sample_items"][0]["url"]) - @property - def content(self): - raise AssertionError("content should not be read") + def test_bad_feed_item_link_is_skipped_without_failing_source(self) -> None: + source = self.service.create_source({ + "name": "mixed-link-feed", + "url": "https://feeds.example.com/rss.xml", + "scope_type": "market", + }) - def raise_for_status(self): - return None + response = Mock() + response.status_code = 200 + response.url = "https://feeds.example.com/rss.xml" + response.headers = {} + response.raise_for_status.return_value = None + response.iter_content.return_value = [BAD_ITEM_LINK_FIXTURE] + + with patch("src.services.intelligence_service.requests.get", return_value=response): + result = self.service.fetch_source(source["id"]) + + self.assertEqual(result["fetched_count"], 1) + self.assertEqual(result["saved_count"], 1) + self.assertEqual(result["sample_items"][0]["url"], "https://news.example.com/good") + + def test_source_templates_can_create_disabled_source(self) -> None: + templates = self.service.list_source_templates(market="hk") + self.assertGreaterEqual(templates["total"], 1) + created = self.service.create_source_from_template("hkex-news", {"enabled": False, "name": "hkex-template-copy"}) + self.assertEqual(created["name"], "hkex-template-copy") + self.assertEqual(created["market"], "hk") + self.assertFalse(created["enabled"]) + + def test_newsnow_source_fetches_json_items(self) -> None: + source = self.service.create_source({ + "name": "newsnow-cls", + "url": "https://newsnow.example.com/api/s?id=cls-hot", + "source_type": "newsnow", + "scope_type": "market", + "market": "cn", + }) - def iter_content(self, chunk_size=1): - yield b"x" * (_MAX_FEED_BYTES + 1) + with patch("src.services.intelligence_service.requests.get", return_value=self._mock_json_response()): + result = self.service.fetch_source(source["id"]) - def close(self): - self.closed = True + self.assertEqual(result["fetched_count"], 2) + self.assertEqual(result["saved_count"], 2) + items = self.service.list_items(market="cn") + self.assertEqual(items["total"], 2) + self.assertEqual(items["items"][0]["source_type"], "newsnow") + self.assertEqual(result["sample_items"][0]["source"], "newsnow-cls") + self.assertEqual(result["sample_items"][0]["summary"], "Capital market hot topic from NewsNow.") + + def test_create_default_sources_is_idempotent(self) -> None: + first = self.service.create_default_sources({"enabled": False}) + second = self.service.create_default_sources({"enabled": False}) + + self.assertGreaterEqual(first["created_count"], 5) + self.assertEqual(second["created_count"], 0) + self.assertEqual(first["total"], second["total"]) + sources = self.service.list_sources(source_type="newsnow", market="cn") + self.assertGreaterEqual(sources["total"], 3) + self.assertTrue(all(not item["source"]["enabled"] for item in first["items"])) + + def test_create_default_sources_are_disabled_by_default(self) -> None: + first = self.service.create_default_sources() + sources = self.service.list_sources() + self.assertEqual(first["created_count"], first["total"]) + self.assertEqual(sources["total"], first["total"]) + self.assertTrue(all(not item["enabled"] for item in sources["items"])) + + def test_same_url_can_be_saved_for_different_scopes(self) -> None: + market = self.service.create_source({ + "name": "market-feed", + "url": "https://feeds.example.com/shared.xml", + "scope_type": "market", + "market": "cn", + }) + symbol = self.service.create_source({ + "name": "symbol-feed", + "url": "https://feeds.example.com/shared.xml", + "scope_type": "symbol", + "scope_value": "600519", + "market": "cn", + }) + response = self._mock_response(source_url="https://feeds.example.com/shared.xml") + with patch("src.services.intelligence_service.requests.get", return_value=response): + market_result = self.service.fetch_source(market["id"]) + symbol_result = self.service.fetch_source(symbol["id"]) + + self.assertEqual(market_result["saved_count"], 2) + self.assertEqual(symbol_result["saved_count"], 2) + rows, total = IntelligenceRepository().list_items() + self.assertEqual(total, 4) + scope_pairs = {(row.scope_type, row.scope_value) for row in rows} + self.assertIn(("market", INTELLIGENCE_ITEM_NULL_SCOPE_VALUE), scope_pairs) + self.assertIn(("symbol", "600519"), scope_pairs) + + def test_redirect_to_private_network_is_blocked(self) -> None: + source = self.service.create_source({ + "name": "redirected-feed", + "url": "https://feeds.example.com/rss.xml", + "scope_type": "market", + }) - response = LargeResponse() - response.closed = False - with self._public_dns(), patch("src.services.intelligence_service.requests.get", return_value=response) as mock_get: - with self.assertRaisesRegex(IntelligenceServiceError, "feed response is too large"): + with patch("src.services.intelligence_service.requests.get", side_effect=[ + self._mock_response_with_redirects(source_url="https://feeds.example.com/rss.xml", next_url="http://localhost/evil.xml"), + self._mock_response(source_url="http://localhost/evil.xml"), + ]): + with self.assertRaises(IntelligenceServiceError): self.service.fetch_source(source["id"]) - self.assertTrue(mock_get.call_args.kwargs["stream"]) - self.assertTrue(response.closed) - def test_fetch_enabled_sources_is_fail_open(self) -> None: - with self._public_dns(): - self.service.create_source({"name": "good-feed", "url": "https://feeds.example.com/rss.xml", "scope_type": "market"}) - bad = self.service.create_source({"name": "bad-feed", "url": "https://bad.example.com/rss.xml", "scope_type": "market"}) + def test_redirect_is_followed_after_dns_validation(self) -> None: + source = self.service.create_source({ + "name": "follow-redirect-feed", + "url": "https://feeds.example.com/rss.xml", + "scope_type": "market", + }) - def fake_get(url, **kwargs): - if "bad" in url: - raise RuntimeError("network token=secret should not leak") - return self._mock_response() - with self._public_dns(), patch("src.services.intelligence_service.requests.get", side_effect=fake_get): - result = self.service.fetch_enabled_sources() - self.assertEqual(result["source_count"], 2) + with patch("src.services.intelligence_service.requests.get", side_effect=[ + self._mock_response_with_redirects(source_url="https://feeds.example.com/rss.xml", next_url="/next.xml"), + self._mock_response(source_url="https://feeds.example.com/next.xml"), + ]): + result = self.service.fetch_source(source["id"]) + + self.assertEqual(result["fetched_count"], 2) self.assertEqual(result["saved_count"], 2) - failures = [item for item in result["results"] if not item["ok"]] - self.assertEqual(len(failures), 1) - self.assertEqual(failures[0]["source_id"], bad["id"]) - self.assertIn("token=***", failures[0]["error"]) - self.assertNotIn("secret", failures[0]["error"]) - def test_sanitize_error_redacts_common_query_secret_names(self) -> None: - sanitized = IntelligenceService._sanitize_error( - RuntimeError( - "failed url=https://feed.example/rss?access_token=first&auth-token=second" - " callback=https://feed.example/rss?api-key=third#secret=fourth" - ) - ) + def test_feed_response_size_limit_is_enforced(self) -> None: + source = self.service.create_source({ + "name": "large-feed", + "url": "https://feeds.example.com/rss.xml", + "scope_type": "market", + }) - self.assertNotIn("first", sanitized) - self.assertNotIn("second", sanitized) - self.assertNotIn("third", sanitized) - self.assertNotIn("fourth", sanitized) - self.assertIn("access_token=***", sanitized) - self.assertIn("auth-token=***", sanitized) - self.assertIn("api-key=***", sanitized) - self.assertIn("secret=***", sanitized) - - def test_fetch_enabled_sources_iterates_every_enabled_source(self) -> None: - with self._public_dns(): - for index in range(101): - self.service.create_source({ - "name": f"feed-{index}", - "url": f"https://feeds{index}.example.com/rss.xml", - "scope_type": "market", - }) - with self._public_dns(), patch("src.services.intelligence_service.requests.get", return_value=self._mock_response()) as mock_get: - result = self.service.fetch_enabled_sources() - self.assertEqual(result["source_count"], 101) - self.assertEqual(len(result["results"]), 101) - self.assertEqual(mock_get.call_count, 101) + large_response = Mock() + large_response.status_code = 200 + large_response.url = "https://feeds.example.com/rss.xml" + large_response.headers = {} + large_response.raise_for_status.return_value = None + large_response.iter_content.return_value = [b"x" * (2 * 1024 * 1024 + 1)] + + with patch("src.services.intelligence_service.requests.get", return_value=large_response): + with self.assertRaises(IntelligenceServiceError): + self.service.fetch_source(source["id"]) def test_retention_removes_old_items(self) -> None: repo = IntelligenceRepository() @@ -251,74 +406,6 @@ def test_retention_removes_old_items(self) -> None: with DatabaseManager.get_instance().get_session() as session: self.assertEqual(session.query(IntelligenceItem).count(), 0) - def test_market_scope_item_uses_non_null_dedupe_key(self) -> None: - repo = IntelligenceRepository() - with self._public_dns(): - source = self.service.create_source({ - "name": "unique-market-feed", - "url": "https://feeds.example.com/rss.xml", - "scope_type": "market", - "market": "cn", - }) - fields = { - "source_id": source["id"], - "source_name": "unique-market-feed", - "source_type": "rss", - "title": "same market item", - "summary": "summary", - "url": "https://news.example.com/unique", - "source": "unique-market-feed", - "published_at": datetime.now(), - "fetched_at": datetime.now(), - "scope_type": "market", - "scope_value": None, - "market": "cn", - } - - self.assertEqual(repo.upsert_items([fields]), 1) - with DatabaseManager.get_instance().get_session() as session: - stored = session.query(IntelligenceItem).one() - self.assertEqual(stored.scope_value, INTELLIGENCE_ITEM_NULL_SCOPE_VALUE) - session.add(IntelligenceItem(**{**fields, "scope_value": INTELLIGENCE_ITEM_NULL_SCOPE_VALUE})) - with self.assertRaises(IntegrityError): - session.flush() - - def test_duplicate_insert_race_does_not_rollback_prior_batch_items(self) -> None: - repo = IntelligenceRepository() - base_fields = { - "source_name": "race-feed", - "source_type": "rss", - "summary": "summary", - "source": "race-feed", - "published_at": datetime.now(), - "fetched_at": datetime.now(), - "scope_type": "market", - "scope_value": None, - "market": "cn", - } - kept = {**base_fields, "title": "kept", "url": "https://news.example.com/kept"} - race = {**base_fields, "title": "race", "url": "https://news.example.com/race"} - after = {**base_fields, "title": "after", "url": "https://news.example.com/after"} - original_flush = Session.flush - raised = False - - def flush_with_duplicate_race(session, *args, **kwargs): - nonlocal raised - if not raised and any( - isinstance(row, IntelligenceItem) and row.url == race["url"] - for row in session.new - ): - raised = True - raise IntegrityError("insert intelligence item", {}, RuntimeError("duplicate url")) - return original_flush(session, *args, **kwargs) - - with patch.object(Session, "flush", flush_with_duplicate_race): - self.assertEqual(repo.upsert_items([kept, race, after]), 2) - - with DatabaseManager.get_instance().get_session() as session: - titles = {row.title for row in session.query(IntelligenceItem).all()} - self.assertEqual(titles, {"kept", "after"}) - if __name__ == "__main__": unittest.main() diff --git a/tests/test_storage.py b/tests/test_storage.py index 10ed97da88a..2ea03c65729 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -2,6 +2,7 @@ import unittest import sys import os +import sqlite3 import tempfile import threading from datetime import date @@ -19,6 +20,121 @@ class TestStorage(unittest.TestCase): + @staticmethod + def _list_sqlite_unique_indexes(db_path: str, table_name: str) -> dict[str, list[str]]: + with sqlite3.connect(db_path) as conn: + rows = conn.execute(f"PRAGMA index_list({table_name})").fetchall() + unique_indexes = {} + for row in rows: + if int(row[2]) != 1: + continue + index_name = row[1] + index_columns = [] + for index_info in conn.execute(f"PRAGMA index_xinfo({index_name})").fetchall(): + column_name = index_info[2] + if column_name is not None: + index_columns.append(column_name) + unique_indexes[index_name] = index_columns + return unique_indexes + + def test_legacy_intelligence_items_url_unique_index_rebuilds_without_collision(self) -> None: + temp_dir = tempfile.TemporaryDirectory() + db_path = os.path.join(temp_dir.name, "legacy_intel.sqlite") + + try: + with sqlite3.connect(db_path) as conn: + conn.execute( + """CREATE TABLE intelligence_sources ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL UNIQUE, + source_type TEXT NOT NULL DEFAULT 'rss', + url TEXT NOT NULL, + scope_type TEXT NOT NULL DEFAULT 'market', + scope_value TEXT, + market TEXT NOT NULL DEFAULT 'cn', + enabled INTEGER NOT NULL DEFAULT 1, + last_status TEXT, + last_error TEXT, + last_fetched_at DATETIME, + created_at DATETIME, + updated_at DATETIME + )""" + ) + conn.execute( + """CREATE TABLE intelligence_items ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source_id INTEGER, + source_name TEXT, + source_type TEXT NOT NULL DEFAULT 'rss', + title TEXT NOT NULL, + summary TEXT, + url TEXT NOT NULL, + source TEXT, + published_at DATETIME, + fetched_at DATETIME, + scope_type TEXT NOT NULL DEFAULT 'market', + scope_value TEXT, + market TEXT NOT NULL DEFAULT 'cn', + raw_payload TEXT + )""" + ) + conn.execute("CREATE UNIQUE INDEX uix_intelligence_item_url_legacy ON intelligence_items(url)") + conn.execute("CREATE INDEX ix_intel_item_scope_time ON intelligence_items(scope_type, scope_value, market, published_at)") + conn.execute("CREATE INDEX ix_intel_item_fetch_time ON intelligence_items(fetched_at)") + conn.execute("INSERT INTO intelligence_sources (name, url) VALUES ('legacy', 'https://legacy.example.com/rss.xml')") + source_id = conn.execute("SELECT id FROM intelligence_sources WHERE name='legacy'").fetchone()[0] + conn.executemany( + """INSERT INTO intelligence_items ( + source_id, + source_name, + source_type, + title, + summary, + url, + source, + published_at, + fetched_at, + scope_type, + scope_value, + market, + raw_payload + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + [ + (source_id, 'legacy-source', 'rss', 'A', 'legacy-a', 'https://legacy.example.com/a', 'legacy', + '2026-01-01 00:00:00', '2026-01-01 00:00:00', 'market', None, 'cn', None), + (source_id, 'legacy-source', 'rss', 'B', 'legacy-b', 'https://legacy.example.com/b', 'legacy', + '2026-01-02 00:00:00', '2026-01-02 00:00:00', 'market', None, 'cn', None), + ], + ) + + unique_indexes_before = self._list_sqlite_unique_indexes(db_path, "intelligence_items") + self.assertIn("uix_intelligence_item_url_legacy", unique_indexes_before) + self.assertEqual(unique_indexes_before["uix_intelligence_item_url_legacy"], ["url"]) + + DatabaseManager.reset_instance() + Config.reset_instance() + DatabaseManager(db_url=f"sqlite:///{db_path}") + + unique_indexes_after = self._list_sqlite_unique_indexes(db_path, "intelligence_items") + self.assertNotIn("uix_intelligence_item_url_legacy", unique_indexes_after) + self.assertIn("uix_intel_item_scope", unique_indexes_after) + self.assertEqual( + unique_indexes_after["uix_intel_item_scope"], + ["source_id", "url", "scope_type", "scope_value", "market"], + ) + with sqlite3.connect(db_path) as conn: + table_count = conn.execute("SELECT COUNT(*) FROM intelligence_items").fetchone()[0] + temp_tables = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'intelligence_items_recreate_tmp_%'" + ).fetchall() + + self.assertEqual(table_count, 2) + self.assertEqual(temp_tables, []) + finally: + DatabaseManager.reset_instance() + Config.reset_instance() + temp_dir.cleanup() + def test_database_initialization_records_schema_version(self): DatabaseManager.reset_instance() db = DatabaseManager(db_url="sqlite:///:memory:")