debug: 创建 save_report_history 函数测试脚本
数据库测试成功,说明问题在 report_generator.py 的保存逻辑中。 创建了 test_save_logic.py 测试脚本: - 直接调用 save_report_history 函数 - 传入 file_path 参数 - 查询刚保存的记录 - 验证 file_path 是否正确 运行方式: ```bash python test_save_logic.py ``` 根据测试结果判断: - 正常:问题在 generate_daily_report 函数的调用逻辑 - 不匹配:问题在 save_report_history 函数本身 Coze-Commit-Type: user Coze-User-ID: 3722323274763196 Coze-Conversation-ID: 9894087
This commit is contained in:
BIN
assets/image_20260709133605216.png
Normal file
BIN
assets/image_20260709133605216.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 78 KiB |
85
test_save_logic.py
Normal file
85
test_save_logic.py
Normal file
@@ -0,0 +1,85 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
测试 report_generator.py 的保存逻辑
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
from lib.report_generator import save_report_history
|
||||
from lib.db import execute_query
|
||||
|
||||
def test_save_logic():
|
||||
"""测试保存逻辑"""
|
||||
print("="*60)
|
||||
print("测试 report_generator 的保存逻辑")
|
||||
print("="*60)
|
||||
|
||||
try:
|
||||
# 1. 保存测试记录
|
||||
print("\n1. 调用 save_report_history...")
|
||||
test_file_path = '/reports/test_from_generator.xlsx'
|
||||
|
||||
print(f" 准备保存:")
|
||||
print(f" - file_path: '{test_file_path}'")
|
||||
print(f" - status: 1")
|
||||
|
||||
history_id = save_report_history(
|
||||
report_date='2025-01-01',
|
||||
start_time='2025-01-01 08:00:00',
|
||||
end_time='2025-01-02 08:00:00',
|
||||
split_type='test',
|
||||
split_value='test_value',
|
||||
split_name='测试名称',
|
||||
config_id=999999,
|
||||
total_orders=10,
|
||||
total_amount=100.50,
|
||||
sum_results={'test_field': 50.0},
|
||||
file_path=test_file_path,
|
||||
status=1
|
||||
)
|
||||
|
||||
print(f" ✓ 保存成功,ID: {history_id}")
|
||||
|
||||
# 2. 查询记录
|
||||
print("\n2. 查询刚保存的记录...")
|
||||
query_sql = "SELECT * FROM t_daily_report_history WHERE id = %s"
|
||||
result = execute_query(query_sql, (history_id,))
|
||||
|
||||
if result and len(result) > 0:
|
||||
record = result[0]
|
||||
print(" ✓ 查询成功")
|
||||
print(f" - ID: {record.get('id')}")
|
||||
print(f" - file_path: '{record.get('file_path')}'")
|
||||
print(f" - status: {record.get('status')}")
|
||||
|
||||
if record.get('file_path') == test_file_path:
|
||||
print("\n✓✓✓ save_report_history 工作正常!")
|
||||
else:
|
||||
print(f"\n❌❌❌ file_path 不匹配!")
|
||||
print(f" 期望: '{test_file_path}'")
|
||||
print(f" 实际: '{record.get('file_path')}'")
|
||||
else:
|
||||
print(" ❌ 查询失败,没有找到数据")
|
||||
|
||||
# 3. 清理测试数据
|
||||
print("\n3. 清理测试数据...")
|
||||
from lib.db import execute_update
|
||||
execute_update("DELETE FROM t_daily_report_history WHERE id = %s", (history_id,))
|
||||
print(" ✓ 已清理")
|
||||
|
||||
print("\n" + "="*60)
|
||||
print("测试完成")
|
||||
print("="*60)
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ 测试失败: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
|
||||
input("\n按回车键退出...")
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_save_logic()
|
||||
Reference in New Issue
Block a user