161 lines
5.5 KiB
Python
161 lines
5.5 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
配置测试脚本
|
|
用于测试 Supabase 和 RageFlow 连接是否正常
|
|
"""
|
|
|
|
import asyncio
|
|
import aiohttp
|
|
import json
|
|
|
|
# ========== 配置区域 - 请修改以下参数 ==========
|
|
SUPABASE_URL = "YOUR_SUPABASE_URL"
|
|
SUPABASE_KEY = "YOUR_SUPABASE_KEY"
|
|
RAGEFLOW_API_KEY = "YOUR_RAGEFLOW_KEY"
|
|
# ===============================================
|
|
|
|
async def test_supabase_connection():
|
|
"""测试 Supabase 连接"""
|
|
print("🔄 测试 Supabase 连接...")
|
|
|
|
headers = {
|
|
"apikey": SUPABASE_KEY,
|
|
"Authorization": f"Bearer {SUPABASE_KEY}",
|
|
}
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
# 测试获取 ak_contents 表
|
|
url = f"{SUPABASE_URL}/rest/v1/ak_contents?select=id,title&limit=1"
|
|
async with session.get(url, headers=headers) as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
print(f"✅ Supabase 连接成功！找到 {len(data)} 条记录")
|
|
if data:
|
|
print(f" 示例内容 ID: {data[0]['id']}")
|
|
return True
|
|
else:
|
|
print(f"❌ Supabase 连接失败: HTTP {response.status}")
|
|
error_text = await response.text()
|
|
print(f" 错误信息: {error_text}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ Supabase 连接异常: {e}")
|
|
return False
|
|
|
|
async def test_rageflow_connection():
|
|
"""测试 RageFlow 连接"""
|
|
print("🔄 测试 RageFlow 连接...")
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {RAGEFLOW_API_KEY}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
test_payload = {
|
|
"model": "gpt-3.5-turbo",
|
|
"messages": [{"role": "user", "content": "Hello, this is a test."}],
|
|
"max_tokens": 50
|
|
}
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
"https://api.rageflow.ai/v1/chat/completions",
|
|
json=test_payload,
|
|
headers=headers
|
|
) as response:
|
|
if response.status == 200:
|
|
data = await response.json()
|
|
reply = data["choices"][0]["message"]["content"]
|
|
print(f"✅ RageFlow 连接成功！")
|
|
print(f" 测试回复: {reply[:50]}...")
|
|
return True
|
|
else:
|
|
print(f"❌ RageFlow 连接失败: HTTP {response.status}")
|
|
error_text = await response.text()
|
|
print(f" 错误信息: {error_text}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ RageFlow 连接异常: {e}")
|
|
return False
|
|
|
|
async def test_translation_tables():
|
|
"""测试翻译相关表结构"""
|
|
print("🔄 测试数据库表结构...")
|
|
|
|
headers = {
|
|
"apikey": SUPABASE_KEY,
|
|
"Authorization": f"Bearer {SUPABASE_KEY}",
|
|
}
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
# 测试 ak_content_translations 表
|
|
url = f"{SUPABASE_URL}/rest/v1/ak_content_translations?select=id&limit=1"
|
|
async with session.get(url, headers=headers) as response:
|
|
if response.status == 200:
|
|
print("✅ ak_content_translations 表存在")
|
|
return True
|
|
elif response.status == 404:
|
|
print("❌ ak_content_translations 表不存在")
|
|
print(" 请创建翻译表，参考 README 中的 SQL")
|
|
return False
|
|
else:
|
|
print(f"❓ 表结构检查异常: HTTP {response.status}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"❌ 表结构检查异常: {e}")
|
|
return False
|
|
|
|
async def run_full_test():
|
|
"""运行完整测试"""
|
|
print("=" * 60)
|
|
print("🧪 翻译服务配置测试")
|
|
print("=" * 60)
|
|
|
|
# 检查配置
|
|
if "YOUR_" in SUPABASE_URL or "YOUR_" in SUPABASE_KEY or "YOUR_" in RAGEFLOW_API_KEY:
|
|
print("❌ 请先配置正确的 API 参数！")
|
|
print("\n请修改脚本顶部的以下参数:")
|
|
print("- SUPABASE_URL: 您的 Supabase 项目 URL")
|
|
print("- SUPABASE_KEY: 您的 Supabase API Key")
|
|
print("- RAGEFLOW_API_KEY: 您的 RageFlow API Key")
|
|
return
|
|
|
|
print(f"📍 Supabase URL: {SUPABASE_URL}")
|
|
print(f"🔑 使用 API Key: {SUPABASE_KEY[:20]}...")
|
|
print(f"🤖 RageFlow Key: {RAGEFLOW_API_KEY[:20]}...")
|
|
print()
|
|
|
|
# 运行测试
|
|
supabase_ok = await test_supabase_connection()
|
|
print()
|
|
|
|
rageflow_ok = await test_rageflow_connection()
|
|
print()
|
|
|
|
tables_ok = await test_translation_tables()
|
|
print()
|
|
|
|
# 总结
|
|
print("=" * 60)
|
|
if supabase_ok and rageflow_ok and tables_ok:
|
|
print("🎉 所有测试通过！可以开始使用翻译服务了")
|
|
print("\n下一步:")
|
|
print("1. 运行 python quick_translate.py 开始翻译")
|
|
print("2. 或者使用 simple_translation_service.py 进行更复杂的操作")
|
|
else:
|
|
print("❌ 部分测试失败，请检查配置:")
|
|
if not supabase_ok:
|
|
print("- 检查 Supabase URL 和 API Key")
|
|
if not rageflow_ok:
|
|
print("- 检查 RageFlow API Key")
|
|
if not tables_ok:
|
|
print("- 检查数据库表结构")
|
|
print("=" * 60)
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(run_full_test())
|