123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596 |
- """
- 文件处理工具模块
- 提供文件操作相关的功能
- """
- import os
- import time
- import datetime
- import shutil
- from .logger import logger
- def get_date_folder(base_folder):
- """
- 获取当前日期对应的文件夹路径,如果不存在则创建
-
- 参数:
- base_folder: 基础文件夹路径
-
- 返回:
- str: 日期文件夹路径
- """
- today = datetime.datetime.now().strftime('%Y-%m-%d')
- date_folder = os.path.join(base_folder, today)
- os.makedirs(date_folder, exist_ok=True)
- return date_folder
- def get_safe_filename(base_name, extension):
- """
- 生成一个带时间戳的安全文件名,避免文件名冲突
-
- 参数:
- base_name: 原始文件名基础部分
- extension: 文件扩展名
-
- 返回:
- str: 安全的文件名
- """
- timestamp = int(time.time())
- return f"{base_name}_{timestamp}{extension}"
- def clean_old_files(output_folder):
- """
- 清理前一天的输出文件
- 每天凌晨2点自动运行
-
- 参数:
- output_folder: 输出文件根目录
- """
- logger.info("开始执行文件清理任务")
- try:
- # 获取前一天的日期
- yesterday = (datetime.datetime.now() - datetime.timedelta(days=1)).strftime('%Y-%m-%d')
- yesterday_folder = os.path.join(output_folder, yesterday)
-
- # 如果文件夹存在,则删除
- if os.path.exists(yesterday_folder) and os.path.isdir(yesterday_folder):
- # 记录删除前的文件数量和大小
- files_count = 0
- total_size = 0
- for root, dirs, files in os.walk(yesterday_folder):
- files_count += len(files)
- for f in files:
- file_path = os.path.join(root, f)
- total_size += os.path.getsize(file_path)
-
- # 删除文件夹
- shutil.rmtree(yesterday_folder)
- logger.info(f"已清理前一天({yesterday})的输出文件夹,共删除 {files_count} 个文件,总大小 {total_size/1024/1024:.2f} MB")
- else:
- logger.info(f"前一天({yesterday})的输出文件夹不存在,无需清理")
- except Exception as e:
- logger.error(f"清理文件时出错: {str(e)}", exc_info=True)
- def allowed_file(file, allowed_extensions):
- """
- 检查上传的文件是否为允许的类型
-
- 参数:
- file: 上传的文件对象
- allowed_extensions: 允许的文件扩展名集合
-
- 返回:
- bool: 文件类型是否允许
- """
- # 检查文件扩展名
- extension = os.path.splitext(file.filename)[1].lower() if file.filename else ''
- if extension not in allowed_extensions:
- return False
-
- # 检查MIME类型
- content_type = file.content_type
- allowed_mime_types = {
- 'application/msword', # .doc
- 'application/vnd.openxmlformats-officedocument.wordprocessingml.document' # .docx
- }
- return content_type in allowed_mime_types
|