94 lines
3.0 KiB
Python
94 lines
3.0 KiB
Python
"""
|
||
测试 API 响应，检查是否有 bytes 序列化问题
|
||
"""
|
||
import pymysql
|
||
import os
|
||
import json
|
||
from dotenv import load_dotenv
|
||
|
||
load_dotenv()
|
||
|
||
# 数据库连接配置
|
||
DB_CONFIG = {
|
||
'host': os.getenv('DB_HOST', '152.136.177.240'),
|
||
'port': int(os.getenv('DB_PORT', 5012)),
|
||
'user': os.getenv('DB_USER', 'finyx'),
|
||
'password': os.getenv('DB_PASSWORD', '6QsGK6MpePZDE57Z'),
|
||
'database': os.getenv('DB_NAME', 'finyx'),
|
||
'charset': 'utf8mb4'
|
||
}
|
||
|
||
def clean_query_result(data):
|
||
"""
|
||
清理查询结果，将 bytes 类型转换为字符串
|
||
用于处理数据库查询结果中的 BLOB 等字段
|
||
"""
|
||
if isinstance(data, bytes):
|
||
try:
|
||
return data.decode('utf-8')
|
||
except UnicodeDecodeError:
|
||
return data.decode('utf-8', errors='ignore')
|
||
elif isinstance(data, dict):
|
||
return {key: clean_query_result(value) for key, value in data.items()}
|
||
elif isinstance(data, list):
|
||
return [clean_query_result(item) for item in data]
|
||
elif isinstance(data, (int, float, str, bool, type(None))):
|
||
return data
|
||
else:
|
||
# 对于其他类型（如 Decimal, datetime），转换为字符串
|
||
try:
|
||
return str(data)
|
||
except:
|
||
return data
|
||
|
||
def test_field_query():
|
||
"""测试字段查询"""
|
||
conn = pymysql.connect(**DB_CONFIG)
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
|
||
try:
|
||
tenant_id = 615873064429507639
|
||
|
||
print("测试查询字段...")
|
||
cursor.execute("""
|
||
SELECT id, name, filed_code, field_type, state
|
||
FROM f_polic_field
|
||
WHERE tenant_id = %s
|
||
ORDER BY field_type, name
|
||
LIMIT 5
|
||
""", (tenant_id,))
|
||
|
||
fields = cursor.fetchall()
|
||
print(f"查询到 {len(fields)} 条记录")
|
||
|
||
# 检查数据类型
|
||
if fields:
|
||
print("\n第一条记录的原始数据:")
|
||
first_field = fields[0]
|
||
for key, value in first_field.items():
|
||
print(f" {key}: {type(value).__name__} = {repr(value)}")
|
||
|
||
# 清理数据
|
||
cleaned_fields = [clean_query_result(field) for field in fields]
|
||
print("\n清理后的第一条记录:")
|
||
first_cleaned = cleaned_fields[0]
|
||
for key, value in first_cleaned.items():
|
||
print(f" {key}: {type(value).__name__} = {repr(value)}")
|
||
|
||
# 测试 JSON 序列化
|
||
print("\n测试 JSON 序列化...")
|
||
try:
|
||
json_str = json.dumps(cleaned_fields, ensure_ascii=False, indent=2, default=str)
|
||
print("✓ JSON 序列化成功")
|
||
print(f"JSON 长度: {len(json_str)} 字符")
|
||
except Exception as e:
|
||
print(f"✗ JSON 序列化失败: {e}")
|
||
print(f"错误类型: {type(e).__name__}")
|
||
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
if __name__ == '__main__':
|
||
test_field_query()
|