ResNet101人脸检测实战案例：电商直播截图中主播人脸自动裁剪与质量评估流程

1. 引言：直播截图处理的痛点与机遇

如果你做过电商直播运营，肯定遇到过这样的场景：一场直播下来，后台自动生成了成百上千张截图。这些截图里，有主播讲解产品的精彩瞬间，有用户互动的有趣画面，也有各种角度、各种光线下的主播形象。

传统做法是什么？运营人员需要一张张翻看，手动把主播的脸部区域裁剪出来，然后判断哪些图片质量好，可以用作封面、宣传素材或者数据分析。这个过程不仅耗时耗力，而且标准不统一——不同的人裁剪的尺寸可能不一样，对“质量好”的判断也可能有差异。

现在，有了基于ResNet101的MogFace人脸检测工具，这一切都可以自动化完成。这个工具能在各种复杂环境下精准找到人脸位置，无论是主播侧身讲解、低头看产品，还是在光线变化、背景杂乱的情况下，都能稳定工作。

本文将带你一步步实现一个完整的电商直播截图处理流程：从批量上传截图，到自动检测并裁剪出所有人脸，再到对裁剪后的人脸图片进行质量评估，最终筛选出高质量的素材。整个过程完全自动化，让你从繁琐的手工操作中解放出来。

2. 工具准备与环境搭建

2.1 核心工具介绍

我们使用的核心工具是MogFace人脸检测模型，它基于ResNet101骨干网络，在CVPR 2022会议上发表。这个模型最大的特点就是“稳”——在各种复杂场景下都能保持高精度的人脸检测能力。

具体来说，它能处理：

• 大角度旋转：主播侧身、回头等姿势
• 部分遮挡：手持产品遮挡部分脸部
• 极小尺寸：远距离拍摄的小人脸
• 密集场景：多人同框的直播画面

工具本身提供了一个Streamlit界面，但我们今天要做的不是简单的单张图片检测，而是构建一个完整的批量处理流水线。

2.2 环境配置步骤

首先确保你的环境已经准备好：

# 创建项目目录
mkdir live-stream-face-processor
cd live-stream-face-processor

# 安装核心依赖
pip install modelscope opencv-python torch streamlit pillow numpy pandas

接下来，我们需要准备模型文件。根据工具要求，模型权重需要放在特定路径：

import os

# 创建模型目录
model_dir = "/root/ai-models/iic/cv_resnet101_face-detection_cvpr22papermogface"
os.makedirs(model_dir, exist_ok=True)

# 检查模型文件是否存在
model_files = ["configuration.json", "pytorch_model.bin"]
for file in model_files:
    file_path = os.path.join(model_dir, file)
    if not os.path.exists(file_path):
        print(f"请将{file}文件放置到: {model_dir}")

如果你没有现成的模型文件，可以通过ModelScope的API直接加载：

from modelscope.pipelines import pipeline
from modelscope.utils.constant import Tasks

# 创建人脸检测pipeline
face_detection = pipeline(
    task=Tasks.face_detection,
    model='damo/cv_resnet101_face-detection_cvpr22papermogface'
)

3. 批量人脸检测与裁剪实现

3.1 直播截图批量处理流程

电商直播的截图通常有以下几个特点：

1. 数量多：一场直播可能生成上千张截图
2. 尺寸统一：通常来自同一个直播平台，尺寸一致
3. 内容重复：很多连续截图内容相似
4. 质量参差：光线、角度、清晰度各不相同

我们的处理流程需要针对这些特点进行优化。下面是一个完整的批量处理脚本：

import cv2
import os
import json
from pathlib import Path
from tqdm import tqdm
import numpy as np

