src/api/search_api.py

from typing import List, Dict
import requests
from abc import ABC, abstractmethod
import logging
import json

logging.basicConfig(level=logging.DEBUG)

class SearchEngine(ABC):
    def __init__(self):
        # 广告域名黑名单
        self.ad_domains = {
            'ads.google.com',
            'doubleclick.net',
            'affiliate.',
            '.ads.',
            'promotion.',
            'sponsored.',
            'partner.',
            'tracking.',
            '.shop.',
            'taobao.com',
            'tmall.com',
            'jd.com',
            'mafengwo.cn',  # 蚂蜂窝
            'ctrip.com',  # 携程
            'tour.aoyou.com',  # 同程
            'wannar.com'  # 玩哪儿
        }

    def is_ad_url(self, url: str) -> bool:
        """检查URL是否为广告链接"""
        url_lower = url.lower()
        return any(ad_domain in url_lower for ad_domain in self.ad_domains)

    def enhance_query(self, query: str) -> str:
        """增强查询词，添加香港旅游关键词"""
        if "Hong Kong" not in query:
            query = f"Hong Kong Tourism{query}"
        return query

    @abstractmethod
    def search(self, query: str) -> List[Dict]:
        pass

class GoogleSearch(SearchEngine):
    def __init__(self, api_key: str, cx: str, proxies: Dict[str, str] = None):
        super().__init__()
        self.api_key = api_key
        self.cx = cx
        self.base_url = "https://www.googleapis.com/customsearch/v1"
        self.proxies = proxies or {}

    def filter_results(self, results: List[Dict]) -> List[Dict]:
        """过滤搜索结果"""
        filtered = []
        for result in results:
            url = result['url'].lower()
            # 只过滤广告域名
            if not self.is_ad_url(url):
                filtered.append(result)
        return filtered

    def search(self, query: str) -> List[Dict]:
        # 增强查询词
        enhanced_query = self.enhance_query(query)
        
        params = {
            'key': self.api_key,
            'cx': self.cx,
            'q': enhanced_query
        }
        response = requests.get(self.base_url, params=params)
        if response.status_code == 200:
            results = response.json()
            return [{
                'title': item['title'],
                'snippet': item['snippet'],
                'url': item['link']
            } for item in results.get('items', [])]
        return []

class BochaSearch(SearchEngine):
    def __init__(self, api_key: str, base_url: str, proxies: Dict[str, str] = None):
        super().__init__()
        self.api_key = api_key
        self.base_url = base_url.rstrip('/')  # 移除末尾可能的斜杠
        self.proxies = proxies or {}

    def search(self, query: str) -> List[Dict]:
        try:
            # 增强查询词
            enhanced_query = self.enhance_query(query)
            
            headers = {
                'Authorization': f'Bearer {self.api_key}',
                'Content-Type': 'application/json',
                'Connection': 'keep-alive',
                'Accept': '*/*'
            }
            
            payload = {
                'query': enhanced_query,
                'stream': False  # 使用非流式返回
            }
            
            # 使用正确的端点
            endpoint = f"{self.base_url}/v1/ai-search"
            
            logging.info(f"正在请求博查API...")
            logging.info(f"增强后的查询词: {enhanced_query}")
            
            response = requests.post(
                endpoint,
                headers=headers,
                json=payload,
                proxies=None
            )
            
            # 详细打印响应信息
            logging.info(f"API响应状态码: {response.status_code}")
            logging.info(f"API响应内容: {response.text[:500]}...")  # 只打印前500个字符
            
            if response.status_code != 200:
                logging.error(f"API请求失败，状态码: {response.status_code}")
                logging.error(f"错误响应: {response.text}")
                return []
                
            response_json = response.json()
            if response_json.get('code') == 200 and 'messages' in response_json:
                messages = response_json['messages']
                if messages and isinstance(messages, list):
                    for msg in messages:
                        if msg.get('type') == 'source' and msg.get('content_type') == 'webpage':
                            try:
                                content = json.loads(msg['content'])
                                if 'value' in content:
                                    return content['value']
                            except json.JSONDecodeError:
                                logging.error(f"无法解析消息内容: {msg['content']}")
                                continue
            
            logging.error(f"API返回数据结构异常: {response_json}")
            return []
        except Exception as e:
            logging.error(f"处理API响应时出错: {str(e)}")
            return []

    def search_images(self, query: str, count: int = 3) -> List[Dict]:
        """搜索相关图片"""
        try:
            headers = {
                'Authorization': f'Bearer {self.api_key}',
                'Content-Type': 'application/json'
            }
            
            # 增强查询词
            enhanced_query = self.enhance_query(query)
            logging.info(f"增强后的图片搜索查询: {enhanced_query}")
            
            payload = {
                'query': enhanced_query,
                'freshness': 'oneYear',
                'count': 10,  # 搜索更多图片以确保有足够的有效结果
                'filter': 'images'
            }
            
            endpoint = f"{self.base_url}/v1/web-search"
            
            response = requests.post(
                endpoint,
                headers=headers,
                json=payload,
                timeout=10
            )
            
            if response.status_code == 200:
                try:
                    data = response.json()
                    logging.info(f"API返回数据结构: {data.keys()}")
                    
                    if data.get('code') == 200 and 'data' in data:
                        data_content = data['data']
                        logging.info(f"data字段内容: {data_content.keys()}")
                        
                        images = []
                        if 'images' in data_content:
                            image_items = data_content['images'].get('value', [])
                            logging.info(f"找到 {len(image_items)} 张图片")
                            
                            for item in image_items:
                                # 简化过滤条件，只检查基本必要条件
                                if (item.get('contentUrl') and 
                                    item.get('width', 0) >= 300 and 
                                    item.get('height', 0) >= 300):
                                    
                                    image_info = {
                                        'url': item['contentUrl'],
                                        'width': item['width'],
                                        'height': item['height']
                                    }
                                    images.append(image_info)
                                    if len(images) >= count:
                                        break
                
                    logging.info(f"最终返回 {len(images)} 张图片")
                    return images[:count]
                    
                except json.JSONDecodeError as e:
                    logging.error(f"JSON解析错误: {str(e)}")
                    return []
                except Exception as e:
                    logging.error(f"处理图片数据时出错: {str(e)}")
                    return []
            
            logging.error(f"API请求失败，状态码: {response.status_code}")
            return []
            
        except Exception as e:
            logging.error(f"图片搜索出错: {str(e)}")
            return []

"""
class BingSearch(SearchEngine):
    def __init__(self, api_key: str):
        super().__init__()
        self.api_key = api_key
        self.base_url = "https://api.bing.microsoft.com/v7.0/search"

    def search(self, query: str) -> List[Dict]:
        # 只添加香港旅游关键词
        enhanced_query = f"香港旅游 {query}"
        
        headers = {'Ocp-Apim-Subscription-Key': self.api_key}
        params = {
            'q': enhanced_query
        }
        
        response = requests.get(
            self.base_url, 
            headers=headers, 
            params=params
        )
        results = response.json()
        
        filtered_results = []
        for item in results.get('webPages', {}).get('value', []):
            if not self.is_ad_url(item['url']):
                filtered_results.append({
                    'title': item['name'],
                    'snippet': item['snippet'],
                    'url': item['url']
                })
        
        return filtered_results

    def is_trusted_domain(self, url: str) -> bool:
        ""检查是否为可信域名""
        return any(
            trusted_domain in url.lower() 
            for trusted_domain in self.config['search_settings']['trusted_domains']
        )
"""