搜索缓存: 可追踪的search_cache.json, webfetch预填充26条

新机制:
- web_search.py按优先级: 缓存→Bing API→Bing抓取
- search_cache.json由opencode webfetch手动填充(不依赖搜索引擎)
- 搜索失效不影响采集器(LLM直接生成选题)
- .gitignore移除search_cache.json(应被版本追踪)
This commit is contained in:
Yuzhiran Dev
2026-05-21 08:29:04 +08:00
parent eb1c4205fa
commit 7d50878c7d
3 changed files with 212 additions and 10 deletions
+38 -9
View File
@@ -2,13 +2,13 @@
"""
网络搜索模块
有两种模式:
1. Bing Web Search API(优先):需设置环境变量 BING_API_KEY
2. 网页搜索回退:从 cn.bing.com 抓取,但服务器环境常反爬拦截
当搜索不可用时,返回空列表。采集器已处理此情况——LLM 直接生成选题。
种模式(优先级从高到低)
1. 本地缓存(opencode webfetch 预填充)
2. Bing Web Search API(设 BING_API_KEY
3. Bing 网页抓取(服务器环境常反爬拦截)
"""
import logging, os, re
import json, logging, os, re
from pathlib import Path
from typing import List, Dict, Optional
from urllib.parse import quote_plus
import requests
@@ -16,8 +16,8 @@ import requests
logger = logging.getLogger(__name__)
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
BING_API_KEY = os.getenv("BING_API_KEY", "")
SEARCH_CACHE_FILE = Path(__file__).parent.parent / "automation" / "data" / "search_cache.json"
def search_api(query: str, max_results: int = 5) -> List[Dict]:
@@ -91,8 +91,37 @@ def search_scrape(query: str, max_results: int = 5) -> List[Dict]:
return []
def search_from_cache(query: str, max_results: int = 5) -> List[Dict]:
"""从 opencode webfetch 预填充的缓存中读取"""
if not SEARCH_CACHE_FILE.exists():
return []
try:
cache = json.loads(SEARCH_CACHE_FILE.read_text(encoding="utf-8"))
results = cache.get(query, [])
return results[:max_results]
except Exception:
return []
def save_to_cache(query: str, results: List[Dict]):
"""保存搜索结果到缓存(供 opencode webfetch 填充时使用)"""
cache = {}
if SEARCH_CACHE_FILE.exists():
try:
cache = json.loads(SEARCH_CACHE_FILE.read_text(encoding="utf-8"))
except Exception:
pass
cache[query] = results
SEARCH_CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
SEARCH_CACHE_FILE.write_text(json.dumps(cache, ensure_ascii=False, indent=2), encoding="utf-8")
def search(query: str, max_results: int = 5) -> List[Dict]:
"""统一搜索接口:API 优先 → 网页抓取回退"""
"""统一搜索接口:缓存 → API → 网页抓取"""
results = search_from_cache(query, max_results)
if results:
return results
if BING_API_KEY:
results = search_api(query, max_results)
if results:
@@ -102,5 +131,5 @@ def search(query: str, max_results: int = 5) -> List[Dict]:
if results:
return results
logger.info(f"搜索 '{query[:20]}' 无结果(服务器环境限制,不影响采集器正常运行)")
logger.info(f"搜索 '{query[:20]}' 无结果")
return []