138 lines
4.7 KiB
Python
138 lines
4.7 KiB
Python
from flask import Flask, render_template, request, redirect, url_for
|
|
from flask_socketio import SocketIO, emit
|
|
import sqlite3
|
|
import os
|
|
import eventlet
|
|
import face_recognition
|
|
import numpy as np
|
|
|
|
app = Flask(__name__)
|
|
app.config['UPLOAD_FOLDER'] = './static/db_image/' # 设置文件上传路径
|
|
socketio = SocketIO(app, async_mode='eventlet')
|
|
|
|
def create_face_database(db_name="face_database.db"):
|
|
"""创建人脸数据库和匹配日志表"""
|
|
conn = sqlite3.connect(db_name)
|
|
c = conn.cursor()
|
|
c.execute('''CREATE TABLE IF NOT EXISTS faces
|
|
(id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL,
|
|
identity TEXT NOT NULL,
|
|
image_path TEXT NOT NULL,
|
|
encoding BLOB NOT NULL)''')
|
|
|
|
c.execute('''CREATE TABLE IF NOT EXISTS match_logs
|
|
(id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL,
|
|
identity TEXT NOT NULL,
|
|
image_path TEXT NOT NULL,
|
|
match_time TEXT NOT NULL)''')
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def add_face_to_database(name, identity, image_path, db_name="face_database.db"):
|
|
"""将人脸信息添加到数据库"""
|
|
conn = sqlite3.connect(db_name)
|
|
c = conn.cursor()
|
|
|
|
# 将图片路径转换为相对路径
|
|
relative_image_path = os.path.relpath(image_path, start='./static')
|
|
|
|
image = face_recognition.load_image_file(image_path)
|
|
face_encodings = face_recognition.face_encodings(image)
|
|
|
|
if face_encodings:
|
|
face_encoding = face_encodings[0]
|
|
encoding_blob = np.array(face_encoding).tobytes()
|
|
c.execute("INSERT INTO faces (name, identity, image_path, encoding) VALUES (?, ?, ?, ?)",
|
|
(name, identity, relative_image_path, encoding_blob))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
def get_match_logs(db_name="face_database.db"):
|
|
conn = sqlite3.connect(db_name)
|
|
c = conn.cursor()
|
|
c.execute("SELECT name, identity, image_path, match_time FROM match_logs")
|
|
logs = c.fetchall()
|
|
conn.close()
|
|
return logs
|
|
|
|
@app.route('/')
|
|
def index():
|
|
logs = get_match_logs()
|
|
return render_template('index.html', logs=logs)
|
|
|
|
@app.route('/info_person', methods=['GET', 'POST'])
|
|
def info_person():
|
|
conn = sqlite3.connect('face_database.db')
|
|
c = conn.cursor()
|
|
|
|
if request.method == 'POST':
|
|
# 添加新人员
|
|
if 'add' in request.form:
|
|
name = request.form['name']
|
|
identity = request.form['identity']
|
|
image = request.files['image_path']
|
|
|
|
if image:
|
|
image_path = os.path.join(app.config['UPLOAD_FOLDER'], image.filename)
|
|
image.save(image_path)
|
|
add_face_to_database(name, identity, image_path)
|
|
return redirect(url_for('info_person'))
|
|
|
|
# 更新人员信息
|
|
elif 'update' in request.form:
|
|
id = request.form['id']
|
|
name = request.form['name']
|
|
identity = request.form['identity']
|
|
image = request.files.get('image_path')
|
|
|
|
if image:
|
|
image_path = os.path.join(app.config['UPLOAD_FOLDER'], image.filename)
|
|
image.save(image_path)
|
|
|
|
# 使用face_recognition生成新的面部编码
|
|
loaded_image = face_recognition.load_image_file(image_path)
|
|
face_encodings = face_recognition.face_encodings(loaded_image)
|
|
|
|
if face_encodings:
|
|
encoding = np.array(face_encodings[0]).tobytes()
|
|
else:
|
|
return "No face detected in the uploaded image."
|
|
else:
|
|
image_path = request.form['current_image_path']
|
|
encoding = request.form['encoding']
|
|
|
|
c.execute("UPDATE faces SET name=?, identity=?, image_path=?, encoding=? WHERE id=?",
|
|
(name, identity, image_path, encoding, id))
|
|
|
|
# 删除人员信息
|
|
elif 'delete' in request.form:
|
|
id = request.form['id']
|
|
c.execute("DELETE FROM faces WHERE id=?", (id,))
|
|
|
|
conn.commit()
|
|
|
|
# 获取所有人员记录
|
|
c.execute("SELECT * FROM faces")
|
|
persons = c.fetchall()
|
|
conn.close()
|
|
|
|
return render_template('info_person.html', persons=persons)
|
|
|
|
@socketio.on('connect')
|
|
def handle_connect():
|
|
print('Client connected')
|
|
emit('update', {'logs': get_match_logs()})
|
|
|
|
def send_updates():
|
|
while True:
|
|
# 模拟实时数据更新
|
|
socketio.emit('update', {'logs': get_match_logs()}, broadcast=True)
|
|
eventlet.sleep(5) # 每 5 秒发送一次更新
|
|
|
|
if __name__ == '__main__':
|
|
create_face_database() # Ensure the database is created
|
|
socketio.start_background_task(send_updates)
|
|
socketio.run(app, debug=True)
|