模糊、黑屏、白屏、水印、曝光异常、尺寸异常、重复图……这些"脏图"会直接拉低模型上限。本页带你用 OpenCV + imagehash 搭建一套可落地的图像质检流水线。
清洗的第一步是知道什么样的图算"脏"。下面是一组真实项目中常见的脏图样本（红框）与正常对照图（绿框）。后面的每一章，就对应解决其中一类问题。
UTF-8 带 BOM（Excel 打开不乱码），两列：filename, issue。可以先用 pandas 读一遍练练手。
模糊图没有清晰的边缘和纹理，模型从中几乎学不到有效特征，反而会引入噪声标签。采集端（爬虫、手机连拍、监控视频抽帧）最常出现这类脏图。
Laplacian 算子对图像求二阶导数，边缘越锐利响应越强。对整图 Laplacian 结果取方差：清晰图方差大，模糊图方差小。方差低于阈值即判模糊。
import cv2
def blur_score(img_path):
"""返回图像的 Laplacian 方差，越小越模糊"""
img = cv2.imread(img_path, cv2.IMREAD_GRAYSCALE) # 直接读成灰度图
if img is None:
return 0.0 # 读图失败按脏图处理
return cv2.Laplacian(img, cv2.CV_64F).var()
# 经验阈值：一般场景 100 起步
# 证件照/商品图等高质量数据集可提到 200~300
# 监控/夜景等天然偏软的图可降到 60~80
score = blur_score("img_001.jpg")
if score < 100:
print(f"模糊图，Laplacian方差={score:.1f}，剔除")
摄像头遮挡、解码失败、生成图崩溃，都会产生全黑或全白的"废图"。它们没有信息，还会让训练 loss 异常波动。
黑/白屏图的本质是像素值极端集中：灰度均值贴近 0（黑）或 255（白），且标准差极小。用"均值 + 标准差"双条件判定，比只看均值更稳（能保住夜景、雪景这类正常图）。
import cv2
import numpy as np
def is_black_or_white(img_path):
"""判定黑屏/白屏图：均值极端 + 标准差极小"""
img = cv2.imread(img_path, cv2.IMREAD_GRAYSCALE)
if img is None:
return True
mean_val = np.mean(img) # 灰度均值：0=纯黑，255=纯白
std_val = np.std(img) # 标准差：越小说明像素越"抱团"
# 黑屏：很暗且几乎没有纹理起伏
if mean_val < 15 and std_val < 10:
return True
# 白屏：很亮且几乎没有纹理起伏
if mean_val > 240 and std_val < 10:
return True
return False
爬虫抓来的图床/素材站图片常带半透明水印。模型会把水印文字当成目标特征学进去——比如"所有狗的照片上都有几个白色字母"，造成严重干扰。
两种轻量思路：① 固定区域边缘密度——水印位置固定（如四角、中央），对该区域做 Canny 边缘检测，文字边缘密度异常高则报警；② 模板匹配——已知水印样式时直接 matchTemplate。
import cv2
import numpy as np
def watermark_score(img_path):
"""中心区域边缘密度：文字水印会让密度异常升高"""
img = cv2.imread(img_path, cv2.IMREAD_GRAYSCALE)
if img is None:
return 1.0
h, w = img.shape
# 裁剪中央 40% 区域（水印最常见的位置）
roi = img[int(h*0.3):int(h*0.7), int(w*0.3):int(w*0.7)]
edges = cv2.Canny(roi, 50, 150) # 边缘检测
return np.mean(edges > 0) # 边缘像素占比
def match_known_watermark(img_path, wm_path):
"""已知水印模板时：归一化互相关匹配，越接近 1 越像"""
img = cv2.imread(img_path, cv2.IMREAD_GRAYSCALE)
wm = cv2.imread(wm_path, cv2.IMREAD_GRAYSCALE)
res = cv2.matchTemplate(img, wm, cv2.TM_CCOEFF_NORMED)
return res.max()
# 经验值：边缘密度 > 0.15 怀疑有文字水印；模板匹配 > 0.6 基本实锤
if watermark_score("img_005.jpg") > 0.15:
print("疑似水印图，剔除或转人工复核")
和黑/白屏不同，曝光异常的图"有内容但看不清"：过曝丢失高光细节，欠曝丢失暗部细节。模型在这种图上标注都困难，训练更是事倍功半。
转 HSV 取 V 通道（亮度），看分位数而不是均值：过曝图的高分位数（如 95%）会顶到 255 附近且占比大；欠曝图的低分位数会贴 0。分位数比均值抗干扰，能保住"大部分正常、局部亮灯"的图。
import cv2
import numpy as np
def exposure_issue(img_path):
"""返回 None（正常）/ 'over'（过曝）/ 'under'（欠曝）"""
img = cv2.imread(img_path)
if img is None:
return 'under'
v = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)[:, :, 2] # 亮度通道
bright_ratio = np.mean(v > 245) # 接近纯白的像素占比
dark_ratio = np.mean(v < 10) # 接近纯黑的像素占比
mean_v = np.mean(v)
# 过曝：超过 40% 像素顶到接近纯白，且整体偏亮
if bright_ratio > 0.40 and mean_v > 180:
return 'over'
# 欠曝：超过 40% 像素贴黑，且整体偏暗
if dark_ratio > 0.40 and mean_v < 70:
return 'under'
return None
几十像素的小图标、缩略图放大后全是马赛克；长宽比 10:1 的条幅图 resize 到方形后内容完全变形。它们混进训练集只会教坏模型。
两条硬规则：① 最小边不低于模型输入尺寸（如 224）；② 长宽比限制在合理区间（如 1/4 ~ 4）。读取尺寸不需要解码像素，用 PIL 的 img.size 秒出结果，适合全库快速过一遍。
from PIL import Image
def size_issue(img_path, min_side=224, max_ratio=4.0):
"""返回 None（正常）或问题描述字符串"""
try:
with Image.open(img_path) as img:
w, h = img.size # 只读元信息，不解码像素，极快
except Exception:
return "文件损坏，无法打开"
if min(w, h) < min_side:
return f"尺寸过小（{w}x{h}，最小边不足 {min_side}px）"
ratio = max(w, h) / min(w, h)
if ratio > max_ratio:
return f"长宽比异常（{ratio:.1f}:1，超过 {max_ratio}:1）"
return None
# 经验值：分类任务 min_side 取模型输入边长；检测任务建议 >= 416
print(size_issue("img_007.jpg"))
同一批图被反复爬取、resize、转存后，MD5 全变了但内容一样。重复样本会让模型"背答案"，还会污染训练/验证集划分，让指标虚高。
用 imagehash 的 pHash（感知哈希）：先 DCT 提取低频特征再生成 64 位指纹，对缩放、压缩、轻微调色都鲁棒。两图哈希的汉明距离 ≤ 5 即判重复。去重后统一转 RGB、统一尺寸、统一存 jpg/png。
from PIL import Image
import imagehash
def phash_dedup(img_paths, threshold=5):
"""pHash 去重：返回保留列表与重复列表"""
seen = {} # 哈希 -> 首次出现的文件名
keep, dup = [], []
for p in img_paths:
h = imagehash.phash(Image.open(p))
# 与已有哈希两两比较（大规模数据请换 BK-Tree / 向量检索）
hit = next((k for k in seen if h - k <= threshold), None)
if hit is None:
seen[h] = p
keep.append(p)
else:
dup.append((p, seen[hit], h - hit)) # 记录它和谁重复、距离多少
return keep, dup
def normalize(img_path, save_path, size=(512, 512)):
"""格式统一：转 RGB -> resize -> 统一存 JPG"""
img = Image.open(img_path).convert("RGB") # 去掉 alpha 通道/灰度模式
img = img.resize(size, Image.LANCZOS)
img.save(save_path, "JPEG", quality=95)
# 经验值：汉明距离 0=完全一样；<=5 重复；6~10 高度相似（建议人工看）；>10 不同
下面左侧是模糊+欠曝的脏图（模拟），右侧是清晰原图。拖动滑块观察：模拟的 Laplacian 方差如何从「应剔除」区间爬升到「健康」区间——这就是 S1 阈值判定的直观含义。
把前面六项检测串成一条流水线：遍历文件夹 → 逐项质检 → 合格图移入 clean/、脏图移入 reject/，并输出一份质检报告 CSV。先出报告、再动文件——真实项目里建议先 dry-run 看报告，确认阈值合理后再执行移动。
"""
图像数据批量质检流水线
用法：python image_qc_pipeline.py --src ./raw_images --report qc_report.csv
"""
import os
import csv
import shutil
import argparse
import cv2
import numpy as np
from PIL import Image
import imagehash
# ---------- 各检测项（阈值均为经验起点，需按数据集标定） ----------
def check_blur(gray):
"""模糊检测：Laplacian 方差过低"""
return "模糊" if cv2.Laplacian(gray, cv2.CV_64F).var() < 100 else None
def check_blackwhite(gray):
"""黑屏/白屏检测：均值极端且标准差极小"""
m, s = np.mean(gray), np.std(gray)
if m < 15 and s < 10:
return "黑屏"
if m > 240 and s < 10:
return "白屏过曝"
return None
def check_watermark(gray):
"""水印初筛：中心区域边缘密度异常"""
h, w = gray.shape
roi = gray[int(h*0.3):int(h*0.7), int(w*0.3):int(w*0.7)]
density = np.mean(cv2.Canny(roi, 50, 150) > 0)
return "疑似水印" if density > 0.15 else None
def check_exposure(bgr):
"""曝光检测：过曝/欠曝看顶格像素占比 + 整体亮度"""
v = cv2.cvtColor(bgr, cv2.COLOR_BGR2HSV)[:, :, 2]
if np.mean(v > 245) > 0.40 and np.mean(v) > 180:
return "过曝"
if np.mean(v < 10) > 0.40 and np.mean(v) < 70:
return "欠曝"
return None
def check_size(w, h):
"""尺寸检测：最小边与长宽比"""
if min(w, h) < 224:
return f"尺寸过小({w}x{h})"
if max(w, h) / min(w, h) > 4.0:
return "长宽比异常"
return None
# ---------- 主流程 ----------
def run(src_dir, report_path, dry_run=True):
clean_dir = os.path.join(src_dir, "clean")
reject_dir = os.path.join(src_dir, "reject")
os.makedirs(clean_dir, exist_ok=True)
os.makedirs(reject_dir, exist_ok=True)
exts = (".jpg", ".jpeg", ".png", ".webp", ".bmp")
files = [f for f in os.listdir(src_dir) if f.lower().endswith(exts)]
rows, seen_hash = [], {} # seen_hash 用于 pHash 去重
for fname in sorted(files):
path = os.path.join(src_dir, fname)
issues = []
try:
pil_img = Image.open(path)
w, h = pil_img.size
except Exception:
issues.append("文件损坏")
w = h = 0
if not issues:
# ① 尺寸（最快，先跑）
r = check_size(w, h)
if r: issues.append(r)
# ② 像素级检测
bgr = cv2.imread(path)
if bgr is None:
issues.append("解码失败")
else:
gray = cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY)
for check in (check_blur, check_blackwhite, check_watermark):
r = check(gray)
if r: issues.append(r)
r = check_exposure(bgr)
if r: issues.append(r)
# ③ pHash 去重（与已保留图比较）
hcode = imagehash.phash(pil_img.convert("RGB"))
hit = next((k for k in seen_hash if hcode - k <= 5), None)
if hit is not None:
issues.append(f"重复图(与{seen_hash[hit]}重复)")
elif not issues:
seen_hash[hcode] = fname # 只把合格图计入哈希库
verdict = "PASS" if not issues else "REJECT"
rows.append([fname, verdict, ";".join(issues)])
if not dry_run: # dry-run 只出报告，确认无误后再移动文件
dst = clean_dir if not issues else reject_dir
shutil.move(path, os.path.join(dst, fname))
# 输出质检报告（BOM 让 Excel 直接打开不乱码）
with open(report_path, "w", newline="", encoding="utf-8-sig") as f:
writer = csv.writer(f)
writer.writerow(["filename", "verdict", "issues"])
writer.writerows(rows)
n_bad = sum(1 for r in rows if r[1] == "REJECT")
print(f"质检完成：共 {len(rows)} 张，剔除 {n_bad} 张，报告 -> {report_path}")
if __name__ == "__main__":
ap = argparse.ArgumentParser()
ap.add_argument("--src", default="./raw_images", help="原始图片目录")
ap.add_argument("--report", default="qc_report.csv", help="质检报告输出路径")
ap.add_argument("--move", action="store_true", help="真正移动文件（默认只出报告）")
args = ap.parse_args()
run(args.src, args.report, dry_run=not args.move)
--move 跑一遍看报告，抽 20 张 REJECT 人工确认没有误杀；② 阈值全部抽成配置项（yaml/argparse），换数据集只改配置不改代码；③ 图库超 10 万张时用多进程（multiprocessing.Pool）并行检测，pHash 比较换 BK-Tree。