BodyBalanceEvaluation/backend/database.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据库管理模块
负责SQLite数据库的创建、连接和数据操作
"""

import sqlite3
import json
import uuid
from datetime import datetime, timezone, timedelta
from typing import List, Dict, Optional, Any, Union
import logging

logger = logging.getLogger(__name__)

class DatabaseManager:
    """数据库管理器"""
    
    def __init__(self, db_path: str):
        self.db_path = db_path
        self.connection = None
        # 设置中国上海时区 (UTC+8)
        self.china_tz = timezone(timedelta(hours=8))
    
    def get_china_time(self) -> str:
        """获取中国时区的当前时间字符串"""
        return datetime.now(self.china_tz).strftime('%Y-%m-%d %H:%M:%S')
    
    def generate_patient_id(self) -> str:
        """生成患者唯一标识(YYYYMMDD0000)年月日+四位序号"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            # 获取当前日期
            china_tz = timezone(timedelta(hours=8))
            today = datetime.now(china_tz).strftime('%Y%m%d')
            
            # 使用循环确保生成唯一ID，防止并发冲突
            for attempt in range(10):  # 最多尝试10次
                # 查询今天已有的最大序号
                cursor.execute('''
                    SELECT id FROM patients 
                    WHERE id LIKE ? 
                    ORDER BY id DESC 
                    LIMIT 1
                ''', (f'{today}%',))
                
                result = cursor.fetchone()
                if result:
                    # 提取序号部分并加1
                    last_id = result[0]
                    if len(last_id) >= 12 and last_id[:8] == today:
                        last_sequence = int(last_id[8:12])
                        sequence = str(last_sequence + 1).zfill(4)
                    else:
                        sequence = '0001'
                else:
                    # 今天第一个患者
                    sequence = '0001'
                
                patient_id = f'{today}{sequence}'
                
                # 检查ID是否已存在
                cursor.execute('SELECT COUNT(*) FROM patients WHERE id = ?', (patient_id,))
                if cursor.fetchone()[0] == 0:
                    return patient_id
            
            # 如果10次尝试都失败，使用UUID作为备用方案
            logger.warning('患者ID生成重试次数过多，使用UUID备用方案')
            return str(uuid.uuid4())
            
        except Exception as e:
            logger.error(f'生成患者ID失败: {e}')
            # 如果生成失败，使用UUID作为备用方案
            return str(uuid.uuid4())
    
    def generate_user_id(self) -> str:
        """生成用户唯一标识，格式为六位数字序号(000001)"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            # 使用循环确保生成唯一ID，防止并发冲突
            for attempt in range(10):  # 最多尝试10次
                # 获取当前最大的用户ID（只考虑六位数字格式的ID）
                cursor.execute('SELECT MAX(CAST(id AS INTEGER)) FROM users WHERE id GLOB "[0-9][0-9][0-9][0-9][0-9][0-9]"')
                result = cursor.fetchone()[0]
                
                if result is None:
                    # 如果没有用户记录，从000001开始
                    next_id = 1
                else:
                    next_id = result + 1
                
                # 格式化为六位数字，前面补零
                user_id = f"{next_id:06d}"
                
                # 检查ID是否已存在
                cursor.execute('SELECT COUNT(*) FROM users WHERE id = ?', (user_id,))
                if cursor.fetchone()[0] == 0:
                    return user_id
            
            # 如果10次尝试都失败，使用时间戳+随机数作为备用方案
            logger.warning('用户ID生成重试次数过多，使用备用方案')
            import time
            import random
            timestamp_suffix = str(int(time.time()))[-4:]
            random_suffix = str(random.randint(10, 99))
            return f"{timestamp_suffix}{random_suffix}"
            
        except Exception as e:
            logger.error(f'生成用户ID失败: {e}')
            # 如果出错，使用时间戳作为备用方案
            import time
            return str(int(time.time()))[-6:]
    
    def generate_session_id(self) -> str:
        """生成会话唯一标识(YYYYMMDDHHMMSS)年月日时分秒"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            # 获取当前时间
            china_tz = timezone(timedelta(hours=8))
            
            # 使用循环确保生成唯一ID，防止并发冲突
            for attempt in range(10):  # 最多尝试10次
                current_time = datetime.now(china_tz)
                session_id = current_time.strftime('%Y%m%d%H%M%S')
                
                # 检查ID是否已存在
                cursor.execute('SELECT COUNT(*) FROM detection_sessions WHERE id = ?', (session_id,))
                if cursor.fetchone()[0] == 0:
                    return session_id
                
                # 如果存在冲突，等待1秒后重试
                import time
                time.sleep(0.001)  # 等待1毫秒
            
            # 如果10次尝试都失败，在时间后添加随机后缀
            import random
            current_time = datetime.now(china_tz)
            base_id = current_time.strftime('%Y%m%d%H%M%S')
            suffix = str(random.randint(10, 99))
            session_id = f'{base_id}{suffix}'
            
            # 最后检查一次
            cursor.execute('SELECT COUNT(*) FROM detection_sessions WHERE id = ?', (session_id,))
            if cursor.fetchone()[0] == 0:
                return session_id
            
            # 如果还是冲突，使用UUID作为备用方案
            logger.warning('会话ID生成重试次数过多，使用UUID备用方案')
            return str(uuid.uuid4())
            
        except Exception as e:
            logger.error(f'生成会话ID失败: {e}')
            # 如果生成失败，使用UUID作为备用方案
            return str(uuid.uuid4())
    
    
    
    def get_connection(self) -> sqlite3.Connection:
        """获取数据库连接"""
        if not self.connection:
            self.connection = sqlite3.connect(
                self.db_path, 
                check_same_thread=False,
                timeout=30.0
            )
            self.connection.row_factory = sqlite3.Row
        return self.connection
    
    def init_database(self):
        """初始化数据库表结构"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            # 创建用户表（医生）
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS users (
                    id TEXT PRIMARY KEY,                                   -- 用户唯一标识(0000000)六位序号
                    name TEXT NOT NULL,                                    -- 用户真实姓名
                    username TEXT UNIQUE NOT NULL,                         -- 用户名（登录名）
                    password TEXT NOT NULL,                                -- 密码
                    register_date TIMESTAMP,                               -- 注册日期
                    is_active BOOLEAN DEFAULT 1,                           -- 账户是否激活（0=未激活，1=已激活）
                    user_type TEXT DEFAULT 'user',                        -- 用户类型（user/admin/doctor）
                    phone TEXT DEFAULT '',                                 -- 联系电话
                    created_at TIMESTAMP,                                 -- 记录创建时间
                    updated_at TIMESTAMP                                  -- 记录更新时间
                )
            ''')   

            # 创建患者表
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS patients (
                    id TEXT PRIMARY KEY,                                    -- 患者唯一标识(YYYYMMDD0000)年月日+四位序号
                    name TEXT NOT NULL,                                    -- 患者姓名
                    gender TEXT,                                           -- 性别
                    birth_date TIMESTAMP,                                  -- 出生日期
                    nationality TEXT,                                      -- 民族
                    residence TEXT,                                        -- 居住地
                    height REAL,                                           -- 身高（cm）
                    weight REAL,                                           -- 体重（kg）
                    shoe_size TEXT,                                        -- 鞋码
                    phone TEXT,                                            -- 电话号码
                    email TEXT,                                            -- 电子邮箱
                    occupation TEXT,                                       -- 职业
                    workplace TEXT,                                        -- 工作单位
                    idcode TEXT,                                           -- 身份证号
                    medical_history TEXT,                                  -- 病史
                    notes TEXT,                                            -- 备注信息
                    created_at TIMESTAMP,                                 -- 记录创建时间
                    updated_at TIMESTAMP                                  -- 记录更新时间
                )
            ''')
            
            # 创建检测会话表
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS detection_sessions (
                    id TEXT PRIMARY KEY,                                   -- 会话唯一标识(YYYYMMDDHHMMSS)年月日时分秒
                    patient_id TEXT NOT NULL,                              -- 患者ID（外键）
                    creator_id TEXT,                                       -- 创建人ID（医生ID，外键）
                    start_time TIMESTAMP,                                  -- 检测开始时间
                    end_time TIMESTAMP,                                    -- 检测结束时间
                    duration INTEGER,                                      -- 检测持续时间（秒）
                    diagnosis_info TEXT,                                   -- 诊断信息
                    treatment_info TEXT,                                   -- 处理信息
                    remark_info TEXT,                                      -- 备注信息
                    detection_report TEXT,                                 -- 生成检测报告的存储路径
                    status TEXT DEFAULT 'checking',                        -- 会话状态（checking/checked/diagnosed/reported)
                    created_at TIMESTAMP,                                 -- 记录创建时间
                    FOREIGN KEY (patient_id) REFERENCES patients (id),    -- 患者表外键约束
                    FOREIGN KEY (creator_id) REFERENCES users (id)        -- 用户表外键约束
                )
            ''')
            
            # 创建检测截图数据表
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS detection_data (
                    id TEXT PRIMARY KEY,                               -- 记录唯一标识(YYYYMMDDHHMMSS)年月日时分秒
                    session_id TEXT NOT NULL,                          -- 检测会话ID（外键）
                    head_pose TEXT ,                                   -- 头部姿态数据（JSON格式）
                    body_pose TEXT ,                                   -- 身体姿态数据（JSON格式）
                    foot_data TEXT ,                                   -- 足部姿态数据（JSON格式）      
                    body_image TEXT,                                   -- 身体视频截图存储路径 
                    screen_image TEXT,                                 -- 屏幕录制视频截图存储路径                                 
                    foot_data_image TEXT,                              -- 足底压力数据图存储路径                    
                    foot1_image TEXT,                                   -- 足部监测视频1截图存储路径
                    foot2_image TEXT,                                   -- 足部监测视频2截图存储路径
                    timestamp TIMESTAMP,                               -- 数据记录时间戳
                    FOREIGN KEY (session_id) REFERENCES detection_sessions (id)  -- 检测会话表外键约束
                )
            ''')            
            # 创建检测视频录制表
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS detection_video (
                    id TEXT PRIMARY KEY,                               -- 记录唯一标识(YYYYMMDDHHMMSS)年月日时分秒
                    session_id TEXT NOT NULL,                          -- 检测会话ID（外键）                
                    screen_video TEXT,                                 -- 屏幕录制视频存储路径
                    body_video TEXT,                                   -- 身体监测视频存储路径                   
                    foot_video1 TEXT,                                  -- 足部监测视频1存储路径
                    foot_video2 TEXT,                                  -- 足部监测视频2存储路径
                    timestamp TIMESTAMP,                               -- 数据记录时间戳
                    FOREIGN KEY (session_id) REFERENCES detection_sessions (id)  -- 检测会话表外键约束
                )
            ''')      
            
            # 创建系统设置表
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS system_settings (
                    key TEXT PRIMARY KEY,                              -- 设置项键名（唯一标识）
                    value TEXT NOT NULL,                               -- 设置项值
                    description TEXT,                                  -- 设置项描述说明
                    updated_at TIMESTAMP                               -- 最后更新时间
                )
            ''')
            
      
           
            
            # 创建索引以提高查询性能
            # 患者表索引
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_patients_name ON patients (name)')        # 患者姓名索引
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_patients_phone ON patients (phone)')      # 患者电话索引
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_patients_email ON patients (email)')      # 患者邮箱索引
            
            # 检测会话表索引
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_sessions_patient ON detection_sessions (patient_id)')    # 患者ID索引
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_sessions_creator ON detection_sessions (creator_id)')    # 创建人ID索引
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_sessions_time ON detection_sessions (start_time)')       # 开始时间索引
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_sessions_status ON detection_sessions (status)')         # 会话状态索引
            
            # 检测截图数据表索引
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_data_session ON detection_data (session_id)')            # 会话ID索引
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_data_timestamp ON detection_data (timestamp)')           # 时间戳索引

            # 检测视频数据表索引
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_video_session ON detection_video (session_id)')            # 会话ID索引
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_video_timestamp ON detection_video (timestamp)')           # 时间戳索引
            # 用户表索引
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_users_username ON users (username)')                     # 用户名索引
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_users_type ON users (user_type)')                       # 用户类型索引
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_users_phone ON users (phone)')                          # 用户电话索引
            cursor.execute('CREATE INDEX IF NOT EXISTS idx_users_active ON users (is_active)')                     # 激活状态索引
            
            # 插入默认管理员账户（如果不存在）
            cursor.execute('SELECT COUNT(*) FROM users WHERE username = ?', ('admin',))
            admin_exists = cursor.fetchone()[0]
            
            if admin_exists == 0:
                admin_id = self.generate_user_id()
                # 默认密码为 admin123，明文存储
                admin_password = 'admin123'
                # 使用中国时区时间
                china_time = self.get_china_time()
                
                cursor.execute('''
                    INSERT INTO users (id, name, username, password, is_active, user_type, register_date, created_at, updated_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                ''', (admin_id, '系统管理员', 'admin', admin_password, 1, 'admin', china_time, china_time, china_time))
                
                logger.info('创建默认管理员账户: admin/admin123')          

            
            conn.commit()
            logger.info('数据库初始化完成')
            
        except Exception as e:
            conn.rollback()
            logger.error(f'数据库初始化失败: {e}')
            raise
    
    # ==================== 患者数据管理 ====================#
    
    def create_patient(self, patient_data: Dict[str, Any]) -> str:
        """创建患者记录"""
        # 验证必填字段
        if not patient_data.get('name'):
            raise ValueError('患者姓名不能为空')
        
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            patient_id = self.generate_patient_id()
            # 使用中国时区时间
            china_time = self.get_china_time()
            
            cursor.execute('''
                INSERT INTO patients (
                    id, name, gender, birth_date, nationality, residence,
                    height, weight, shoe_size, phone, email, occupation, workplace,
                    medical_history,idcode, notes, created_at, updated_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?)
            ''', (
                patient_id,
                patient_data.get('name'),
                patient_data.get('gender'),
                patient_data.get('birth_date'),
                patient_data.get('nationality'),
                patient_data.get('residence'),
                patient_data.get('height'),
                patient_data.get('weight'),
                patient_data.get('shoe_size'),
                patient_data.get('phone'),
                patient_data.get('email'),
                patient_data.get('occupation'),
                patient_data.get('workplace'),
                patient_data.get('medical_history'),
                patient_data.get('idcode'),
                patient_data.get('notes'),
                china_time,
                china_time
            ))
            
            conn.commit()
            logger.info(f'创建患者记录: {patient_id}')
            return patient_id
            
        except Exception as e:
            conn.rollback()
            logger.error(f'创建患者记录失败: {e}')
            raise
    
    def get_patients(self, page: int = 1, size: int = 10, keyword: str = '') -> List[Dict]:
        """获取患者列表"""
        # 验证分页参数
        if page < 1:
            page = 1
        if size < 1 or size > 100:
            size = 10
        
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            offset = (page - 1) * size
            
            if keyword:
                cursor.execute('''
                    SELECT p.*, COALESCE(ds.session_count, 0) AS detection_count
                    FROM patients p
                    LEFT JOIN (
                        SELECT patient_id, COUNT(*) AS session_count
                        FROM detection_sessions
                        GROUP BY patient_id
                    ) ds ON ds.patient_id = p.id
                    WHERE p.name LIKE ? OR p.phone LIKE ? OR p.email LIKE ?
                    ORDER BY p.created_at DESC
                    LIMIT ? OFFSET ?
                ''', (f'%{keyword}%', f'%{keyword}%', f'%{keyword}%', size, offset))
            else:
                cursor.execute('''
                    SELECT p.*, COALESCE(ds.session_count, 0) AS detection_count
                    FROM patients p
                    LEFT JOIN (
                        SELECT patient_id, COUNT(*) AS session_count
                        FROM detection_sessions
                        GROUP BY patient_id
                    ) ds ON ds.patient_id = p.id
                    ORDER BY p.created_at DESC
                    LIMIT ? OFFSET ?
                ''', (size, offset))
            
            rows = cursor.fetchall()
            return [dict(row) for row in rows]
            
        except Exception as e:
            logger.error(f'获取患者列表失败: {e}')
            raise
    
    def get_patients_count(self, keyword: str = '') -> int:
        """获取患者总数"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            if keyword:
                cursor.execute('''
                    SELECT COUNT(*) FROM patients 
                    WHERE name LIKE ? OR phone LIKE ?
                ''', (f'%{keyword}%', f'%{keyword}%'))
            else:
                cursor.execute('SELECT COUNT(*) FROM patients')
            
            return cursor.fetchone()[0]
            
        except Exception as e:
            logger.error(f'获取患者总数失败: {e}')
            return 0
    
    def get_patient(self, patient_id: str) -> Optional[Dict]:
        """获取单个患者信息"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            cursor.execute('SELECT * FROM patients WHERE id = ?', (patient_id,))
            row = cursor.fetchone()
            return dict(row) if row else None
            
        except Exception as e:
            logger.error(f'获取患者信息失败: {e}')
            return None
    
    def update_patient(self, patient_id: str, patient_data: Dict[str, Any]):
        """更新患者信息"""
        # 验证必填字段
        if not patient_data.get('name'):
            raise ValueError('患者姓名不能为空')
        
        # 验证患者是否存在
        if not self.get_patient(patient_id):
            raise ValueError(f'患者不存在: {patient_id}')
        
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            # 使用中国时区时间
            china_time = self.get_china_time()
            
            cursor.execute('''
                UPDATE patients SET 
                    name = ?, gender = ?, birth_date = ?, nationality = ?,
                    residence = ?, height = ?, weight = ?, shoe_size = ?, phone = ?, email = ?,
                    occupation = ?, workplace = ?, medical_history = ?, idcode = ?, notes = ?, updated_at = ?
                WHERE id = ?
            ''', (
                patient_data.get('name'),
                patient_data.get('gender'),
                patient_data.get('birth_date'),
                patient_data.get('nationality'),
                patient_data.get('residence'),
                patient_data.get('height'),
                patient_data.get('weight'),
                patient_data.get('shoe_size'),
                patient_data.get('phone'),
                patient_data.get('email'),
                patient_data.get('occupation'),
                patient_data.get('workplace'),
                patient_data.get('medical_history'),
                patient_data.get('idcode'),
                patient_data.get('notes'),
                china_time,
                patient_id
            ))
            
            conn.commit()
            logger.info(f'更新患者信息: {patient_id}')
            
        except Exception as e:
            conn.rollback()
            logger.error(f'更新患者信息失败: {e}')
            raise
    
    def delete_patient(self, patient_id: str):
        """删除患者记录"""
        # 验证患者是否存在
        if not self.get_patient(patient_id):
            raise ValueError(f'患者不存在: {patient_id}')
        
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            # 删除相关的检测截图数据
            cursor.execute('''
                DELETE FROM detection_data 
                WHERE session_id IN (
                    SELECT id FROM detection_sessions WHERE patient_id = ?
                )
            ''', (patient_id,))

             # 删除相关的检测视频数据
            cursor.execute('''
                DELETE FROM detection_video 
                WHERE session_id IN (
                    SELECT id FROM detection_sessions WHERE patient_id = ?
                )
            ''', (patient_id,))
            
            # 删除检测会话
            cursor.execute('DELETE FROM detection_sessions WHERE patient_id = ?', (patient_id,))
            
            # 删除患者记录
            cursor.execute('DELETE FROM patients WHERE id = ?', (patient_id,))
            
            conn.commit()
            logger.info(f'删除患者记录: {patient_id}')
            
        except Exception as e:
            conn.rollback()
            logger.error(f'删除患者记录失败: {e}')
            raise  
    # ==================== 检测会话数据管理 ==================== # 
    
    def create_detection_session(self, patient_id: str, settings: Dict[str, Any], creator_id: str = None) -> str:
        """创建检测会话"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            session_id = self.generate_session_id()
            
            # 使用中国时区时间
            china_time = self.get_china_time()
            
            cursor.execute('''
                INSERT INTO detection_sessions (
                    id, patient_id, creator_id, status, 
                    start_time, created_at
                ) VALUES (?, ?, ?, ?, ?, ?)
            ''', (
                session_id,
                patient_id,
                creator_id,                
                'checking',                
                china_time,
                china_time
            ))
            
            conn.commit()
            logger.info(f'创建检测会话: {session_id}')
            return session_id
            
        except Exception as e:
            conn.rollback()
            logger.error(f'创建检测会话失败: {e}')
            raise
    
    def update_session_status(self, session_id: str, status: str) -> bool:
        """更新会话状态        
        Returns:
            bool: 更新成功返回True，失败返回False
        """
        conn = self.get_connection()
        cursor = conn.cursor()
        try:
            # 首先获取会话对应的患者ID
            cursor.execute('SELECT patient_id FROM detection_sessions WHERE id = ?', (session_id,))
            result = cursor.fetchone()
            if not result:
                logger.error(f'会话不存在: {session_id}')
                return False          
            cursor.execute('''
                    UPDATE detection_sessions SET 
                        status = ?
                    WHERE id = ?
                ''', (status, session_id))            
            conn.commit()           
            return True            
        except Exception as e:
            conn.rollback()
            logger.error(f'更新会话状态失败: {e}')
            return False
    

    def update_session_endcheck(self, session_id: str, diagnosis_info: Optional[str] = None,
                                 treatment_info: Optional[str] = None,
                                 suggestion_info: Optional[str] = None) -> bool:
        """结束检测：根据 start_time 与当前时间计算持续时长并写入结束信息

        行为：
        - 计算并保存 `duration`
        - 保存 `end_time` 为当前中国时区时间
        - 可选更新 `diagnosis_info`、`treatment_info`、`suggestion_info`
        - 设置 `status = 'checked'`
        - 同步更新患者 `updated_at`
        """
        conn = self.get_connection()
        cursor = conn.cursor()

        try:
            # 读取会话的开始时间与患者ID
            cursor.execute('SELECT start_time, patient_id FROM detection_sessions WHERE id = ?', (session_id,))
            row = cursor.fetchone()
            if not row:
                logger.error(f'会话不存在: {session_id}')
                return False

            start_time_str, patient_id = row
            if not start_time_str:
                logger.error(f'会话缺少开始时间: {session_id}')
                return False

            # 计算持续时间（秒）
            now_str = self.get_china_time()
            try:
                start_dt = datetime.strptime(start_time_str, '%Y-%m-%d %H:%M:%S')
                now_dt = datetime.strptime(now_str, '%Y-%m-%d %H:%M:%S')
                duration_seconds = max(0, int((now_dt - start_dt).total_seconds()))
            except Exception as e:
                logger.error(f'解析时间失败: start_time={start_time_str}, now={now_str}, error={e}')
                return False

            # 构造更新语句
            update_fields = [
                'duration = ?',
                'end_time = ?',
                'status = ?'
            ]
            update_values = [duration_seconds, now_str, 'checked']

            if diagnosis_info is not None:
                update_fields.append('diagnosis_info = ?')
                update_values.append(diagnosis_info)

            if treatment_info is not None:
                update_fields.append('treatment_info = ?')
                update_values.append(treatment_info)

            if suggestion_info is not None:
                update_fields.append('suggestion_info = ?')
                update_values.append(suggestion_info)

            # 添加会话ID到参数列表
            update_values.append(session_id)

            sql = f'''UPDATE detection_sessions SET {', '.join(update_fields)} WHERE id = ?'''
            cursor.execute(sql, update_values)

            # 同步更新患者表的updated_at时间
            cursor.execute('''UPDATE patients SET updated_at = ? WHERE id = ?''', (now_str, patient_id))

            conn.commit()
            logger.info(f'结束检测并更新会话: {session_id}, duration={duration_seconds}s, status=checked')
            return True

        except Exception as e:
            conn.rollback()
            logger.error(f'结束检测更新失败: {e}')
            return False    
    
        
    def update_session_all_info(self, session_id: str, diagnosis_info: str = None, treatment_info: str = None, suggestion_info: str = None, status: str = None):
        """同时更新会话的诊断信息、处理信息、建议信息和状态"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            # 构建动态SQL语句，只更新非None的字段
            update_fields = []
            update_values = []
            
            if diagnosis_info is not None:
                update_fields.append('diagnosis_info = ?')
                update_values.append(diagnosis_info)
            
            if treatment_info is not None:
                update_fields.append('treatment_info = ?')
                update_values.append(treatment_info)
            
            if suggestion_info is not None:
                update_fields.append('suggestion_info = ?')
                update_values.append(suggestion_info)
            
            if status is not None:
                update_fields.append('status = ?')
                update_values.append(status)
            
            if not update_fields:
                logger.warning(f'没有提供要更新的信息: {session_id}')
                return
            
            # 添加session_id到参数列表
            update_values.append(session_id)
            
            sql = f'''
                UPDATE detection_sessions SET {', '.join(update_fields)}
                WHERE id = ?
            '''
            
            cursor.execute(sql, update_values)
            conn.commit()
            
            updated_info = []
            if diagnosis_info is not None:
                updated_info.append('诊断信息')
            if treatment_info is not None:
                updated_info.append('处理信息')
            if suggestion_info is not None:
                updated_info.append('建议信息')
            if status is not None:
                updated_info.append(f'状态({status})')
            
            logger.info(f'批量更新会话信息成功: {session_id}, 更新字段: {", ".join(updated_info)}')
            
        except Exception as e:
            conn.rollback()
            logger.error(f'批量更新会话信息失败: {e}')
            raise   
    
    def get_detection_sessions(self, page: int = 1, size: int = 10, patient_id: str = None) -> List[Dict]:
        """获取检测会话列表"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            offset = (page - 1) * size            
            cursor.execute('''
                        SELECT s.id, s.status, s.start_time, u.name as creator_name
                        FROM detection_sessions s                   
                        LEFT JOIN users u ON s.creator_id = u.id
                        WHERE s.patient_id = ?
                        ORDER BY s.start_time DESC
                        LIMIT ? OFFSET ?
                    ''', (patient_id, size, offset))            
            rows = cursor.fetchall()
            sessions = []            
            return sessions
            
        except Exception as e:
            logger.error(f'获取检测会话列表失败: {e}')
            raise
    
    def get_sessions_count(self, patient_id: str = None) -> int:
        """获取会话总数"""
        conn = self.get_connection()
        cursor = conn.cursor()        
        try:
            icursor.execute('SELECT COUNT(*) FROM detection_sessions WHERE patient_id = ?', (patient_id,))            
            return cursor.fetchone()[0]            
        except Exception as e:
            logger.error(f'获取会话总数失败: {e}')
            return 0
    
    def delete_detection_session(self, session_id: str):
        """删除检测会话及其相关的检测数据"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            # 验证会话是否存在
            cursor.execute('SELECT COUNT(*) FROM detection_sessions WHERE id = ?', (session_id,))
            if cursor.fetchone()[0] == 0:
                raise ValueError(f'检测会话不存在: {session_id}')
            
            # 先删除相关的检测截图数据
            cursor.execute('DELETE FROM detection_data WHERE session_id = ?', (session_id,))
            deleted_data_count = cursor.rowcount
             # 先删除相关的检测视频数据
            cursor.execute('DELETE FROM detection_video WHERE session_id = ?', (session_id,))
            deleted_video_count = cursor.rowcount
            # 再删除检测会话
            cursor.execute('DELETE FROM detection_sessions WHERE id = ?', (session_id,))            
            conn.commit()
            logger.info(f'删除检测会话: {session_id}, 同时删除了 {deleted_data_count} 条检测截图数据, {deleted_video_count} 条检测视频数据')
            
        except Exception as e:
            conn.rollback()
            logger.error(f'删除检测会话失败: {e}')
            raise
    
    def get_session_data(self, session_id: str) -> Optional[Dict]:
        """获取会话详细数据"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            # 获取会话基本信息
            cursor.execute('''
                SELECT s.*, p.name as patient_name, u.name as creator_name
                FROM detection_sessions s
                LEFT JOIN patients p ON s.patient_id = p.id
                LEFT JOIN users u ON s.creator_id = u.id
                WHERE s.id = ?
            ''', (session_id,))
            
            session_row = cursor.fetchone()
            if not session_row:
                return None
            
            session = dict(session_row)         
            
            # 获取检测数据
            cursor.execute('''
                SELECT * FROM detection_data 
                WHERE session_id = ?
                ORDER BY timestamp
            ''', (session_id,))
            
            data_rows = cursor.fetchall()
            session['data'] = []
            
            for row in data_rows:
                data_point = dict(row)
                # 解析数据JSON
                try:
                    data_point['data_value'] = json.loads(data_point['data_value'])
                except:
                    pass
                session['data'].append(data_point)
            
            # 获取检测视频数据
            cursor.execute('''
                SELECT * FROM detection_video 
                WHERE session_id = ?
                ORDER BY timestamp
            ''', (session_id,))

            video_rows = cursor.fetchall()
            session['videos'] = []

            for vrow in video_rows:
                video_item = dict(vrow)
                session['videos'].append(video_item)
            
            return session
            
        except Exception as e:
            logger.error(f'获取会话数据失败: {e}')
            return None
    
    # ==================== 检测截图数据管理 ==================== # 
    def generate_detection_data_id(self) -> str:
        """生成检测数据记录唯一标识(YYYYMMDDHHMMSS)年月日时分秒"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            # 获取当前时间
            china_tz = timezone(timedelta(hours=8))
            
            # 使用循环确保生成唯一ID，防止并发冲突
            for attempt in range(10):  # 最多尝试10次
                current_time = datetime.now(china_tz)
                data_id = current_time.strftime('%Y%m%d%H%M%S')
                
                # 检查ID是否已存在
                cursor.execute('SELECT COUNT(*) FROM detection_data WHERE id = ?', (data_id,))
                if cursor.fetchone()[0] == 0:
                    return data_id
                
                # 如果存在冲突，等待1毫秒后重试
                import time
                time.sleep(0.001)
            
            # 如果10次尝试都失败，在时间后添加随机后缀
            import random
            current_time = datetime.now(china_tz)
            base_id = current_time.strftime('%Y%m%d%H%M%S')
            suffix = str(random.randint(100, 999))
            data_id = f'{base_id}{suffix}'
            
            # 最后检查一次
            cursor.execute('SELECT COUNT(*) FROM detection_data WHERE id = ?', (data_id,))
            if cursor.fetchone()[0] == 0:
                return data_id
            
            # 如果还是冲突，使用UUID作为备用方案
            logger.warning('检测数据ID生成重试次数过多，使用UUID备用方案')
            return str(uuid.uuid4())
            
        except Exception as e:
            logger.error(f'生成检测数据ID失败: {e}')
            # 如果生成失败，使用UUID作为备用方案
            return str(uuid.uuid4())

    def save_detection_data(self, session_id: str, data: Dict[str, Any]) -> bool:
        """保存检测数据（与 detection_data 表结构保持一致）"""
        conn = self.get_connection()
        cursor = conn.cursor()

        try:
            # 使用中国时区时间
            china_time = self.get_china_time()

            # 生成检测数据记录ID
            data_id = self.generate_detection_data_id()

            # 根据表结构保存数据
            cursor.execute('''
                INSERT INTO detection_data (
                    id, session_id, head_pose, body_pose, foot_data,
                    body_image, screen_image, foot_data_image, foot1_image, foot2_image, timestamp
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ''', (
                data_id,
                session_id,
                json.dumps(data.get('head_pose')) if data.get('head_pose') else None,
                json.dumps(data.get('body_pose')) if data.get('body_pose') else None,
                json.dumps(data.get('foot_data')) if data.get('foot_data') else None,
                data.get('body_image'),
                data.get('screen_image'),
                data.get('foot_data_image'),
                data.get('foot1_image'),
                data.get('foot2_image'),
                china_time
            ))

            conn.commit()
            logger.info(f'保存检测数据: {data_id}')
            return True

        except Exception as e:
            conn.rollback()
            logger.error(f'保存检测数据失败: {e}')
            return False

    def get_detection_data_by_ids(self, data_ids: Union[str, List[str]]) -> List[Dict[str, Any]]:
            """根据1个或2个ID查询检测数据记录，返回列表

            - 支持传入单个字符串ID或ID列表（建议最多2个）
            - 解析 `data_value` 字段为对象（若解析失败则保留原字符串）
            """
            conn = self.get_connection()
            cursor = conn.cursor()

            # 规范化为列表
            ids: List[str] = [data_ids] if isinstance(data_ids, str) else list(data_ids or [])
            if not ids:
                return []

            try:
                placeholders = ','.join(['?'] * len(ids))
                sql = f'''SELECT * FROM detection_data WHERE id IN ({placeholders}) ORDER BY timestamp'''
                cursor.execute(sql, ids)
                rows = cursor.fetchall()

                results: List[Dict[str, Any]] = []
                for row in rows:
                    item = dict(row)
                    # 解析 JSON 字段
                    try:
                        item['data_value'] = json.loads(item.get('data_value', '{}'))
                    except Exception:
                        pass
                    results.append(item)

                logger.info(f'查询检测数据 {len(results)} 条，IDs: {ids}')
                return results
            except Exception as e:
                logger.error(f'查询检测数据失败: {e}')
                return []

    def delete_detection_data(self, data_ids: Union[str, List[str]]) -> bool:
        """删除检测数据记录（支持单个或多个ID）"""
        conn = self.get_connection()
        cursor = conn.cursor()

        try:
            # 规范化为列表
            ids: List[str] = [data_ids] if isinstance(data_ids, str) else list(data_ids or [])
            if not ids:
                logger.info('未提供需要删除的检测数据ID，跳过')
                return True

            placeholders = ','.join(['?'] * len(ids))
            sql = f'DELETE FROM detection_data WHERE id IN ({placeholders})'
            cursor.execute(sql, ids)
            conn.commit()
            logger.info(f'删除检测数据 {len(ids)} 条: {ids}')
            return True
        except Exception as e:
            conn.rollback()
            logger.error(f'删除检测数据失败: {e}')
            return False

    def has_session_detection_data(self, session_id: str) -> bool:
        """检查指定会话是否存在检测数据，用于判断单次检测是否有效"""
        conn = self.get_connection()
        cursor = conn.cursor()
        try:
            cursor.execute('SELECT COUNT(1) FROM detection_data WHERE session_id = ?', (session_id,))
            row = cursor.fetchone()
            count = row[0] if row else 0
            exists = count > 0
            logger.info(f'会话 {session_id} 检测数据存在: {exists} (count={count})')
            return exists
        except Exception as e:
            logger.error(f'检查会话检测数据存在失败: {e}')
            return False
   
# ==================== 检测视频数据管理 ==================== #
    def generate_detection_video_id(self) -> str:
        """生成检测视频记录唯一标识(YYYYMMDDHHMMSS)年月日时分秒"""
        conn = self.get_connection()
        cursor = conn.cursor()

        try:
            china_tz = timezone(timedelta(hours=8))
            for attempt in range(10):
                current_time = datetime.now(china_tz)
                video_id = current_time.strftime('%Y%m%d%H%M%S')
                cursor.execute('SELECT COUNT(*) FROM detection_video WHERE id = ?', (video_id,))
                if cursor.fetchone()[0] == 0:
                    return video_id
                import time
                time.sleep(0.001)

            import random
            base_id = datetime.now(china_tz).strftime('%Y%m%d%H%M%S')
            suffix = str(random.randint(100, 999))
            video_id = f'{base_id}{suffix}'
            cursor.execute('SELECT COUNT(*) FROM detection_video WHERE id = ?', (video_id,))
            if cursor.fetchone()[0] == 0:
                return video_id
            logger.warning('检测视频ID生成重试次数过多，使用UUID备用方案')
            return str(uuid.uuid4())
        except Exception as e:
            logger.error(f'生成检测视频ID失败: {e}')
            return str(uuid.uuid4())

    def save_detection_video(self, session_id: str, video: Dict[str, Any]) -> bool:
        """保存检测视频记录（与 detection_video 表结构保持一致）"""
        conn = self.get_connection()
        cursor = conn.cursor()

        try:
            video_id = self.generate_detection_video_id()
            china_time = self.get_china_time()

            cursor.execute('''
                INSERT INTO detection_video (
                    id, session_id, screen_video, body_video, foot_video1, foot_video2, timestamp
                ) VALUES (?, ?, ?, ?, ?, ?, ?)
            ''', (
                video_id,
                session_id,
                video.get('screen_video_path'),
                video.get('femtobolt_video_path'),
                video.get('camera1_video_path'),
                video.get('camera2_video_path'),
                china_time
            ))

            conn.commit()
            logger.info(f'保存检测视频: {video_id}')
            return True
        except Exception as e:
            conn.rollback()
            logger.error(f'保存检测视频失败: {e}')
            return False

    def delete_detection_video(self, video_ids: Union[str, List[str]]) -> bool:
        """删除检测视频记录（支持单个或多个ID）"""
        conn = self.get_connection()
        cursor = conn.cursor()

        try:
            # 规范化为列表
            ids: List[str] = [video_ids] if isinstance(video_ids, str) else list(video_ids or [])
            if not ids:
                logger.info('未提供需要删除的检测视频ID，跳过')
                return True

            placeholders = ','.join(['?'] * len(ids))
            sql = f'DELETE FROM detection_video WHERE id IN ({placeholders})'
            cursor.execute(sql, ids)
            conn.commit()
            logger.info(f'删除检测视频 {len(ids)} 条: {ids}')
            return True
        except Exception as e:
            conn.rollback()
            logger.error(f'删除检测视频失败: {e}')
            return False

    
    # ==================== 系统设置管理 ====================
    
    def get_setting(self, key: str, default_value: Any = None) -> Any:
        """获取系统设置"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            cursor.execute('SELECT value FROM system_settings WHERE key = ?', (key,))
            row = cursor.fetchone()
            
            if row:
                try:
                    return json.loads(row['value'])
                except:
                    return row['value']
            
            return default_value
            
        except Exception as e:
            logger.error(f'获取系统设置失败: {e}')
            return default_value
    
    def set_setting(self, key: str, value: Any, description: str = ''):
        """设置系统设置"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            value_str = json.dumps(value) if not isinstance(value, str) else value
            
            # 使用中国时区时间
            china_time = self.get_china_time()
            
            cursor.execute('''
                INSERT OR REPLACE INTO system_settings (key, value, description, updated_at)
                VALUES (?, ?, ?, ?)
            ''', (key, value_str, description, china_time))
            
            conn.commit()
            logger.info(f'设置系统设置: {key}')
            
        except Exception as e:
            conn.rollback()
            logger.error(f'设置系统设置失败: {e}')
            raise
    
    # ==================== 用户管理 ====================
    
    def register_user(self, user_data: Dict[str, Any]) -> Dict[str, Any]:
        """用户注册"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            # 检查手机号是否已存在（如果提供了手机号）
            if user_data.get('phone'):
                cursor.execute('SELECT COUNT(*) FROM users WHERE phone = ?', (user_data['phone'],))
                if cursor.fetchone()[0] > 0:
                    return {
                        'success': False,
                        'message': '手机号已存在'
                    }
            
            user_id = self.generate_user_id()
            # 密码明文存储
            password = user_data['password']
            # 使用中国时区时间
            china_time = self.get_china_time()
            
            cursor.execute('''
                INSERT INTO users (id, name, username, password, phone, is_active, user_type, register_date, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ''', (
                user_id,
                user_data['name'],
                user_data['username'],
                password,
                user_data.get('phone'),  # 手机号可选
                1,  # 新注册用户默认激活
                'user',
                china_time,
                china_time,
                china_time
            ))
            
            conn.commit()
            logger.info(f'用户注册成功: {user_data["username"]}')
            return {
                'success': True,
                'user_id': user_id,
                'message': '注册成功'
            }
            
        except sqlite3.IntegrityError:
            conn.rollback()
            logger.error(f'用户名已存在: {user_data["username"]}')
            return {
                'success': False,
                'message': '用户名已存在'
            }
        except Exception as e:
            conn.rollback()
            logger.error(f'用户注册失败: {e}')
            return {
                'success': False,
                'message': '注册失败'
            }
    
    def authenticate_user(self, username: str, password: str) -> Optional[Dict]:
        """用户登录验证"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            cursor.execute('''
                SELECT * FROM users 
                WHERE username = ? AND password = ? AND is_active = 1
            ''', (username, password))
            
            row = cursor.fetchone()
            if row:
                user = dict(row)
                # 不返回密码
                del user['password']
                logger.info(f'用户登录成功: {username}')
                return user
            else:
                logger.warning(f'用户登录失败: {username}')
                return None
                
        except Exception as e:
            logger.error(f'用户验证失败: {e}')
            return None
    
    def get_user_by_phone(self, phone: str) -> Optional[Dict]:
        """根据手机号查询用户"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            cursor.execute('SELECT * FROM users WHERE phone = ?', (phone,))
            row = cursor.fetchone()
            if row:
                user = dict(row)
                # 不返回密码
                del user['password']
                return user
            return None
        except Exception as e:
            logger.error(f'根据手机号查询用户失败: {e}')
            return None
    
    def get_users(self, page: int = 1, size: int = 10, status: str = 'all') -> List[Dict]:
        """获取用户列表"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            offset = (page - 1) * size
            
            if status == 'pending':
                cursor.execute('''
                    SELECT id, name, username, phone, register_date, is_active, user_type, created_at
                    FROM users 
                    WHERE is_active = 0 AND user_type = 'user'
                    ORDER BY created_at DESC
                    LIMIT ? OFFSET ?
                ''', (size, offset))
            elif status == 'active':
                cursor.execute('''
                    SELECT id, name, username, phone, register_date, is_active, user_type, created_at
                    FROM users 
                    WHERE is_active = 1
                    ORDER BY created_at DESC
                    LIMIT ? OFFSET ?
                ''', (size, offset))
            else:
                cursor.execute('''
                    SELECT id, name, username, phone, register_date, is_active, user_type, created_at
                    FROM users 
                    ORDER BY created_at DESC
                    LIMIT ? OFFSET ?
                ''', (size, offset))
            
            rows = cursor.fetchall()
            return [dict(row) for row in rows]
            
        except Exception as e:
            logger.error(f'获取用户列表失败: {e}')
            return []
    
    def get_users_count(self, status: str = 'all') -> int:
        """获取用户总数"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            if status == 'pending':
                cursor.execute('SELECT COUNT(*) FROM users WHERE is_active = 0 AND user_type = "user"')
            elif status == 'active':
                cursor.execute('SELECT COUNT(*) FROM users WHERE is_active = 1')
            else:
                cursor.execute('SELECT COUNT(*) FROM users')
            
            return cursor.fetchone()[0]
            
        except Exception as e:
            logger.error(f'获取用户总数失败: {e}')
            return 0
    
    def approve_user(self, user_id: str, approved: bool = True):
        """审核用户"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            # 使用中国时区时间
            china_time = self.get_china_time()
            
            cursor.execute('''
                UPDATE users SET 
                    is_active = ?, 
                    updated_at = ?
                WHERE id = ?
            ''', (1 if approved else 0, china_time, user_id))
            
            conn.commit()
            status = '通过' if approved else '拒绝'
            logger.info(f'用户审核{status}: {user_id}')
            
        except Exception as e:
            conn.rollback()
            logger.error(f'用户审核失败: {e}')
            raise
    
    def get_user(self, user_id: str) -> Optional[Dict]:
        """获取单个用户信息"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            cursor.execute('''
                SELECT id, name, username, phone, register_date, is_active, user_type, created_at, updated_at
                FROM users WHERE id = ?
            ''', (user_id,))
            
            row = cursor.fetchone()
            return dict(row) if row else None
            
        except Exception as e:
            logger.error(f'获取用户信息失败: {e}')
            return None
    
    def update_user(self, user_id: str, user_data: Dict[str, Any]):
        """更新用户信息"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            # 密码明文存储，无需加密处理
            
            # 构建更新语句
            fields = []
            values = []
            
            for key, value in user_data.items():
                if key in ['name', 'username', 'password', 'is_active', 'user_type','phone']:
                    fields.append(f'{key} = ?')
                    values.append(value)
            
            if fields:
                # 使用中国时区时间
                china_time = self.get_china_time()
                fields.append('updated_at = ?')
                values.append(china_time)
                values.append(user_id)
                
                sql = f"UPDATE users SET {', '.join(fields)} WHERE id = ?"
                cursor.execute(sql, values)
                
                conn.commit()
                logger.info(f'更新用户信息: {user_id}')
            
        except Exception as e:
            conn.rollback()
            logger.error(f'更新用户信息失败: {e}')
            raise
    
    def delete_user(self, user_id: str):
        """删除用户"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            cursor.execute('DELETE FROM users WHERE id = ?', (user_id,))
            conn.commit()
            logger.info(f'删除用户: {user_id}')
            
        except Exception as e:
            conn.rollback()
            logger.error(f'删除用户失败: {e}')
            raise
    
    def get_system_setting(self, key: str, default_value: str = None) -> Optional[str]:
        """获取系统设置值"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            cursor.execute('SELECT value FROM system_settings WHERE key = ?', (key,))
            result = cursor.fetchone()
            
            if result:
                return result[0]
            else:
                logger.warning(f'系统设置项不存在: {key}')
                return default_value
                
        except Exception as e:
            logger.error(f'获取系统设置失败: {e}')
            return default_value
    
    def set_system_setting(self, key: str, value: str, description: str = None):
        """设置系统设置值"""
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            china_time = self.get_china_time()
            
            # 检查设置项是否存在
            cursor.execute('SELECT COUNT(*) FROM system_settings WHERE key = ?', (key,))
            exists = cursor.fetchone()[0]
            
            if exists:
                # 更新现有设置
                cursor.execute('''
                    UPDATE system_settings SET value = ?, updated_at = ?
                    WHERE key = ?
                ''', (value, china_time, key))
            else:
                # 插入新设置
                cursor.execute('''
                    INSERT INTO system_settings (key, value, description, updated_at)
                    VALUES (?, ?, ?, ?)
                ''', (key, value, description, china_time))
            
            conn.commit()
            logger.info(f'设置系统配置: {key} = {value}')
            
        except Exception as e:
            conn.rollback()
            logger.error(f'设置系统配置失败: {e}')
            raise
    
    def is_empty_session(self, session_id: str) -> bool:
        """检查是否为空白会话
        
        空白会话的条件：
        1. status 为 'created'
        2. screen_video_path 为空
        3. detection_data 表中没有对应的数据
        
        Args:
            session_id: 会话ID
            
        Returns:
            bool: True表示是空白会话，False表示不是
        """
        conn = self.get_connection()
        cursor = conn.cursor()
        
        try:
            # 检查会话状态和screen_video_path
            cursor.execute('''
                SELECT status, screen_video_path 
                FROM detection_sessions 
                WHERE id = ?
            ''', (session_id,))
            
            session_row = cursor.fetchone()
            if not session_row:
                logging.info(f"[is_empty_session] 会话 {session_id} 不存在")
                return False  # 会话不存在
            
            session_data = dict(session_row)
            status = session_data.get('status')
            screen_video_path = session_data.get('screen_video_path')
            
            logging.info(f"[is_empty_session] 会话 {session_id} 状态检查: status='{status}', screen_video_path='{screen_video_path}'")
            
            # 检查条件1：status是否为created
            if status != 'checking':
                logging.info(f"[is_empty_session] 会话 {session_id} 状态不是'checking'，当前状态: '{status}' - 不是空白会话")
                return False
            
            # 检查条件2：screen_video_path是否为空
            if screen_video_path and screen_video_path.strip():
                logging.info(f"[is_empty_session] 会话 {session_id} 已有录屏路径: '{screen_video_path}' - 不是空白会话")
                return False
            
            logging.info(f"[is_empty_session] 会话 {session_id} 通过状态和录屏路径检查，继续检查检测数据")
            
            # 检查条件3：detection_data表中是否有对应数据
            cursor.execute('''
                SELECT COUNT(*) as count 
                FROM detection_data 
                WHERE session_id = ?
            ''', (session_id,))
            
            count_row = cursor.fetchone()
            data_count = count_row['count'] if count_row else 0
            
            logging.info(f"[is_empty_session] 会话 {session_id} 检测数据数量: {data_count}")
            
            # 如果没有检测数据，则认为是空白会话
            return data_count == 0
            
        except Exception as e:
            logger.error(f'检查空白会话失败: {e}')
            return False        

    def close(self):
        """关闭数据库连接"""
        if self.connection:
            self.connection.close()
            self.connection = None
            logger.info('数据库连接已关闭')