Blame view

data_crawling/test_api.py 5.65 KB
8f6f14da   tangwang   test data prepare:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
  #!/usr/bin/env python3
  # -*- coding: utf-8 -*-
  """
  API连接测试脚本
  用于测试万邦API是否配置正确并可以正常工作
  """
  
  import requests
  import json
  import sys
  
  def test_api(api_key: str, api_secret: str):
      """
      测试API连接
      
      Args:
          api_key: API Key
          api_secret: API Secret
      """
      print("=" * 70)
      print("万邦API连接测试")
      print("=" * 70)
      
      # 测试查询
      test_query = "Mobile Phone Holder"
      
      print(f"\n测试查询: {test_query}")
      print(f"API Key: {api_key[:10]}..." if len(api_key) > 10 else api_key)
      print(f"API Secret: {api_secret[:10]}..." if len(api_secret) > 10 else api_secret)
      
      # 构建请求
      url = "https://api-gw.onebound.cn/amazon/item_search"
      params = {
          'key': api_key,
          'secret': api_secret,
          'q': test_query,
          'cache': 'yes',
          'result_type': 'json',
          'lang': 'cn',
          'page_size': 10  # 只获取10个结果用于测试
      }
      
      print(f"\n请求URL: {url}")
      print("发送请求...")
      
      try:
          response = requests.get(url, params=params, timeout=30)
          response.raise_for_status()
          
          data = response.json()
          
          print("\n" + "=" * 70)
          print("响应结果")
          print("=" * 70)
          
          # 显示基本信息
          error_code = data.get('error_code', 'N/A')
          reason = data.get('reason', 'N/A')
          
          print(f"\n错误代码: {error_code}")
          print(f"返回信息: {reason}")
          
          if error_code == '0000':
              print("\n✓ API连接成功!")
              
              # 显示商品信息
              items_data = data.get('items', {})
              items = items_data.get('item', [])
              total_results = items_data.get('real_total_results', 0)
              
              print(f"\n查询词: {items_data.get('q', 'N/A')}")
              print(f"总结果数: {total_results}")
              print(f"当前页数量: {len(items)}")
              
              if items:
                  print(f"\n前3个商品示例:")
                  for i, item in enumerate(items[:3], 1):
                      print(f"\n  [{i}] {item.get('title', 'N/A')[:60]}...")
                      print(f"      价格: ${item.get('price', 'N/A')}")
                      print(f"      评分: {item.get('stars', 'N/A')} ({item.get('reviews', 'N/A')} 评论)")
              
              # 显示API配额信息
              api_info = data.get('api_info', 'N/A')
              print(f"\nAPI配额信息: {api_info}")
              
              # 保存完整响应
              test_file = "test_api_response.json"
              with open(test_file, 'w', encoding='utf-8') as f:
                  json.dump(data, f, ensure_ascii=False, indent=2)
              print(f"\n完整响应已保存至: {test_file}")
              
          else:
              print(f"\n✗ API返回错误")
              print(f"错误信息: {data.get('error', 'Unknown error')}")
              
              # 常见错误提示
              if 'key' in reason.lower() or 'secret' in reason.lower():
                  print("\n提示: 请检查API Key和Secret是否正确")
              elif 'quota' in reason.lower() or 'limit' in reason.lower():
                  print("\n提示: API配额可能已用完,请检查配额或等待重置")
          
          print("\n" + "=" * 70)
          
      except requests.exceptions.Timeout:
          print("\n✗ 请求超时")
          print("提示: 请检查网络连接")
      except requests.exceptions.ConnectionError:
          print("\n✗ 连接失败")
          print("提示: 请检查网络连接或API服务是否可用")
      except requests.exceptions.HTTPError as e:
          print(f"\n✗ HTTP错误: {e}")
      except json.JSONDecodeError:
          print("\n✗ JSON解析失败")
          print("响应内容:")
          print(response.text[:500])
      except Exception as e:
          print(f"\n✗ 未知错误: {str(e)}")
  
  
  def main():
      """主函数"""
      import argparse
      import os
      
      parser = argparse.ArgumentParser(description='测试万邦API连接')
      parser.add_argument('--key', type=str, help='API Key')
      parser.add_argument('--secret', type=str, help='API Secret')
      
      args = parser.parse_args()
      
      # 获取API密钥
      api_key = args.key
      api_secret = args.secret
      
      # 从配置文件读取
      if not api_key or not api_secret:
          try:
              import config
              api_key = api_key or getattr(config, 'API_KEY', None)
              api_secret = api_secret or getattr(config, 'API_SECRET', None)
          except ImportError:
              pass
      
      # 从环境变量读取
      if not api_key or not api_secret:
          api_key = api_key or os.getenv('ONEBOUND_API_KEY')
          api_secret = api_secret or os.getenv('ONEBOUND_API_SECRET')
      
      # 交互式输入
      if not api_key or not api_secret:
          print("未找到API密钥配置")
          print("\n请输入API密钥(或按Ctrl+C退出):")
          try:
              api_key = input("API Key: ").strip()
              api_secret = input("API Secret: ").strip()
          except KeyboardInterrupt:
              print("\n\n已取消")
              return
      
      # 验证
      if not api_key or not api_secret or \
         api_key == "your_api_key_here" or api_secret == "your_api_secret_here":
          print("\n错误: 无效的API密钥")
          print("\n使用方法:")
          print("  1. python test_api.py --key YOUR_KEY --secret YOUR_SECRET")
          print("  2. 配置 config.py 文件后直接运行")
          print("  3. 设置环境变量 ONEBOUND_API_KEY 和 ONEBOUND_API_SECRET")
          return
      
      # 执行测试
      test_api(api_key, api_secret)
  
  
  if __name__ == "__main__":
      main()