98 lines
3.7 KiB
Python
98 lines
3.7 KiB
Python
import sqlite3
|
||
import os
|
||
import datetime
|
||
|
||
# --- 配置 ---
|
||
DB_PATH = 'prediction_history.db'
|
||
# 用户指定的时间：2025年7月25日下午两点 (14:00)
|
||
# 数据库中存储的是本地时区的ISO格式字符串，所以我们直接用本地时间构建字符串
|
||
CUTOFF_DATETIME_LOCAL = datetime.datetime(2025, 7, 25, 14, 0, 0)
|
||
# 生成与数据库格式匹配的ISO字符串
|
||
CUTOFF_ISO_STRING = CUTOFF_DATETIME_LOCAL.isoformat()
|
||
|
||
def get_db_connection():
|
||
"""创建并返回数据库连接"""
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
return conn
|
||
|
||
def main():
|
||
"""主执行函数"""
|
||
print("--- 开始清理旧的预测历史数据 ---")
|
||
print(f"数据库路径: {os.path.abspath(DB_PATH)}")
|
||
print(f"时间分界点 (本地时间): {CUTOFF_ISO_STRING}")
|
||
|
||
conn = None
|
||
try:
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
|
||
# 1. 找出所有需要删除的记录
|
||
query = "SELECT id, file_path, created_at FROM prediction_history WHERE created_at < ?"
|
||
cursor.execute(query, (CUTOFF_ISO_STRING,))
|
||
records_to_delete = cursor.fetchall()
|
||
|
||
if not records_to_delete:
|
||
print("\n[INFO] 没有找到需要清理的旧数据。")
|
||
return
|
||
|
||
print(f"\n[INFO] 发现了 {len(records_to_delete)} 条需要清理的记录。")
|
||
|
||
ids_to_delete_from_db = []
|
||
files_deleted_count = 0
|
||
files_not_found_count = 0
|
||
|
||
# 2. 遍历记录，删除关联的JSON文件
|
||
for record in records_to_delete:
|
||
record_id = record['id']
|
||
file_path = record['file_path']
|
||
created_at = record['created_at']
|
||
|
||
print(f" - 正在处理记录 ID: {record_id} (创建于: {created_at})")
|
||
|
||
if file_path and isinstance(file_path, str):
|
||
# 确保我们处理的是项目根目录下的相对路径
|
||
absolute_file_path = os.path.abspath(file_path)
|
||
if os.path.exists(absolute_file_path):
|
||
try:
|
||
os.remove(absolute_file_path)
|
||
print(f" [SUCCESS] 已删除文件: {absolute_file_path}")
|
||
files_deleted_count += 1
|
||
except OSError as e:
|
||
print(f" [ERROR] 删除文件失败: {e}")
|
||
else:
|
||
print(f" [WARNING] 文件未找到 (可能已被删除): {absolute_file_path}")
|
||
files_not_found_count += 1
|
||
else:
|
||
print(f" [INFO] 此记录没有关联的文件路径。")
|
||
|
||
ids_to_delete_from_db.append(record_id)
|
||
|
||
# 3. 从数据库中删除记录
|
||
if ids_to_delete_from_db:
|
||
# 使用元组来传递给 a, b, c IN (?, ?, ?)
|
||
placeholders = ','.join('?' for _ in ids_to_delete_from_db)
|
||
delete_query = f"DELETE FROM prediction_history WHERE id IN ({placeholders})"
|
||
|
||
cursor.execute(delete_query, ids_to_delete_from_db)
|
||
conn.commit()
|
||
print(f"\n[SUCCESS] 已从数据库中删除 {cursor.rowcount} 条记录。")
|
||
|
||
print("\n--- 清理完成 ---")
|
||
print(f"总计处理记录: {len(records_to_delete)}")
|
||
print(f"成功删除文件: {files_deleted_count}")
|
||
print(f"未找到的文件: {files_not_found_count}")
|
||
|
||
except sqlite3.Error as e:
|
||
print(f"\n[FATAL] 数据库操作失败: {e}")
|
||
except Exception as e:
|
||
print(f"\n[FATAL] 发生未知错误: {e}")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
print("\n数据库连接已关闭。")
|
||
|
||
if __name__ == '__main__':
|
||
# 确保脚本在项目根目录执行
|
||
# os.chdir(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
main() |