class LiveStreamFaceProcessor:
    def __init__(self, model_pipeline, output_dir="./processed"):
        self.model = model_pipeline
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(exist_ok=True)
        
        # 创建子目录
        self.faces_dir = self.output_dir / "cropped_faces"
        self.faces_dir.mkdir(exist_ok=True)
        
        self.metadata_dir = self.output_dir / "metadata"
        self.metadata_dir.mkdir(exist_ok=True)
    
    def process_single_image(self, image_path, min_confidence=0.8):
        """处理单张图片，返回检测到的人脸信息"""
        # 读取图片
        image = cv2.imread(str(image_path))
        if image is None:
            print(f"无法读取图片: {image_path}")
            return []
        
        # 转换颜色空间（BGR转RGB）
        image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        
        # 人脸检测
        result = self.model(image_rgb)
        
        faces_info = []
        if 'boxes' in result:
            for i, box in enumerate(result['boxes']):
                confidence = result['scores'][i] if 'scores' in result else 1.0
                
                # 过滤低置信度的人脸
                if confidence < min_confidence:
                    continue
                
                # 获取坐标（模型输出可能是不同格式）
                x1, y1, x2, y2 = self._parse_box_coordinates(box)
                
                # 确保坐标在图片范围内
                h, w = image.shape[:2]
                x1, y1 = max(0, int(x1)), max(0, int(y1))
                x2, y2 = min(w, int(x2)), min(h, int(y2))
                
                # 裁剪人脸
                face_img = image[y1:y2, x1:x2]
                
                # 保存裁剪后的人脸
                face_filename = f"{image_path.stem}_face_{i}.jpg"
                face_path = self.faces_dir / face_filename
                cv2.imwrite(str(face_path), face_img)
                
                # 收集人脸信息
                face_info = {
                    'image_name': image_path.name,
                    'face_index': i,
                    'confidence': float(confidence),
                    'bbox': [x1, y1, x2, y2],
                    'face_size': (x2 - x1) * (y2 - y1),
                    'face_path': str(face_path.relative_to(self.output_dir))
                }
                faces_info.append(face_info)
        
        return faces_info
    
    def _parse_box_coordinates(self, box):
        """解析边界框坐标，适配不同格式"""
        if isinstance(box, list) and len(box) == 4:
            return box
        elif hasattr(box, 'tolist'):
            return box.tolist()
        else:
            # 默认格式 [x1, y1, x2, y2]
            return box
    
    def process_batch(self, image_dir, min_confidence=0.8):
        """批量处理目录中的所有图片"""
        image_dir = Path(image_dir)
        image_files = list(image_dir.glob("*.jpg")) + list(image_dir.glob("*.png"))
        
        all_faces_info = []
        
        print(f"开始处理 {len(image_files)} 张图片...")
        for image_file in tqdm(image_files, desc="处理进度"):
            faces_info = self.process_single_image(image_file, min_confidence)
            all_faces_info.extend(faces_info)
            
            # 每处理10张图片保存一次元数据（防止程序中断丢失数据）
            if len(all_faces_info) % 10 == 0:
                self._save_metadata(all_faces_info)
        
        # 最终保存元数据
        self._save_metadata(all_faces_info)
        
        print(f"处理完成！共检测到 {len(all_faces_info)} 个人脸")
        print(f"人脸图片保存在: {self.faces_dir}")
        
        return all_faces_info
    
    def _save_metadata(self, faces_info):
        """保存元数据到JSON文件"""
        metadata_path = self.metadata_dir / "faces_metadata.json"
        with open(metadata_path, 'w', encoding='utf-8') as f:
            json.dump(faces_info, f, ensure_ascii=False, indent=2)

3.2 实际应用示例

假设你有一场电商直播的所有截图，都放在 live_screenshots 文件夹里：

# 使用示例
if __name__ == "__main__":
    # 初始化人脸检测pipeline
    from modelscope.pipelines import pipeline
    from modelscope.utils.constant import Tasks
    
    print("正在加载人脸检测模型...")
    face_detection = pipeline(
        task=Tasks.face_detection,
        model='damo/cv_resnet101_face-detection_cvpr22papermogface'
    )
    
    # 创建处理器
    processor = LiveStreamFaceProcessor(face_detection)
    
    # 批量处理直播截图
    screenshots_dir = "./live_screenshots"  # 你的截图目录
    faces_info = processor.process_batch(screenshots_dir, min_confidence=0.7)
    
    # 查看处理结果
    print(f"\n处理统计:")
    print(f"- 总人脸数: {len(faces_info)}")
    
    # 按置信度排序
    high_confidence_faces = [f for f in faces_info if f['confidence'] > 0.9]
    print(f"- 高置信度人脸(>0.9): {len(high_confidence_faces)}")
    
    # 按人脸大小排序
    large_faces = [f for f in faces_info if f['face_size'] > 10000]  # 面积大于10000像素
    print(f"- 大人脸(>10000像素): {len(large_faces)}")

