Python实现指定编号JPG/JSON文件批量复制与规则分类排序

核心实现逻辑 #

• 产品ID解析：支持多段范围、单ID混合输入，自动去重、兼容前后空格、范围起止顺序写反等异常输入
• 文件匹配：通过正则严格匹配[产品ID]_[批次号]_[文件序号]_[时间戳]命名规则，自动区分JPEG、JSON格式
• 分组规则：单个产品的JPEG文件按文件序号升序排列后，按10-5-10的数量规则切分为3组，组号从1开始计数，不足25张时按现有顺序顺延分组
• 属性判定：读取同名JSON标注文件的属性字段，区分NG（异常）/OK（正常）分类，可根据实际JSON结构灵活调整读取逻辑
• 目录自动创建：严格按照「处理日期根目录→属性二级目录→gray_images/Json分类子目录→批次号目录」的层级自动创建路径，无需手动新建
• 实时日志：逐文件输出复制状态，无匹配文件、标注解析失败、目标文件已存在等场景均有明确提示

完整可运行脚本 #

import os
import re
import shutil
import json
from datetime import datetime
from collections import defaultdict

# 文件名匹配正则，对应[产品ID]_[批次号]_[文件序号]_[时间戳].后缀格式
FILE_PATTERN = re.compile(r'^(\d+)_([^_]+)_(\d+)_(\d+)\.(jpe?g|json)$', re.IGNORECASE)
# 单产品图片分组切分规则
GROUP_SPLIT = [10, 5, 10]

def parse_pid_range(input_str: str) -> set:
    """解析用户输入的多段产品ID范围，返回去重后的ID集合"""
    pid_set = set()
    segments = [s.strip() for s in input_str.split(';') if s.strip()]
    for seg in segments:
        if '-' in seg:
            start, end = seg.split('-', 1)
            start, end = int(start.strip()), int(end.strip())
            if start > end:
                start, end = end, start
            pid_set.update(range(start, end + 1))
        else:
            pid_set.add(int(seg.strip()))
    return pid_set

def get_label_status(json_path: str) -> bool:
    """读取JSON标注，返回True为NG异常，False为OK正常，可根据实际JSON结构修改此函数逻辑"""
    try:
        with open(json_path, 'r', encoding='utf-8') as f:
            label = json.load(f)
        # 示例默认读取is_defect字段，值为True代表异常，可按需替换字段名、判断规则
        return label.get('is_defect', False)
    except Exception as e:
        print(f"[警告] 解析标注文件{os.path.basename(json_path)}失败，默认归类为OK：{str(e)}")
        return False

def get_group_num(file_seq: int, sorted_seq: list) -> int:
    """根据文件序号在同产品所有图片中的排序位置，返回所属组号（从1开始）"""
    idx = sorted_seq.index(file_seq)
    pos = 0
    for g_num, count in enumerate(GROUP_SPLIT, start=1):
        pos += count
        if idx < pos:
            return g_num
    # 超出25张的文件默认归到第3组
    return 3

def main():
    source_dir = input("请输入源文件夹路径：").strip()
    target_root = input("请输入目标文件夹根路径：").strip()
    pid_input = input("请输入产品ID范围（格式示例：965-1000;100-110; 71）：").strip()

    if not os.path.isdir(source_dir):
        print("源文件夹路径不存在，程序退出")
        return
    os.makedirs(target_root, exist_ok=True)

    target_pids = parse_pid_range(pid_input)
    process_date = datetime.now().strftime('%Y%m%d')
    # 文件存储结构：{产品ID: {批次号: {文件序号: {后缀: 文件绝对路径}}}}
    file_map = defaultdict(lambda: defaultdict(dict))
    found_pids = set()

    # 扫描源文件夹收集匹配文件
    print("\n开始扫描源文件夹...")
    for root, _, files in os.walk(source_dir):
        for file in files:
            match = FILE_PATTERN.match(file)
            if not match:
                continue
            pid, batch_no, seq, ts, suffix = match.groups()
            pid, seq = int(pid), int(seq)
            suffix = suffix.lower()
            if pid not in target_pids:
                continue
            file_path = os.path.join(root, file)
            file_map[pid][batch_no][seq] = file_map[pid][batch_no].get(seq, {})
            file_map[pid][batch_no][seq][suffix] = file_path
            found_pids.add(pid)

    # 提示无匹配文件的ID
    missing_pids = target_pids - found_pids
    for pid in missing_pids:
        print(f"[提示] 产品ID {pid} 无对应匹配文件")
    if not found_pids:
        print("未找到任何匹配目标文件，程序退出")
        return

    # 按规则复制归档
    print("\n开始执行文件复制归档...")
    for pid in found_pids:
        for batch_no, seq_files in file_map[pid].items():
            all_seqs = sorted(seq_files.keys())
            jpg_seqs = [s for s in all_seqs if 'jpg' in seq_files[s] or 'jpeg' in seq_files[s]]
            if len(jpg_seqs) < 25:
                print(f"[警告] 产品ID {pid} 批次{batch_no} 图片数量不足25张，将按现有顺序分组")
            
            for seq in all_seqs:
                file_item = seq_files[seq]
                # 判定NG/OK属性，优先读取同序号JSON
                is_ng = False
                if 'json' in file_item:
                    is_ng = get_label_status(file_item['json'])
                group_num = get_group_num(seq, jpg_seqs)
                # 拼接二级目录名
                tag = 'NG' if is_ng else 'OK'
                dir_suffix = 'anomaly' if is_ng else 'nomaly'
                second_dir = f"{process_date}_XXXX_{tag}_{group_num}_{dir_suffix}"

                # 复制JPEG文件
                for img_suf in ('jpg', 'jpeg'):
                    if img_suf in file_item:
                        src = file_item[img_suf]
                        target_dir = os.path.join(target_root, process_date, second_dir, 'gray_images', batch_no)
                        os.makedirs(target_dir, exist_ok=True)
                        target_path = os.path.join(target_dir, os.path.basename(src))
                        if os.path.exists(target_path):
                            print(f"[跳过] 目标已存在同名文件：{target_path}")
                            continue
                        shutil.copy2(src, target_path)
                        print(f"[成功] 复制图片：{os.path.basename(src)} -> {target_path}")
                
                # 复制JSON文件
                if 'json' in file_item:
                    src = file_item['json']
                    target_dir = os.path.join(target_root, process_date, second_dir, 'Json', batch_no)
                    os.makedirs(target_dir, exist_ok=True)
                    target_path = os.path.join(target_dir, os.path.basename(src))
                    if os.path.exists(target_path):
                        print(f"[跳过] 目标已存在同名文件：{target_path}")
                        continue
                    shutil.copy2(src, target_path)
                    print(f"[成功] 复制标注：{os.path.basename(src)} -> {target_path}")
    print("\n所有文件处理完成")

if __name__ == "__main__":
    main()

使用说明 #

• 运行脚本后按提示依次输入源文件夹路径、目标存储根路径、待处理产品ID范围即可
• 如果实际JSON文件中存储OK/NG属性的字段、判断规则和脚本默认逻辑不一致，直接修改get_label_status函数内的读取判断代码即可
• 脚本默认不会覆盖目标路径下已存在的同名文件，若需要覆盖可删除对应路径存在判断逻辑
• 命名不符合规则的文件会被自动跳过，不会中断整体处理流程

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