重写web_search: Bing API优先+抓取回退+诚实失败

- Bing Web Search API支持（设BING_API_KEY环境变量即可）
- 抓取回退但诚实面对反爬限制
- 搜索为空时采集器正常运行（LLM直接生成选题）
- 降低搜索依赖为可选增强

scripts/web_search.py

@@ -1,152 +1,106 @@
 #!/usr/bin/env python3
-"""实时搜索模块：使用 Bing 搜索获取真实结果，为研究提供数据支撑"""
+"""
+网络搜索模块
 
-import sys, json, logging, re
+有两种模式：
-from pathlib import Path
+1. Bing Web Search API（优先）：需设置环境变量 BING_API_KEY
-from typing import List, Dict
+2. 网页搜索回退：从 cn.bing.com 抓取，但服务器环境常反爬拦截
-from datetime import datetime
+
+当搜索不可用时，返回空列表。采集器已处理此情况——LLM 直接生成选题。
+"""
+import logging, os, re
+from typing import List, Dict, Optional
+from urllib.parse import quote_plus
+import requests
 
-PROJECT_ROOT = Path(__file__).parent.parent
-DATA_DIR = PROJECT_ROOT / "automation" / "data"
 logger = logging.getLogger(__name__)
 
-CACHE_FILE = DATA_DIR / "search_cache.json"
+UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
-CACHE_TTL = 3600 * 6
-_skip_domains = ['baike.baidu.com', 'zdic.net', 'hanyu.baidu.com', 'dict.cn']
 
-_cache = None
+BING_API_KEY = os.getenv("BING_API_KEY", "")
 
-def _load_cache() -> Dict:
+
-    global _cache
+def search_api(query: str, max_results: int = 5) -> List[Dict]:
-    if _cache is not None:
+    """Bing Web Search API（需要 BING_API_KEY 环境变量）"""
-        return _cache
+    if not BING_API_KEY:
-    if CACHE_FILE.exists():
+        return []
     try:
-            _cache = json.loads(CACHE_FILE.read_text(encoding='utf-8'))
+        resp = requests.get(
-            return _cache
+            "https://api.bing.microsoft.com/v7.0/search",
-        except:
+            params={"q": query, "count": max_results, "mkt": "zh-CN"},
-            pass
+            headers={"Ocp-Apim-Subscription-Key": BING_API_KEY},
-    _cache = {}
+            timeout=10
-    return _cache
+        )
-
+        if resp.status_code != 200:
-def _save_cache():
+            logger.warning(f"Bing API 返回 {resp.status_code}")
-    global _cache
+            return []
-    CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
+        data = resp.json()
-    CACHE_FILE.write_text(json.dumps(_cache, ensure_ascii=False, indent=2), encoding='utf-8')
-
-def _is_relevant(result: Dict, query: str) -> bool:
-    url = result.get('url', '')
-    if any(d in url for d in _skip_domains):
-        return False
-    text = result.get('title', '') + result.get('snippet', '')
-    query_words = [w for w in re.split(r'[\s,，]+', query) if len(w) >= 2]
-    if not query_words:
-        return True
-    match_count = sum(1 for w in query_words if w in text)
-    return match_count >= max(1, int(len(query_words) * 0.3))
-
-def search(query: str, max_results: int = 5, use_cache: bool = True) -> List[Dict]:
-    cache_key = f"{query}_{max_results}"
-    if use_cache:
-        cache = _load_cache()
-        if cache_key in cache:
-            entry = cache[cache_key]
-            if datetime.now().timestamp() - entry.get('ts', 0) < CACHE_TTL:
-                logger.info(f"缓存命中: {query[:30]}")
-                return entry['results']
-    try:
-        from bs4 import BeautifulSoup
-        import requests
-        headers = {
-            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
-            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
-        }
-        url = 'https://cn.bing.com/search?q=' + requests.utils.quote(query) + '&setlang=zh-cn&cc=cn'
-        r = requests.get(url, headers=headers, timeout=15)
-        soup = BeautifulSoup(r.text, 'html.parser')
         results = []
-        for li in soup.find_all('li', class_='b_algo'):
+        for item in data.get("webPages", {}).get("value", [])[:max_results]:
-            if len(results) >= max_results:
+            results.append({
-                break
+                "title": item.get("name", "")[:120],
-            h2 = li.find('h2')
+                "url": item.get("url", ""),
-            if not h2:
+                "content": item.get("snippet", "")[:300],
-                continue
+                "source": "bing_api"
-            a = h2.find('a')
+            })
-            if not a or not a.get('href'):
+        logger.info(f"Bing API '{query[:20]}': {len(results)} 条")
-                continue
-            title = a.get_text(strip=True)
-            href = a['href']
-            p = li.find('p')
-            snippet = p.get_text(strip=True) if p else ''
-            if title and href and not href.startswith('javascript'):
-                r = {'title': title, 'url': href, 'snippet': snippet[:200]}
-                if _is_relevant(r, query):
-                    results.append(r)
-        if use_cache:
-            _load_cache()
-            _cache[cache_key] = {'ts': datetime.now().timestamp(), 'results': results}
-            _save_cache()
-        logger.info(f"搜索完成: {query[:40]} -> {len(results)}条")
         return results
     except Exception as e:
