109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
"""
|
|
简单的搜索API测试脚本 - 直接测试HTTP请求
|
|
"""
|
|
import asyncio
|
|
import os
|
|
from dotenv import load_dotenv
|
|
|
|
# 加载环境变量
|
|
load_dotenv('.env.dev')
|
|
|
|
try:
|
|
import httpx
|
|
except ImportError:
|
|
print("错误: 缺少 httpx 模块")
|
|
print("请运行: pip install httpx")
|
|
exit(1)
|
|
|
|
|
|
async def test_search_api():
|
|
"""直接测试SerpAPI"""
|
|
|
|
api_key = os.getenv('SEARCH_API_KEY', '')
|
|
|
|
print("=" * 60)
|
|
print("搜索API测试")
|
|
print("=" * 60)
|
|
print(f"API Key: {'已配置' if api_key else '未配置'}")
|
|
|
|
if not api_key:
|
|
print("\n错误: SEARCH_API_KEY 未配置")
|
|
print("请在 .env.dev 中设置 SEARCH_API_KEY")
|
|
return
|
|
|
|
# 测试查询
|
|
query = "北京天气"
|
|
|
|
print(f"\n测试查询: {query}")
|
|
print("-" * 60)
|
|
|
|
# 构建请求
|
|
url = "https://serpapi.com/search"
|
|
params = {
|
|
'q': query,
|
|
'api_key': api_key,
|
|
'engine': 'google',
|
|
'hl': 'zh-cn',
|
|
'gl': 'cn',
|
|
'num': 5,
|
|
}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15.0) as client:
|
|
print("发送请求...")
|
|
response = await client.get(url, params=params)
|
|
|
|
print(f"状态码: {response.status_code}")
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
|
|
print("\n✓ 搜索成功!")
|
|
|
|
# 显示答案框
|
|
answer_box = data.get('answer_box', {})
|
|
if answer_box:
|
|
answer = answer_box.get('answer') or answer_box.get('snippet', '')
|
|
if answer:
|
|
print(f"\n【快速答案】")
|
|
print(answer)
|
|
|
|
# 显示搜索结果
|
|
organic_results = data.get('organic_results', [])
|
|
if organic_results:
|
|
print(f"\n【搜索结果】({len(organic_results)}条)")
|
|
for idx, item in enumerate(organic_results[:3], 1):
|
|
print(f"\n{idx}. {item.get('title', 'N/A')}")
|
|
print(f" {item.get('link', 'N/A')}")
|
|
snippet = item.get('snippet', '')
|
|
if snippet:
|
|
print(f" {snippet[:100]}...")
|
|
|
|
# 显示API使用情况
|
|
search_metadata = data.get('search_metadata', {})
|
|
if search_metadata:
|
|
print(f"\n【API信息】")
|
|
print(f"总耗时: {search_metadata.get('total_time_taken', 'N/A')}秒")
|
|
print(f"请求ID: {search_metadata.get('id', 'N/A')}")
|
|
|
|
else:
|
|
print(f"\n✗ 请求失败")
|
|
try:
|
|
error_data = response.json()
|
|
print(f"错误: {error_data.get('error', response.text)}")
|
|
except:
|
|
print(f"错误: {response.text}")
|
|
|
|
except httpx.RequestError as e:
|
|
print(f"\n✗ 网络错误: {e}")
|
|
except Exception as e:
|
|
print(f"\n✗ 未知错误: {type(e).__name__}: {e}")
|
|
|
|
print("\n" + "=" * 60)
|
|
print("测试完成")
|
|
print("=" * 60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(test_search_api())
|