greater_wms/backup/backup_utils.py

# 数据库备份核心模块：backup_utils.py
import os
import shutil
from datetime import datetime
from django.conf import settings
import sqlite3
from pathlib import Path
import logging

logger = logging.getLogger(__name__)

def backup_database():
    """执行数据库备份，返回备份路径"""
    try:
        # 源数据库路径（根据您的项目配置调整）
        source_db = Path(settings.BASE_DIR) / 'db.sqlite3'
        
        if not source_db.exists():
            raise FileNotFoundError(f"数据库文件不存在: {source_db}")
        
        # 生成备份目录路径
        now = datetime.now()
        year_month = now.strftime("%Y%m")  # 202507
        day_time = now.strftime("%m%d_%H_%M")  # 0728_14_45
        backup_dir = Path(f"E:/code/backup/{year_month}/{day_time}")

        
        # 创建备份目录
        backup_dir.mkdir(parents=True, exist_ok=True)
        
        # 目标备份路径
        backup_path = backup_dir / f"db.sqlite3"
        
        # 使用SQLite在线备份API确保备份完整性
        con = sqlite3.connect(source_db)
        bck = sqlite3.connect(backup_path)
        with bck:
            con.backup(bck, pages=1)
        bck.close()
        con.close()
        
        logger.info(f"数据库备份成功: {backup_path}")
        return str(backup_path)
    
    except Exception as e:
        logger.error(f"数据库备份失败: {str(e)}")
        raise