151 lines
5.1 KiB
Python
151 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
测试日期时间解析功能
|
|
"""
|
|
|
|
import sqlite3
|
|
import json
|
|
from datetime import datetime, timezone
|
|
|
|
def test_datetime_formats():
|
|
"""测试不同的日期时间格式"""
|
|
# 连接到数据库
|
|
db_path = r"C:\Users\imeep\AppData\Roaming\mixvideo\mixvideoV2.db"
|
|
conn = sqlite3.connect(db_path)
|
|
cursor = conn.cursor()
|
|
|
|
try:
|
|
# 查询最新的测试数据
|
|
cursor.execute("""
|
|
SELECT created_at, started_at, completed_at
|
|
FROM uni_comfyui_task
|
|
WHERE workflow_run_id LIKE 'run_%'
|
|
ORDER BY id DESC
|
|
LIMIT 3
|
|
""")
|
|
|
|
tasks = cursor.fetchall()
|
|
|
|
print("=== 数据库中的日期时间格式 ===")
|
|
for i, task in enumerate(tasks):
|
|
print(f"任务 {i+1}:")
|
|
print(f" created_at: {task[0]}")
|
|
print(f" started_at: {task[1]}")
|
|
print(f" completed_at: {task[2]}")
|
|
print()
|
|
|
|
# 测试Python解析这些格式
|
|
print("=== Python解析测试 ===")
|
|
for i, task in enumerate(tasks):
|
|
print(f"任务 {i+1}:")
|
|
for field_name, field_value in [("created_at", task[0]), ("started_at", task[1]), ("completed_at", task[2])]:
|
|
if field_value:
|
|
try:
|
|
# 尝试解析
|
|
dt = datetime.fromisoformat(field_value.replace('Z', '+00:00'))
|
|
print(f" {field_name}: ✅ {field_value} -> {dt}")
|
|
except Exception as e:
|
|
print(f" {field_name}: ❌ {field_value} -> {e}")
|
|
else:
|
|
print(f" {field_name}: NULL")
|
|
print()
|
|
|
|
except Exception as e:
|
|
print(f"❌ 测试失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
finally:
|
|
conn.close()
|
|
|
|
def create_test_with_different_formats():
|
|
"""创建使用不同日期格式的测试数据"""
|
|
db_path = r"C:\Users\imeep\AppData\Roaming\mixvideo\mixvideoV2.db"
|
|
conn = sqlite3.connect(db_path)
|
|
cursor = conn.cursor()
|
|
|
|
try:
|
|
# 测试不同的日期格式
|
|
test_formats = [
|
|
("format1", datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S.%f')), # 标准格式
|
|
("format2", datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%f')), # ISO格式
|
|
("format3", datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%fZ')), # ISO格式带Z
|
|
("format4", datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S')), # 不带毫秒
|
|
]
|
|
|
|
print("=== 创建不同格式的测试数据 ===")
|
|
|
|
for format_name, date_str in test_formats:
|
|
workflow_run_id = f"test_{format_name}"
|
|
|
|
print(f"创建 {format_name}: {date_str}")
|
|
|
|
cursor.execute("""
|
|
INSERT OR REPLACE INTO uni_comfyui_task
|
|
(workflow_run_id, workflow_name, task_type, status, input_params,
|
|
created_at, started_at, completed_at, server_url)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
workflow_run_id, "test_datetime_formats", "single", "completed",
|
|
json.dumps({"test": "datetime_format"}),
|
|
date_str, date_str, date_str,
|
|
"http://test"
|
|
))
|
|
|
|
conn.commit()
|
|
print("✅ 测试数据创建完成")
|
|
|
|
# 验证数据
|
|
cursor.execute("""
|
|
SELECT workflow_run_id, created_at, started_at, completed_at
|
|
FROM uni_comfyui_task
|
|
WHERE workflow_run_id LIKE 'test_format%'
|
|
ORDER BY workflow_run_id
|
|
""")
|
|
|
|
results = cursor.fetchall()
|
|
print("\n=== 验证创建的数据 ===")
|
|
for result in results:
|
|
print(f"{result[0]}: {result[1]}")
|
|
|
|
except Exception as e:
|
|
conn.rollback()
|
|
print(f"❌ 创建失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
finally:
|
|
conn.close()
|
|
|
|
def cleanup_datetime_test_data():
|
|
"""清理日期时间测试数据"""
|
|
db_path = r"C:\Users\imeep\AppData\Roaming\mixvideo\mixvideoV2.db"
|
|
conn = sqlite3.connect(db_path)
|
|
cursor = conn.cursor()
|
|
|
|
try:
|
|
cursor.execute("DELETE FROM uni_comfyui_task WHERE workflow_run_id LIKE 'test_format%'")
|
|
deleted = cursor.rowcount
|
|
conn.commit()
|
|
print(f"✅ 清理完成: 删除了 {deleted} 个测试任务")
|
|
except Exception as e:
|
|
conn.rollback()
|
|
print(f"❌ 清理失败: {e}")
|
|
finally:
|
|
conn.close()
|
|
|
|
if __name__ == "__main__":
|
|
print("=== 日期时间解析测试 ===")
|
|
|
|
# 1. 测试现有数据的格式
|
|
test_datetime_formats()
|
|
|
|
# 2. 创建不同格式的测试数据
|
|
create_test_with_different_formats()
|
|
|
|
# 3. 再次测试
|
|
print("\n=== 重新测试所有格式 ===")
|
|
test_datetime_formats()
|
|
|
|
# 4. 清理测试数据
|
|
print("\n=== 清理测试数据 ===")
|
|
cleanup_datetime_test_data()
|