前言

在前面的【重识别系列01】和【重识别系列02】中，我们依靠 ResNet50 和一系列 Tricks，在标准的 Market1501 数据集上取得了还不错的成绩。但是，如果你觉得掌握了这些就可以直接去降维打击真实的电动车追踪任务，那就太天真了。

Market1501 是一个高度清洗过的学术数据集，每个人都被完美地框了出来，标签也绝对正确。

但在真实的智慧城市监控中，我们通过目标检测（YOLO）和多目标追踪（DeepSORT）自动截取下来的原始电动车数据，简直就是一场灾难：

• 车牌识别错误：由于反光、模糊，自动抓拍系统经常把车牌认错。
• 追踪串线（ID Switch）：两辆车交汇时，追踪算法把 A 车的 ID 赋给了 B 车，导致同一个文件夹里其实装了两辆完全不同的车。
• 套牌问题：真实的物理世界里，同号不同车的情况屡见不鲜。

“Garbage in, garbage out.”（垃圾进，垃圾出）

如果在这种充满噪声的数据集上强行训练，无论你的三元组损失写得多优雅，模型最终都会学到一堆自相矛盾的废特征。因此，在真正开始电动车重识别的训练之前，我们必须进行一次彻底的“大扫除”。

本文将分享笔者编写的两个自动化数据清洗脚本：

• ocr.py：利用本地部署的 Qwen 视觉大模型进行车牌复核。
• clean_same.py：利用预训练残差网络进行物理特征比对。

第一件神器：用大模型做粗筛（ocr.py）

这段代码展示了整个 ocr.py 的实现逻辑，主要解决三个问题：

• 图像到文本：把电动车号牌图片送入本地部署的 Qwen 视觉模型，提取数字结果。
• 结果校验与精准剔除：通过编辑距离判断识别结果是否可信，一旦发现劣质图片，直接将其“连根拔起”移至隔离区，而不是误杀整个文件夹。
• 自动化清理与容错：支持基于 TXT 文件的断点续传、空文件夹自动销毁、以及照片移走后的序号自动重排，极大节省了硬盘空间和人工整理的精力。

这段代码比较长，但你可以把它看成四个核心模块：

1. ElectricBikeOCR 类定义：封装 OCR 调用、数字提取、编辑距离、日志记录、单图移动与序号重排等功能。
2. recognize_plate_numbers()：调用本地 Qwen 模型做视觉 OCR，直接取出纯数字结果。
3. batch_recognize_to_csv()：遍历目录，判定失效，把有问题的单张图片移动到./L，并负责善后工作（删空文件夹或重排序）。
4. 断点续传：load_processed_folders() 将已处理文件夹记入 CSV，避免重复处理。

下面先看完整实现，再继续拆解几个最关键的部分。

import os
import base64
import shutil
from pathlib import Path
import re
import argparse
from datetime import datetime
from typing import Optional, Tuple
from openai import OpenAI

