小溪
2ee12bae8e
- 新增 user_export_skill.md 完整导出技能说明 - 支持导出指定账户ID或角色ID的完整学习行为数据 - 包含6个sheet：音频数据、互动组件、课程巩固、单元挑战、单元总结、汇总统计 - 已成功验证导出两个用户数据，功能正常可用
145 lines
4.8 KiB
Python
145 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
||
"""单独测试角色12698的导出，查看具体报错"""
|
||
|
||
import os
|
||
import json
|
||
import sys
|
||
import datetime
|
||
from typing import Any, Dict, List
|
||
|
||
# 加载环境变量
|
||
def load_env():
|
||
env_path = os.path.join(os.getcwd(), ".env")
|
||
if os.path.exists(env_path):
|
||
with open(env_path, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
k, v = line.split("=", 1)
|
||
os.environ[k.strip()] = v.strip().strip('"').strip("'")
|
||
|
||
load_env()
|
||
|
||
import psycopg2
|
||
from psycopg2.extras import RealDictCursor
|
||
import pymysql
|
||
import requests
|
||
from requests.auth import HTTPBasicAuth
|
||
import warnings
|
||
warnings.filterwarnings('ignore')
|
||
|
||
def test_role_12698():
|
||
print("="*60)
|
||
print("单独测试角色ID=12698的查询")
|
||
print("="*60)
|
||
|
||
# 连接PG
|
||
try:
|
||
conn = psycopg2.connect(
|
||
host=os.getenv("PG_DB_HOST"),
|
||
port=int(os.getenv("PG_DB_PORT")),
|
||
user=os.getenv("PG_DB_USER"),
|
||
password=os.getenv("PG_DB_PASSWORD"),
|
||
dbname=os.getenv("PG_DB_DATABASE"),
|
||
connect_timeout=10
|
||
)
|
||
print("✅ PG连接成功")
|
||
except Exception as e:
|
||
print(f"❌ PG连接失败: {e}")
|
||
return
|
||
|
||
user_id = "12698"
|
||
|
||
# 测试第一个查询：user_component_play_record_0
|
||
print(f"\n测试查询表 user_component_play_record_0，user_id={user_id}")
|
||
try:
|
||
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
||
sql = f"""
|
||
SELECT user_id, component_unique_code, session_id, c_type, c_id,
|
||
play_result, user_behavior_info, updated_at
|
||
FROM user_component_play_record_0
|
||
WHERE user_id = %s
|
||
ORDER BY updated_at DESC
|
||
"""
|
||
cur.execute(sql, (user_id,))
|
||
rows = cur.fetchall()
|
||
print(f"✅ 查询成功，返回{len(rows)}条记录")
|
||
except Exception as e:
|
||
print(f"❌ 查询失败: {e}")
|
||
print(f"错误类型: {type(e).__name__}")
|
||
|
||
# 回滚事务
|
||
print("\n尝试回滚事务...")
|
||
try:
|
||
conn.rollback()
|
||
print("✅ 事务回滚成功")
|
||
except Exception as e2:
|
||
print(f"❌ 回滚失败: {e2}")
|
||
|
||
# 测试查询课程巩固记录表
|
||
print(f"\n测试查询表 user_unit_review_question_result，user_id={user_id}")
|
||
try:
|
||
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
||
sql = f"""
|
||
SELECT user_id, story_id, chapter_id, question_list, updated_at
|
||
FROM user_unit_review_question_result
|
||
WHERE user_id = %s
|
||
ORDER BY updated_at DESC
|
||
"""
|
||
cur.execute(sql, (user_id,))
|
||
rows = cur.fetchall()
|
||
print(f"✅ 查询成功，返回{len(rows)}条记录")
|
||
except Exception as e:
|
||
print(f"❌ 查询失败: {e}")
|
||
print(f"错误类型: {type(e).__name__}")
|
||
|
||
# 回滚事务
|
||
print("\n尝试回滚事务...")
|
||
try:
|
||
conn.rollback()
|
||
print("✅ 事务回滚成功")
|
||
except Exception as e2:
|
||
print(f"❌ 回滚失败: {e2}")
|
||
|
||
# 测试查询单元挑战记录表
|
||
print(f"\n测试查询表 user_unit_challenge_question_result，user_id={user_id}")
|
||
try:
|
||
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
||
sql = f"""
|
||
SELECT user_id, story_id, category, score_text, question_list, updated_at
|
||
FROM user_unit_challenge_question_result
|
||
WHERE user_id = %s
|
||
ORDER BY updated_at DESC
|
||
"""
|
||
cur.execute(sql, (user_id,))
|
||
rows = cur.fetchall()
|
||
print(f"✅ 查询成功，返回{len(rows)}条记录")
|
||
except Exception as e:
|
||
print(f"❌ 查询失败: {e}")
|
||
print(f"错误类型: {type(e).__name__}")
|
||
|
||
# 测试查询单元总结记录表
|
||
print(f"\n测试查询表 user_unit_summary_record，user_id={user_id}")
|
||
try:
|
||
with conn.cursor(cursor_factory=RealDictCursor) as cur:
|
||
sql = f"""
|
||
SELECT id, user_id, unit_id, updated_at, km_id, km_type, play_time_seconds
|
||
FROM user_unit_summary_record
|
||
WHERE user_id = %s
|
||
ORDER BY updated_at DESC
|
||
"""
|
||
cur.execute(sql, (user_id,))
|
||
rows = cur.fetchall()
|
||
print(f"✅ 查询成功，返回{len(rows)}条记录")
|
||
except Exception as e:
|
||
print(f"❌ 查询失败: {e}")
|
||
print(f"错误类型: {type(e).__name__}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
conn.close()
|
||
|
||
if __name__ == "__main__":
|
||
test_role_12698()
|