-        logger.warning(f"搜索失败: {query[:30]} -> {e}")
+        logger.warning(f"Bing API 失败: {e}")
         return []
 
-def _extract_keywords(text: str, max_words: int = 4) -> str:
+
-    sep = r'[：:，。！？\s()（）""\u201c\u201d#+\-*\d]'
+def search_scrape(query: str, max_results: int = 5) -> List[Dict]:
-    words = [w.strip() for w in re.split(sep, text) if len(w.strip()) >= 2]
+    """从 cn.bing.com 抓取搜索结果（服务器环境常遭受反爬，返回空为正常）"""
+    try:
+        resp = requests.get(
+            "https://cn.bing.com/search",
+            params={"q": query, "setlang": "zh-cn", "cc": "cn", "count": "15"},
+            headers={"User-Agent": UA, "Accept-Language": "zh-CN,zh;q=0.9"},
+            timeout=15
+        )
+        if resp.status_code != 200:
+            return []
+
+        html = resp.text
+        results = []
         seen = set()
-    result = []
-    for w in words:
-        if w not in seen:
-            seen.add(w)
-            result.append(w)
-    return ' '.join(result[:max_words])
 
-def _make_queries(topic: Dict) -> List[str]:
+        for m in re.finditer(
-    title = topic.get('title', '')
+            r'<li[^>]*class="[^"]*b_algo[^"]*"[^>]*>.*?<a[^>]*href="([^"]*)"[^>]*>(.*?)</a>',
-    field = topic.get('field', '')
+            html, re.DOTALL
-    core = topic.get('core_concept', '')
+        ):
-    pain = topic.get('audience_pain', '')
+            href, title_raw = m.group(1), m.group(2)
-    queries = []
+            title = re.sub(r'<[^>]+>', '', title_raw).strip()
-    kw = _extract_keywords(title)
+            if not title or len(title) < 8 or href in seen:
-    if kw:
+                continue
-        queries.append(kw)
+            if re.search(r'(bing\.com|microsoft\.com|beian\.miit|beian\.mps)', href, re.I):
-    core_kw = _extract_keywords(core, 3)
+                continue
-    if core_kw and core_kw != kw:
+            if re.search(r'(zdic|hanyu|hancibao|chengyu|dict\.|iciba|bishun)', href, re.I):
-        queries.append(core_kw)
+                continue
-    if field and kw:
+            if len(title) <= 5:
-        queries.append(field + ' ' + kw.split()[0] if kw.split() else field)
+                continue
-    if pain:
+            seen.add(href)
-        pain_kw = _extract_keywords(pain, 3)
+            results.append({"title": title[:120], "url": href, "content": "", "source": "bing"})
-        if pain_kw and pain_kw not in queries:
+            if len(results) >= max_results:
-            queries.append(pain_kw)
+                break
-    if kw:
-        queries.append(kw + ' 2025 2026')
-        queries.append(kw + ' 案例')
-    return queries
 
-def enrich_topic_research(topic: Dict) -> str:
+        return results
-    queries = _make_queries(topic)
+    except Exception as e:
-    seen_urls = set()
+        logger.warning(f"Bing 抓取失败: {e}")
-    all_results = []
+        return []
-    for q in queries:
-        for r in search(q, max_results=3):
-            if r['url'] not in seen_urls:
-                seen_urls.add(r['url'])
-                all_results.append(r)
-    if not all_results:
-        return ""
-    lines = ["\n## 实时搜索数据\n"]
-    for r in all_results[:6]:
-        lines.append("- **" + r['title'] + "**")
-        lines.append("  " + r['url'])
-        if r.get('snippet'):
-            lines.append("  > " + r['snippet'][:200])
-        lines.append("")
-    return '\n'.join(lines)
 
-if __name__ == "__main__":
+
-    logging.basicConfig(level=logging.INFO)
+def search(query: str, max_results: int = 5) -> List[Dict]:
-    results = search("Obsidian AI 第二大脑 搭建")
+    """统一搜索接口：API 优先 → 网页抓取回退"""
-    print(json.dumps(results, ensure_ascii=False, indent=2))
+    if BING_API_KEY:
+        results = search_api(query, max_results)
+        if results:
+            return results
+
+    results = search_scrape(query, max_results)
+    if results:
+        return results
+
+    logger.info(f"搜索 '{query[:20]}' 无结果（服务器环境限制，不影响采集器正常运行）")
+    return []