Python实现按数据库批量数定时移动文件的技术咨询

核心改进实现 #

• 从数据库获取每次移动的文件数量X
• 按批次移动文件，每批处理X个
• 批次间添加指定时长的等待逻辑

修改后的完整代码 #

import os
import shutil
import time
# 根据你的数据库类型替换导入，比如pymysql/sqlalchemy等
import sqlite3

def get_batch_size_from_db():
    """从数据库读取每次移动的文件量X"""
    # 示例：SQLite连接逻辑，实际替换为你的数据库配置
    conn = sqlite3.connect('your_config.db')
    cursor = conn.cursor()
    # 假设你的数据库中有存储批次大小的配置表
    cursor.execute("SELECT batch_count FROM move_config WHERE id = 1")
    result = cursor.fetchone()
    conn.close()
    return result[0] if result else 10  # 查询失败时默认返回10

def move_files_in_batches(source_dir, dest_dir, batch_size, wait_seconds):
    # 仅获取源目录下的文件（排除子文件夹）
    all_files = [f for f in os.listdir(source_dir) if os.path.isfile(os.path.join(source_dir, f))]
    total = len(all_files)
    print(f"待移动文件总数：{total}")

    # 按批次切割文件列表并处理
    for idx in range(0, total, batch_size):
        batch = all_files[idx:idx+batch_size]
        print(f"处理第{idx//batch_size + 1}批，共{len(batch)}个文件")
        
        # 移动当前批次文件
        for file_name in batch:
            src_path = os.path.join(source_dir, file_name)
            dest_path = os.path.join(dest_dir, file_name)
            try:
                shutil.move(src_path, dest_path)
                print(f"已移动：{file_name}")
            except Exception as e:
                print(f"移动失败[{file_name}]：{str(e)}")
        
        # 非最后一批则等待指定时间
        if idx + batch_size < total:
            print(f"等待{wait_seconds}秒后继续...")
            time.sleep(wait_seconds)

if __name__ == "__main__":
    source = 'callfiles/processed/'
    destination = 'callfiles/created/'
    
    # 从数据库获取批次大小X
    batch_X = get_batch_size_from_db()
    # 设置间隔时间（秒），如需分钟则乘以60，比如5分钟=300秒
    wait_interval = 60
    
    # 确保目标文件夹存在
    os.makedirs(destination, exist_ok=True)
    
    move_files_in_batches(source, destination, batch_X, wait_interval)

关键细节说明 #

• 数据库交互：get_batch_size_from_db()函数需要根据你的实际数据库类型（MySQL/PostgreSQL等）修改连接和查询逻辑，示例用SQLite做演示。
• 文件过滤：代码中仅处理源目录下的文件，若需要移动子文件夹可去掉os.path.isfile()的判断。
• 异常处理：单个文件移动失败不会终止整个脚本，会打印错误信息后继续处理其他文件。
• 文件夹检查：添加了os.makedirs(destination, exist_ok=True)确保目标文件夹存在，避免报错。

内容的提问来源于stack exchange，提问作者Chetan Jha