登录搞定了,这次写核心业务:图片怎么过滤、怎么挪、怎么改名。
FileHandler、路径构建、文件移动、日志记录——一篇讲完。
一、概述
本文介绍图片归档工具的核心业务逻辑,涉及两个核心模块:
| 模块 | 职责 |
|---|---|
file_handler.py | 图片过滤、目录构建、文件移动 |
db_handler.py | 操作日志写入数据库 |
二、FileHandler 类
2.1 类结构
python
import os import shutil import time from typing import List, Tuple, Dict SUPPORTED_EXTENSIONS = ('.bmp', '.jpg', '.jpeg') class FileHandler: def __init__(self, source_dir: str, share_root_dir: str): self.source_dir = source_dir self.share_root_dir = share_root_dir2.2 图片过滤
python
def filter_images(self) -> List[str]: """过滤源目录中的图片文件""" images = [] if not os.path.isdir(self.source_dir): return images for entry in os.listdir(self.source_dir): entry_path = os.path.join(self.source_dir, entry) if os.path.isfile(entry_path): ext = os.path.splitext(entry)[1].lower() if ext in SUPPORTED_EXTENSIONS: images.append(entry_path) return sorted(images)
关键点:
| 要点 | 说明 |
|---|---|
| 支持格式 | BMP、JPG、JPEG |
| 返回格式 | 绝对路径列表 |
| 排序 | 按文件名排序,保证处理顺序 |
三、核心方法
3.1 目标路径构建
归档目录结构:共享根目录/客户/日期/批次号
python
def build_target_path(self, customer: str, lot_no: str, op_date: str) -> str: """构建目标目录路径""" return os.path.join( self.share_root_dir, customer, op_date, lot_no )
3.2 目录创建
python
def ensure_directory(self, target_path: str) -> bool: """确保目标目录存在""" try: os.makedirs(target_path, exist_ok=True) return True except Exception as e: raise Exception(f"创建目录失败: {str(e)}")3.3 文件名生成
文件名格式:操作员_原文件名_年月日时分秒
python
def generate_unique_filename(self, target_dir: str, original_name: str) -> str: """生成带时间戳的唯一文件名""" name, ext = os.path.splitext(original_name) timestamp_str = time.strftime('%Y%m%d%H%M%S') new_name = f"{name}_{timestamp_str}{ext}" new_path = os.path.join(target_dir, new_name) # 无冲突直接返回 if not os.path.exists(new_path): return new_name # 冲突时追加序号 counter = 1 while counter <= 100: new_name = f"{name}_{timestamp_str}_{counter}{ext}" new_path = os.path.join(target_dir, new_name) if not os.path.exists(new_path): return new_name counter += 1 return original_name # 兜底3.4 文件移动
python
def move_file(self, src_path: str, target_dir: str, operator: str) -> Tuple[bool, str, str]: """移动文件到目标目录""" try: original_name = os.path.basename(src_path) name, ext = os.path.splitext(original_name) new_name = f"{operator}_{name}{ext}" target_path = os.path.join(target_dir, new_name) # 冲突处理 if os.path.exists(target_path): new_name = self.generate_unique_filename(target_dir, new_name) target_path = os.path.join(target_dir, new_name) shutil.move(src_path, target_path) # 验证移动成功 if os.path.exists(target_path) and not os.path.exists(src_path): return True, target_path, new_name else: return False, target_path, new_name except Exception as e: return False, "", str(e)四、归档流程(archive_images)
python
def archive_images(self, customer: str, lot_no: str, op_date: str, operator: str) -> Dict[str, int]: """归档所有图片""" images = self.filter_images() if not images: return {'success': 0, 'failed': 0, 'total': 0, 'logs': []} # 构建目标目录 target_dir = self.build_target_path(customer, lot_no, op_date) self.ensure_directory(target_dir) success_count = 0 failed_count = 0 logs = [] for src_path in images: op_timestamp = time.time() file_name = os.path.basename(src_path) file_suffix = os.path.splitext(file_name)[1].lower() success, dst_path, final_name = self.move_file(src_path, target_dir, operator) if success: success_count += 1 status = '成功' else: failed_count += 1 status = f'失败: {dst_path}' dst_path = '' # 记录每张图的日志 logs.append({ 'customer': customer, 'operator': operator, 'lot_no': lot_no, 'op_date': op_date, 'src_path': src_path, 'dst_path': dst_path, 'file_name': final_name, 'file_suffix': file_suffix, 'op_timestamp': op_timestamp, 'status': status }) return { 'success': success_count, 'failed': failed_count, 'total': len(images), 'logs': logs }五、归档流程图
text
┌─────────────────────────────────────────┐ │ 开始归档操作 │ └─────────────────┬───────────────────────┘ ▼ ┌─────────────────────────────────────────┐ │ 验证参数(客户、操作人、批次号) │ └─────────────────┬───────────────────────┘ ▼ ┌─────────────────────────────────────────┐ │ 过滤源目录图片文件 │ └─────────────────┬───────────────────────┘ ▼ ┌─────────────────────────────────────────┐ │ 构建目标路径:客户/日期/批次号 │ └─────────────────┬───────────────────────┘ ▼ ┌─────────────────────────────────────────┐ │ 逐个移动文件 │ │ ┌─────────────────────────────────┐ │ │ │ 生成文件名:操作员_原文件名_时间戳 │ │ │ │ 移动文件到目标目录 │ │ │ │ 记录操作日志 │ │ │ └─────────────────────────────────┘ │ └─────────────────┬───────────────────────┘ ▼ ┌─────────────────────────────────────────┐ │ 返回归档结果 │ └─────────────────────────────────────────┘
六、MainWindow 调用
python
def execute_archive(self): customer = self.customer_input.text().strip() operator = self.operator_input.text().strip() lot_no = self.lot_no_input.text().strip() # 参数验证 if not customer: QMessageBox.warning(self, "参数错误", "客户名称不能为空") return if not operator: QMessageBox.warning(self, "参数错误", "操作人不能为空") return if not lot_no: QMessageBox.warning(self, "参数错误", "批次号不能为空") return source_dir = self.config_manager.get_source_dir() share_dir = self.config_manager.get_share_root_dir() # 目录验证 valid, msg = self.config_manager.validate_directory(source_dir) if not valid: QMessageBox.warning(self, "目录错误", msg) return self.add_log(f"开始归档 - 客户:{customer}, 操作人:{operator}, 批次号:{lot_no}") op_date = format_date() file_handler = FileHandler(source_dir, share_dir) result = file_handler.archive_images(customer, lot_no, op_date, operator) if result['total'] == 0: self.add_log("未找到可归档的图片", "WARN") QMessageBox.information(self, "提示", "源目录中没有找到可归档的图片") return # 写入数据库 self.db_handler = DBHandler(share_dir) if result['logs']: self.db_handler.batch_insert_logs(result['logs']) self.add_log(f"数据库记录已写入") msg = f"归档完成!\n\n成功: {result['success']} 张\n失败: {result['failed']} 张" QMessageBox.information(self, "归档完成", msg) self.refresh_data()七、目录结构示例
归档前(源目录):
text
源目录 (Source): ├── image1.bmp ├── image2.jpg └── document.pdf (忽略,不支持)
归档后(目标目录):
text
目标目录 (Target): └── 客户A/ └── 2026-06-24/ └── LOT001/ ├── 张三_image1_20260624103015.bmp └── 张三_image2_20260624103015.jpg
八、踩坑记录
文件名冲突:同一秒内多个文件移动可能重名,加了时间戳 + 序号双重保障
移动后验证:
shutil.move成功不代表真的成功,用os.path.exists双重验证目录自动创建:
os.makedirs(exist_ok=True)确保多级目录一次性创建空目录处理:源目录为空或无图片时,给出明确提示,不要崩溃
不支持格式静默忽略:PDF等非图片文件不报错,只归档图片
下篇预告
下一篇写数据库设计与操作日志管理:SQLite建表、批量插入、日志查询。
如果对文件处理有不同思路,评论区聊。