集成Anspire search (#982)

* 集成Anspire search

* fix: add anspire_api_keys to config and handle safely

* fix: add anspire_api_keys to config and handle safely

* 修改单元测试文件、硬编码问题、systemConfigI18n描述

---------

Co-authored-by: xiaowei <4775987@qq.com>

.env.example

@@ -112,6 +112,9 @@ GEMINI_API_KEY=
 # ===================================
 # 搜索引擎配置（用于获取股票新闻）
 # ===================================
+#Anspire Search API Keys （支持多个，逗号分隔）
+# 获取: https://aisearch.anspire.cn/
+ANSPIRE_API_KEYS=
 # Bocha API Keys（中文搜索优化，支持AI摘要，支持多个，逗号分隔）
 # 获取: https://open.bocha.cn/
 # BOCHA_API_KEYS=your_bocha_key_here

README.md

@@ -70,7 +70,7 @@
 |------|------|
 | AI 模型 | [AIHubMix](https://aihubmix.com/?aff=CfMq)、Gemini、OpenAI 兼容、DeepSeek、通义千问、Claude、Ollama 本地模型 等（统一通过 [LiteLLM](https://github.com/BerriAI/litellm) 调用，支持多 Key 负载均衡）|
 | 行情数据 | AkShare、Tushare、Pytdx、Baostock、YFinance、[Longbridge](https://open.longbridge.com/)（美股/港股首选数据源） |
-| 新闻搜索 | Tavily、SerpAPI、Bocha、Brave、MiniMax |
+| 新闻搜索 | Anspire、Tavily、SerpAPI、Bocha、Brave、MiniMax |
 | 社交舆情 | [Stock Sentiment API](https://api.adanos.org/docs)（Reddit / X / Polymarket，仅美股，可选） |
 
 > **长桥优先策略（仅美/港股）**：在配置 `LONGBRIDGE_APP_KEY` / `LONGBRIDGE_APP_SECRET` / `LONGBRIDGE_ACCESS_TOKEN` 的前提下，美股与港股的 **日线 K 线** 与 **实时行情** 由 **Longbridge 优先拉取**；若长桥失败或部分字段缺失，再由 **YFinance（美股）/ AkShare（港股）** 兜底或合并补全字段。**未配置长桥凭据时不会调用 Longbridge**，美股/港股仍以 YFinance / AkShare 为主数据源（与未集成长桥前的行为一致）。**美股大盘指数**（如 SPX）始终以 YFinance 优先（长桥不提供指数行情）。**A 股**路由不变，仍为 Efinance → AkShare → Tushare → Pytdx → Baostock。详见 `.env.example` 与 [完整指南](docs/full-guide.md) 中长桥说明。
@@ -179,6 +179,7 @@
 |------------|------|:----:|
 | `STOCK_LIST` | 自选股代码，如 `600519,hk00700,AAPL,TSLA` | ✅ |
 | `TAVILY_API_KEYS` | [Tavily](https://tavily.com/) 搜索 API（新闻搜索） | 推荐 |
+| `ANSPIRE_API_KEYS` | [Anspire AI Search](https://aisearch.anspire.cn/) 针对中文内容特别优化 (可有效增强A股分析效果) | 可选 |
 | `MINIMAX_API_KEYS` | [MiniMax](https://platform.minimaxi.com/) Coding Plan Web Search（结构化搜索结果） | 可选 |
 | `SERPAPI_API_KEYS` | [SerpAPI](https://serpapi.com/baidu-search-api?utm_source=github_daily_stock_analysis) 全渠道搜索 | 可选 |
 | `BOCHA_API_KEYS` | [博查搜索](https://open.bocha.cn/) Web Search API（中文搜索优化，支持AI摘要，多个key用逗号分隔） | 可选 |

apps/dsa-web/src/utils/systemConfigI18n.ts

@@ -27,6 +27,7 @@ const fieldTitleMap: Record<string, string> = {
   TUSHARE_TOKEN: 'Tushare Token',
   BOCHA_API_KEYS: 'Bocha API Keys',
   TAVILY_API_KEYS: 'Tavily API Keys',
+  ANSPIRE_API_KEYS: 'Anspire API Keys',
   SERPAPI_API_KEYS: 'SerpAPI API Keys',
   BRAVE_API_KEYS: 'Brave API Keys',
   SEARXNG_BASE_URLS: 'SearXNG Base URLs',
@@ -88,6 +89,7 @@ const fieldDescriptionMap: Record<string, string> = {
   TUSHARE_TOKEN: '用于接入 Tushare Pro 数据服务的凭据。',
   BOCHA_API_KEYS: '用于新闻检索的 Bocha 密钥，支持逗号分隔多个（最高优先级）。',
   TAVILY_API_KEYS: '用于新闻检索的 Tavily 密钥，支持逗号分隔多个。',
+  ANSPIRE_API_KEYS: '用于新闻检索的 Anspire 密钥，支持逗号分隔多个。',
   SERPAPI_API_KEYS: '用于新闻检索的 SerpAPI 密钥，支持逗号分隔多个。',
   BRAVE_API_KEYS: '用于新闻检索的 Brave Search 密钥，支持逗号分隔多个。',
   SEARXNG_BASE_URLS: 'SearXNG 自建实例地址（逗号分隔，无配额兜底，需在 settings.yml 启用 format: json）。',

bot/commands/market.py

@@ -85,6 +85,7 @@ class MarketCommand(BotCommand):
             search_service = None
             if config.has_search_capability_enabled():
                 search_service = SearchService(
+                    anspire_keys=config.anspire_api_keys,
                     bocha_keys=config.bocha_api_keys,
                     tavily_keys=config.tavily_api_keys,
                     brave_keys=config.brave_api_keys,

docs/CHANGELOG.md

@@ -30,6 +30,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 - [修复] SQLite 主写入链路现在对 `stock_daily(code,date)` 使用批量原子 upsert，并在文件型 SQLite 连接上默认启用 `WAL`、`busy_timeout` 与有限写入重试，降低批量分析和并发回写场景下的锁竞争与吞吐抖动，返回值中的“新增数”改为按本次真正插入窗口计算（并发场景不再把并行写入行误算入当前调用）。
 - [修复] 优化多 Agent 与单 Agent 的预算护栏语义：当后续阶段/步骤剩余预算低于最小阈值（首阶段除外）时会主动跳过并进行降级处理；若当前已完成阶段可支持构建降级报告，则返回 `success=True` 并携带非空内容；否则返回 `success=False`、`content=""`；`run_agent_loop` 预算过低时当前仍返回失败降级语义（`success=False`、`content=""`），`AgentExecutor` 保持统一下游契约。
 
+- [新功能] 集成 Anspire Search 作为可选语义搜索后端; 配置 `ANSPIRE_*` 可使用Anspire Search获取实时行情及新闻资讯，未配置时行为与此前一致。Anspire Search请使用 `tests/test_anspire_search.py`（手动脚本）。
+
 ## [3.12.0] - 2026-04-01
 
 ### 发布亮点

docs/README_CHT.md

@@ -55,7 +55,7 @@
 |------|------|
 | AI 模型 | Gemini（免費）、OpenAI 兼容、DeepSeek、通義千問、Claude、Ollama |
 | 行情數據 | AkShare、Tushare、Pytdx、Baostock、YFinance、[Longbridge](https://open.longbridge.com/)（美股/港股首選數據源） |
-| 新聞搜索 | Tavily、SerpAPI、Bocha、Brave、MiniMax |
+| 新聞搜索 | Tavily、Anspire、SerpAPI、Bocha、Brave、MiniMax |
 
 > **長橋優先策略（僅美／港股）**：在已設定 `LONGBRIDGE_APP_KEY` / `LONGBRIDGE_APP_SECRET` / `LONGBRIDGE_ACCESS_TOKEN` 的前提下，美股與港股的 **日線** 與 **即時行情** 由 **Longbridge 優先**；長橋失敗或欄位不足時再由 **YFinance／AkShare** 兜底或合併補欄。**未設定長橋憑證時不會呼叫 Longbridge**，美／港股仍以 YFinance／AkShare 為主（與未整合長橋前一致）。**美股大盤指數**始終以 YFinance 優先（長橋不提供指數行情）。**A 股**路由不變。詳見 `.env.example` 與 [完整指南](./full-guide.md)。
 
@@ -134,6 +134,7 @@
 |------------|------|:----:|
 | `STOCK_LIST` | 自選股代碼，如 `600519,hk00700,AAPL,TSLA` | ✅ |
 | `TAVILY_API_KEYS` | [Tavily](https://tavily.com/) 搜索 API（新聞搜索） | 推薦 |
+| `ANSPIRE_API_KEYS` | [Anspire AI Search](https://aisearch.anspire.cn/) 針對中文內容特別優化（可有效增強A股分析效果） | 可選 |
 | `MINIMAX_API_KEYS` | [MiniMax](https://platform.minimaxi.com/) Coding Plan Web Search（結構化搜索結果） | 可選 |
 | `BOCHA_API_KEYS` | [博查搜索](https://open.bocha.cn/) Web Search API（中文搜索優化，支持AI摘要，多個key用逗號分隔） | 可選 |
 | `BRAVE_API_KEYS` | [Brave Search](https://brave.com/search/api/) API（隱私優先，美股優化，多個key用逗號分隔） | 可選 |

docs/README_EN.md

@@ -58,7 +58,7 @@ English | [简体中文](../README.md) | [繁體中文](README_CHT.md)
 |------|----------|
 | LLMs | Gemini (free), OpenAI-compatible, DeepSeek, Qwen, Claude, Ollama |
 | Market Data | AkShare, Tushare, Pytdx, Baostock, YFinance, [Longbridge](https://open.longbridge.com/) (primary for US/HK when configured) |
-| News Search | Tavily, SerpAPI, Bocha, Brave, MiniMax |
+| News Search | Tavily, Anspire、SerpAPI, Bocha, Brave, MiniMax |
 
 > **Longbridge-first (US/HK only):** With `LONGBRIDGE_APP_KEY` / `LONGBRIDGE_APP_SECRET` / `LONGBRIDGE_ACCESS_TOKEN` set, **daily bars and realtime quotes** for US & HK stocks are fetched from **Longbridge first**; **YFinance / AkShare** are used for **fallback** or **field merge** when Longbridge fails or returns incomplete fields. **If Longbridge is not configured, it is not called** — US/HK still use YFinance / AkShare as before. **US market indices** (e.g. SPX) always prefer **YFinance** (indices are not supported on Longbridge). **A-share** routing is unchanged. See `.env.example` and the [full guide](./full-guide_EN.md).
 
@@ -138,6 +138,7 @@ Go to your forked repo → `Settings` → `Secrets and variables` → `Actions`
 |------------|------|:----:|
 | `STOCK_LIST` | Watchlist codes, e.g., `600519,AAPL,hk00700` | ✅ |
 | `TAVILY_API_KEYS` | [Tavily](https://tavily.com/) Search API (for news) | Recommended |
+| `ANSPIRE_API_KEYS` | [Anspire AI Search](https://aisearch.anspire.cn/) Specially optimized for Chinese content (effectively enhances A-share analysis) | Optional |
 | `MINIMAX_API_KEYS` | [MiniMax](https://platform.minimaxi.com/) Coding Plan Web Search (structured search results) | Optional |
 | `BRAVE_API_KEYS` | [Brave Search](https://brave.com/search/api/) API (privacy-focused, US stocks optimized) | Optional |
 | `SERPAPI_API_KEYS` | [SerpAPI](https://serpapi.com/baidu-search-api?utm_source=github_daily_stock_analysis) Backup search | Optional |

docs/full-guide.md

@@ -122,6 +122,7 @@ daily_stock_analysis/
 |------------|------|:----:|
 | `STOCK_LIST` | 自选股代码，如 `600519,300750,002594` | ✅ |
 | `TAVILY_API_KEYS` | [Tavily](https://tavily.com/) 搜索 API（新闻搜索） | 推荐 |
+| `ANSPIRE_API_KEYS` | [Anspire AI Search](https://aisearch.anspire.cn/) 针对中文内容特别优化 (可有效增强A股分析效果) | 可选 |
 | `MINIMAX_API_KEYS` | [MiniMax](https://platform.minimaxi.com/) Coding Plan Web Search（结构化搜索结果） | 可选 |
 | `BOCHA_API_KEYS` | [博查搜索](https://open.bocha.cn/) Web Search API（中文搜索优化，支持AI摘要，多个key用逗号分隔） | 可选 |
 | `BRAVE_API_KEYS` | [Brave Search](https://brave.com/search/api/) API（隐私优先，美股优化，多个key用逗号分隔） | 可选 |
@@ -258,6 +259,7 @@ daily_stock_analysis/
 | 变量名 | 说明 | 必填 |
 |--------|------|:----:|
 | `TAVILY_API_KEYS` | Tavily 搜索 API Key（推荐） | 推荐 |
+| `ANSPIRE_API_KEYS` | Anspire 搜索 API Key（可有效增强A股分析效果） | 可选 | 
 | `MINIMAX_API_KEYS` | MiniMax Coding Plan Web Search（结构化搜索结果） | 可选 |
 | `BOCHA_API_KEYS` | 博查搜索 API Key（中文优化） | 可选 |
 | `BRAVE_API_KEYS` | Brave Search API Key（美股优化） | 可选 |

docs/full-guide_EN.md

@@ -117,6 +117,7 @@ Go to your forked repo → `Settings` → `Secrets and variables` → `Actions`
 |------------|------|:----:|
 | `STOCK_LIST` | Watchlist codes, e.g., `600519,300750,002594` | ✅ |
 | `TAVILY_API_KEYS` | [Tavily](https://tavily.com/) Search API (for news search) | Recommended |
+| `ANSPIRE_API_KEYS` | [Anspire AI Search](https://aisearch.anspire.cn/) Specially optimized for Chinese content (effectively enhances A-share analysis) | Optional |
 | `MINIMAX_API_KEYS` | [MiniMax](https://platform.minimaxi.com/) Coding Plan Web Search (structured search results) | Optional |
 | `BOCHA_API_KEYS` | [Bocha Search](https://open.bocha.cn/) Web Search API (Chinese search optimized, supports AI summaries, multiple keys comma-separated) | Optional |
 | `SERPAPI_API_KEYS` | [SerpAPI](https://serpapi.com/baidu-search-api?utm_source=github_daily_stock_analysis) Backup search | Optional |
@@ -234,6 +235,7 @@ Default schedule: Every weekday at **18:00 (Beijing Time)** automatic execution.
 | Variable | Description | Required |
 |--------|------|:----:|
 | `TAVILY_API_KEYS` | Tavily Search API Key (recommended) | Recommended |
+| `ANSPIRE_API_KEYS` | Anspire Search API Key (effectively enhances A-share analysis) | Optional | 
 | `MINIMAX_API_KEYS` | MiniMax Coding Plan Web Search (structured results) | Optional |
 | `BOCHA_API_KEYS` | Bocha Search API Key (Chinese optimized) | Optional |
 | `BRAVE_API_KEYS` | Brave Search API Key (US stocks optimized) | Optional |

main.py

@@ -852,6 +852,7 @@ def main() -> int:
                 search_service = SearchService(
                     bocha_keys=config.bocha_api_keys,
                     tavily_keys=config.tavily_api_keys,
+                    anspire_keys=config.anspire_api_keys,
                     brave_keys=config.brave_api_keys,
                     serpapi_keys=config.serpapi_keys,
                     minimax_keys=config.minimax_api_keys,

src/config.py

@@ -498,6 +498,7 @@ class Config:
     vision_provider_priority: str = "gemini,anthropic,openai"
 
     # === 搜索引擎配置（支持多 Key 负载均衡）===
+    anspire_api_keys: List[str] = field(default_factory=list)  # Anspire Search API Keys
     bocha_api_keys: List[str] = field(default_factory=list)  # Bocha API Keys
     minimax_api_keys: List[str] = field(default_factory=list)  # MiniMax API Keys
     tavily_api_keys: List[str] = field(default_factory=list)  # Tavily API Keys
@@ -1019,6 +1020,10 @@ class Config:
         )
 
         # 解析搜索引擎 API Keys（支持多个 key，逗号分隔）
+        # Anspire Search
+        anspire_keys_str = os.getenv('ANSPIRE_API_KEYS', '')
+        anspire_api_keys = [k.strip() for k in anspire_keys_str.split(',') if k.strip()]
+
         bocha_keys_str = os.getenv('BOCHA_API_KEYS', '')
         bocha_api_keys = [k.strip() for k in bocha_keys_str.split(',') if k.strip()]
 
@@ -1151,6 +1156,7 @@ class Config:
                 or ""
             ),
             vision_provider_priority=os.getenv('VISION_PROVIDER_PRIORITY', 'gemini,anthropic,openai'),
+            anspire_api_keys=anspire_api_keys,
             bocha_api_keys=bocha_api_keys,
             minimax_api_keys=minimax_api_keys,
             tavily_api_keys=tavily_api_keys,
@@ -1899,7 +1905,8 @@ class Config:
     def has_search_capability_enabled(self) -> bool:
         """Whether any search provider is configured or SearXNG fallback is enabled."""
         return bool(
-            self.bocha_api_keys
+            self.anspire_api_keys
+            or self.bocha_api_keys
             or self.minimax_api_keys
             or self.tavily_api_keys
             or self.brave_api_keys

src/core/config_registry.py

@@ -269,6 +269,20 @@ _FIELD_DEFINITIONS: Dict[str, Dict[str, Any]] = {
         "validation": {},
         "display_order": 21,
     },
+    "ANSPIRE_API_KEYS": {
+        "title": "Anspire API Keys",
+        "description": "Comma-separated Anspire Search API keys.",
+        "category": "data_source",
+        "data_type": "string",
+        "ui_control": "password",
+        "is_sensitive": True,
+        "is_required": False,
+        "is_editable": True,
+        "default_value": None,
+        "options": [],
+        "validation": {"multi_value": True, "delimiter": ","},
+        "display_order": 22,
+    },
     "TAVILY_API_KEYS": {
         "title": "Tavily API Keys",
         "description": "Comma-separated Tavily API keys.",
@@ -1890,6 +1904,7 @@ def _infer_category(key: str) -> str:
             "SERPAPI",
             "BRAVE",
             "BOCHA",
+            "ANSPIRE",
             "SEARXNG",
             "NEWS_",
             "BIAS_",

src/core/pipeline.py

@@ -107,6 +107,7 @@ class StockAnalysisPipeline:
             self.search_service = SearchService(
                 bocha_keys=self.config.bocha_api_keys,
                 tavily_keys=self.config.tavily_api_keys,
+                anspire_keys=self.config.anspire_api_keys,
                 brave_keys=self.config.brave_api_keys,
                 serpapi_keys=self.config.serpapi_keys,
                 minimax_keys=self.config.minimax_api_keys,

src/search_service.py

@@ -1058,6 +1058,194 @@ class BochaSearchProvider(BaseSearchProvider):
             return '未知来源'
 
 
+class AnspireSearchProvider(BaseSearchProvider):
+    """
+    Anspire Search 搜索引擎
+    
+    特点：
+    - 面向AI生态的下一代实时智能搜索引擎
+    - 结果精准、响应快速
+    - 适用于股票新闻和市场情报搜索
+    
+    文档: https://open.anspire.cn/document/docs/searchApi/
+    """
+    
+    def __init__(self, api_keys: List[str]):
+        super().__init__(api_keys, "Anspire")
+    
+    def _do_search(self, query: str, api_key: str, max_results: int, days: int = 7) -> SearchResponse:
+        """执行 Anspire 搜索"""
+        try:
+            import requests
+        except ImportError:
+            return SearchResponse(
+                query=query,
+                results=[],
+                provider=self.name,
+                success=False,
+                error_message="requests 未安装，请运行：pip install requests"
+            )
+        
+        try:
+            # API 端点
+            url = "https://plugin.anspire.cn/api/ntsearch/search"
+            
+            # 请求头
+            headers = {
+                'Authorization': f'Bearer {api_key}'
+            }
+
+            # 请求参数
+            payload = {
+                "query": query,
+                "top_k": min(max_results,50), 
+                "FromTime": (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S"),
+                "ToTime": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            }
+            
+            # 执行搜索
+            response = _get_with_retry(url, headers=headers, params=payload, timeout=10)
+            
+            # 检查 HTTP 状态码
+            if response.status_code != 200:
+                # 尝试解析错误信息
+                try:
+                    if response.headers.get('content-type', '').startswith('application/json'):
+                        error_data = response.json()
+                        error_message = error_data.get('message', response.text)
+                    else:
+                        error_message = response.text
+                except Exception:
+                    error_message = response.text
+                
+                # 根据错误码处理
+                if response.status_code == 403:
+                    error_msg = f"余额不足或权限不足：{error_message}"
+                elif response.status_code == 401:
+                    error_msg = f"API KEY 无效：{error_message}"
+                elif response.status_code == 400:
+                    error_msg = f"请求参数错误：{error_message}"
+                else:
+                    error_msg = f"HTTP {response.status_code}: {error_message}"
+                
+                logger.warning(f"[Anspire] 搜索失败：{error_msg}")
+                
+                return SearchResponse(
+                    query=query,
+                    results=[],
+                    provider=self.name,
+                    success=False,
+                    error_message=error_msg
+                )
+            
+            # 解析响应
+            try:
+                data = response.json()
+            except ValueError as e:
+                error_msg = f"响应 JSON 解析失败：{str(e)}"
+                logger.error(f"[Anspire] {error_msg}")
+                return SearchResponse(
+                    query=query,
+                    results=[],
+                    provider=self.name,
+                    success=False,
+                    error_message=error_msg
+                )
+            
+            if 'code' in data and data.get('code') != 200:
+                error_msg = data.get('msg') or f"API 返回错误码：{data.get('code')}"
+                logger.warning(f"[Anspire] 搜索失败：{error_msg}")
+                return SearchResponse(
+                    query=query,
+                    results=[],
+                    provider=self.name,
+                    success=False,
+                    error_message=error_msg
+                )
+            
+            if 'results' not in data:
+                error_msg = "响应中缺少 results 字段"
+                logger.error(f"[Anspire] {error_msg}，原始响应：{data}")
+                return SearchResponse(
+                    query=query,
+                    results=[],
+                    provider=self.name,
+                    success=False,
+                    error_message=error_msg
+                )
+            
+            # 记录原始响应到日志
+            logger.info(f"[Anspire] 搜索完成，query='{query}'")
+            logger.debug(f"[Anspire] 原始响应：{data}")
+            
+            results = []
+            value_list = data.get('results', [])
+            
+            for item in value_list[:max_results]:
+                snippet = item.get('content')
+                if snippet and isinstance(snippet, str) and len(snippet) > 500:
+                    snippet = snippet[:500] + "..."
+                
+                results.append(SearchResult(
+                    title=item.get('title', ''),
+                    snippet=snippet,
+                    url=item.get('url', ''),
+                    source=self._extract_domain(item.get('url', '')),
+                    published_date=item.get('date', '')
+                ))
+            
+            logger.info(f"[Anspire] 成功解析 {len(results)} 条结果")
+            
+            return SearchResponse(
+                query=query,
+                results=results,
+                provider=self.name,
+                success=True,
+            )
+            
+        except requests.exceptions.Timeout:
+            error_msg = "请求超时"
+            logger.error(f"[Anspire] {error_msg}")
+            return SearchResponse(
+                query=query,
+                results=[],
+                provider=self.name,
+                success=False,
+                error_message=error_msg
+            )
+        except requests.exceptions.RequestException as e:
+            error_msg = f"网络请求失败：{str(e)}"
+            logger.error(f"[Anspire] {error_msg}")
+            return SearchResponse(
+                query=query,
+                results=[],
+                provider=self.name,
+                success=False,
+                error_message=error_msg
+            )
+        except Exception as e:
+            error_msg = f"未知错误：{str(e)}"
+            logger.error(f"[Anspire] {error_msg}")
+            return SearchResponse(
+                query=query,
+                results=[],
+                provider=self.name,
+                success=False,
+                error_message=error_msg
+            )
+    
+    @staticmethod
+    def _extract_domain(url: str) -> str:
+        """从 URL 提取域名作为来源"""
+        try:
+            from urllib.parse import urlparse
+            parsed = urlparse(url)
+            domain = parsed.netloc.replace('www.', '')
+            return domain or '未知来源'
+        except Exception:
+            return '未知来源'
+
+
 class MiniMaxSearchProvider(BaseSearchProvider):
     """
     MiniMax Web Search (Coding Plan API)
@@ -1933,6 +2121,7 @@ class SearchService:
         self,
         bocha_keys: Optional[List[str]] = None,
         tavily_keys: Optional[List[str]] = None,
+        anspire_keys: Optional[List[str]] = None,
         brave_keys: Optional[List[str]] = None,
         serpapi_keys: Optional[List[str]] = None,
         minimax_keys: Optional[List[str]] = None,
@@ -1947,6 +2136,7 @@ class SearchService:
         Args:
             bocha_keys: 博查搜索 API Key 列表
             tavily_keys: Tavily API Key 列表
+            anspire_keys: Anspire Search API Key 列表
             brave_keys: Brave Search API Key 列表
             serpapi_keys: SerpAPI Key 列表
             minimax_keys: MiniMax API Key 列表
@@ -2011,6 +2201,11 @@ class SearchService:
             else:
                 logger.info("已启用 SearXNG 公共实例自动发现模式")
 
+        # 7. Anspire Search（实时智能搜索优化）
+        if anspire_keys:
+            self._providers.insert(0, AnspireSearchProvider(anspire_keys))
+            logger.info(f"已配置 Anspire Search 搜索，共 {len(anspire_keys)} 个 API Key")
+            
         if not self._providers:
             logger.warning("未配置任何搜索能力，新闻搜索功能将不可用")
 
@@ -3243,6 +3438,7 @@ def get_search_service() -> SearchService:
                 _search_service = SearchService(
                     bocha_keys=config.bocha_api_keys,
                     tavily_keys=config.tavily_api_keys,
+                    anspire_keys=config.anspire_api_keys,
                     brave_keys=config.brave_api_keys,
                     serpapi_keys=config.serpapi_keys,
                     minimax_keys=config.minimax_api_keys,

tests/test_anspire_search.py

@@ -0,0 +1,633 @@
+# -*- coding: utf-8 -*-
+"""
+Anspire Search 搜索引擎测试套件
+
+测试覆盖范围:
+1. 配置加载测试 - 验证 anspire_api_keys 是否正确从环境变量加载
+2. 服务初始化测试 - 验证 SearchService 是否正确初始化 AnspireSearchProvider
+3. API 调用测试 - 实际调用 Anspire API 验证返回结果
+4. 故障转移测试 - 验证无效 Key 时的错误处理和降级机制
+5. 搜索功能测试 - 测试股票新闻搜索和通用搜索功能
+
+运行方式:
+```bash
+# Windows PowerShell
+$env:ANSPIRE_API_KEYS="your_test_api_key"
+python -m pytest tests/test_anspire_search.py -v
+
+# Linux/Mac
+export ANSPIRE_API_KEYS="your_test_api_key"
+python -m pytest tests/test_anspire_search.py -v
+```
+"""
+
+import os
+import sys
+import unittest
+from datetime import datetime, timedelta
+from pathlib import Path
+from types import ModuleType
+from unittest.mock import MagicMock, patch
+
+import pytest
+from dotenv import load_dotenv
+load_dotenv()
+
+# 添加项目根目录到 Python 路径，解决模块导入问题
+project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if project_root not in sys.path:
+    sys.path.insert(0, project_root)
+
+# Mock newspaper before search_service import (optional dependency)
+if "newspaper" not in sys.modules:
+    mock_np = MagicMock()
+    mock_np.Article = MagicMock()
+    mock_np.Config = MagicMock()
+    sys.modules["newspaper"] = mock_np
+
+from src.config import Config, get_config
+from src.search_service import (
+    AnspireSearchProvider,
+    SearchService,
+    get_search_service,
+    reset_search_service,
+)
+
+
+class _FakeResponse:
+    """模拟 HTTP 响应对象"""
+    def __init__(self, status_code=200, json_data=None, text="", headers=None):
+        self.status_code = status_code
+        self._json_data = json_data or {}
+        self.text = text
+        self.headers = headers or {'content-type': 'application/json'}
+    
+    def json(self):
+        return self._json_data
+
+
+class TestAnspireConfigLoading(unittest.TestCase):
+    """Test Anspire configuration loading from environment variables."""
+    
+    def setUp(self):
+        """保存并清除环境变量（不操作 .env 文件）"""
+        # ✅ 保存原始值，测试后恢复
+        self._original_anspire_keys = os.environ.get('ANSPIRE_API_KEYS')
+        
+        # 清除环境变量
+        if 'ANSPIRE_API_KEYS' in os.environ:
+            del os.environ['ANSPIRE_API_KEYS']
+        
+        # 重置 Config 单例
+        Config._Config__instance = None
+        reset_search_service()
+
+    def tearDown(self):
+        """恢复原始环境变量"""
+        # ✅ 恢复原始值
+        if self._original_anspire_keys is not None:
+            os.environ['ANSPIRE_API_KEYS'] = self._original_anspire_keys
+        elif 'ANSPIRE_API_KEYS' in os.environ:
+            del os.environ['ANSPIRE_API_KEYS']
+        
+        # 重置 Config 单例
+        Config._Config__instance = None
+        reset_search_service()
+
+    def test_anspire_keys_loaded_from_env(self):
+        """Test that ANSPIRE_API_KEYS is correctly parsed from environment."""
+        # ✅ 使用 patch.dict 临时设置，测试后自动恢复
+        with patch.dict(os.environ, {'ANSPIRE_API_KEYS': 'key1,key2,key3'}):
+            config = Config._load_from_env()
+            
+            self.assertEqual(len(config.anspire_api_keys), 3)
+            self.assertIn('key1', config.anspire_api_keys)
+            self.assertIn('key2', config.anspire_api_keys)
+            self.assertIn('key3', config.anspire_api_keys)
+
+    def test_anspire_keys_single_key(self):
+        """Test single API Key parsing."""
+        with patch.dict(os.environ, {'ANSPIRE_API_KEYS': 'single_key_test'}):
+            config = Config._load_from_env()
+            
+            self.assertEqual(len(config.anspire_api_keys), 1)
+            self.assertEqual(config.anspire_api_keys[0], 'single_key_test')
+
+    def test_anspire_keys_empty_env(self):
+        """Test empty environment variable handling."""
+        with patch.dict(os.environ, {'ANSPIRE_API_KEYS': ''}):
+            config = Config._load_from_env()
+            
+            self.assertEqual(len(config.anspire_api_keys), 0)
+
+    def test_anspire_keys_whitespace_handling(self):
+        """Test whitespace trimming in API Keys."""
+        with patch.dict(os.environ, {'ANSPIRE_API_KEYS': ' key1 , key2 , key3 '}):
+            config = Config._load_from_env()
+            
+            self.assertEqual(len(config.anspire_api_keys), 3)
+            self.assertEqual(config.anspire_api_keys, ['key1', 'key2', 'key3'])
+
+
+class TestAnspireSearchProvider(unittest.TestCase):
+    """Anspire Search Provider 单元测试"""
+    
+    def setUp(self):
+        """测试前准备"""
+        # ✅ 使用明确的测试占位符，不是真实密钥形态
+        self.test_api_key = "sk-test-anspire-placeholder-key-12345"
+        self.provider = AnspireSearchProvider([self.test_api_key])
+        # 保存原始 requests 模块
+        self._original_requests = sys.modules.get('requests')
+    
+    def tearDown(self):
+        """测试后清理"""
+        # 恢复原始 requests 模块
+        if self._original_requests is not None:
+            sys.modules['requests'] = self._original_requests
+    
+    def test_provider_initialization(self):
+        """测试 Provider 初始化"""
+        provider = AnspireSearchProvider(["key1", "key2"])
+        self.assertEqual(provider.name, "Anspire")
+        if hasattr(provider, 'api_keys'):
+            self.assertEqual(len(provider.api_keys), 2)
+        elif hasattr(provider, '_api_keys'):
+            self.assertEqual(len(provider._api_keys), 2)
+        self.assertTrue(provider.is_available)
+    
+    def test_provider_name(self):
+        """测试 Provider 名称"""
+        self.assertEqual(self.provider.name, "Anspire")
+    
+    def test_provider_availability(self):
+        """测试 Provider 可用性检测"""
+        # 有 API Key 时应可用
+        provider_with_keys = AnspireSearchProvider(["key1"])
+        self.assertTrue(provider_with_keys.is_available)
+        
+        # 无 API Key 时不可用
+        provider_without_keys = AnspireSearchProvider([])
+        self.assertFalse(provider_without_keys.is_available)
+    
+    def test_extract_domain(self):
+        """测试域名提取功能"""
+        test_cases = [
+            ("https://www.example.com/article", "example.com"),
+            ("https://finance.sina.com.cn/stock/", "finance.sina.com.cn"),
+            ("http://www.10jqka.com.cn/news", "10jqka.com.cn"),
+            ("invalid_url", "未知来源"),
+            ("", "未知来源"),
+        ]
+        
+        for url, expected in test_cases:
+            result = AnspireSearchProvider._extract_domain(url)
+            self.assertEqual(result, expected, f"Failed for URL: {url}")
+    
+    @patch('src.search_service.requests')
+    def test_search_success_response(self, mock_requests):
+        """测试成功响应处理"""
+        # 设置 mock exceptions
+        try:
+            import requests as real_requests
+            mock_requests.exceptions = real_requests.exceptions
+        except ImportError:
+            pass
+        
+        fake_response = _FakeResponse(
+            status_code=200,
+            json_data={
+                "code": 200,
+                "msg": "success",
+                "results": [
+                    {
+                        "title": "贵州茅台今日股价上涨",
+                        "url": "https://finance.sina.com.cn/stock/600519",
+                        "content": "贵州茅台 (600519) 今日收盘股价上涨 2.5%，成交量放大...",
+                    },
+                    {
+                        "title": "白酒板块持续走强",
+                        "url": "https://www.10jqka.com.cn/baijiu",
+                        "content": "白酒板块今日表现强势，贵州茅台、五粮液等个股涨幅居前...",
+                    }
+                ]
+            }
+        )
+        
+        mock_requests.get = MagicMock(return_value=fake_response)
+        
+        response = self.provider.search("贵州茅台 股票新闻", max_results=5, days=7)
+        
+        # 验证结果
+        self.assertTrue(response.success)
+        self.assertEqual(response.provider, "Anspire")
+        self.assertEqual(len(response.results), 2)
+        self.assertEqual(response.results[0].title, "贵州茅台今日股价上涨")
+        # 假设 source 是从 url 提取的域名
+        self.assertEqual(response.results[0].source, "finance.sina.com.cn")
+        
+        # 验证 API 调用参数
+        mock_requests.get.assert_called_once()
+        call_args = mock_requests.get.call_args
+        # 检查 URL 是否包含 anspire 相关域名 (具体 URL 需根据实际实现调整)
+        # self.assertIn("plugin.anspire.cn", call_args[0][0]) 
+        self.assertIn("Authorization", call_args[1]["headers"])
+        # 验证使用 params 而非 json
+        self.assertIn("params", call_args[1])
+        self.assertNotIn("json", call_args[1])
+    
+    @patch('src.search_service.requests')
+    def test_search_invalid_api_key(self, mock_requests):
+        """测试无效 API Key 的错误处理"""
+        try:
+            import requests as real_requests
+            mock_requests.exceptions = real_requests.exceptions
+        except ImportError:
+            pass
+        
+        fake_response = _FakeResponse(
+            status_code=401,
+            json_data={"message": "Invalid API key"},
+            text="Unauthorized"
+        )
+        
+        mock_requests.get = MagicMock(return_value=fake_response)
+        
+        response = self.provider.search("测试查询", max_results=3)
+        
+        self.assertFalse(response.success)
+        self.assertEqual(response.provider, "Anspire")
+        self.assertEqual(len(response.results), 0)
+        # 错误消息可能因实现而异，这里做宽松检查
+        self.assertTrue("API" in response.error_message or "KEY" in response.error_message or "无效" in response.error_message)
+    
+    @patch('src.search_service.requests')
+    def test_search_timeout_error(self, mock_requests):
+        """测试超时错误处理"""
+        try:
+            import requests as real_requests
+            mock_requests.exceptions = real_requests.exceptions
+            timeout_exc = mock_requests.exceptions.Timeout
+        except ImportError:
+            mock_requests.exceptions = MagicMock()
+            timeout_exc = Exception
+            
+        mock_requests.get = MagicMock(side_effect=timeout_exc())
+        
+        response = self.provider.search("测试查询", max_results=3)
+        
+        self.assertFalse(response.success)
+        self.assertEqual(response.provider, "Anspire")
+        self.assertEqual(len(response.results), 0)
+        # 错误消息检查
+        self.assertTrue("超时" in response.error_message or "Timeout" in response.error_message)
+    
+    @patch('src.search_service.requests')
+    def test_search_network_error(self, mock_requests):
+        """测试网络错误处理"""
+        try:
+            import requests as real_requests
+            mock_requests.exceptions = real_requests.exceptions
+            conn_exc = mock_requests.exceptions.ConnectionError
+        except ImportError:
+            mock_requests.exceptions = MagicMock()
+            conn_exc = Exception
+
+        mock_requests.get = MagicMock(side_effect=conn_exc())
+        
+        response = self.provider.search("测试查询", max_results=3)
+        
+        self.assertFalse(response.success)
+        self.assertEqual(response.provider, "Anspire")
+        self.assertEqual(len(response.results), 0)
+        self.assertTrue("网络" in response.error_message or "Connection" in response.error_message)
+    
+    @patch('src.search_service.requests')
+    def test_search_empty_results(self, mock_requests):
+        """测试空结果处理"""
+        try:
+            import requests as real_requests
+            mock_requests.exceptions = real_requests.exceptions
+        except ImportError:
+            mock_requests.exceptions = MagicMock()
+        
+        fake_response = _FakeResponse(
+            status_code=200,
+            json_data={"code": 200, "msg": "success", "results": []}
+        )
+        
+        mock_requests.get = MagicMock(return_value=fake_response)
+        
+        response = self.provider.search("不存在的股票 XYZ", max_results=5)
+        
+        self.assertTrue(response.success)
+        self.assertEqual(response.provider, "Anspire")
+        self.assertEqual(len(response.results), 0)
+    
+    @patch('src.search_service.requests')
+    def test_search_content_truncation(self, mock_requests):
+        """测试长内容截断功能"""
+        try:
+            import requests as real_requests
+            mock_requests.exceptions = real_requests.exceptions
+        except ImportError:
+            mock_requests.exceptions = MagicMock()
+        
+        long_content = "这是一段非常长的内容，" * 100  # 超过 500 字符
+        
+        fake_response = _FakeResponse(
+            status_code=200,
+            json_data={
+                "code": 200,
+                "msg": "success",
+                "results": [{
+                    "title": "长内容测试",
+                    "url": "https://example.com/long",
+                    "content": long_content
+                }]
+            }
+        )
+        
+        mock_requests.get = MagicMock(return_value=fake_response)
+        
+        response = self.provider.search("测试", max_results=1)
+        
+        self.assertTrue(response.success)
+        self.assertEqual(len(response.results), 1)
+        # 验证内容被截断到 500 字符以内
+        if response.results[0].snippet:
+            self.assertLessEqual(len(response.results[0].snippet), 503)  # 500 + "..."
+            self.assertTrue(response.results[0].snippet.endswith("..."))
+    
+    @patch('src.search_service.requests')
+    def test_search_time_range(self, mock_requests):
+        """测试时间范围参数"""
+        try:
+            import requests as real_requests
+            mock_requests.exceptions = real_requests.exceptions
+        except ImportError:
+            mock_requests.exceptions = MagicMock()
+        
+        fake_response = _FakeResponse(status_code=200, json_data={"code": 200, "results": []})
+        mock_requests.get = MagicMock(return_value=fake_response)
+        
+        # 测试 7 天范围
+        self.provider.search("测试", max_results=3, days=7)
+        
+        # 验证时间参数
+        call_args = mock_requests.get.call_args
+        if call_args and len(call_args) > 1 and 'params' in call_args[1]:
+            params = call_args[1]["params"]
+                
+            # 验证时间参数存在 (具体字段名取决于实现)
+            # 这里假设使用了 FromTime/ToTime 或类似字段，若无则跳过具体字段检查
+            # self.assertIn("FromTime", params)
+            # self.assertIn("ToTime", params)
+
+
+class TestAnspireSearchService(unittest.TestCase):
+    """SearchService 中 Anspire 集成测试"""
+    
+    def setUp(self):
+        Config._Config__instance = None
+        reset_search_service()
+
+    def test_search_service_with_anspire(self):
+        """测试 SearchService 正确初始化 Anspire Provider"""
+        service = SearchService(
+            anspire_keys=["test_key"],
+            bocha_keys=[],
+            tavily_keys=[],
+            searxng_public_instances_enabled=False,
+            news_max_age_days=3,
+            news_strategy_profile="short"
+        )
+        
+        self.assertTrue(hasattr(service, '_providers'))
+        self.assertGreater(len(service._providers), 0)
+        
+        first_provider = service._providers[0]
+        self.assertIsInstance(first_provider, AnspireSearchProvider)
+        self.assertEqual(first_provider.name, "Anspire")
+    
+    def test_search_service_without_anspire(self):
+        """测试未配置 Anspire 时的行为"""
+        service = SearchService(
+            anspire_keys=[],
+            tavily_keys=["tavily_key"],
+            bocha_keys=[],
+            searxng_public_instances_enabled=False,
+            news_max_age_days=3,
+            news_strategy_profile="short"
+        )
+        
+        # 验证没有 Anspire Provider
+        anspire_providers = [p for p in service._providers if isinstance(p, AnspireSearchProvider)]
+        self.assertEqual(len(anspire_providers), 0)
+    
+    def test_search_service_priority(self):
+        """测试 Anspire 优先级"""
+        service = SearchService(
+            anspire_keys=["anspire_key"],
+            bocha_keys=["bocha_key"],
+            tavily_keys=["tavily_key"],
+            searxng_public_instances_enabled=False,
+            news_max_age_days=3,
+            news_strategy_profile="short"
+        )
+        
+        self.assertIsInstance(service._providers[0], AnspireSearchProvider)
+
+
+class TestAnspireIntegration(unittest.TestCase):
+    """Anspire 集成测试（需要真实 API Key）"""
+    
+    @classmethod
+    def setUpClass(cls):
+        """Check if API Key is configured."""
+        cls.api_keys = [k.strip() for k in os.getenv('ANSPIRE_API_KEYS', '').split(',') if k.strip()]
+        cls.has_api_key = len(cls.api_keys) > 0
+        
+        if cls.has_api_key:
+            reset_search_service()
+            cls.service = get_search_service()
+
+    @unittest.skipIf(
+        not os.environ.get("ANSPIRE_API_KEYS"),
+        "未设置 ANSPIRE_API_KEYS 环境变量，跳过集成测试"
+    )
+    @pytest.mark.network
+    def test_real_api_call_stock_news(self):
+        """真实 API 调用测试 - 股票新闻搜索"""
+        # 确保服务已重置
+        reset_search_service()
+        service = get_search_service()
+        
+        # 验证 Anspire 已配置
+        anspire_provider = None
+        for provider in service._providers:
+            if isinstance(provider, AnspireSearchProvider):
+                anspire_provider = provider
+                break
+        
+        if not anspire_provider:
+            self.skipTest("Anspire Provider 未初始化")
+        
+        # 测试 A 股搜索
+        response = service.search_stock_news("600519", "贵州茅台", max_results=3)
+        
+        print(f"\n=== Anspire 真实 API 测试结果 ===")
+        print(f"搜索状态：{'成功' if response.success else '失败'}")
+        print(f"搜索引擎：{response.provider}")
+        print(f"结果数量：{len(response.results)}")
+        print(f"耗时：{response.search_time:.2f}s")
+        
+        # 基本验证
+        self.assertTrue(response.success, f"搜索失败：{response.error_message}")
+        self.assertEqual(response.provider, "Anspire")
+        self.assertGreater(len(response.results), 0, "应至少返回一条结果")
+        
+        # 验证结果格式
+        for result in response.results:
+            self.assertIsNotNone(result.title)
+            self.assertIsNotNone(result.url)
+            # snippet 可能为空，视具体实现而定
+            # self.assertIsNotNone(result.snippet)
+    
+    @unittest.skipIf(
+        not os.environ.get("ANSPIRE_API_KEYS"),
+        "未设置 ANSPIRE_API_KEYS 环境变量，跳过集成测试"
+    )
+    @pytest.mark.network
+    def test_real_api_call_general_search(self):
+        """真实 API 调用测试 - 通用搜索"""
+        reset_search_service()
+        service = get_search_service()
+        
+        anspire_provider = None
+        for provider in service._providers:
+            if isinstance(provider, AnspireSearchProvider):
+                anspire_provider = provider
+                break
+        
+        if not anspire_provider:
+            self.skipTest("Anspire Provider 未初始化")
+        
+        # 测试通用搜索
+        response = anspire_provider.search("人工智能最新发展", max_results=5, days=7)
+        
+        print(f"\n=== Anspire 通用搜索结果 ===")
+        print(f"搜索状态：{'成功' if response.success else '失败'}")
+        print(f"结果数量：{len(response.results)}")
+        
+        self.assertTrue(response.success)
+        self.assertGreater(len(response.results), 0)
+
+
+def run_manual_test():
+    """手动测试函数（用于快速验证）"""
+    import logging
+    from src.config import get_config
+    
+    # 配置日志
+    logging.basicConfig(
+        level=logging.INFO,
+        format='%(asctime)s | %(levelname)-8s | %(message)s'
+    )
+    
+    print("=" * 60)
+    print("Anspire Search 快速测试")
+    print("=" * 60)
+    
+    # 检查配置
+    config = get_config()
+    if not config.anspire_api_keys:
+        print("\n❌ 未检测到 Anspire API Keys")
+        print("请设置环境变量：")
+        print("  Windows PowerShell: $env:ANSPIRE_API_KEYS=\"your_api_key\"")
+        print("  Linux/Mac: export ANSPIRE_API_KEYS=\"your_api_key\"")
+        return False
+    
+    print(f"\n✅ 已配置 {len(config.anspire_api_keys)} 个 Anspire API Key")
+    
+    # 创建服务
+    service = SearchService(
+        anspire_keys=config.anspire_api_keys,
+        bocha_keys=config.bocha_api_keys,
+        tavily_keys=config.tavily_keys,
+        searxng_public_instances_enabled=False,
+        news_max_age_days=3,
+        news_strategy_profile="short"
+    )
+    
+    # 验证 Provider
+    anspire_provider = service._providers[0] if service._providers else None
+    if not anspire_provider or not isinstance(anspire_provider, AnspireSearchProvider):
+        print("\n❌ Anspire Provider 未正确初始化")
+        return False
+    
+    print(f"✅ Anspire Provider 初始化成功")
+    print(f"   Provider 名称：{anspire_provider.name}")
+    if hasattr(anspire_provider, 'api_keys'):
+        print(f"   API Keys 数量：{len(anspire_provider.api_keys)}")
+    elif hasattr(anspire_provider, '_api_keys'):
+        print(f"   API Keys 数量：{len(anspire_provider._api_keys)}")
+    
+    # 执行测试搜索
+    print("\n" + "=" * 60)
+    print("执行测试搜索：贵州茅台 (600519)")
+    print("=" * 60)
+    
+    response = service.search_stock_news("600519", "贵州茅台", max_results=3)
+    
+    print(f"\n搜索结果:")
+    print(f"  状态：{'✅ 成功' if response.success else '❌ 失败'}")
+    print(f"  搜索引擎：{response.provider}")
+    print(f"  结果数量：{len(response.results)}")
+    print(f"  耗时：{response.search_time:.2f}s")
+    
+    if response.error_message:
+        print(f"  错误信息：{response.error_message}")
+    
+    if response.results:
+        print(f"\n前 {min(2, len(response.results))} 条结果预览:")
+        for i, result in enumerate(response.results[:2], 1):
+            print(f"\n  [{i}] {result.title}")
+            print(f"      来源：{result.source}")
+            print(f"      URL: {result.url}")
+            if result.snippet:
+                snippet_preview = result.snippet[:100] + "..." if len(result.snippet) > 100 else result.snippet
+                print(f"      摘要：{snippet_preview}")
+    
+    print("\n" + "=" * 60)
+    print("测试完成!")
+    print("=" * 60)
+    
+    return response.success
+
+
+if __name__ == "__main__":
+    # 如果设置了环境变量，运行完整测试
+    if os.environ.get("ANSPIRE_API_KEYS"):
+        print("检测到 ANSPIRE_API_KEYS 环境变量，运行完整测试套件...")
+        unittest.main(verbosity=2)
+    else:
+        # 否则只运行单元测试，跳过集成测试
+        print("未设置 ANSPIRE_API_KEYS 环境变量，仅运行单元测试（跳过集成测试）...")
+        print("如需运行完整测试，请设置环境变量:")
+        print("  Windows PowerShell: $env:ANSPIRE_API_KEYS=\"your_api_key\"")
+        print("  Linux/Mac: export ANSPIRE_API_KEYS=\"your_api_key\"")
+        print()
+        
+        # 运行单元测试
+        suite = unittest.TestLoader().loadTestsFromTestCase(TestAnspireConfigLoading)
+        suite.addTests(unittest.TestLoader().loadTestsFromTestCase(TestAnspireSearchProvider))
+        suite.addTests(unittest.TestLoader().loadTestsFromTestCase(TestAnspireSearchService))
+        runner = unittest.TextTestRunner(verbosity=2)
+        runner.run(suite)
+        
+        # 提供手动测试选项
+        print("\n" + "=" * 60)
+        choice = input("是否运行手动测试（需要有效的 API Key）? (y/n): ").strip().lower()
+        if choice == 'y':
+            run_manual_test()

tests/test_search_service_concurrency.py

@@ -237,6 +237,7 @@ class SearchServiceConcurrencyTestCase(unittest.TestCase):
             searxng_public_instances_enabled=False,
             news_max_age_days=3,
             news_strategy_profile="short",
+            anspire_api_keys=[],
         )
 
         created = []