class ElectricBikeOCR:
    def __init__(self, base_url: str = "http://localhost:8000/v1", model: str = "Qwen/Qwen3.5-9B"):
        """
        初始化OCR识别器，使用本地部署的Qwen模型（OpenAI兼容接口）
        """
        self.base_url = base_url
        self.model = model
        self.client = OpenAI(api_key="EMPTY", base_url=base_url)
        
    def encode_image(self, image_path: str) -> str:
        with open(image_path, "rb") as image_file:
            img_base64 = base64.b64encode(image_file.read()).decode('utf-8')
        return f"data:image/jpeg;base64,{img_base64}"
    
    def extract_numbers(self, text: str) -> str:
        if not text:
            return ""
        return re.sub(r'[^0-9]', '', text)
    
    def levenshtein_distance(self, s1: str, s2: str) -> int:
        if len(s1) < len(s2):
            return self.levenshtein_distance(s2, s1)
        if len(s2) == 0:
            return len(s1)
        
        previous_row = list(range(len(s2) + 1))
        for i, c1 in enumerate(s1):
            current_row = [i + 1]
            for j, c2 in enumerate(s2):
                insertions = previous_row[j + 1] + 1
                deletions = current_row[j] + 1
                substitutions = previous_row[j] + (c1 != c2)
                current_row.append(min(insertions, deletions, substitutions))
            previous_row = current_row
        return previous_row[-1]
    
    def recognize_plate_numbers(self, image_path: str) -> Tuple[Optional[str], float]:
        try:
            img_data_url = self.encode_image(image_path)
            messages = [
                {
                    "role": "user",
                    "content": [
                        {"type": "image_url", "image_url": {"url": img_data_url}},
                        {"type": "text", "text": (
                            "请仔细识别这张电动车号牌上的所有数字。"
                            "注意：电动车号牌通常包含数字，请忽略汉字和字母，只提取数字部分。"
                            "例如：如果号牌是'冀AU17N13'，请输出'1713'；"
                            "如果号牌是'蒙C100139'，请输出'100139'。"
                            "直接输出纯数字，不要输出任何其他字符和说明文字。"
                            "如果看不清，只输出能确定的数字。"
                        )}
                    ]
                }
            ]
            
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=0,
                max_tokens=200,
                extra_body={"chat_template_kwargs": {"enable_thinking": False}}
            )
            
            result_text = response.choices[0].message.content
            numbers_only = self.extract_numbers(result_text)
            
            if numbers_only:
                expected_len = 6
                len_ratio = min(len(numbers_only), expected_len) / expected_len
                confidence = min(0.7 + (len_ratio * 0.3), 0.95)
            else:
                confidence = 0.0
            
            return numbers_only if numbers_only else None, confidence
        except Exception as e:
            return None, 0.0

    def write_log(self, log_file: str, message: str):
        """写入日志并打印到控制台"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        log_msg = f"[{timestamp}] {message}"
        print(log_msg)
        with open(log_file, 'a', encoding='utf-8') as f:
            f.write(log_msg + '\n')

    def resequence_images(self, folder_path: Path):
        """将文件夹内的图片重新按 k1.jpg, k2.jpg ... 排序命名"""
        images = []
        for ext in ['*.jpg', '*.jpeg', '*.png', '*.bmp']:
            images.extend(list(folder_path.glob(ext)))
            
        def extract_num(path):
            m = re.search(r'\d+', path.stem)
            return int(m.group()) if m else 0
            
        images.sort(key=extract_num)

        temp_files = []
        for i, img in enumerate(images, 1):
            temp_name = folder_path / f"temp_rename_{i}{img.suffix}"
            img.rename(temp_name)
            temp_files.append(temp_name)

        for i, temp_img in enumerate(temp_files, 1):
            final_name = folder_path / f"k{i}{temp_img.suffix}"
            temp_img.rename(final_name)

    def load_processed_folders(self, tracking_file: str) -> set:
        """从TXT文件中读取已处理的文件夹名"""
        processed = set()
        if os.path.exists(tracking_file):
            with open(tracking_file, 'r', encoding='utf-8') as f:
                for line in f:
                    folder_name = line.strip()
                    if folder_name:
                        processed.add(folder_name)
        return processed

    def batch_recognize_and_clean(self, base_dir: str, failed_dir: str, max_edit_distance: int = 2):
        base_path = Path(base_dir)
        tracking_file = "processed_folders.txt"
        log_file = "ocr_log.txt"
        
        if not base_path.exists():
            print(f"错误：输入目录 {base_dir} 不存在")
            return
            
        self.write_log(log_file, "=" * 60)
        self.write_log(log_file, "🚀 启动电动车号牌清洗程序 (图片移动模式)")
        
        processed_folders = self.load_processed_folders(tracking_file)
        all_subdirs = [d for d in base_path.iterdir() if d.is_dir()]
        all_subdirs.sort()
        
        pending_subdirs = [d for d in all_subdirs if d.name not in processed_folders]
        
        self.write_log(log_file, f"总文件夹数: {len(all_subdirs)} | 已跳过: {len(processed_folders)} | 待处理: {len(pending_subdirs)}")
        self.write_log(log_file, f"容错率: 编辑距离 > {max_edit_distance} 将被移走")
        self.write_log(log_file, "=" * 60)

        if not pending_subdirs:
            self.write_log(log_file, "🎉 所有文件夹已处理完毕！")
            return

        with open(tracking_file, 'a', encoding='utf-8') as tracking_f:
            for idx, folder in enumerate(pending_subdirs, 1):
                truth_plate = folder.name
                truth_numbers = self.extract_numbers(truth_plate)
                
                self.write_log(log_file, f"[{idx}/{len(pending_subdirs)}] 正在扫描: {truth_plate} (真值: {truth_numbers})")
                
                image_files = []
                for ext in ['*.jpg', '*.jpeg', '*.png', '*.bmp']:
                    image_files.extend(list(folder.glob(ext)))
                
                image_files.sort(key=lambda x: (int(re.search(r'\d+', x.stem).group()) if re.search(r'\d+', x.stem) else 0))
                
                if not image_files:
                    self.write_log(log_file, f"  ⚠️ 空文件夹，执行删除: {folder}")
                    folder.rmdir()
                    tracking_f.write(truth_plate + '\n')
                    tracking_f.flush()
                    continue

                bad_images = []
                
                for img_file in image_files:
                    ocr_numbers, _ = self.recognize_plate_numbers(str(img_file))
                    
                    if ocr_numbers and truth_numbers:
                        edit_dist = self.levenshtein_distance(ocr_numbers, truth_numbers)
                    else:
                        edit_dist = len(truth_numbers) if truth_numbers else 999
                    
                    is_good_ocr = edit_dist <= max_edit_distance
                    
                    if not is_good_ocr:
                        self.write_log(log_file, f"    ✗ 发现劣质图片 [{img_file.name}] -> OCR识别: {ocr_numbers} (编辑距离: {edit_dist})")
                        bad_images.append(img_file)
                    else:
                        print(f"    ✓ 正常图片 [{img_file.name}] -> OCR: {ocr_numbers}")

                if bad_images:
                    target_folder = Path(failed_dir) / truth_plate
                    target_folder.mkdir(parents=True, exist_ok=True)
                    
                    for bad_img in bad_images:
                        target_path = target_folder / bad_img.name
                        shutil.move(str(bad_img), str(target_path))
                        self.write_log(log_file, f"  👉 已移动: {bad_img.name} -> {target_folder}")

                remaining_files = list(folder.glob('*'))
                if not remaining_files:
                    folder.rmdir()
                    self.write_log(log_file, f"  🗑️ 照片已全部移走，彻底删除空文件夹: {folder}")
                elif bad_images:
                    self.resequence_images(folder)
                    self.write_log(log_file, f"  🔄 文件夹内发生裁员，已重新排序剩余照片: {folder}")
                
                tracking_f.write(truth_plate + '\n')
                tracking_f.flush()

        self.write_log(log_file, "=" * 60)
        self.write_log(log_file, "🎉 全部数据清洗完毕！详细记录请查看 ocr_log.txt")

def main():
    parser = argparse.ArgumentParser(description="电动车号牌大扫除脚本 (移动模式)")
    parser.add_argument("--base-url", default="http://localhost:8000/v1", help="本地模型服务地址")
    parser.add_argument("--model", default="Qwen/Qwen3.5-9B", help="模型名称")
    parser.add_argument("--input-dir", default="./K", help="包含车牌子文件夹的根目录")
    parser.add_argument("--failed-dir", default="./L", help="废片隔离区根目录")
    parser.add_argument("--max-edit-distance", type=int, default=2, 
                       help="最大允许编辑距离（默认2）。设为0表示必须完全匹配")
    args = parser.parse_args()
    
    ocr = ElectricBikeOCR(base_url=args.base_url, model=args.model)
    ocr.batch_recognize_and_clean(args.input_dir, args.failed_dir, args.max_edit_distance)

if __name__ == "__main__":
    main()

电动车号牌的一个痛点是包含大量的汉字、字母和毫无规律的排版。传统的OCR算法在面对模糊和变形时常常力不从心。

既然现在是大模型时代，我们不妨直接使用具备强大零样本（Zero-shot）能力的视觉大语言模型（VLM）来帮我们提取车牌。为了兼顾隐私和成本（主要是成本，现在token的价格太贵了），我们通过 OpenAI 兼容接口调用了本地部署的 Qwen/Qwen3.5-9B 模型。

1. Prompt工程

在这个脚本中，我们没有让模型做开放式问答，而是通过极度明确的 Prompt 将其限制为一个纯粹的“数字提取器”。

prompt = (
    "请仔细识别这张电动车号牌上的所有数字。"
    "注意：电动车号牌通常包含数字，请忽略汉字和字母，只提取数字部分。"
    "例如：如果号牌是'冀AU17N13'，请输出'1713'；"
    "如果号牌是'蒙C100139'，请输出'100139'。"
    "直接输出纯数字，不要输出任何其他字符和说明文字。"
    "如果看不清，只输出能确定的数字。"
)

通过对比大模型增加深度思考与否最终识别的准确度，我们发现其实不加深度思考的准确度也在我们理想范围，而且速度增加到极其可观的程度：开了深度思考可能一分钟才只能处理两张图，两千多张图要将近三十个小时。于是我们决定关闭深度思考。我们设置（enable_thinking: False）并将 temperature 设为 0，拿到模型返回的结果后，再用正则表达式 re.sub(r'[^0-9]', '', text) 兜底，剔除所有非数字字符。

2. 引入编辑距离

拿到提取的数字后，我们要将其与该文件夹的真实名称进行比对。由于大模型识别电动车牌会存在一定的识别损失，部分车牌也存在被遮挡的情况，所以我们不能简单地用 == 来判断对错。

我们引入编辑距离算法：计算将模型识别出的字符串转换成真实值字符串，最少需要几次插入、删除或替换操作。

    def levenshtein_distance(self, s1: str, s2: str) -> int:
        if len(s1) < len(s2):
            return self.levenshtein_distance(s2, s1)
        if len(s2) == 0:
            return len(s1)
        
        previous_row = list(range(len(s2) + 1))
        for i, c1 in enumerate(s1):
            current_row = [i + 1]
            for j, c2 in enumerate(s2):
                insertions = previous_row[j + 1] + 1
                deletions = current_row[j] + 1
                substitutions = previous_row[j] + (c1 != c2)
                current_row.append(min(insertions, deletions, substitutions))
            previous_row = current_row
        return previous_row[-1]

edit_dist = self.levenshtein_distance(ocr_numbers, truth_numbers)
is_good_ocr = edit_dist <= max_edit_distance

如果某张图片的识别结果与真实数字的编辑距离大于 max_edit_distance（脚本中默认为 2），我们就判定这是一张“劣质图片”。

        if bad_images:
            target_folder = Path(failed_dir) / truth_plate
            target_folder.mkdir(parents=True, exist_ok=True)
            
            for bad_img in bad_images:
                target_path = target_folder / bad_img.name
                shutil.move(str(bad_img), str(target_path))

def resequence_images(self, folder_path: Path):
        """将文件夹内的图片重新按 k1.jpg, k2.jpg ... 排序命名"""
        images = []
        for ext in ['*.jpg', '*.jpeg', '*.png', '*.bmp']:
            images.extend(list(folder_path.glob(ext)))
            
        # 根据原文件名中的数字大小进行排序，防止 k10.jpg 排在 k2.jpg 前面
        def extract_num(path):
            m = re.search(r'\d+', path.stem)
            return int(m.group()) if m else 0
            
        images.sort(key=extract_num)

        # 第一步：先全部重命名为临时文件，避免 k1 变成 k2 时发生覆盖冲突
        temp_files = []
        for i, img in enumerate(images, 1):
            temp_name = folder_path / f"temp_rename_{i}{img.suffix}"
            img.rename(temp_name)
            temp_files.append(temp_name)

        # 第二步：将临时文件重命名为最终的 kn.jpg
        for i, temp_img in enumerate(temp_files, 1):
            final_name = folder_path / f"k{i}{temp_img.suffix}"
            temp_img.rename(final_name)

我们用脚本将其剔除：直接将这张有问题的图片移动到失败目录（如 ./L/蒙C123456/）下。移动后，程序还会进行善后处理：如果原本的文件夹空了，直接将其删除；如果文件夹里还有幸存的优质照片，则会自动进行重排序（将名称规整为 k1.jpg, k2.jpg 等），确保后续 DataLoader 读取时拥有完美的连续性。

3. 断点续传

对于这种大规模目录的 OCR 清洗，重跑一次整个流程的代价非常高。为此，我们利用一个极其轻量的.txt文件加入了断点续传机制。

脚本会实时将已经处理完的车牌号写入processed_folders.txt，并在下次启动时读取过滤。这样即使程序中断、断电或者你临时想跑别的任务，下次继续时也能无缝衔接。

        processed_folders = self.load_processed_folders(tracking_file)
        all_subdirs = [d for d in base_path.iterdir() if d.is_dir()]
        all_subdirs.sort()
        
        pending_subdirs = [d for d in all_subdirs if d.name not in processed_folders]

这里的关键函数是 load_processed_folders：

def load_processed_folders(self, tracking_file: str) -> set:
    processed = set()
    if os.path.exists(tracking_file):
        with open(tracking_file, 'r', encoding='utf-8') as f:
            for line in f:
                folder_name = line.strip()
                if folder_name:
                    processed.add(folder_name)
    return processed

这种方式让整个清洗流程更加稳定，尤其适合处理上万、甚至几十万张图片的场景。

第二件神器：同号不同车排查（clean_same.py）

源代码如下：

import os
import shutil
import torch
import torch.nn.functional as F
import torchvision.transforms as transforms
from torchvision.models import resnet50, ResNet50_Weights
from PIL import Image
from pathlib import Path
from itertools import combinations
from tqdm import tqdm

class SamePlateDifferentCarCleaner:
    def __init__(self, threshold=0.70, crop_top_ratio=0.5):
        """
        :param threshold: 相似度阈值 (0.70 是个不错的预演起点)
        :param crop_top_ratio: 裁掉图片上半部分的比例 (0.5 表示砍掉上半身)
        """
        self.threshold = threshold
        self.crop_top_ratio = crop_top_ratio
        
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        print(f"🚀 正在加载 ResNet50 预训练模型至 {self.device}...")
        
        self.model = resnet50(weights=ResNet50_Weights.IMAGENET1K_V1)
        self.model.fc = torch.nn.Identity() # 剥离分类头，只取特征
        self.model.eval().to(self.device)
        
        self.transform = transforms.Compose([
            transforms.Resize((224, 224)),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
        ])

    def extract_feature(self, image_path):
        """读取图片 -> 裁掉上半部分 -> 提取特征"""
        try:
            img = Image.open(image_path).convert('RGB')
            w, h = img.size
            
            # ✂️ 核心物理隔离：砍掉上半身（人），只留车尾
            top = int(h * self.crop_top_ratio)
            img_cropped = img.crop((0, top, w, h))
            
            # 转为 Tensor 提取特征
            tensor = self.transform(img_cropped).unsqueeze(0).to(self.device)
            with torch.no_grad():
                feature = self.model(tensor)
                feature = F.normalize(feature, p=2, dim=1) # L2归一化
            return feature
        except Exception as e:
            print(f"  ❌ 读取或处理图片 {image_path} 失败: {e}")
            return None

    def process_dataset(self, input_dir, suspect_dir):
        """
        :param input_dir: 原始车牌文件夹的根目录 (K/)
        :param suspect_dir: 嫌疑目标移动到的复查目录 (M/)
        """
        base_path = Path(input_dir)
        m_path = Path(suspect_dir)
        
        # 确保 M 文件夹存在
        m_path.mkdir(parents=True, exist_ok=True)

        if not base_path.exists():
            print(f"错误：输入目录 {input_dir} 不存在，请检查路径！")
            return

        # 遍历 K/ 下的所有文件夹
        folders = [d for d in base_path.iterdir() if d.is_dir()]
        total_folders = len(folders)
        moved_count = 0
        
        print(f"\n🔍 启动【同号不同车】排查程序")
        print(f"原始数据目录: {input_dir}")
        print(f"复查隔离目录: {suspect_dir}")
        print(f"共发现 {total_folders} 个车牌文件夹待检")
        print(f"相似度及格线: {self.threshold} (裁掉上半 {self.crop_top_ratio*100}%)")
        print("="*60)

        for folder in tqdm(folders, desc="排查进度"):
            images = []
            for ext in ['*.jpg', '*.jpeg', '*.png', '*.bmp']:
                images.extend(list(folder.glob(ext)))
            
            # 只有1张图的没法比对，直接跳过
            if len(images) <= 1:
                continue
                
            features = {}
            for img_path in images:
                feat = self.extract_feature(img_path)
                if feat is not None:
                    features[img_path.name] = feat
            
            if len(features) < 2:
                continue

            # 寻找嫌疑对子
            is_suspect = False
            worst_pair = ("", "")
            lowest_sim = 1.0

            img_names = list(features.keys())
            for img1, img2 in combinations(img_names, 2):
                sim = F.cosine_similarity(features[img1], features[img2]).item()
                
                if sim < lowest_sim:
                    lowest_sim = sim
                    worst_pair = (img1, img2)
                
                # 一旦发现一对低于阈值的，直接判定为嫌疑文件夹
                if sim < self.threshold:
                    is_suspect = True

            # 如果是嫌疑文件夹，执行移动和重命名打标
            if is_suspect:
                tqdm.write(f"\n⚠️ 发现嫌疑车辆: {folder.name} (最低相似度: {lowest_sim:.3f})")
                
                target_folder = m_path / folder.name
                # 如果 M 里已经有这个文件夹了，先删掉避免冲突
                if target_folder.exists():
                    shutil.rmtree(target_folder)
                    
                try:
                    # 移动整个文件夹从 K/ 到 M/
                    shutil.move(str(folder), str(target_folder))
                    moved_count += 1
                    
                    # 移动过去之后，在 M/ 文件夹里找到最不像的那两张图，重命名打上分数
                    img1_path = target_folder / worst_pair[0]
                    img2_path = target_folder / worst_pair[1]
                    
                    if img1_path.exists():
                        new_name1 = f"{img1_path.stem}_sim{lowest_sim:.2f}_with_{img2_path.stem}{img1_path.suffix}"
                        img1_path.rename(target_folder / new_name1)
                        
                    tqdm.write(f"  👉 已移入 {suspect_dir} 并在文件名打上 {lowest_sim:.2f} 的分数标记")
                except Exception as e:
                    tqdm.write(f"  ❌ 移动文件夹失败: {e}")

        print("\n" + "="*60)
        print("🎉 预演排查完成！")
        print(f"总计检查: {total_folders} 个文件夹")
        print(f"移入嫌疑区 ({suspect_dir}/): {moved_count} 个文件夹")
        print(f"你可以前往 {suspect_dir} 目录查看带有 sim 分数标记的照片进行人工复核。")
        print("="*60)

if __name__ == "__main__":
    # 根据你肉眼的观察，调整阈值（0.70 到 0.75 之间微调）
    cleaner = SamePlateDifferentCarCleaner(threshold=0.70, crop_top_ratio=0.5)
    
    # 明确设置输入输出路径
    INPUT_DIR = "./K" 
    M_DIR = "./M"
    
    cleaner.process_dataset(INPUT_DIR, M_DIR)

搞定了车牌文字标注的问题，我们发现由于早期电动车标准不规范，同时存在绿牌和白牌的情况，而不同颜色的牌照在不同的车上竟然有同一个号码。

同一个身份（PID）下混入了不同人的照片，这是度量学习（如中心损失和三元组损失）极其不能接受的点，它会直接撕裂模型在特征空间中的聚类。

为了找出这些“伪装者”，我们编写了 clean_same.py。我们的原则是“宁可错杀一千，也不放过一个”。只要找到同牌不同车的，我们就把这个车牌号的整个文件夹删除。

1. 核心物理隔离：砍掉上半身

如果我们直接把整张图片送进模型比对，会遇到一个致命的干扰项：人。

• 如果是同一个骑手，今天穿红衣服，明天穿黄衣服，模型可能会认为这是两辆不同的车；
• 反之，如果是两个不同的人骑了相似的车，也可能引发误判。

为了让比对纯粹聚焦在“车辆”本身，我们在提取特征前做了一个极其有效的物理裁剪：

# ✂️ 核心物理隔离：砍掉上半身（人），只留车尾
top = int(h * self.crop_top_ratio)
img_cropped = img.crop((0, top, w, h))

通过设置 crop_top_ratio=0.5，我们直接将图片的下半部分（车身、车牌、后轮）裁切出来，从而在物理层面隔绝了人员着装带来的特征干扰。

2. 借用预训练模型的火眼金睛

这里我们不需要从头训练任何东西。我们直接加载了 torchvision 官方在 ImageNet 上预训练的 ResNet50 骨干网络（ResNet50_Weights.IMAGENET1K_V1）。

我们将最后的分类层剥离：

self.model.fc = torch.nn.Identity()

然后直接提取出经过 L_2 归一化的特征向量。

3. 寻找最可疑的对子

接着，我们在同一个车牌文件夹内，对所有图片进行两两组合（combinations），并计算它们特征向量的余弦相似度（F.cosine_similarity）。

sim = F.cosine_similarity(features[img1], features[img2]).item()
if sim < self.threshold:
    is_suspect = True

如果在这个文件夹里，发现任何一对图片的相似度低于我们设定的及格线（脚本预设为 0.70），系统就会直接拉响警报：判定这个文件夹为“嫌疑车辆”。

随后，脚本会将这个有嫌疑的文件夹整个移动到复查目录（如 ./M）中，并把那一对相似度最低的“罪魁祸首”图片重命名，将它们的相似度分数（如 sim0.65）直接打在文件名上。这样人工复核时，一眼就能看出哪两张图出了问题。

结语

通过 ocr.py 筛除车牌不符的错标数据，再通过 clean_same.py 剔除同号不同车的脏数据，我们终于为即将到来的大模型训练铺平了道路。

在深度学习的工程实践中，数据清洗往往占据了 70% 的工作量。磨刀不误砍柴工，拥有了一个高质量的电动车特征数据集，我们在下一篇真正切入电动车重识别模型微调时，才能做到游刃有余。