76 lines
2.2 KiB
Python
76 lines
2.2 KiB
Python
import hashlib
|
|
import httpx
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
class SearchAPIError(Exception):
|
|
"""搜索API调用异常"""
|
|
|
|
def __init__(self, status_code: int, message: str):
|
|
self.status_code = status_code
|
|
self.message = message
|
|
super().__init__(f"Search API error {status_code}: {message}")
|
|
|
|
|
|
class SearchAPIClient:
|
|
"""搜索API客户端 - 使用SerpAPI"""
|
|
|
|
def __init__(self, base_url: str, api_key: str, timeout: float = 15.0):
|
|
self.base_url = base_url.rstrip('/') if base_url else ''
|
|
self.api_key = api_key
|
|
self.timeout = timeout
|
|
|
|
async def search(
|
|
self,
|
|
query: str,
|
|
*,
|
|
engine: str = 'google',
|
|
lang: str = 'zh-cn',
|
|
country: str = 'cn',
|
|
num_results: int = 5,
|
|
) -> Dict[str, Any]:
|
|
"""执行搜索请求
|
|
|
|
Args:
|
|
query: 搜索查询词
|
|
engine: 搜索引擎类型 (google, bing等)
|
|
lang: 搜索语言
|
|
country: 搜索国家/地区
|
|
num_results: 返回结果数量
|
|
|
|
Returns:
|
|
搜索结果JSON
|
|
|
|
Raises:
|
|
SearchAPIError: API调用失败时抛出
|
|
"""
|
|
if not self.api_key:
|
|
raise SearchAPIError(401, '搜索服务未配置 API 密钥')
|
|
|
|
endpoint = f"{self.base_url}/search" if self.base_url else "https://serpapi.com/search"
|
|
params = {
|
|
'q': query,
|
|
'api_key': self.api_key,
|
|
'engine': engine,
|
|
'hl': lang,
|
|
'gl': country,
|
|
'num': num_results,
|
|
}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(endpoint, params=params)
|
|
except httpx.RequestError as exc:
|
|
raise SearchAPIError(0, f'搜索请求失败: {str(exc)}') from exc
|
|
|
|
if response.status_code != 200:
|
|
detail = ''
|
|
try:
|
|
payload = response.json()
|
|
detail = payload.get('error', '')
|
|
except ValueError:
|
|
detail = response.text
|
|
raise SearchAPIError(response.status_code, detail or '搜索服务调用失败')
|
|
|
|
return response.json()
|