diff --git a/scripts/web_search.py b/scripts/web_search.py index fa440a8..67becd4 100644 --- a/scripts/web_search.py +++ b/scripts/web_search.py @@ -1,152 +1,106 @@ #!/usr/bin/env python3 -"""实时搜索模块：使用 Bing 搜索获取真实结果，为研究提供数据支撑""" +""" +网络搜索模块 -import sys, json, logging, re -from pathlib import Path -from typing import List, Dict -from datetime import datetime +有两种模式： +1. Bing Web Search API（优先）：需设置环境变量 BING_API_KEY +2. 网页搜索回退：从 cn.bing.com 抓取，但服务器环境常反爬拦截 + +当搜索不可用时，返回空列表。采集器已处理此情况——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" -CACHE_TTL = 3600 * 6 -_skip_domains = ['baike.baidu.com', 'zdic.net', 'hanyu.baidu.com', 'dict.cn'] +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 = None +BING_API_KEY = os.getenv("BING_API_KEY", "") -def _load_cache() -> Dict: - global _cache - if _cache is not None: - return _cache - if CACHE_FILE.exists(): - try: - _cache = json.loads(CACHE_FILE.read_text(encoding='utf-8')) - return _cache - except: - pass - _cache = {} - return _cache -def _save_cache(): - global _cache - CACHE_FILE.parent.mkdir(parents=True, exist_ok=True) - 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'] +def search_api(query: str, max_results: int = 5) -> List[Dict]: + """Bing Web Search API（需要 BING_API_KEY 环境变量）""" + if not BING_API_KEY: + return [] 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') + resp = requests.get( + "https://api.bing.microsoft.com/v7.0/search", + params={"q": query, "count": max_results, "mkt": "zh-CN"}, + headers={"Ocp-Apim-Subscription-Key": BING_API_KEY}, + timeout=10 + ) + if resp.status_code != 200: + logger.warning(f"Bing API 返回 {resp.status_code}") + return [] + data = resp.json() results = [] - for li in soup.find_all('li', class_='b_algo'): - if len(results) >= max_results: - break - h2 = li.find('h2') - if not h2: - continue - a = h2.find('a') - if not a or not a.get('href'): - 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)}条") + for item in data.get("webPages", {}).get("value", [])[:max_results]: + results.append({ + "title": item.get("name", "")[:120], + "url": item.get("url", ""), + "content": item.get("snippet", "")[:300], + "source": "bing_api" + }) + logger.info(f"Bing API '{query[:20]}': {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]' - words = [w.strip() for w in re.split(sep, text) if len(w.strip()) >= 2] - 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]: - title = topic.get('title', '') - field = topic.get('field', '') - core = topic.get('core_concept', '') - pain = topic.get('audience_pain', '') - queries = [] - kw = _extract_keywords(title) - if kw: - queries.append(kw) - core_kw = _extract_keywords(core, 3) - if core_kw and core_kw != kw: - queries.append(core_kw) - if field and kw: - queries.append(field + ' ' + kw.split()[0] if kw.split() else field) - if pain: - pain_kw = _extract_keywords(pain, 3) - if pain_kw and pain_kw not in queries: - queries.append(pain_kw) - if kw: - queries.append(kw + ' 2025 2026') - queries.append(kw + ' 案例') - return queries +def search_scrape(query: str, max_results: int = 5) -> List[Dict]: + """从 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 [] -def enrich_topic_research(topic: Dict) -> str: - queries = _make_queries(topic) - seen_urls = set() - all_results = [] - 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) + html = resp.text + results = [] + seen = set() -if __name__ == "__main__": - logging.basicConfig(level=logging.INFO) - results = search("Obsidian AI 第二大脑 搭建") - print(json.dumps(results, ensure_ascii=False, indent=2)) + for m in re.finditer( + r']*class="[^"]*b_algo[^"]*"[^>]*>.*?]*href="([^"]*)"[^>]*>(.*?)', + html, re.DOTALL + ): + href, title_raw = m.group(1), m.group(2) + title = re.sub(r'<[^>]+>', '', title_raw).strip() + if not title or len(title) < 8 or href in seen: + continue + if re.search(r'(bing\.com|microsoft\.com|beian\.miit|beian\.mps)', href, re.I): + continue + if re.search(r'(zdic|hanyu|hancibao|chengyu|dict\.|iciba|bishun)', href, re.I): + continue + if len(title) <= 5: + continue + seen.add(href) + results.append({"title": title[:120], "url": href, "content": "", "source": "bing"}) + if len(results) >= max_results: + break + + return results + except Exception as e: + logger.warning(f"Bing 抓取失败: {e}") + return [] + + +def search(query: str, max_results: int = 5) -> List[Dict]: + """统一搜索接口：API 优先 → 网页抓取回退""" + 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 []