fix: [recorder] 修复视频录制时设置key的续约时间异常

isc_video_record/recorder.py

@@ -94,6 +94,7 @@ class StreamRecorder:
         on_message_callback = functools.partial(
             on_message, args=(self.connection, threads))
 
+        # 设置消费能力
         self.channel.basic_qos(prefetch_count=dynaconf.settings.get('PREFETCH_COUNT', 10))
         self.channel.basic_consume(on_message_callback=on_message_callback,
                                    queue=self.queue_name)
@@ -127,25 +128,30 @@ class ProcessMessage:
         log.info('thread_id:%s:%s: events count: %s', self.thread_id, self.body['camera_code'], len(events))
 
         for inx, event in enumerate(events):
-            _continue = False
-            expire_time = int(self.body['ex'] * 2 + 1)
+            expire_time = int((event['end_time'] - event['start_time']).total_seconds() * 2 + 1)
             pipe = redis_connect()
 
-            if pipe.ttl(self.camera_key) == -2:
-                if pipe.set(self.camera_key, self.thread_id, nx=True, ex=expire_time):
-                    _continue = True
+            if pipe.ttl(self.camera_key) > expire_time:
+                _continue = True
             elif -1 < pipe.ttl(self.camera_key) < expire_time \
                     and pipe.get(self.camera_key) == str(self.thread_id) \
                     and pipe.expire(self.camera_key, expire_time):
                 _continue = True
+            elif pipe.ttl(self.camera_key) == -2 \
+                    and pipe.set(self.camera_key, self.thread_id, nx=True, ex=expire_time):
+                _continue = True
+            else:
+                _continue = False
 
             if _continue:
+                # 续约成功，本次任务继续
                 # 判定当前分布式锁是本线程设置的
                 log.info('thread_id:%s:%s,ttl:%s', self.thread_id, self.camera_key,
                          pipe.ttl(self.camera_key))
                 self.recording(event)
                 continue
             else:
+                log.info('thread_id:%s:%s, expire failed', self.thread_id, self.camera_key)
                 break
 
         else:

isc_video_record/utils/isc_client.py

@@ -179,9 +179,8 @@ class HikVisionClient(object):
 
     def get_camera_online(self, camera_index_code):
         """
-        获取监控点在线状态(貌似无用)
+        获取监控点在线状态
         """
-        # uri = '/artemis/api/nms/v1/online/encode_device/get'
         uri = '/artemis/api/nms/v1/online/camera/get'
         body = {
             'indexCodes': [camera_index_code],

tests/test_playback.py

@@ -5,7 +5,7 @@ from intelab_python_sdk.logger import log_init
 from dynaconf import settings
 
 from isc_video_record.utils.isc_client import HikVisionClient
-from isc_video_record.recorder import StreamRecorder
+from isc_video_record.utils.record_utils import record_thread, get_video_duration
 
 
 tz = pytz.timezone('Asia/Shanghai')
@@ -20,10 +20,36 @@ client = HikVisionClient(config.get('KEY'), config.get('SECRET'),
 
 
 def main():
-    # rtmp_20210426T003128_20210426T003247.mp4
-    start_time = datetime(2021, 5, 4, 15, 53, 52)
-    end_time = datetime(2021, 5, 4, 15, 55, 23)
-    StreamRecorder.recorder('9e6768059bd74f6085eec605b7658e8f', start_time, end_time, 'D86639983')
+
+    start_time = datetime(2021, 5, 18, 16, 7, 0).astimezone(tz)
+    end_time = datetime(2021, 5, 18, 16, 15, 0).astimezone(tz)
+    res = client.get_cameras_playback_urls('3dd6c37895a54e5084a390e33a7c806a',
+                                           client.iso_format(start_time), client.iso_format(end_time))
+    print(res)
+    for event in res:
+        cur_start_time = max(event['start_time'], start_time)
+        cur_end_time = min(event['end_time'], end_time)
+        print(event)
+        file_info, error_log = stream_record(event['stream_url'], cur_start_time, cur_end_time)
+        print(file_info)
+
+
+def stream_record(stream, start_time, end_time):
+    start_time = start_time.strftime('%Y%m%dT%H%M%S')
+    end_time = end_time.strftime('%Y%m%dT%H%M%S')
+
+    if stream['protocol'] == 'rtsp':
+        stream_url = stream['url']
+    else:
+        stream_url = '{}?beginTime={}&endTime={}&{}'.format(stream['url'],
+                                                            start_time,
+                                                            end_time, stream['extra_args'])
+    file_name = os.path.join(
+        video_path, 'rtmp_{}_{}_{}.mp4'.format('y', start_time, end_time))
+    # TODO 多进程处理
+    print(stream_url, start_time, end_time)
+    record_thread(stream_url, file_name, thread_name='y')
+    return get_video_duration(file_name)
 
 
 if __name__ == '__main__':