这个脚本会自动：

1. 读取 live_screenshots 目录下的所有图片
2. 用MogFace模型检测每张图片中的人脸
3. 根据置信度过滤（默认0.7以上）
4. 裁剪出每个人脸并保存
5. 记录所有人脸的元数据（位置、大小、置信度等）

4. 人脸质量评估体系

4.1 为什么要评估人脸质量？

裁剪出人脸只是第一步，更重要的是判断哪些人脸图片是“高质量”的。在电商直播场景中，高质量的人脸图片应该具备：

1. 清晰度高：主播面部特征清晰可见
2. 角度合适：正面或微侧面，便于识别
3. 光线良好：不过曝也不欠曝
4. 表情自然：主播表情符合宣传需求
5. 尺寸足够：人脸在图片中占比适中

手动评估这些标准既主观又低效。我们可以用计算机视觉的方法来自动评估。

4.2 多维度质量评估实现

下面是一个完整的人脸质量评估类，它从多个维度给人脸图片打分：

import cv2
import numpy as np
from scipy import ndimage
from pathlib import Path

class FaceQualityAssessor:
    def __init__(self):
        # 初始化人脸关键点检测器（用于评估姿态）
        # 这里使用dlib，你也可以用其他方法
        try:
            import dlib
            self.detector = dlib.get_frontal_face_detector()
            predictor_path = "shape_predictor_68_face_landmarks.dat"
            self.predictor = dlib.shape_predictor(predictor_path)
            self.has_dlib = True
        except:
            print("未安装dlib，将使用简化版质量评估")
            self.has_dlib = False
    
    def assess_single_face(self, face_image_path):
        """评估单张人脸图片的质量"""
        # 读取图片
        img = cv2.imread(str(face_image_path))
        if img is None:
            return {"error": "无法读取图片"}
        
        # 转换为灰度图用于部分计算
        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
        
        # 计算各项质量指标
        scores = {
            'brightness_score': self._calculate_brightness_score(img),
            'contrast_score': self._calculate_contrast_score(gray),
            'sharpness_score': self._calculate_sharpness_score(gray),
            'face_size_score': self._calculate_face_size_score(img),
            'aspect_ratio_score': self._calculate_aspect_ratio_score(img),
        }
        
        # 如果有dlib，计算姿态分数
        if self.has_dlib:
            scores['pose_score'] = self._calculate_pose_score(img)
        
        # 计算综合分数（加权平均）
        weights = {
            'brightness_score': 0.2,
            'contrast_score': 0.2,
            'sharpness_score': 0.3,
            'face_size_score': 0.15,
            'aspect_ratio_score': 0.15,
        }
        
        if 'pose_score' in scores:
            weights['pose_score'] = 0.2
            # 调整其他权重
            for key in weights:
                if key != 'pose_score':
                    weights[key] *= 0.8
        
        total_score = sum(scores[key] * weights.get(key, 0) for key in scores)
        scores['total_score'] = total_score
        
        # 质量等级
        if total_score >= 0.8:
            scores['quality_level'] = '优秀'
        elif total_score >= 0.6:
            scores['quality_level'] = '良好'
        elif total_score >= 0.4:
            scores['quality_level'] = '一般'
        else:
            scores['quality_level'] = '较差'
        
        return scores
    
    def _calculate_brightness_score(self, img):
        """计算亮度分数"""
        # 转换为HSV色彩空间，取V通道（亮度）
        hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
        v_channel = hsv[:,:,2]
        
        # 计算平均亮度
        avg_brightness = np.mean(v_channel) / 255.0
        
        # 理想亮度在0.4-0.6之间
        if 0.4 <= avg_brightness <= 0.6:
            score = 1.0
        elif 0.3 <= avg_brightness <= 0.7:
            score = 0.8
        elif 0.2 <= avg_brightness <= 0.8:
            score = 0.6
        else:
            score = 0.3
        
        return score
    
    def _calculate_contrast_score(self, gray_img):
        """计算对比度分数"""
        # 计算局部对比度（使用标准差）
        contrast = np.std(gray_img) / 255.0
        
        # 理想对比度在0.2-0.4之间
        if 0.2 <= contrast <= 0.4:
            score = 1.0
        elif 0.15 <= contrast <= 0.45:
            score = 0.8
        elif 0.1 <= contrast <= 0.5:
            score = 0.6
        else:
            score = 0.3
        
        return score
    
    def _calculate_sharpness_score(self, gray_img):
        """计算清晰度分数（使用拉普拉斯方差）"""
        # 拉普拉斯算子可以检测边缘，方差越大说明图像越清晰
        laplacian_var = cv2.Laplacian(gray_img, cv2.CV_64F).var()
        
        # 归一化处理
        if laplacian_var > 1000:
            score = 1.0
        elif laplacian_var > 500:
            score = 0.8
        elif laplacian_var > 200:
            score = 0.6
        elif laplacian_var > 50:
            score = 0.4
        else:
            score = 0.2
        
        return score
    
    def _calculate_face_size_score(self, img):
        """计算人脸尺寸分数"""
        h, w = img.shape[:2]
        face_area = h * w
        
        # 理想人脸尺寸（根据经验值）
        # 电商直播中，人脸通常占图片较大比例
        if face_area > 40000:  # 200x200像素以上
            score = 1.0
        elif face_area > 10000:  # 100x100像素以上
            score = 0.8
        elif face_area > 2500:   # 50x50像素以上
            score = 0.6
        else:
            score = 0.3
        
        return score
    
    def _calculate_aspect_ratio_score(self, img):
        """计算宽高比分数"""
        h, w = img.shape[:2]
        aspect_ratio = w / h
        
        # 理想的人脸宽高比接近1:1（正面）
        # 但允许一定的范围
        if 0.8 <= aspect_ratio <= 1.2:
            score = 1.0
        elif 0.7 <= aspect_ratio <= 1.3:
            score = 0.8
        elif 0.6 <= aspect_ratio <= 1.4:
            score = 0.6
        else:
            score = 0.3
        
        return score
    
    def _calculate_pose_score(self, img):
        """计算姿态分数（使用dlib检测关键点）"""
        try:
            import dlib
            
            # 检测人脸
            gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
            faces = self.detector(gray)
            
            if len(faces) == 0:
                return 0.5  # 未检测到关键点，给中间分数
            
            # 获取第一个检测到的人脸
            face = faces[0]
            landmarks = self.predictor(gray, face)
            
            # 计算眼睛、鼻子、嘴巴的位置关系来判断姿态
            # 这里简化处理，实际可以根据68个关键点计算更精确的姿态
            left_eye = (landmarks.part(36).x, landmarks.part(36).y)
            right_eye = (landmarks.part(45).x, landmarks.part(45).y)
            nose = (landmarks.part(30).x, landmarks.part(30).y)
            
            # 计算眼睛连线的角度（判断头部倾斜）
            eye_dx = right_eye[0] - left_eye[0]
            eye_dy = right_eye[1] - left_eye[1]
            eye_angle = np.degrees(np.arctan2(eye_dy, eye_dx))
            
            # 计算鼻子相对于眼睛中点的位置（判断俯仰）
            eye_center = ((left_eye[0] + right_eye[0]) / 2, 
                         (left_eye[1] + right_eye[1]) / 2)
            nose_offset = nose[1] - eye_center[1]
            
            # 综合判断姿态
            # 角度接近0度（水平），鼻子在眼睛下方适中位置为最佳
            angle_score = 1.0 - min(abs(eye_angle) / 30, 1.0)
            position_score = 1.0 - min(abs(nose_offset) / (img.shape[0] * 0.3), 1.0)
            
            total_pose_score = (angle_score + position_score) / 2
            return total_pose_score
            
        except Exception as e:
            print(f"姿态评估出错: {e}")
            return 0.5
    
    def assess_batch(self, faces_dir, output_csv="face_quality_scores.csv"):
        """批量评估人脸图片质量"""
        faces_dir = Path(faces_dir)
        face_files = list(faces_dir.glob("*.jpg")) + list(faces_dir.glob("*.png"))
        
        results = []
        
        print(f"开始评估 {len(face_files)} 张人脸图片质量...")
        for face_file in tqdm(face_files, desc="质量评估进度"):
            scores = self.assess_single_face(face_file)
            
            if 'error' not in scores:
                result = {
                    'face_file': face_file.name,
                    'total_score': scores['total_score'],
                    'quality_level': scores['quality_level'],
                    'brightness': scores['brightness_score'],
                    'contrast': scores['contrast_score'],
                    'sharpness': scores['sharpness_score'],
                    'face_size': scores['face_size_score'],
                    'aspect_ratio': scores['aspect_ratio_score'],
                }
                
                if 'pose_score' in scores:
                    result['pose'] = scores['pose_score']
                
                results.append(result)
        
        # 保存结果到CSV
        import pandas as pd
        df = pd.DataFrame(results)
        
        # 按总分排序
        df = df.sort_values('total_score', ascending=False)
        
        df.to_csv(output_csv, index=False, encoding='utf-8-sig')
        print(f"评估完成！结果已保存到: {output_csv}")
        
        # 打印统计信息
        print(f"\n质量分布:")
        quality_counts = df['quality_level'].value_counts()
        for level, count in quality_counts.items():
            percentage = count / len(df) * 100
            print(f"- {level}: {count}张 ({percentage:.1f}%)")
        
        return df

