yu-zhi-ran/scripts/web_search.py

#!/usr/bin/env python3
"""实时搜索模块：使用 Bing 搜索获取真实结果，为研究提供数据支撑"""

import sys, json, logging, re
from pathlib import Path
from typing import List, Dict
from datetime import datetime

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']

_cache = None

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']
    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'):
            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)}条")
        return results
    except Exception as e:
        logger.warning(f"搜索失败: {query[:30]} -> {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 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)

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    results = search("Obsidian AI 第二大脑 搭建")
    print(json.dumps(results, ensure_ascii=False, indent=2))