462 lines
15 KiB
Python
462 lines
15 KiB
Python
|
"""
|
|||
|
|
云端同步服务
|
|||
|
|
|
|||
|
|
实现"云端为主、本地为辅"的双层数据存储架构：
|
|||
|
|
- 配置双向同步
|
|||
|
|
- 报警单向上报
|
|||
|
|
- 设备状态上报
|
|||
|
|
- 断网容错机制
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import os
|
|||
|
|
import sys
|
|||
|
|
import time
|
|||
|
|
import threading
|
|||
|
|
import logging
|
|||
|
|
from datetime import datetime
|
|||
|
|
from typing import Optional, List, Dict, Any
|
|||
|
|
from queue import Queue, Empty
|
|||
|
|
from dataclasses import dataclass
|
|||
|
|
from enum import Enum
|
|||
|
|
|
|||
|
|
import requests
|
|||
|
|
from sqlalchemy.orm import Session
|
|||
|
|
|
|||
|
|
# 添加项目根目录到路径
|
|||
|
|
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|||
|
|
sys.path.insert(0, project_root)
|
|||
|
|
|
|||
|
|
from config import get_config
|
|||
|
|
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class SyncStatus(Enum):
|
|||
|
|
"""同步状态"""
|
|||
|
|
PENDING = "pending"
|
|||
|
|
SYNCING = "syncing"
|
|||
|
|
SUCCESS = "success"
|
|||
|
|
FAILED = "failed"
|
|||
|
|
RETRY = "retry"
|
|||
|
|
|
|||
|
|
|
|||
|
|
class EntityType(Enum):
|
|||
|
|
"""实体类型"""
|
|||
|
|
CAMERA = "camera"
|
|||
|
|
ROI = "roi"
|
|||
|
|
ALARM = "alarm"
|
|||
|
|
STATUS = "status"
|
|||
|
|
|
|||
|
|
|
|||
|
|
@dataclass
|
|||
|
|
class SyncTask:
|
|||
|
|
"""同步任务"""
|
|||
|
|
entity_type: EntityType
|
|||
|
|
entity_id: int
|
|||
|
|
operation: str # create, update, delete
|
|||
|
|
data: Optional[Dict[str, Any]] = None
|
|||
|
|
status: SyncStatus = SyncStatus.PENDING
|
|||
|
|
retry_count: int = 0
|
|||
|
|
error_message: Optional[str] = None
|
|||
|
|
created_at: datetime = None
|
|||
|
|
|
|||
|
|
def __post_init__(self):
|
|||
|
|
if self.created_at is None:
|
|||
|
|
self.created_at = datetime.utcnow()
|
|||
|
|
|
|||
|
|
|
|||
|
|
class CloudAPIClient:
|
|||
|
|
"""云端 API 客户端"""
|
|||
|
|
|
|||
|
|
def __init__(self, base_url: str, api_key: str, device_id: str):
|
|||
|
|
self.base_url = base_url.rstrip('/')
|
|||
|
|
self.api_key = api_key
|
|||
|
|
self.device_id = device_id
|
|||
|
|
self.session = requests.Session()
|
|||
|
|
self.session.headers.update({
|
|||
|
|
'Authorization': f'Bearer {api_key}',
|
|||
|
|
'Content-Type': 'application/json',
|
|||
|
|
'X-Device-ID': device_id
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
def request(self, method: str, path: str, **kwargs) -> requests.Response:
|
|||
|
|
"""发送 API 请求"""
|
|||
|
|
url = f"{self.base_url}{path}"
|
|||
|
|
response = self.session.request(method, url, **kwargs)
|
|||
|
|
response.raise_for_status()
|
|||
|
|
return response
|
|||
|
|
|
|||
|
|
def get(self, path: str, **kwargs):
|
|||
|
|
return self.request('GET', path, **kwargs)
|
|||
|
|
|
|||
|
|
def post(self, path: str, **kwargs):
|
|||
|
|
return self.request('POST', path, **kwargs)
|
|||
|
|
|
|||
|
|
def put(self, path: str, **kwargs):
|
|||
|
|
return self.request('PUT', path, **kwargs)
|
|||
|
|
|
|||
|
|
def delete(self, path: str, **kwargs):
|
|||
|
|
return self.request('DELETE', path, **kwargs)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class SyncService:
|
|||
|
|
"""云端同步服务"""
|
|||
|
|
|
|||
|
|
def __init__(self):
|
|||
|
|
config = get_config()
|
|||
|
|
self.config = config
|
|||
|
|
|
|||
|
|
# 云端配置
|
|||
|
|
self.cloud_enabled = config.cloud.enabled
|
|||
|
|
self.cloud_url = config.cloud.api_url
|
|||
|
|
self.api_key = config.cloud.api_key
|
|||
|
|
self.device_id = config.cloud.device_id
|
|||
|
|
|
|||
|
|
# 同步配置
|
|||
|
|
self.sync_interval = config.cloud.sync_interval
|
|||
|
|
self.alarm_retry_interval = config.cloud.alarm_retry_interval
|
|||
|
|
self.status_report_interval = config.cloud.status_report_interval
|
|||
|
|
self.max_retries = config.cloud.max_retries
|
|||
|
|
|
|||
|
|
# 客户端
|
|||
|
|
self.client: Optional[CloudAPIClient] = None
|
|||
|
|
if self.cloud_enabled:
|
|||
|
|
self.client = CloudAPIClient(
|
|||
|
|
self.cloud_url,
|
|||
|
|
self.api_key,
|
|||
|
|
self.device_id
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# 任务队列
|
|||
|
|
self.sync_queue: Queue = Queue()
|
|||
|
|
self.alarm_queue: Queue = Queue()
|
|||
|
|
|
|||
|
|
# 状态
|
|||
|
|
self.running = False
|
|||
|
|
self.threads: List[threading.Thread] = []
|
|||
|
|
self.network_status = "disconnected"
|
|||
|
|
|
|||
|
|
# 重试配置
|
|||
|
|
self.retry_delays = [60, 300, 900, 3600] # 1分钟, 5分钟, 15分钟, 1小时
|
|||
|
|
|
|||
|
|
def start(self):
|
|||
|
|
"""启动同步服务"""
|
|||
|
|
if self.running:
|
|||
|
|
logger.warning("同步服务已在运行")
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
self.running = True
|
|||
|
|
|
|||
|
|
if self.cloud_enabled:
|
|||
|
|
logger.info(f"启动云端同步服务，设备ID: {self.device_id}")
|
|||
|
|
else:
|
|||
|
|
logger.info("云端同步已禁用，使用本地模式")
|
|||
|
|
|
|||
|
|
# 启动工作线程
|
|||
|
|
self.threads.append(threading.Thread(target=self._sync_worker, daemon=True))
|
|||
|
|
self.threads.append(threading.Thread(target=self._alarm_worker, daemon=True))
|
|||
|
|
self.threads.append(threading.Thread(target=self._status_worker, daemon=True))
|
|||
|
|
|
|||
|
|
for thread in self.threads:
|
|||
|
|
thread.start()
|
|||
|
|
|
|||
|
|
logger.info("同步服务已启动")
|
|||
|
|
|
|||
|
|
def stop(self):
|
|||
|
|
"""停止同步服务"""
|
|||
|
|
self.running = False
|
|||
|
|
|
|||
|
|
for thread in self.threads:
|
|||
|
|
if thread.is_alive():
|
|||
|
|
thread.join(timeout=5)
|
|||
|
|
|
|||
|
|
logger.info("同步服务已停止")
|
|||
|
|
|
|||
|
|
def _sync_worker(self):
|
|||
|
|
"""配置同步工作线程"""
|
|||
|
|
while self.running:
|
|||
|
|
try:
|
|||
|
|
task = self.sync_queue.get(timeout=1)
|
|||
|
|
self._execute_sync(task)
|
|||
|
|
except Empty:
|
|||
|
|
self._check_network_status()
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"同步工作线程异常: {e}")
|
|||
|
|
|
|||
|
|
def _alarm_worker(self):
|
|||
|
|
"""报警上报工作线程"""
|
|||
|
|
while self.running:
|
|||
|
|
try:
|
|||
|
|
alarm_id = self.alarm_queue.get(timeout=1)
|
|||
|
|
self._upload_alarm(alarm_id)
|
|||
|
|
except Empty:
|
|||
|
|
continue
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"报警上报工作线程异常: {e}")
|
|||
|
|
|
|||
|
|
def _status_worker(self):
|
|||
|
|
"""状态上报工作线程"""
|
|||
|
|
while self.running:
|
|||
|
|
try:
|
|||
|
|
if self.network_status == "connected":
|
|||
|
|
self._report_status()
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"状态上报失败: {e}")
|
|||
|
|
time.sleep(self.status_report_interval)
|
|||
|
|
|
|||
|
|
def _check_network_status(self):
|
|||
|
|
"""检查网络状态"""
|
|||
|
|
if not self.cloud_enabled:
|
|||
|
|
self.network_status = "disabled"
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
self.client.get('/health')
|
|||
|
|
self.network_status = "connected"
|
|||
|
|
except:
|
|||
|
|
self.network_status = "disconnected"
|
|||
|
|
|
|||
|
|
def _execute_sync(self, task: SyncTask):
|
|||
|
|
"""执行同步任务"""
|
|||
|
|
logger.info(f"执行同步任务: {task.entity_type.value}/{task.entity_id} ({task.operation})")
|
|||
|
|
|
|||
|
|
task.status = SyncStatus.SYNCING
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
if task.entity_type == EntityType.CAMERA:
|
|||
|
|
self._sync_camera(task)
|
|||
|
|
elif task.entity_type == EntityType.ROI:
|
|||
|
|
self._sync_roi(task)
|
|||
|
|
|
|||
|
|
task.status = SyncStatus.SUCCESS
|
|||
|
|
logger.info(f"同步成功: {task.entity_type.value}/{task.entity_id}")
|
|||
|
|
|
|||
|
|
except requests.exceptions.RequestException as e:
|
|||
|
|
self._handle_sync_error(task, str(e))
|
|||
|
|
except Exception as e:
|
|||
|
|
task.status = SyncStatus.FAILED
|
|||
|
|
task.error_message = str(e)
|
|||
|
|
logger.error(f"同步失败: {task.entity_type.value}/{task.entity_id}: {e}")
|
|||
|
|
|
|||
|
|
def _handle_sync_error(self, task: SyncTask, error: str):
|
|||
|
|
"""处理同步错误"""
|
|||
|
|
task.retry_count += 1
|
|||
|
|
|
|||
|
|
if task.retry_count < self.max_retries:
|
|||
|
|
task.status = SyncStatus.RETRY
|
|||
|
|
delay = self.retry_delays[task.retry_count - 1]
|
|||
|
|
task.error_message = f"第{task.retry_count}次失败: {error}"
|
|||
|
|
logger.warning(f"同步重试 ({task.retry_count}/{self.max_retries}): {task.entity_type.value}/{task.entity_id}")
|
|||
|
|
# 重新入队
|
|||
|
|
time.sleep(delay)
|
|||
|
|
self.sync_queue.put(task)
|
|||
|
|
else:
|
|||
|
|
task.status = SyncStatus.FAILED
|
|||
|
|
task.error_message = f"已超过最大重试次数: {error}"
|
|||
|
|
logger.error(f"同步失败，已达最大重试次数: {task.entity_type.value}/{task.entity_id}")
|
|||
|
|
|
|||
|
|
def _sync_camera(self, task: SyncTask):
|
|||
|
|
"""同步摄像头配置"""
|
|||
|
|
if task.operation == 'update':
|
|||
|
|
self.client.put(f"/api/v1/cameras/{task.entity_id}", json=task.data)
|
|||
|
|
elif task.operation == 'delete':
|
|||
|
|
self.client.delete(f"/api/v1/cameras/{task.entity_id}")
|
|||
|
|
|
|||
|
|
def _sync_roi(self, task: SyncTask):
|
|||
|
|
"""同步 ROI 配置"""
|
|||
|
|
if task.operation == 'update':
|
|||
|
|
self.client.put(f"/api/v1/rois/{task.entity_id}", json=task.data)
|
|||
|
|
elif task.operation == 'delete':
|
|||
|
|
self.client.delete(f"/api/v1/rois/{task.entity_id}")
|
|||
|
|
|
|||
|
|
def _upload_alarm(self, alarm_id: int):
|
|||
|
|
"""上传报警记录"""
|
|||
|
|
from db.crud import get_alarm_by_id, update_alarm_status
|
|||
|
|
from db.models import get_session_factory
|
|||
|
|
|
|||
|
|
SessionLocal = get_session_factory()
|
|||
|
|
db = SessionLocal()
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
alarm = get_alarm_by_id(db, alarm_id)
|
|||
|
|
if not alarm:
|
|||
|
|
logger.warning(f"报警记录不存在: {alarm_id}")
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
# 准备数据
|
|||
|
|
alarm_data = {
|
|||
|
|
'device_id': self.device_id,
|
|||
|
|
'camera_id': alarm.camera_id,
|
|||
|
|
'alarm_type': alarm.event_type,
|
|||
|
|
'confidence': alarm.confidence,
|
|||
|
|
'timestamp': alarm.created_at.isoformat() if alarm.created_at else None,
|
|||
|
|
'region': alarm.region_data
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 上传图片
|
|||
|
|
if alarm.snapshot_path and os.path.exists(alarm.snapshot_path):
|
|||
|
|
with open(alarm.snapshot_path, 'rb') as f:
|
|||
|
|
files = {'file': f}
|
|||
|
|
response = self.client.post('/api/v1/alarms/images', files=files)
|
|||
|
|
alarm_data['image_url'] = response.json().get('data', {}).get('url')
|
|||
|
|
|
|||
|
|
# 上报报警
|
|||
|
|
response = self.client.post('/api/v1/alarms/report', json=alarm_data)
|
|||
|
|
cloud_id = response.json().get('data', {}).get('alarm_id')
|
|||
|
|
|
|||
|
|
# 更新本地状态
|
|||
|
|
update_alarm_status(db, alarm_id, status='uploaded', cloud_id=cloud_id)
|
|||
|
|
logger.info(f"报警上报成功: {alarm_id} -> 云端ID: {cloud_id}")
|
|||
|
|
|
|||
|
|
except requests.exceptions.RequestException as e:
|
|||
|
|
update_alarm_status(db, alarm_id, status='retry', error_message=str(e))
|
|||
|
|
self.alarm_queue.put(alarm_id) # 重试
|
|||
|
|
except Exception as e:
|
|||
|
|
update_alarm_status(db, alarm_id, status='failed', error_message=str(e))
|
|||
|
|
logger.error(f"报警处理失败: {alarm_id}: {e}")
|
|||
|
|
finally:
|
|||
|
|
db.close()
|
|||
|
|
|
|||
|
|
def _report_status(self):
|
|||
|
|
"""上报设备状态"""
|
|||
|
|
import psutil
|
|||
|
|
from db.crud import get_active_camera_count
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
metrics = {
|
|||
|
|
'device_id': self.device_id,
|
|||
|
|
'cpu_percent': psutil.cpu_percent(),
|
|||
|
|
'memory_percent': psutil.virtual_memory().percent,
|
|||
|
|
'disk_usage': psutil.disk_usage('/').percent,
|
|||
|
|
'timestamp': datetime.utcnow().isoformat()
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
self.client.post('/api/v1/devices/status', json=metrics)
|
|||
|
|
logger.debug(f"设备状态上报成功: CPU={metrics['cpu_percent']}%")
|
|||
|
|
except requests.exceptions.RequestException as e:
|
|||
|
|
logger.warning(f"设备状态上报失败: {e}")
|
|||
|
|
|
|||
|
|
# 公共接口
|
|||
|
|
|
|||
|
|
def queue_camera_sync(self, camera_id: int, operation: str = 'update', data: Dict[str, Any] = None):
|
|||
|
|
"""将摄像头同步加入队列"""
|
|||
|
|
task = SyncTask(
|
|||
|
|
entity_type=EntityType.CAMERA,
|
|||
|
|
entity_id=camera_id,
|
|||
|
|
operation=operation,
|
|||
|
|
data=data
|
|||
|
|
)
|
|||
|
|
self.sync_queue.put(task)
|
|||
|
|
|
|||
|
|
def queue_roi_sync(self, roi_id: int, operation: str = 'update', data: Dict[str, Any] = None):
|
|||
|
|
"""将 ROI 同步加入队列"""
|
|||
|
|
task = SyncTask(
|
|||
|
|
entity_type=EntityType.ROI,
|
|||
|
|
entity_id=roi_id,
|
|||
|
|
operation=operation,
|
|||
|
|
data=data
|
|||
|
|
)
|
|||
|
|
self.sync_queue.put(task)
|
|||
|
|
|
|||
|
|
def queue_alarm_upload(self, alarm_id: int):
|
|||
|
|
"""将报警上传加入队列"""
|
|||
|
|
self.alarm_queue.put(alarm_id)
|
|||
|
|
|
|||
|
|
def sync_config_from_cloud(self, db: Session) -> Dict[str, int]:
|
|||
|
|
"""从云端拉取配置"""
|
|||
|
|
result = {'cameras': 0, 'rois': 0}
|
|||
|
|
|
|||
|
|
if not self.cloud_enabled:
|
|||
|
|
logger.info("云端同步已禁用，跳过配置拉取")
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
logger.info("从云端拉取配置...")
|
|||
|
|
|
|||
|
|
# 拉取设备配置
|
|||
|
|
response = self.client.get(f"/api/v1/devices/{self.device_id}/config")
|
|||
|
|
config = response.json().get('data', {})
|
|||
|
|
|
|||
|
|
# 处理摄像头
|
|||
|
|
cameras = config.get('cameras', [])
|
|||
|
|
for cloud_cam in cameras:
|
|||
|
|
self._merge_camera(db, cloud_cam)
|
|||
|
|
result['cameras'] += 1
|
|||
|
|
|
|||
|
|
logger.info(f"从云端拉取配置完成: {result['cameras']} 个摄像头")
|
|||
|
|
|
|||
|
|
except requests.exceptions.RequestException as e:
|
|||
|
|
logger.error(f"从云端拉取配置失败: {e}")
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"处理云端配置时出错: {e}")
|
|||
|
|
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
def _merge_camera(self, db: Session, cloud_data: Dict[str, Any]):
|
|||
|
|
"""合并摄像头配置"""
|
|||
|
|
from db.crud import get_camera_by_cloud_id, create_camera, update_camera
|
|||
|
|
from db.models import Camera
|
|||
|
|
|
|||
|
|
cloud_id = cloud_data.get('id')
|
|||
|
|
existing = get_camera_by_cloud_id(db, cloud_id)
|
|||
|
|
|
|||
|
|
if existing:
|
|||
|
|
# 更新现有记录
|
|||
|
|
if not existing.pending_sync:
|
|||
|
|
update_camera(db, existing.id, {
|
|||
|
|
'name': cloud_data.get('name'),
|
|||
|
|
'rtsp_url': cloud_data.get('rtsp_url'),
|
|||
|
|
'enabled': cloud_data.get('enabled', True),
|
|||
|
|
'fps_limit': cloud_data.get('fps_limit', 30),
|
|||
|
|
'process_every_n_frames': cloud_data.get('process_every_n_frames', 3),
|
|||
|
|
})
|
|||
|
|
else:
|
|||
|
|
# 创建新记录
|
|||
|
|
camera = create_camera(db, {
|
|||
|
|
'name': cloud_data.get('name'),
|
|||
|
|
'rtsp_url': cloud_data.get('rtsp_url'),
|
|||
|
|
'enabled': cloud_data.get('enabled', True),
|
|||
|
|
'fps_limit': cloud_data.get('fps_limit', 30),
|
|||
|
|
'process_every_n_frames': cloud_data.get('process_every_n_frames', 3),
|
|||
|
|
})
|
|||
|
|
# 更新 cloud_id
|
|||
|
|
camera.cloud_id = cloud_id
|
|||
|
|
db.commit()
|
|||
|
|
|
|||
|
|
def get_status(self) -> Dict[str, Any]:
|
|||
|
|
"""获取同步服务状态"""
|
|||
|
|
return {
|
|||
|
|
'running': self.running,
|
|||
|
|
'cloud_enabled': self.cloud_enabled,
|
|||
|
|
'network_status': self.network_status,
|
|||
|
|
'device_id': self.device_id,
|
|||
|
|
'pending_sync': self.sync_queue.qsize(),
|
|||
|
|
'pending_alarms': self.alarm_queue.qsize(),
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 单例
|
|||
|
|
_sync_service: Optional[SyncService] = None
|
|||
|
|
|
|||
|
|
|
|||
|
|
def get_sync_service() -> SyncService:
|
|||
|
|
"""获取同步服务单例"""
|
|||
|
|
global _sync_service
|
|||
|
|
if _sync_service is None:
|
|||
|
|
_sync_service = SyncService()
|
|||
|
|
return _sync_service
|
|||
|
|
|
|||
|
|
|
|||
|
|
def start_sync_service():
|
|||
|
|
"""启动同步服务"""
|
|||
|
|
service = get_sync_service()
|
|||
|
|
service.start()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def stop_sync_service():
|
|||
|
|
"""停止同步服务"""
|
|||
|
|
global _sync_service
|
|||
|
|
if _sync_service:
|
|||
|
|
_sync_service.stop()
|
|||
|
|
_sync_service = None
|