4.3 质量评估实战应用

现在我们可以把前面裁剪的人脸图片进行质量评估：

# 使用质量评估器
if __name__ == "__main__":
    # 初始化质量评估器
    assessor = FaceQualityAssessor()
    
    # 评估裁剪后的人脸图片
    faces_directory = "./processed/cropped_faces"
    quality_df = assessor.assess_batch(faces_directory)
    
    # 查看质量最好的10张人脸
    print("\n质量最好的10张人脸:")
    top_faces = quality_df.head(10)
    for idx, row in top_faces.iterrows():
        print(f"{row['face_file']}: 总分{row['total_score']:.3f} ({row['quality_level']})")
    
    # 筛选出高质量人脸（总分>0.7）
    high_quality_faces = quality_df[quality_df['total_score'] > 0.7]
    print(f"\n高质量人脸(>0.7): {len(high_quality_faces)}张")
    
    # 可以进一步筛选特定要求的人脸
    # 例如：清晰度高且尺寸足够的人脸
    sharp_and_large = quality_df[
        (quality_df['sharpness'] > 0.8) & 
        (quality_df['face_size'] > 0.8)
    ]
    print(f"清晰且尺寸大的人脸: {len(sharp_and_large)}张")

5. 完整工作流与自动化脚本

5.1 一键式处理流水线

