191 lines
6.3 KiB
Python
191 lines
6.3 KiB
Python
#!/usr/bin/env python3
|
||
import sys
|
||
import requests
|
||
import json
|
||
import os
|
||
from datetime import datetime
|
||
|
||
# 配置项
|
||
API_URL = "https://api.valavala.com/v2/user/unit/learn/info"
|
||
OPERATOR = "Kingson"
|
||
# 存储目录配置
|
||
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
OUTPUT_DIR = os.path.join(BASE_DIR, "output")
|
||
TEMPLATE_PATH = os.path.join(BASE_DIR, "assets/template.html")
|
||
TEST_DATA_PATH = os.path.join(BASE_DIR, "assets/test.json")
|
||
|
||
def init_dirs():
|
||
"""初始化必要的目录"""
|
||
if not os.path.exists(OUTPUT_DIR):
|
||
os.makedirs(OUTPUT_DIR)
|
||
|
||
def parse_params():
|
||
"""解析并验证输入参数
|
||
|
||
支持两种模式：
|
||
1. 正常模式：python analysis.py [角色ID] [Level] [Unit]
|
||
2. 测试模式：python analysis.py test
|
||
"""
|
||
if len(sys.argv) == 2 and sys.argv[1].lower() == "test":
|
||
# 测试模式
|
||
return "test", None, None
|
||
|
||
if len(sys.argv) != 4:
|
||
print("参数错误！使用方式：")
|
||
print(" 正常模式：python analysis.py [角色ID] [Level] [Unit]")
|
||
print(" 测试模式：python analysis.py test")
|
||
sys.exit(1)
|
||
|
||
role_id = sys.argv[1]
|
||
level = sys.argv[2]
|
||
unit = sys.argv[3]
|
||
|
||
# 验证角色ID是数字
|
||
if not role_id.isdigit():
|
||
print("错误：角色ID必须是数字")
|
||
sys.exit(1)
|
||
|
||
# 处理Level参数，支持Level1/L1/1等格式
|
||
level = level.lower().replace("level", "").replace("l", "")
|
||
if not level.isdigit() or int(level) not in [1, 2]:
|
||
print("错误：Level只能是1或2，支持格式：Level1/L1/1/Level2/L2/2")
|
||
sys.exit(1)
|
||
|
||
# 处理Unit参数，支持Unit1/U1/1等格式
|
||
unit = unit.lower().replace("unit", "").replace("u", "")
|
||
if not unit.isdigit() or int(unit) < 1:
|
||
print("错误：Unit必须是正整数，支持格式：Unit1/U1/1")
|
||
sys.exit(1)
|
||
|
||
return int(role_id), int(level), int(unit)
|
||
|
||
def request_api(role_id, level, unit):
|
||
"""请求学情分析API
|
||
|
||
调用接口: https://api.valavala.com/v2/user/unit/learn/info
|
||
参数:
|
||
- operator: Kingson (固定值)
|
||
- level: L1 或 L2 (根据传入的level参数)
|
||
- userId: 用户角色ID
|
||
- unitIndex: 单元号
|
||
"""
|
||
params = {
|
||
"operator": OPERATOR,
|
||
"level": f"L{level}",
|
||
"userId": role_id,
|
||
"unitIndex": unit
|
||
}
|
||
|
||
try:
|
||
response = requests.get(API_URL, params=params, timeout=30)
|
||
response.raise_for_status()
|
||
return response.json()
|
||
except requests.exceptions.Timeout:
|
||
print("API请求超时，请稍后重试")
|
||
sys.exit(1)
|
||
except requests.exceptions.ConnectionError:
|
||
print("API连接失败，请检查网络连接")
|
||
sys.exit(1)
|
||
except requests.exceptions.HTTPError as e:
|
||
print(f"API返回错误状态码：{e.response.status_code}")
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
print(f"API请求失败：{str(e)}")
|
||
sys.exit(1)
|
||
|
||
def load_test_data():
|
||
"""加载测试数据"""
|
||
try:
|
||
with open(TEST_DATA_PATH, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except FileNotFoundError:
|
||
print(f"错误：测试数据文件不存在 {TEST_DATA_PATH}")
|
||
sys.exit(1)
|
||
except json.JSONDecodeError as e:
|
||
print(f"错误：测试数据文件格式不正确 - {str(e)}")
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
print(f"加载测试数据失败：{str(e)}")
|
||
sys.exit(1)
|
||
|
||
def save_data(role_id, level, unit, data):
|
||
"""保存API返回的原始数据"""
|
||
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
|
||
filename = f"{role_id}_L{level}_U{unit}_{timestamp}.json"
|
||
file_path = os.path.join(OUTPUT_DIR, filename)
|
||
|
||
with open(file_path, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
|
||
return file_path
|
||
|
||
def save_test_data(data):
|
||
"""保存测试数据副本"""
|
||
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
|
||
filename = f"test_data_{timestamp}.json"
|
||
file_path = os.path.join(OUTPUT_DIR, filename)
|
||
|
||
with open(file_path, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
|
||
return file_path
|
||
|
||
def generate_html(role_id, level, unit, data):
|
||
"""生成可视化HTML报告"""
|
||
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
|
||
filename = f"study_report_{role_id}_L{level}_U{unit}_{timestamp}.html"
|
||
html_path = os.path.join(OUTPUT_DIR, filename)
|
||
|
||
# 读取模板
|
||
with open(TEMPLATE_PATH, "r", encoding="utf-8") as f:
|
||
template = f.read()
|
||
|
||
# 替换模板中的数据占位符
|
||
html_content = template.replace("{{DATA}}", json.dumps(data, ensure_ascii=False))
|
||
html_content = html_content.replace("{{ROLE_ID}}", str(role_id))
|
||
html_content = html_content.replace("{{LEVEL}}", str(level))
|
||
html_content = html_content.replace("{{UNIT}}", str(unit))
|
||
html_content = html_content.replace("{{GENERATE_TIME}}", datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
|
||
|
||
# 保存HTML文件
|
||
with open(html_path, "w", encoding="utf-8") as f:
|
||
f.write(html_content)
|
||
|
||
return html_path
|
||
|
||
def main():
|
||
init_dirs()
|
||
role_id, level, unit = parse_params()
|
||
|
||
# 判断是否为测试模式
|
||
is_test_mode = role_id == "test"
|
||
|
||
if is_test_mode:
|
||
print("【测试模式】使用本地测试数据生成报告...")
|
||
api_data = load_test_data()
|
||
# 测试模式使用默认值或从数据中读取
|
||
role_id = api_data.get("role_id", 99999)
|
||
level = api_data.get("level", 1)
|
||
unit = api_data.get("unit", 1)
|
||
print(f"测试数据：用户{role_id} Level{level} Unit{unit}")
|
||
else:
|
||
print(f"正在分析用户{role_id} Level{level} Unit{unit}的学习情况...")
|
||
# 请求API
|
||
api_data = request_api(role_id, level, unit)
|
||
print("API数据获取成功")
|
||
|
||
# 保存原始数据
|
||
if is_test_mode:
|
||
data_path = save_test_data(api_data)
|
||
else:
|
||
data_path = save_data(role_id, level, unit, api_data)
|
||
print(f"原始数据已保存到：{data_path}")
|
||
|
||
# 生成HTML报告
|
||
html_path = generate_html(role_id, level, unit, api_data)
|
||
print(f"可视化报告已生成：{html_path}")
|
||
print(f"请访问以下地址查看报告：file://{html_path}")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|