Python + MySQL + OpenCV 人脸识别门禁考勤系统
这是一套集人脸检测、识别、打卡记录与异常报警于一体的完整系统
🏗️ 系统架构概览
[摄像头] → [OpenCV人脸检测] → [特征提取(FaceNet/Dlib)] → [特征比对] ↓ ┌──────────────┐ │ MySQL数据库 │ │ - 员工信息 │ │ - 人脸特征 │ │ - 打卡记录 │ └──────────────┘ ↓ ┌──────────────┐ │ 异常报警 │ │ - 陌生人报警 │ │ - 多次失败 │ │ - 超时未签退 │ └──────────────┘📁 项目目录结构
face_attendance/ ├── main.py # 主程序入口 ├── face_recognition.py # 人脸识别核心模块 ├── database.py # MySQL数据库操作 ├── alarm.py # 异常报警模块 ├── config.py # 配置文件 ├── templates/ # Web管理界面模板 │ ├── index.html │ └── records.html ├── static/ # 静态资源 ├── known_faces/ # 已知员工照片 └── unknown_captures/ # 陌生人抓拍🗄️ 数据库设计 (MySQL)
-- 员工表 CREATE TABLE employees ( id INT AUTO_INCREMENT PRIMARY KEY, employee_id VARCHAR(20) UNIQUE NOT NULL, -- 工号 name VARCHAR(50) NOT NULL, department VARCHAR(50), phone VARCHAR(20), photo_path VARCHAR(255), -- 注册照片路径 face_feature BLOB, -- 128维特征向量（二进制存储） registered_at DATETIME DEFAULT CURRENT_TIMESTAMP, status TINYINT DEFAULT 1 -- 1在职 0离职 ); -- 打卡记录表 CREATE TABLE attendance_records ( id BIGINT AUTO_INCREMENT PRIMARY KEY, employee_id VARCHAR(20), check_time DATETIME DEFAULT CURRENT_TIMESTAMP, direction ENUM('in', 'out'), -- 进/出 confidence FLOAT, -- 识别置信度 snapshot_path VARCHAR(255), -- 抓拍照片路径 device_id VARCHAR(50), -- 设备编号 INDEX idx_emp_time (employee_id, check_time) ); -- 异常日志表 CREATE TABLE anomaly_logs ( id BIGINT AUTO_INCREMENT PRIMARY KEY, event_type ENUM('unknown_face', 'low_confidence', 'multiple_fail', 'abnormal_time'), description TEXT, snapshot_path VARCHAR(255), occurred_at DATETIME DEFAULT CURRENT_TIMESTAMP, handled TINYINT DEFAULT 0 ); -- 考勤汇总表（每日统计） CREATE TABLE daily_summary ( id INT AUTO_INCREMENT PRIMARY KEY, employee_id VARCHAR(20), date DATE, first_check_in DATETIME, last_check_out DATETIME, total_hours DECIMAL(4,2), status ENUM('normal', 'late', 'early_leave', 'absent'), UNIQUE KEY uk_emp_date (employee_id, date) );🔧 核心代码实现
1. 人脸特征提取与比对 (face_recognition.py)
import cv2 import numpy as np import face_recognition # 基于dlib的封装库 from config import Config class FaceRecognizer: def __init__(self): self.known_encodings = [] # 已知员工特征列表 self.known_ids = [] # 对应工号 self.tolerance = 0.45 # 匹配阈值（越小越严格） def load_known_faces(self, db_cursor): """从数据库加载所有在职员工的人脸特征""" db_cursor.execute("SELECT employee_id, face_feature FROM employees WHERE status=1") rows = db_cursor.fetchall() for emp_id, feature_blob in rows: encoding = np.frombuffer(feature_blob, dtype=np.float64) self.known_encodings.append(encoding) self.known_ids.append(emp_id) def extract_encoding(self, image): """提取单张图片的人脸特征（128维向量）""" rgb_img = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) face_locations = face_recognition.face_locations(rgb_img) if len(face_locations) == 0: return None, None # 取最大人脸（离镜头最近的） largest = max(face_locations, key=lambda rect: (rect[2]-rect[0])*(rect[1]-rect[3])) encodings = face_recognition.face_encodings(rgb_img, [largest]) return encodings[0], largest def recognize(self, frame): """实时帧识别，返回 (employee_id, confidence, face_location)""" encoding, location = self.extract_encoding(frame) if encoding is None: return None, 0, None if len(self.known_encodings) == 0: return None, 0, location # 计算与所有已知人脸的距离 distances = face_recognition.face_distance(self.known_encodings, encoding) min_idx = np.argmin(distances) min_dist = distances[min_idx] if min_dist < self.tolerance: confidence = round((1 - min_dist) * 100, 2) return self.known_ids[min_idx], confidence, location else: return None, 0, location2. 主循环与打卡逻辑 (main.py)
import cv2 import time from datetime import datetime from database import Database from face_recognition import FaceRecognizer from alarm import AlarmSystem class AttendanceSystem: def __init__(self): self.db = Database() self.recognizer = FaceRecognizer() self.alarm = AlarmSystem() self.last_check_time = {} # 防止同一人短时间内重复打卡 self.cooldown = 5 # 冷却时间（秒） def initialize(self): """初始化：连接数据库、加载已知人脸、打开摄像头""" self.db.connect() cursor = self.db.get_cursor() self.recognizer.load_known_faces(cursor) self.cap = cv2.VideoCapture(0) # 0为默认摄像头 if not self.cap.isOpened(): raise Exception("无法打开摄像头") def process_frame(self, frame): """处理每一帧""" emp_id, confidence, face_loc = self.recognizer.recognize(frame) if emp_id: # 检查冷却时间 now = time.time() if emp_id in self.last_check_time and \ now - self.last_check_time[emp_id] < self.cooldown: return # 跳过 self.last_check_time[emp_id] = now # 判断进出方向（可根据人脸框位置或单独按钮决定） direction = self.determine_direction(frame, face_loc) # 保存抓拍快照 snapshot_path = self.save_snapshot(frame, emp_id) # 写入打卡记录 self.db.insert_record(emp_id, direction, confidence, snapshot_path) # 更新每日汇总 self.db.update_daily_summary(emp_id, direction) print(f"[打卡] {emp_id} {direction} 置信度:{confidence}%") else: # 陌生人检测 if face_loc: self.alarm.unknown_face_warning(frame, face_loc) def run(self): """主循环""" self.initialize() while True: ret, frame = self.cap.read() if not ret: break # 镜像翻转（更自然） frame = cv2.flip(frame, 1) # 处理 self.process_frame(frame) # 显示画面（可选） cv2.imshow("Attendance System", frame) if cv2.waitKey(1) & 0xFF == ord('q'): break self.cap.release() cv2.destroyAllWindows() self.db.close()3. 异常报警模块 (alarm.py)
import cv2 import smtplib import requests from email.mime.text import MIMEText from email.mime.image import MIMEImage from email.mime.multipart import MIMEMultipart from config import Config class AlarmSystem: def __init__(self): self.fail_count = {} # 连续识别失败计数 self.max_fails = 5 # 触发报警阈值 def unknown_face_warning(self, frame, face_loc): """陌生人报警：保存截图 + 发送通知""" # 保存陌生人截图 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") path = f"unknown_captures/unknown_{timestamp}.jpg" cv2.imwrite(path, frame) # 记录到异常日志 db.insert_anomaly('unknown_face', f"陌生人闯入 {path}", path) # 发送邮件/钉钉通知 self.send_email_alert("陌生人检测", path) self.send_dingtalk_alert(f"⚠️ 检测到陌生人！截图：{path}") def low_confidence_warning(self, emp_id, confidence): """低置信度报警""" db.insert_anomaly('low_confidence', f"员工{emp_id}置信度过低({confidence}%)") def send_email_alert(self, subject, image_path): """发送带截图的邮件""" msg = MIMEMultipart() msg['Subject'] = f"[门禁报警] {subject}" msg['From'] = Config.EMAIL_FROM msg['To'] = Config.EMAIL_TO body = MIMEText(f"时间：{datetime.now()}\n请立即查看监控。") msg.attach(body) with open(image_path, 'rb') as f: img = MIMEImage(f.read()) img.add_header('Content-ID', '<image1>') msg.attach(img) server = smtplib.SMTP(Config.SMTP_SERVER, Config.SMTP_PORT) server.login(Config.EMAIL_USER, Config.EMAIL_PASS) server.send_message(msg) server.quit() def send_dingtalk_alert(self, message): """钉钉机器人通知""" webhook = Config.DINGTALK_WEBHOOK data = {"msgtype": "text", "text": {"content": message}} requests.post(webhook, json=data)🌐 Web管理界面（Flask示例）
from flask import Flask, render_template, jsonify from database import Database app = Flask(__name__) @app.route('/') def index(): return render_template('index.html') @app.route('/api/today_records') def today_records(): db = Database() records = db.get_today_records() return jsonify(records) @app.route('/api/daily_summary') def daily_summary(): db = Database() summary = db.get_daily_summary() return jsonify(summary) @app.route('/api/anomalies') def anomalies(): db = Database() logs = db.get_recent_anomalies(limit=20) return jsonify(logs) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000, debug=True)前端使用ECharts展示当日打卡趋势、部门出勤率、异常事件分布等。
🚀 进阶功能与优化
功能 | 实现方式 |
|---|---|
活体检测 | 眨眼检测、头部转动、红外双目摄像头 |
多人同时识别 | 遍历所有人脸框，分别比对 |
口罩识别 | 训练口罩分类器，或使用眼部以上特征 |
离线模式 | 本地缓存特征库，断网自动切换 |
语音播报 | 识别成功后通过TTS播报姓名 |
远程开门 | 对接继电器控制电磁锁 |
📊 性能优化建议
特征预加载：启动时将所有人脸特征加载到内存字典
GPU加速：使用CUDA版本的dlib/OpenCV
异步写入：打卡记录通过消息队列（Redis/RabbitMQ）异步写入MySQL
定时清理：定期删除超过30天的抓拍截图
索引优化：
attendance_records表建立复合索引(employee_id, check_time)