把前面的所有功能整合起来，创建一个完整的自动化处理脚本：

import argparse
from pathlib import Path
import pandas as pd

class LiveStreamFacePipeline:
    def __init__(self, model_name='damo/cv_resnet101_face-detection_cvpr22papermogface'):
        """初始化完整的人脸处理流水线"""
        print("初始化人脸处理流水线...")
        
        # 加载人脸检测模型
        from modelscope.pipelines import pipeline
        from modelscope.utils.constant import Tasks
        
        self.face_detector = pipeline(
            task=Tasks.face_detection,
            model=model_name
        )
        
        # 初始化处理器和评估器
        self.processor = LiveStreamFaceProcessor(self.face_detector)
        self.assessor = FaceQualityAssessor()
        
        print("流水线初始化完成！")
    
    def run_full_pipeline(self, input_dir, output_base="./output", 
                         min_confidence=0.7, quality_threshold=0.6):
        """运行完整处理流程"""
        # 创建输出目录
        output_dir = Path(output_base)
        output_dir.mkdir(exist_ok=True)
        
        print("=" * 50)
        print("开始处理直播截图...")
        print("=" * 50)
        
        # 步骤1: 批量检测并裁剪人脸
        print("\n步骤1: 人脸检测与裁剪")
        print("-" * 30)
        faces_info = self.processor.process_batch(
            input_dir, 
            min_confidence=min_confidence
        )
        
        # 步骤2: 质量评估
        print("\n步骤2: 人脸质量评估")
        print("-" * 30)
        faces_dir = Path(self.processor.faces_dir)
        quality_df = self.assessor.assess_batch(
            faces_dir,
            output_csv=str(output_dir / "quality_scores.csv")
        )
        
        # 步骤3: 筛选高质量人脸
        print("\n步骤3: 筛选高质量人脸")
        print("-" * 30)
        high_quality_df = quality_df[quality_df['total_score'] > quality_threshold]
        
        # 创建高质量人脸目录
        high_quality_dir = output_dir / "high_quality_faces"
        high_quality_dir.mkdir(exist_ok=True)
        
        # 复制高质量人脸图片
        import shutil
        high_quality_files = []
        for _, row in high_quality_df.iterrows():
            src_file = faces_dir / row['face_file']
            dst_file = high_quality_dir / row['face_file']
            shutil.copy2(src_file, dst_file)
            high_quality_files.append(row['face_file'])
        
        # 步骤4: 生成处理报告
        print("\n步骤4: 生成处理报告")
        print("-" * 30)
        self._generate_report(
            input_dir, 
            faces_info, 
            quality_df, 
            high_quality_df,
            output_dir
        )
        
        print("\n" + "=" * 50)
        print("处理完成！")
        print("=" * 50)
        print(f"输入图片: {len(list(Path(input_dir).glob('*.jpg')))}张")
        print(f"检测到人脸: {len(faces_info)}个")
        print(f"高质量人脸: {len(high_quality_df)}个")
        print(f"输出目录: {output_dir}")
        
        return {
            'total_faces': len(faces_info),
            'high_quality_faces': len(high_quality_df),
            'output_dir': str(output_dir),
            'quality_csv': str(output_dir / "quality_scores.csv"),
            'high_quality_dir': str(high_quality_dir)
        }
    
    def _generate_report(self, input_dir, faces_info, quality_df, 
                        high_quality_df, output_dir):
        """生成处理报告"""
        report_path = output_dir / "processing_report.txt"
        
        with open(report_path, 'w', encoding='utf-8') as f:
            f.write("=" * 60 + "\n")
            f.write("电商直播截图人脸处理报告\n")
            f.write("=" * 60 + "\n\n")
            
            f.write("1. 处理概览\n")
            f.write("-" * 40 + "\n")
            f.write(f"输入目录: {input_dir}\n")
            f.write(f"输出目录: {output_dir}\n")
            f.write(f"处理时间: {pd.Timestamp.now()}\n\n")
            
            f.write("2. 人脸检测结果\n")
            f.write("-" * 40 + "\n")
            f.write(f"总人脸数: {len(faces_info)}\n")
            
            # 按置信度统计
            conf_groups = {'高(>0.9)': 0, '中(0.7-0.9)': 0, '低(<0.7)': 0}
            for face in faces_info:
                conf = face['confidence']
                if conf > 0.9:
                    conf_groups['高(>0.9)'] += 1
                elif conf > 0.7:
                    conf_groups['中(0.7-0.9)'] += 1
                else:
                    conf_groups['低(<0.7)'] += 1
            
            for group, count in conf_groups.items():
                if count > 0:
                    percentage = count / len(faces_info) * 100
                    f.write(f"- {group}置信度: {count}个 ({percentage:.1f}%)\n")
            
            f.write("\n3. 质量评估结果\n")
            f.write("-" * 40 + "\n")
            f.write(f"评估人脸数: {len(quality_df)}\n")
            
            # 质量分布
            quality_dist = quality_df['quality_level'].value_counts()
            for level, count in quality_dist.items():
                percentage = count / len(quality_df) * 100
                f.write(f"- {level}: {count}张 ({percentage:.1f}%)\n")
            
            f.write(f"\n平均质量分数: {quality_df['total_score'].mean():.3f}\n")
            f.write(f"最高质量分数: {quality_df['total_score'].max():.3f}\n")
            f.write(f"最低质量分数: {quality_df['total_score'].min():.3f}\n")
            
            f.write("\n4. 高质量人脸筛选\n")
            f.write("-" * 40 + "\n")
            f.write(f"筛选阈值: >0.6\n")
            f.write(f"高质量人脸数: {len(high_quality_df)}\n")
            f.write(f"筛选比例: {len(high_quality_df)/len(quality_df)*100:.1f}%\n\n")
            
            f.write("5. 文件输出\n")
            f.write("-" * 40 + "\n")
            f.write(f"所有人脸图片: {output_dir}/processed/cropped_faces/\n")
            f.write(f"高质量人脸图片: {output_dir}/high_quality_faces/\n")
            f.write(f"质量评分表: {output_dir}/quality_scores.csv\n")
            f.write(f"处理元数据: {output_dir}/processed/metadata/faces_metadata.json\n")
        
        print(f"处理报告已生成: {report_path}")

