Spaces:
Paused
Paused
File size: 5,330 Bytes
fd21f34 |
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 |
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
测试 LongCat API 连接性
"""
import asyncio
import httpx
import json
# LongCat API 端点
LONGCAT_API_ENDPOINT = "https://longcat.chat/api/v1/chat-completion-oversea"
async def test_longcat_api():
"""测试 LongCat API 连接"""
print(f"🧪 测试 LongCat API 连接...")
print(f"📡 API 端点: {LONGCAT_API_ENDPOINT}")
headers = {
'accept': 'text/event-stream,application/json',
'content-type': 'application/json',
'origin': 'https://longcat.chat',
'referer': 'https://longcat.chat/t',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36'
}
payload = {
"stream": True,
"temperature": 0.7,
"content": "Hello",
"messages": [
{
"role": "user",
"content": "Hello"
}
]
}
print(f"📤 发送请求...")
print(f"📋 Headers: {json.dumps(headers, indent=2)}")
print(f"📋 Payload: {json.dumps(payload, indent=2)}")
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
LONGCAT_API_ENDPOINT,
headers=headers,
json=payload
)
print(f"📡 响应状态码: {response.status_code}")
print(f"📋 响应头: {dict(response.headers)}")
if not response.is_success:
error_text = await response.atext()
print(f"❌ API 错误: {error_text}")
return False
print(f"✅ 连接成功，开始读取流数据...")
line_count = 0
async for line in response.aiter_lines():
line_count += 1
line = line.strip()
print(f"📥 第 {line_count} 行: {line}")
if line_count > 10: # 只读取前10行
print(f"⏹️ 停止读取（已读取 {line_count} 行）")
break
if line.startswith('data:'):
data_str = line[5:].strip()
if data_str == '[DONE]':
print(f"🏁 收到结束标记")
break
try:
data = json.loads(data_str)
print(f"📦 解析成功: {json.dumps(data, ensure_ascii=False, indent=2)}")
except json.JSONDecodeError as e:
print(f"❌ JSON 解析失败: {e}")
print(f"✅ 测试完成，共读取 {line_count} 行")
return True
except httpx.TimeoutException:
print(f"❌ 请求超时")
return False
except httpx.ConnectError as e:
print(f"❌ 连接错误: {e}")
return False
except Exception as e:
print(f"❌ 未知错误: {e}")
import traceback
print(f"❌ 错误堆栈: {traceback.format_exc()}")
return False
async def test_simple_request():
"""测试简单的非流式请求"""
print(f"\n🧪 测试简单的非流式请求...")
headers = {
'accept': 'application/json',
'content-type': 'application/json',
'origin': 'https://longcat.chat',
'referer': 'https://longcat.chat/t',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
payload = {
"stream": False,
"temperature": 0.7,
"content": "Hello",
"messages": [
{
"role": "user",
"content": "Hello"
}
]
}
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
LONGCAT_API_ENDPOINT,
headers=headers,
json=payload
)
print(f"📡 响应状态码: {response.status_code}")
if response.is_success:
content = await response.atext()
print(f"✅ 响应内容: {content[:500]}...")
return True
else:
error_text = await response.atext()
print(f"❌ 错误响应: {error_text}")
return False
except Exception as e:
print(f"❌ 请求失败: {e}")
return False
async def main():
"""运行所有测试"""
print("🚀 开始 LongCat API 连接测试...\n")
# 测试流式请求
stream_result = await test_longcat_api()
# 测试非流式请求
simple_result = await test_simple_request()
print(f"\n📊 测试结果:")
print(f" 流式请求: {'✅ 成功' if stream_result else '❌ 失败'}")
print(f" 非流式请求: {'✅ 成功' if simple_result else '❌ 失败'}")
if stream_result and simple_result:
print(f"🎉 所有测试通过！")
else:
print(f"⚠️ 部分测试失败，请检查网络连接和 API 端点")
if __name__ == "__main__":
asyncio.run(main())
|