搭建本地人脸识别系统：解析FastAPI+InsightFace+FAISS全栈实现原理与优化方案26.6

import os
import cv2
import faiss
import numpy as np
import jwt
import redis
import pymysql
from fastapi import FastAPI, UploadFile, File, Form, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from insightface.app import FaceAnalysis
from datetime import datetime, timedelta

# ========== 全局配置 ==========
app = FastAPI(title="本地人脸登录系统")
# 跨域配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
# 托管静态前端页面
app.mount("/static", StaticFiles(directory="static"), name="static")

JWT_SECRET = "change-strong-secret-2026"
JWT_ALGORITHM = "HS256"
JWT_EXPIRE_SEC = 3600
FACE_THRESHOLD = 0.65
MAX_FAIL_COUNT = 5
LOCK_MINUTES = 10
DIM = 512

# Redis连接
r = redis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True)
# MySQL连接
def get_mysql_conn():
    conn = pymysql.connect(host="127.0.0.1", user="root", password="123456", database="daysurgery", autocommit=True)
    return conn

# 加载InsightFace人脸模型
face_app = FaceAnalysis(name="buffalo_l")
face_app.prepare(ctx_id=0, det_size=(640,640))

# 初始化FAISS向量库
index_file = "face_index.bin"
if os.path.exists(index_file):
    index = faiss.read_index(index_file)
    print(f"加载人脸向量库，总数据量：{index.ntotal}")
else:
    index = faiss.IndexIDMap2(faiss.IndexFlatL2(DIM))
    faiss.write_index(index, index_file)

# 人脸特征归一化核心示例
import numpy as np
# feat为InsightFace提取的512维人脸特征向量
feat = face.embedding.astype(np.float32)
# L2归一化，消除模长影响
feat = feat / np.linalg.norm(feat)
feat = feat.reshape(1, -1)

def get_face_feature(img_bgr):
    faces = face_app.get(img_bgr)
    if len(faces) == 0:
        raise HTTPException(status_code=400, detail="图片未检测到人脸")
    feat = faces[0].embedding.astype(np.float32)
    feat = feat / np.linalg.norm(feat)
    return feat.reshape(1, -1)

# FAISS索引初始化基础示例
import faiss
# 固定512维人脸向量
DIM = 512
# 创建带ID映射的平面L2索引
index = faiss.IndexIDMap2(faiss.IndexFlatL2(DIM))
# 加载本地持久化向量库
if os.path.exists("face_index.bin"):
    index = faiss.read_index("face_index.bin")
    print(f"加载人脸向量库，总数据量：{index.ntotal}")

# JWT令牌生成基础示例
import jwt
from datetime import datetime, timedelta
JWT_SECRET = "change-strong-secret-2026"
JWT_ALGORITHM = "HS256"
JWT_EXPIRE_SEC = 3600

def create_jwt(user_id: int, username: str):
    payload = {
        "sub": user_id,
        "username": username,
        "exp": datetime.utcnow() + timedelta(seconds=JWT_EXPIRE_SEC)
    }
    token = jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
    # Redis缓存令牌
    r.setex(f"login:token:{token}", JWT_EXPIRE_SEC, str(user_id))
    return token

// 画面灰度采样基础示例
function sampleGray() {
    mcv.width = 48;
    mcv.height = 36;
    mctx.drawImage(video, 0, 0, mcv.width, mcv.height);
    const d = mctx.getImageData(0, 0, mcv.width, mcv.height).data;
    let sum = 0;
    for (let i = 0; i < d.length; i += 4) 
        sum += (d[i] + d[i + 1] + d[i + 2]) / 3;
    return sum / (mcv.width * mcv.height);
}

<script>
const video = document.getElementById('video');
const mcv = document.getElementById('mcv');
const mctx = mcv.getContext('2d');
let lastGray = 0;
let preheatTime = 2500;
let ready = false;
let coolDown = 0;
const startTs = Date.now();

function detectMotion(){
    if(coolDown > Date.now()) return;
    const now = Date.now();
    if(now - startTs < preheatTime) return;
    const gray = sampleGray();
    const diff = Math.abs(gray - lastGray);
    lastGray = gray;
    if(diff > 2){
        // 触发刷脸请求
        fetch("/face_login", {
            method:"POST",
            body: getImageFormData()
        }).then(res=>res.json()).then(ret=>{
            if(ret.success){
                alert("登录成功");
                coolDown = Date.now() + 6000;
            }else{
                alert("失败："+ret.msg);
                coolDown = Date.now() + 3000;
            }
        })
    }
}
// 定时器，200ms采样一次
setInterval(detectMotion, 200);
</script>

@app.post("/face_register")
async def face_register(username: str = Form(...), file: UploadFile = File(...)):
    # 读取图片字节流转opencv图像
    data = await file.read()
    arr = np.frombuffer(data, np.uint8)
    img_bgr = cv2.imdecode(arr, cv2.IMREAD_COLOR)
    feat = get_face_feature(img_bgr)

    conn = get_mysql_conn()
    cur = conn.cursor()
    # 查询该用户旧人脸ID，存在则删除FAISS旧向量
    cur.execute("SELECT face_id FROM user WHERE username=%s", (username,))
    row = cur.fetchone()
    if row and row[0] is not None:
        old_faceid = row[0]
        index.remove_ids(np.array([old_faceid], dtype=np.int64))

    # 全局查重，防止人脸绑定其他账号
    D, I = index.search(feat, 1)
    if D[0][0] < FACE_THRESHOLD:
        match_faceid = int(I[0][0])
        cur.execute("SELECT username FROM user WHERE face_id=%s", (match_faceid,))
        r = cur.fetchone()
        if r:
            raise HTTPException(status_code=400, detail=f"该人脸已绑定账号：{r[0]}")

    # 生成新人脸ID，写入FAISS
    new_faceid = int(datetime.now().timestamp())
    index.add_with_ids(feat, np.array([new_faceid], dtype=np.int64))
    faiss.write_index(index, "face_index.bin")

    # 写入MySQL
    cur.execute("""
        INSERT INTO user(username,face_id,fail_count,lock_until) VALUES(%s,%s,0,NULL)
        ON DUPLICATE KEY UPDATE face_id=%s,fail_count=0,lock_until=NULL
    """, (username, new_faceid, new_faceid))
    cur.close()
    conn.close()
    return {"success":True, "msg":"人脸注册成功"}