# 命令行接口
def main():
    parser = argparse.ArgumentParser(description='电商直播截图人脸处理流水线')
    parser.add_argument('input_dir', help='直播截图目录路径')
    parser.add_argument('--output_dir', default='./output', help='输出目录路径')
    parser.add_argument('--min_confidence', type=float, default=0.7, 
                       help='人脸检测最小置信度 (默认: 0.7)')
    parser.add_argument('--quality_threshold', type=float, default=0.6,
                       help='高质量人脸筛选阈值 (默认: 0.6)')
    
    args = parser.parse_args()
    
    # 运行完整流水线
    pipeline = LiveStreamFacePipeline()
    results = pipeline.run_full_pipeline(
        input_dir=args.input_dir,
        output_base=args.output_dir,
        min_confidence=args.min_confidence,
        quality_threshold=args.quality_threshold
    )
    
    print(f"\n处理结果摘要:")
    print(f"- 总人脸数: {results['total_faces']}")
    print(f"- 高质量人脸: {results['high_quality_faces']}")
    print(f"- 质量评分表: {results['quality_csv']}")
    print(f"- 高质量人脸目录: {results['high_quality_dir']}")

if __name__ == "__main__":
    main()

5.2 使用方式

保存上面的代码为 live_face_pipeline.py，然后通过命令行运行：

