要将SQL语句嵌入代码中，特别是集成到您现有的“基于OpenCV的校园人脸识别考勤系统”中，以实现用户数据、考勤记录的存储与管理，核心在于利用数据库操作库建立程序与数据库的连接，并在适当的逻辑位置执行SQL语句。 ### 一、技术选型与依赖库 对于Python后端的系统，您不应使用传统C语言的“嵌入式SQL”预编译方式（如`EXEC SQL BEGIN DECLARE SECTION;`，这主要适用于C、COBOL等语言与数据库的深度集成[ref_1][ref_2][ref_5]）。Python中通常采用 **ORM（对象关系映射）框架** 或 **数据库连接驱动** 来执行SQL，这种方式在概念上也属于广义的“将SQL嵌入到宿主语言”[ref_3]。 推荐方案及所需库： | 方案 | 核心库 | 特点 | 适用场景 | | :--- | :--- | :--- | :--- | | **原生驱动** | `mysql-connector-python` 或 `pymysql` (MySQL) / `psycopg2` (PostgreSQL) / `sqlite3` (内置) | 直接执行SQL字符串，灵活，性能好，但需手动处理连接、参数和结果集映射。 | 需要精细控制SQL、或项目结构简单时。 | | **ORM框架** | `SQLAlchemy` 或 `Django ORM` (如果使用Django框架) | 通过Python类与对象操作数据库，代码更面向对象，可迁移性好，但有一定学习成本。 | 中大型项目、需要良好的代码结构和可维护性时。 | 对于毕业设计，考虑到易用性和普遍性，以下示例将采用 `pymysql`（连接MySQL）和内置的 `sqlite3` 作为演示。 首先安装必要库： ```bash pip install pymysql # sqlite3 是Python标准库，无需安装 ``` ### 二、SQL应加入到哪一段代码：核心模块与位置 一个典型的人脸识别考勤系统后端（使用Flask/FastAPI等框架）代码结构如下，SQL语句主要嵌入在以下几个关键模块： ``` your_project/ ├── app.py # 应用主入口，路由定义 ├── database.py # **数据库连接与初始化模块 (关键)** ├── models.py # **数据模型/表结构定义 (关键)** ├── services/ # 业务逻辑服务层 │ ├── user_service.py # **用户相关业务逻辑 (关键)** │ ├── attendance_service.py # **考勤相关业务逻辑 (关键)** │ └── face_service.py # 人脸特征处理逻辑 ├── utils/ │ ├── face_utils.py # OpenCV人脸识别工具函数 │ └── config.py # 配置文件 └── requirements.txt ``` **SQL语句具体嵌入的位置如下表所示：** | 代码模块 | 嵌入的SQL类型 | 目的与时机 | 代码示例位置 | | :--- | :--- | :--- | :--- | | **`database.py`** | DDL (CREATE TABLE) | 应用启动时，初始化数据库，创建用户表、考勤记录表等。 | 在创建数据库连接后，执行建表语句。 | | **`models.py`** | DDL (隐含) | 使用ORM时，定义Python类来映射表结构，ORM会**自动生成并执行**对应的CREATE TABLE SQL。 | 类定义中。若使用原生驱动，此文件可能不直接包含SQL。 | | **`user_service.py`** | DML (INSERT, SELECT, UPDATE, DELETE) | 实现用户注册、登录、信息查询、角色（学生/教师/管理员）权限校验等功能。 | 在每个具体的业务函数中，如`register_user`, `get_user_by_id`。 | | **`attendance_service.py`** | DML (INSERT, SELECT) | 实现考勤打卡（插入记录）、按日期/人员查询考勤统计等功能。 | 在打卡处理函数、查询统计函数中。 | | **`app.py` 中的视图函数** | **不直接嵌入** | 视图函数应调用 `services/` 下的业务逻辑函数，**自身不直接包含SQL**，以保持代码清晰。 | 在路由处理函数内部，调用服务层方法。 | ### 三、具体代码嵌入示例 假设我们使用 **MySQL数据库** 和 **原生`pymysql`驱动**。 #### 1. 数据库连接与初始化 (`database.py`) 此模块负责建立数据库连接池或全局连接，并确保表结构存在。 ```python # database.py import pymysql from pymysql import cursors from utils.config import DB_CONFIG # 假设配置从config.py读取 def get_db_connection(): """获取数据库连接""" try: connection = pymysql.connect( host=DB_CONFIG['host'], user=DB_CONFIG['user'], password=DB_CONFIG['password'], database=DB_CONFIG['database'], charset='utf8mb4', cursorclass=cursors.DictCursor # 返回字典格式的结果 ) return connection except pymysql.MySQLError as e: print(f"数据库连接失败: {e}") return None def init_database(): """初始化数据库表结构 (嵌入DDL SQL)""" create_user_table_sql = """ CREATE TABLE IF NOT EXISTS `user` ( `id` INT AUTO_INCREMENT PRIMARY KEY, `username` VARCHAR(50) UNIQUE NOT NULL COMMENT '用户名/工号/学号', `password_hash` VARCHAR(255) NOT NULL COMMENT '密码哈希', `name` VARCHAR(50) NOT NULL COMMENT '真实姓名', `role` ENUM('student', 'teacher', 'admin') NOT NULL DEFAULT 'student' COMMENT '角色', `face_feature` BLOB COMMENT '人脸特征向量', `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户表'; """ create_attendance_table_sql = """ CREATE TABLE IF NOT EXISTS `attendance` ( `id` INT AUTO_INCREMENT PRIMARY KEY, `user_id` INT NOT NULL COMMENT '用户ID', `check_in_time` TIMESTAMP NULL DEFAULT NULL COMMENT '签到时间', `check_out_time` TIMESTAMP NULL DEFAULT NULL COMMENT '签退时间', `date` DATE NOT NULL COMMENT '考勤日期', `status` ENUM('present', 'late', 'absent', 'leave') DEFAULT 'present' COMMENT '考勤状态', `location` VARCHAR(255) COMMENT '打卡地点', FOREIGN KEY (`user_id`) REFERENCES `user`(`id`) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='考勤记录表'; """ conn = get_db_connection() if conn: try: with conn.cursor() as cursor: cursor.execute(create_user_table_sql) # 执行建表SQL cursor.execute(create_attendance_table_sql) conn.commit() print("数据库表初始化成功。") except pymysql.MySQLError as e: print(f"初始化表结构失败: {e}") conn.rollback() finally: conn.close() ``` #### 2. 用户服务层 (`services/user_service.py`) 在此模块的业务函数中嵌入 **DML SQL**，实现增删改查。 ```python # services/user_service.py from database import get_db_connection import hashlib def register_user(username, password, name, role='student'): """用户注册 (嵌入INSERT SQL)""" # 对密码进行哈希处理（生产环境应使用更安全的方法如bcrypt） password_hash = hashlib.sha256(password.encode()).hexdigest() insert_sql = """ INSERT INTO `user` (`username`, `password_hash`, `name`, `role`) VALUES (%s, %s, %s, %s) """ conn = get_db_connection() if not conn: return False, "数据库连接失败" try: with conn.cursor() as cursor: cursor.execute(insert_sql, (username, password_hash, name, role)) conn.commit() user_id = cursor.lastrowid return True, f"用户注册成功，ID: {user_id}" except pymysql.IntegrityError: return False, "用户名已存在" except pymysql.MySQLError as e: conn.rollback() return False, f"数据库错误: {e}" finally: conn.close() def get_user_by_username(username): """根据用户名查询用户 (嵌入SELECT SQL)""" select_sql = "SELECT `id`, `username`, `name`, `role` FROM `user` WHERE `username` = %s" conn = get_db_connection() if not conn: return None try: with conn.cursor() as cursor: cursor.execute(select_sql, (username,)) user = cursor.fetchone() # 获取一条记录 return user except pymysql.MySQLError as e: print(f"查询用户失败: {e}") return None finally: conn.close() def verify_user_credentials(username, password): """验证用户登录凭证 (嵌入SELECT SQL并进行比对)""" select_sql = "SELECT `id`, `username`, `password_hash`, `role` FROM `user` WHERE `username` = %s" conn = get_db_connection() if not conn: return False, None, None try: with conn.cursor() as cursor: cursor.execute(select_sql, (username,)) user = cursor.fetchone() if user and user['password_hash'] == hashlib.sha256(password.encode()).hexdigest(): return True, user['id'], user['role'] # 验证成功，返回用户ID和角色 else: return False, None, None except pymysql.MySQLError as e: print(f"登录验证失败: {e}") return False, None, None finally: conn.close() # 新增：根据角色获取用户列表 (用于管理员功能) def get_users_by_role(role=None): """获取用户列表，可按角色过滤 (嵌入带条件的SELECT SQL)""" base_sql = "SELECT `id`, `username`, `name`, `role`, `created_at` FROM `user`" params = () if role: base_sql += " WHERE `role` = %s" params = (role,) conn = get_db_connection() if not conn: return [] try: with conn.cursor() as cursor: cursor.execute(base_sql, params) users = cursor.fetchall() # 获取所有记录 return users except pymysql.MySQLError as e: print(f"获取用户列表失败: {e}") return [] finally: conn.close() ``` #### 3. 考勤服务层 (`services/attendance_service.py`) ```python # services/attendance_service.py from database import get_db_connection from datetime import datetime, date def record_attendance(user_id, check_type='in', location=None): """记录考勤打卡 (嵌入INSERT或UPDATE SQL)""" today = date.today() current_time = datetime.now() conn = get_db_connection() if not conn: return False, "数据库连接失败" try: with conn.cursor() as cursor: # 首先检查今天是否已有记录 check_sql = "SELECT `id`, `check_in_time`, `check_out_time` FROM `attendance` WHERE `user_id` = %s AND `date` = %s" cursor.execute(check_sql, (user_id, today)) record = cursor.fetchone() if check_type == 'in': if record and record['check_in_time']: return False, "今日已签到" if record: # 有记录但未签到，更新签到时间 update_sql = "UPDATE `attendance` SET `check_in_time` = %s, `location` = %s WHERE `id` = %s" cursor.execute(update_sql, (current_time, location, record['id'])) else: # 无记录，插入新记录 insert_sql = """ INSERT INTO `attendance` (`user_id`, `check_in_time`, `date`, `location`) VALUES (%s, %s, %s, %s) """ cursor.execute(insert_sql, (user_id, current_time, today, location)) elif check_type == 'out': if not record: return False, "请先完成签到" if record and record['check_out_time']: return False, "今日已签退" update_sql = "UPDATE `attendance` SET `check_out_time` = %s WHERE `id` = %s" cursor.execute(update_sql, (current_time, record['id'])) conn.commit() return True, f"{'签到' if check_type == 'in' else '签退'}成功" except pymysql.MySQLError as e: conn.rollback() return False, f"打卡失败: {e}" finally: conn.close() def get_attendance_statistics(start_date, end_date, user_id=None): """获取考勤统计 (嵌入复杂SELECT SQL，使用JOIN和聚合)""" base_sql = """ SELECT u.`id`, u.`username`, u.`name`, u.`role`, a.`date`, a.`check_in_time`, a.`check_out_time`, a.`status` FROM `attendance` a JOIN `user` u ON a.`user_id` = u.`id` WHERE a.`date` BETWEEN %s AND %s """ params = [start_date, end_date] if user_id: base_sql += " AND a.`user_id` = %s" params.append(user_id) base_sql += " ORDER BY a.`date` DESC, u.`role`, u.`id`" conn = get_db_connection() if not conn: return [] try: with conn.cursor() as cursor: cursor.execute(base_sql, tuple(params)) records = cursor.fetchall() return records except pymysql.MySQLError as e: print(f"获取考勤统计失败: {e}") return [] finally: conn.close() ``` #### 4. 在主应用或初始化脚本中调用 在应用启动文件（如`app.py`或单独的`init_db.py`）中，调用初始化函数。 ```python # app.py (Flask示例) 或 main.py from database import init_database from flask import Flask app = Flask(__name__) # 应用启动时初始化数据库（仅需一次，生产环境通常用迁移工具） with app.app_context(): init_database() # ... 定义路由，在路由处理函数中调用上述 services 里的函数 ... # 例如： # @app.route('/api/register', methods=['POST']) # def register(): # data = request.get_json() # success, message = register_user(data['username'], data['password'], data['name'], data.get('role', 'student')) # return jsonify({'success': success, 'message': message}) ``` ### 四、关键要点与安全建议 1. **防止SQL注入**：务必使用参数化查询（如`cursor.execute(sql, (param1, param2))`），**绝对不要**使用字符串拼接（如`f"SELECT * FROM user WHERE name='{user_input}'"`）[ref_3][ref_4]。 2. **连接管理**：确保数据库连接在使用后正确关闭（使用`try...finally`或上下文管理器），避免连接泄漏。对于Web应用，考虑使用连接池。 3. **错误处理**：对所有数据库操作进行`try...except`捕获，并对事务进行`commit`或`rollback`[ref_3][ref_6]。 4. **密码存储**：示例中的密码哈希仅为演示，生产环境务必使用专为密码哈希设计的算法，如`bcrypt`或`argon2`。 5. **配置文件**：数据库主机、端口、用户名、密码等敏感信息应存储在环境变量或单独的配置文件中（如`utils/config.py`），切勿硬编码在代码中。 通过以上方式，您可以将SQL语句安全、结构化和可维护地嵌入到Python人脸识别考勤系统的各个业务逻辑层中，为后续增加教师和管理员角色（通过`user`表的`role`字段区分）的功能打下坚实的数据持久化基础。管理员后台可以调用`get_users_by_role`、`get_attendance_statistics`等函数实现用户管理和数据统计视图。