chore: 修改不规范语法和静态类方法

isc_video_record/merger.py

@@ -40,7 +40,8 @@ class EventMergerJob:
                      time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(next_run_at)))
             time.sleep(15 * 60)
 
-    def get_scan_time(self, camera, now):
+    @staticmethod
+    def get_scan_time(camera, now):
         last_video_info = mysql.get_last_video_info(
             camera['db_table'], camera['device_code'])
         if last_video_info:
@@ -49,11 +50,14 @@ class EventMergerJob:
             last_check_time = now - timedelta(days=1)
         return last_check_time
 
-    def get_camera_local_events(self, pipe, camera, now):
+    @staticmethod
+    def get_camera_local_events(camera, now):
         """
         获取摄像头本地移动侦测事件
+        :param camera:
+        :param now: uct now
         """
-        last_check_time = self.get_scan_time(camera, now)
+        last_check_time = EventMergerJob.get_scan_time(camera, now)
         events = []
         try:
             log.info('查询摄像头%s在%s,%s的本地视频文件', camera['device_code'],
@@ -70,26 +74,23 @@ class EventMergerJob:
                 # TODO 需要重新处理事件
                 now = last_check_time + timedelta(hours=2)
                 log.warning('查询%s事件跨度较大.', camera['device_code'])
-                return self.get_camera_local_events(pipe, camera, now)
+                return EventMergerJob.get_camera_local_events(camera, now)
         except Exception as e:
             # TODO 网络请求错误重试
             log.exception(e)
-        else:
-            last_check_time_key = LAST_CHECK_TIME_KEY.format(
-                camera['device_code'])
-            res = pipe.set(last_check_time_key,
-                           now.strftime('%Y-%m-%d %H:%M:%S'))
-            if not res:
-                events = []
+
         return events
 
-    def get_alarm_events(self, pipe, camera, now):
+    @staticmethod
+    def get_alarm_events(camera, now):
         """
         摄像头录制计划为全天录制时
-
         查询指定时间段内摄像头是否有移动告警消息并合并成事件
+
+        :param camera:
+        :param now: uct now
         """
-        last_check_time = self.get_scan_time(camera, now)
+        last_check_time = EventMergerJob.get_scan_time(camera, now)
 
         pre_event = PreEvent(
             last_check_time.strftime('%Y-%m-%d %H:%M:%S'),
@@ -108,14 +109,9 @@ class EventMergerJob:
                 'end_time': now.astimezone(tz)
             }]
 
-        last_check_time_key = LAST_CHECK_TIME_KEY.format(camera['device_code'])
-        res = pipe.set(last_check_time_key,
-                       now.strftime('%Y-%m-%d %H:%M:%S'))
-        if not res:
-            events = []
         return events
 
-    def process_camera(self, pipe, camera):
+    def process_camera(self, camera):
         body = {
             'camera_code': camera['device_code'],
             'camera_index': camera['point_index_code'],
@@ -135,7 +131,7 @@ class EventMergerJob:
         call_get_events_func = self.get_alarm_events if camera['video_plan_type'] == 1 \
             else self.get_camera_local_events
 
-        for event in call_get_events_func(pipe, camera, now):
+        for event in call_get_events_func(camera, now):
             insert_video_info(
                 camera['db_table'], camera['device_code'],
                 event['start_time'].astimezone(pytz.utc),
@@ -157,13 +153,13 @@ class EventMergerJob:
 
     def run(self):
 
-        pipe = redis_connect()
         total_video_duration = 0
         camera_count = 0
 
         for camera in get_camera_info():
+            # TODO 根据摄像头AI配置启动移动侦测回放录制
             try:
-                camera_event_duration = self.process_camera(pipe, camera)
+                camera_event_duration = self.process_camera(camera)
                 total_video_duration += camera_event_duration
                 if camera_event_duration > 10:
                     camera_count += 1
@@ -174,15 +170,14 @@ class EventMergerJob:
 
             log.info('=====================')
 
-        pipe.close()
-
         try:
             self.write_video_duration_to_influx(
                 total_video_duration, camera_count)
         except Exception as e:
             log.exception(e)
 
-    def write_video_duration_to_influx(self, total_video_duration, camera_count):
+    @staticmethod
+    def write_video_duration_to_influx(total_video_duration, camera_count):
         influxdb.reconnect()
         influxdb.write_points([{
             'measurement': 'pre_video_duration',

isc_video_record/recorder.py

@@ -57,7 +57,7 @@ class StreamRecorder:
             try:
                 if pipe.set(camera_key, thread_id, nx=True, ex=100):
                     log.info('setnx:%s:ttl:100', camera_key)
-                    process_message = ProcessMessage(camera_key, thread_id, body)
+                    process_message = ProcessMessage(camera_key, body)
                     ack = process_message.process()
 
                 else:
@@ -115,9 +115,9 @@ class StreamRecorder:
 
 class ProcessMessage:
 
-    def __init__(self, camera_key, thread_id, body):
+    def __init__(self, camera_key, body):
         self.camera_key = camera_key
-        self.thread_id = thread_id
+        self.thread_id = threading.get_ident()
         self.body = body
 
     def process(self):

setup.py

@@ -24,7 +24,7 @@ requires = [
 
 setuptools.setup(
     name='isc-video-record',
-    version='1.0.0b8',
+    version='1.0.0b9',
     description='ISC motion detection playback video stream recording service.',
     long_description=long_description,
     long_description_content_type='text/markdown',