# 基本用法
python live_face_pipeline.py ./live_screenshots

# 自定义输出目录和质量阈值
python live_face_pipeline.py ./live_screenshots --output_dir ./my_results --quality_threshold 0.7

# 调整人脸检测置信度
python live_face_pipeline.py ./live_screenshots --min_confidence 0.8

这个脚本会自动完成所有工作：

1. 检测并裁剪所有人脸
2. 评估每个人脸图片的质量
3. 筛选出高质量的人脸
4. 生成详细的处理报告

6. 总结

通过本文的实战案例，我们实现了一个完整的电商直播截图人脸处理流程。这个流程的核心价值在于：

自动化程度高：从原始的直播截图到最终筛选出的高质量人脸图片，整个过程完全自动化，无需人工干预。

处理质量稳定：基于ResNet101的MogFace模型在各种复杂场景下都能稳定检测人脸，确保不会漏掉重要画面。

评估标准客观：通过多维度质量评估体系，用统一的标准筛选人脸图片，避免了主观判断的差异。

灵活可配置：你可以根据实际需求调整置信度阈值、质量评估权重、筛选标准等参数。

扩展性强：这个框架可以轻松扩展到其他应用场景，比如：

• 社交媒体图片中的人物提取
• 视频会议截图的表情分析
• 安防监控中的人脸检测与质量评估
• 证件照自动筛选与裁剪

在实际的电商直播运营中，这个工具可以帮助你：

1. 快速制作宣传素材：自动筛选出主播表情好、角度佳、清晰度高的图片
2. 分析主播表现：通过人脸出现的频率、角度变化等分析主播的活跃度
3. 优化直播效果：识别光线不足、角度不佳的截图，指导直播设备调整
4. 建立素材库：自动整理和分类所有直播中的人脸图片

整个方案基于开源的ModelScope平台和ResNet101 MogFace模型，你可以在自己的服务器上部署运行，完全掌控数据隐私。无论是处理几十张还是上万张截图，都能高效完成。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。