124 lines
3.6 KiB
Python
124 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
测试脚本 - 测试grade_assignments.py功能
|
||
使用CSV格式保存测试结果
|
||
"""
|
||
|
||
import os
|
||
import random
|
||
import re
|
||
import yaml
|
||
from grade_assignments import (
|
||
read_excel_submissions,
|
||
find_assignment_yml,
|
||
parse_yml_file,
|
||
call_dify_api,
|
||
save_results
|
||
)
|
||
|
||
def test_grade_assignments():
|
||
"""测试作业批改全流程，包含Excel结果写入"""
|
||
print("=== 开始测试 ===")
|
||
|
||
# 1. 创建测试YML文件
|
||
test_yml = "ai作业/作业/第3题_测试_20250331_测试作业_1.yml"
|
||
test_data = {
|
||
"work_description": "测试工作流程描述",
|
||
"solution": "测试解决方案设计",
|
||
"additional_field": "测试额外字段"
|
||
}
|
||
|
||
with open(test_yml, 'w', encoding='utf-8') as f:
|
||
yaml.dump(test_data, f, allow_unicode=True)
|
||
print(f"已创建测试YML文件: {test_yml}")
|
||
|
||
# 2. 测试API调用
|
||
print("\n[测试1] 调用Dify API...")
|
||
api_response = call_dify_api(test_yml, test_data)
|
||
|
||
if not api_response:
|
||
print("API调用失败")
|
||
return
|
||
|
||
print("API调用成功")
|
||
print(f"响应结果: {api_response}")
|
||
|
||
# 3. 测试结果写入Excel
|
||
print("\n[测试2] 测试结果写入Excel...")
|
||
# 从API响应中提取评分结果
|
||
api_output = api_response['data']['outputs']['text']
|
||
score_match = re.search(r'总分：(\d+)分', api_output)
|
||
feedback_match = re.search(r'优点和不足之处(.*?)改进建议', api_output, re.DOTALL)
|
||
|
||
test_result = {
|
||
"name": "测试学生",
|
||
"score": int(score_match.group(1)) if score_match else 0,
|
||
"feedback": feedback_match.group(1).strip() if feedback_match else "无反馈",
|
||
"status": "已完成",
|
||
"api_response": api_response
|
||
}
|
||
|
||
# 临时结果文件路径 - 使用Excel格式
|
||
from openpyxl import Workbook
|
||
import uuid
|
||
test_xlsx = f"results/测试结果_{uuid.uuid4().hex[:8]}.xlsx"
|
||
os.makedirs("results", exist_ok=True)
|
||
|
||
try:
|
||
# 创建Excel工作簿
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "测试结果"
|
||
|
||
# 写入表头
|
||
ws.append(['姓名', '分数', '反馈'])
|
||
|
||
# 写入数据
|
||
ws.append([
|
||
test_result['name'],
|
||
test_result['score'],
|
||
test_result['feedback']
|
||
])
|
||
|
||
# 保存Excel文件
|
||
wb.save(test_xlsx)
|
||
print(f"测试结果已写入Excel: {test_xlsx}")
|
||
|
||
# 验证文件是否存在
|
||
if os.path.exists(test_xlsx):
|
||
print("Excel文件写入验证成功")
|
||
else:
|
||
print("Excel文件写入失败")
|
||
except Exception as e:
|
||
print(f"写入Excel文件时出错: {str(e)}")
|
||
|
||
# 清理测试文件 - 添加重试机制处理文件锁定
|
||
import time
|
||
max_retries = 3
|
||
retry_delay = 1 # 秒
|
||
|
||
def safe_remove(filepath):
|
||
for i in range(max_retries):
|
||
try:
|
||
if os.path.exists(filepath):
|
||
os.remove(filepath)
|
||
print(f"已清理文件: {filepath}")
|
||
return True
|
||
except Exception as e:
|
||
if i == max_retries - 1:
|
||
print(f"无法清理文件 {filepath}: {str(e)}")
|
||
return False
|
||
time.sleep(retry_delay)
|
||
return False
|
||
|
||
print("\n[清理] 删除测试文件...")
|
||
safe_remove(test_yml)
|
||
safe_remove(test_xlsx)
|
||
|
||
print("\n=== 测试完成 ===")
|
||
|
||
if __name__ == "__main__":
|
||
test_grade_assignments()
|