feat: [merger] 根据摄像头的录像计划使用不同的方案

isc_video_record/db/mysql.py

@@ -89,7 +89,8 @@ def get_camera_info(cursor, conn):
                    'video_data_motion_', mod(biz_type, 8)
             ) as 'db_table',
             ai_config_support, device_code, service_type, biz_type,
-            region_path_name
+            region_path_name,
+            1 as video_plan_type
         from camera_info
             join camera_ai_config cac
             on camera_info.id = cac.camera_info_id

isc_video_record/merger.py

@@ -40,10 +40,7 @@ class EventMergerJob:
                      time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(next_run_at)))
             time.sleep(15 * 60)
 
-    def get_camera_local_events(self, pipe, camera, now):
-        """
-        获取摄像头本地移动侦测事件
-        """
+    def get_scan_time(self, pipe, camera, now):
         last_check_time_key = LAST_CHECK_TIME_KEY.format(camera['device_code'])
         last_check_time = pipe.get(last_check_time_key)
         if not last_check_time:
@@ -60,7 +57,13 @@ class EventMergerJob:
             if now - last_check_time > timedelta(days=1):
                 # 时间跨度超过一天，则分割成两个小时
                 now = last_check_time + timedelta(hours=2)
+        return last_check_time, now
 
+    def get_camera_local_events(self, pipe, camera, now):
+        """
+        获取摄像头本地移动侦测事件
+        """
+        last_check_time, now = self.get_scan_time(pipe, camera, now)
         events = []
         try:
             log.info('查询摄像头%s在%s,%s的本地视频文件', camera['device_code'],
@@ -82,6 +85,7 @@ class EventMergerJob:
             # 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:
@@ -101,12 +105,14 @@ class EventMergerJob:
             self.send_mq_message(body)
         return body.get('ex', 0)
 
-    def get_new_events(self, last_check_time, now, camera):
+    def get_alarm_events(self, pipe, camera, now):
         """
         摄像头录制计划为全天录制时
 
         查询指定时间段内摄像头是否有移动告警消息并合并成事件
         """
+        last_check_time, now = self.get_scan_time(pipe, camera, now)
+
         pre_event = PreEvent(
             last_check_time.strftime('%Y-%m-%d %H:%M:%S'),
             now.strftime('%Y-%m-%d %H:%M:%S')
@@ -135,7 +141,10 @@ class EventMergerJob:
 
         event_duration = self.look_untreated_events(camera, body)
         if event_duration == 0:
-            for event in self.get_camera_local_events(pipe, camera, now):
+            call_get_events_func = self.get_alarm_events if camera['video_plan_type'] == 1 else  \
+                    self.get_cameras_playback_urls
+
+            for event in call_get_events_func(pipe, camera, now):
                 insert_video_info(
                     camera['db_table'], camera['device_code'],
                     event['start_time'].astimezone(pytz.utc),

isc_video_record/recorder.py

@@ -21,7 +21,7 @@ from isc_video_record.db import mysql
 
 tz = pytz.timezone('Asia/Shanghai')
 
-video_path = '/data/videos/isc-record'
+video_path = '/tmp/videos/isc-record'
 os.makedirs(video_path, exist_ok=True)
 api_helper = IntelabApiHelper()
 

isc_video_record/utils/isc_client.py

@@ -10,6 +10,7 @@ from time import mktime
 from email.utils import formatdate
 from requests.packages import urllib3
 from intelab_python_sdk.logger import log
+from isc_video_record.utils.api_helper import PlaybackUrlException
 
 
 class HikVisionClient(object):
@@ -51,16 +52,16 @@ class HikVisionClient(object):
         response = requests.post(url, headers=self._sign(uri, body, 'POST'),
                                  json=body, verify=False, timeout=30)
         data = {}
+        response.raise_for_status()
+
         if response.status_code == 200:
             data['code'] = 200
             res_json = response.json()
-            if res_json.get('code', '0x000000') == '0':
+            if res_json.get('code', '0') != '0':
+                raise PlaybackUrlException(res_json)
+
             if 'data' in res_json:
                 data = res_json['data']
-            else:
-                log.warning(res_json['msg'])
-        else:
-            log.error('error: %s', response.text)
         return data
 
     def get_camera_preview_url(self, camera_index_code,
@@ -128,7 +129,7 @@ class HikVisionClient(object):
             'recordLocation': 1,
             'protocol': protocol,
             'expand': 'fileSize=1024',
-            'streamform': streamform
+            'streamform': streamform,
         }
         log.info('requests body: {}'.format(body))
         expired_time = datetime.now() + timedelta(minutes=5)

isc_video_record/utils/pre_event.py

@@ -44,7 +44,7 @@ class PreEvent(object):
             #  合并告警消息最小单位60s
             # TODO 事件的开始和结束需要依赖于status，status=1触发，status=2结束
             if len(events) > 0 \
-               and alarm_time - events[-1]['end_time'] < timedelta(seconds=60) \
+               and alarm_time - events[-1]['end_time'] < timedelta(seconds=30) \
                and events[-1]['end_time'] - events[-1]['start_time'] < timedelta(hours=0.5):
                 events[-1]['end_time'] = alarm_time + timedelta(seconds=10)
                 continue

setup.py

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

tests/test_playback.py

@@ -6,11 +6,13 @@ from dynaconf import settings
 
 from isc_video_record.utils.isc_client import HikVisionClient
 from isc_video_record.utils.record_utils import record_thread, get_video_duration
+from isc_video_record.utils.api_helper import PlaybackUrlException
+from isc_video_record.utils.pre_event import PreEvent
 
 
 tz = pytz.timezone('Asia/Shanghai')
 log_init(__name__, True, './log')
-video_path = '/data/videos/isc-record'
+video_path = '/tmp/videos/isc-record'
 os.makedirs(video_path, exist_ok=True)
 
 config = settings.get('ISC')
@@ -21,19 +23,43 @@ client = HikVisionClient(config.get('KEY'), config.get('SECRET'),
 
 def main():
 
-    start_time = datetime(2021, 5, 27, 9, 10, 30).astimezone(tz)
+    start_time = datetime(2021, 6, 11, 9, 27, 21).astimezone(tz)
     # start_time = datetime(2021, 5, 28, 9, 10, 59).astimezone(tz)
-    end_time = datetime(2021, 5, 27, 9, 30, 50).astimezone(tz)
-    res = client.get_cameras_playback_urls('8f50e406cad6489fac443e034d29a66f',
-                                           client.iso_format(start_time), client.iso_format(end_time))
+    end_time = datetime(2021, 6, 11, 18, 30, 50).astimezone(tz)
+    camera_index = 'bb954401e1e9420d91e4b3691566f161'
+
+    results = []
+    try:
+        res = client.get_cameras_playback_urls(
+            camera_index,
+            client.iso_format(start_time), client.iso_format(end_time)
+        )
+        results.extend(res)
+    except PlaybackUrlException as e:
+        if 'Internal Error' in e.msg:
+            pre_events = PreEvent(start_time.astimezone(pytz.utc).strftime('%Y-%m-%d %H:%M:%S'),
+                                  end_time.astimezone(pytz.utc).strftime('%Y-%m-%d %H:%M:%S'))
+            unverified_events = pre_events.merge_alarm_to_event(pre_events.get_alarm_list(camera_index))
+            for event in unverified_events:
+                try:
+                    res = client.get_cameras_playback_urls(
+                        camera_index, client.iso_format(event['start_time']),
+                        client.iso_format(event['end_time'])
+                    )
+                    results.extend(res)
+                except Exception as e:
+                    print('ERROR:', e)
+                else:
                     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(cur_start_time, cur_end_time)
-        # print(event)
-        file_info, error_log = stream_record(event['stream_url'], cur_start_time, cur_end_time)
-        print(file_info)
+
+    print(results)
+    # for event in res:
+    #     cur_start_time = max(event['start_time'], start_time)
+    #     cur_end_time = min(event['end_time'], end_time)
+    #     print(cur_start_time, cur_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):