@app.post("/face_login")
async def face_login(username: str|None = Form(None), file: UploadFile = File(...)):
    data = await file.read()
    arr = np.frombuffer(data, np.uint8)
    img_bgr = cv2.imdecode(arr, cv2.IMREAD_COLOR)
    feat = get_face_feature(img_bgr)
    conn = get_mysql_conn()
    cur = conn.cursor()

    target_faceid = None
    if username:
        # 指定用户名登录模式
        cur.execute("SELECT id,face_id,fail_count,lock_until FROM user WHERE username=%s",(username,))
        user = cur.fetchone()
        if not user:
            raise HTTPException(400, "用户不存在")
        uid, fid, fail_cnt, lock_time = user
        target_faceid = fid
        # 判断账号锁定
        if lock_time and datetime.now() < lock_time:
            raise HTTPException(400, "账号临时锁定，请稍后重试")
        if not fid:
            raise HTTPException(400, "该用户未注册人脸")
        # 只检索指定人脸
        D,I = index.search(feat,1)
        if int(I[0][0]) != fid or D[0][0] >= FACE_THRESHOLD:
            # 失败计数
            cur.execute("UPDATE user SET fail_count=fail_count+1 WHERE id=%s",(uid,))
            cur.execute("SELECT fail_count FROM user WHERE id=%s",(uid,))
            new_fail = cur.fetchone()[0]
            if new_fail >= MAX_FAIL_COUNT:
                lock_end = datetime.now() + timedelta(minutes=LOCK_MINUTES)
                cur.execute("UPDATE user SET lock_until=%s WHERE id=%s",(lock_end,uid))
            cur.close();conn.close()
            return {"success":False,"msg":"人脸匹配失败","distance":float(D[0][0])}
    else:
        # 全局刷脸模式
        D,I = index.search(feat,1)
        if D[0][0] >= FACE_THRESHOLD:
            return {"success":False,"msg":"未匹配到人脸","distance":float(D[0][0])}
        target_faceid = int(I[0][0])
        cur.execute("SELECT id,username,fail_count,lock_until FROM user WHERE face_id=%s",(target_faceid,))
        user = cur.fetchone()
        if not user:
            return {"success":False,"msg":"匹配到人脸，但无绑定账号"}
        uid,username,fail_cnt,lock_time = user
        if lock_time and datetime.now() < lock_time:
            raise HTTPException(400,"账号锁定")

    # 登录成功，重置失败次数，生成token
    cur.execute("UPDATE user SET fail_count=0,lock_until=NULL WHERE face_id=%s",(target_faceid,))
    token = create_jwt(uid, username)
    cur.close()
    conn.close()
    return {"success":True, "token":token, "username":username, "distance":float(D[0][0])}

@app.post("/logout")
async def logout(authorization:str|None = None):
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(401,"未携带令牌")
    token = authorization.replace("Bearer ","")
    exists = r.exists(f"login:token:{token}")
    if not exists:
        raise HTTPException(401,"令牌无效")
    r.delete(f"login:token:{token}")
    return {"success":True, "msg":"已登出"}

# 向量库手动保存接口
@app.post("/save_index")
async def save_index():
    faiss.write_index(index, "face_index.bin")
    return {"success":True, "total":index.ntotal}

<script>
const video = document.getElementById('video');
const canvas = document.getElementById('canvas');
const ctx = canvas.getContext('2d');
let useFile = false;

// 摄像头截图
function captureImage(){
    canvas.width = video.videoWidth;
    canvas.height = video.videoHeight;
    ctx.save();
    ctx.scale(-1,1);
    ctx.drawImage(video, -canvas.width,0, canvas.width, canvas.height);
    ctx.restore();
    return new Promise(resolve=>canvas.toBlob(resolve, "image/jpeg"));
}
// 注册按钮
document.getElementById("btnReg").onclick = async ()=>{
    const username = document.getElementById("username").value.trim();
    if(!username){alert("请输入用户名");return;}
    const blob = await captureImage();
    const fd = new FormData();
    fd.append("username", username);
    fd.append("file", blob, "face.jpg");
    const res = await fetch("/face_register", {method:"POST", body:fd});
    const ret = await res.json();
    alert(ret.msg || ret.detail);
}
</script>

pip install fastapi uvicorn insightface faiss-cpu pymysql redis opencv-python numpy pyjwt

CREATE DATABASE IF NOT EXISTS daysurgery;
USE daysurgery;
CREATE TABLE IF NOT EXISTS user(
    id INT AUTO_INCREMENT PRIMARY KEY,
    username VARCHAR(50) NOT NULL UNIQUE,
    face_id BIGINT NULL,
    fail_count INT DEFAULT 0,
    lock_until DATETIME NULL
);

uvicorn main:app --host 0.0.0.0 --port 8000