1553 lines
64 KiB
Python
1553 lines
64 KiB
Python
#!/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('数据库连接已关闭') |