XW-AIOT/security-ai-edge
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: SQLite 添加设备配置表

Browse Source 
- 新增 camera_configs 表（摄像头配置，永不删除）
- 新增 roi_configs 表（ROI配置，永不删除）
- 添加配置表的增删改查方法
- 只清理 alert_records 表（7天过期）
This commit is contained in:
16337
2026-01-30 11:42:12 +08:00
parent ccb49e9eb9
commit 816a67f33e
1 changed files with 190 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

190
config/database.py
View File

@@ -138,6 +138,34 @@ class SQLiteManager:
ON alert_records(status)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS camera_configs (
camera_id TEXT PRIMARY KEY,
rtsp_url TEXT NOT NULL,
camera_name TEXT,
status BOOLEAN DEFAULT 1,
enabled BOOLEAN DEFAULT 1,
location TEXT,
extra_params TEXT,
updated_at TEXT
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS roi_configs (
roi_id TEXT PRIMARY KEY,
camera_id TEXT NOT NULL,
roi_type TEXT NOT NULL,
coordinates TEXT NOT NULL,
algorithm_type TEXT NOT NULL,
alert_threshold INTEGER DEFAULT 3,
alert_cooldown INTEGER DEFAULT 300,
enabled BOOLEAN DEFAULT 1,
extra_params TEXT,
updated_at TEXT
)
""")
self._conn.commit()
def _start_background_threads(self):
@@ -368,6 +396,168 @@ class SQLiteManager:
self._conn.close()
logger.info("SQLite 数据库已关闭")
def save_camera_config(self, camera_id: str, rtsp_url: str, **kwargs) -> bool:
"""保存摄像头配置"""
try:
cursor = self._conn.cursor()
now = datetime.now().isoformat()
cursor.execute("""
INSERT OR REPLACE INTO camera_configs (
camera_id, rtsp_url, camera_name, status, enabled,
location, extra_params, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (
camera_id, rtsp_url,
kwargs.get('camera_name'),
kwargs.get('status', True),
kwargs.get('enabled', True),
kwargs.get('location'),
str(kwargs.get('extra_params')) if kwargs.get('extra_params') else None,
now
))
self._conn.commit()
return True
except Exception as e:
logger.error(f"保存摄像头配置失败: {e}")
return False
def get_camera_config(self, camera_id: str) -> Optional[Dict[str, Any]]:
"""获取摄像头配置"""
try:
cursor = self._conn.cursor()
cursor.execute("SELECT * FROM camera_configs WHERE camera_id = ?", (camera_id,))
row = cursor.fetchone()
if row:
columns = ['camera_id', 'rtsp_url', 'camera_name', 'status',
'enabled', 'location', 'extra_params', 'updated_at']
return dict(zip(columns, row))
return None
except Exception as e:
logger.error(f"获取摄像头配置失败: {e}")
return None
def get_all_camera_configs(self) -> List[Dict[str, Any]]:
"""获取所有摄像头配置"""
try:
cursor = self._conn.cursor()
cursor.execute("SELECT * FROM camera_configs ORDER BY camera_id")
columns = ['camera_id', 'rtsp_url', 'camera_name', 'status',
'enabled', 'location', 'extra_params', 'updated_at']
return [dict(zip(columns, row)) for row in cursor.fetchall()]
except Exception as e:
logger.error(f"获取所有摄像头配置失败: {e}")
return []
def delete_camera_config(self, camera_id: str) -> bool:
"""删除摄像头配置"""
try:
cursor = self._conn.cursor()
cursor.execute("DELETE FROM camera_configs WHERE camera_id = ?", (camera_id,))
self._conn.commit()
return cursor.rowcount > 0
except Exception as e:
logger.error(f"删除摄像头配置失败: {e}")
return False
def save_roi_config(self, roi_id: str, camera_id: str, roi_type: str,
coordinates: List, algorithm_type: str, **kwargs) -> bool:
"""保存ROI配置"""
try:
cursor = self._conn.cursor()
now = datetime.now().isoformat()
cursor.execute("""
INSERT OR REPLACE INTO roi_configs (
roi_id, camera_id, roi_type, coordinates, algorithm_type,
alert_threshold, alert_cooldown, enabled, extra_params, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
roi_id, camera_id, roi_type, str(coordinates), algorithm_type,
kwargs.get('alert_threshold', 3),
kwargs.get('alert_cooldown', 300),
kwargs.get('enabled', True),
str(kwargs.get('extra_params')) if kwargs.get('extra_params') else None,
now
))
self._conn.commit()
return True
except Exception as e:
logger.error(f"保存ROI配置失败: {e}")
return False
def get_roi_config(self, roi_id: str) -> Optional[Dict[str, Any]]:
"""获取ROI配置"""
try:
cursor = self._conn.cursor()
cursor.execute("SELECT * FROM roi_configs WHERE roi_id = ?", (roi_id,))
row = cursor.fetchone()
if row:
columns = ['roi_id', 'camera_id', 'roi_type', 'coordinates',
'algorithm_type', 'alert_threshold', 'alert_cooldown',
'enabled', 'extra_params', 'updated_at']
result = dict(zip(columns, row))
try:
result['coordinates'] = eval(result['coordinates'])
except:
pass
return result
return None
except Exception as e:
logger.error(f"获取ROI配置失败: {e}")
return None
def get_rois_by_camera(self, camera_id: str) -> List[Dict[str, Any]]:
"""获取指定摄像头的所有ROI配置"""
try:
cursor = self._conn.cursor()
cursor.execute("SELECT * FROM roi_configs WHERE camera_id = ?", (camera_id,))
columns = ['roi_id', 'camera_id', 'roi_type', 'coordinates',
'algorithm_type', 'alert_threshold', 'alert_cooldown',
'enabled', 'extra_params', 'updated_at']
results = []
for row in cursor.fetchall():
r = dict(zip(columns, row))
try:
r['coordinates'] = eval(r['coordinates'])
except:
pass
results.append(r)
return results
except Exception as e:
logger.error(f"获取ROI配置失败: {e}")
return []
def get_all_roi_configs(self) -> List[Dict[str, Any]]:
"""获取所有ROI配置"""
try:
cursor = self._conn.cursor()
cursor.execute("SELECT * FROM roi_configs ORDER BY camera_id, roi_id")
columns = ['roi_id', 'camera_id', 'roi_type', 'coordinates',
'algorithm_type', 'alert_threshold', 'alert_cooldown',
'enabled', 'extra_params', 'updated_at']
results = []
for row in cursor.fetchall():
r = dict(zip(columns, row))
try:
r['coordinates'] = eval(r['coordinates'])
except:
pass
results.append(r)
return results
except Exception as e:
logger.error(f"获取所有ROI配置失败: {e}")
return []
def delete_roi_config(self, roi_id: str) -> bool:
"""删除ROI配置"""
try:
cursor = self._conn.cursor()
cursor.execute("DELETE FROM roi_configs WHERE roi_id = ?", (roi_id,))
self._conn.commit()
return cursor.rowcount > 0
except Exception as e:
logger.error(f"删除ROI配置失败: {e}")
return False
def get_sqlite_manager() -> SQLiteManager: