chore: 测试脚本

tests/test_playback.py

@@ -3,6 +3,7 @@ import shutil
 import pytz
 from datetime import datetime, timedelta
 from intelab_python_sdk.logger import log_init, log
+from intelab_python_sdk.ffmpeg.ffmpeg_concat import concat
 from dynaconf import settings
 
 from hikvision_isc_client.client import HikVisionClient
@@ -19,46 +20,19 @@ config = settings.get('ISC')
 client = HikVisionClient(config.get('KEY'), config.get('SECRET'),
                          config.get('HOST'), config.get('PORT'))
 
-event_start_time = datetime(2021, 4, 16, 13, 30)
-event_end_time = datetime(2021, 4, 16, 14, 10)
-# playback_urls = client.get_cameras_playback_urls(
-#     'c440c77c2a344799bd92948dc67ba805',
-#     HikVisionClient.iso_format(event_start_time),
-#     HikVisionClient.iso_format(event_end_time),
-# )
-# print(playback_urls)
-
-# print(client.get_camera_preview_url('c440c77c2a344799bd92948dc67ba805', protocol='hls'))
-
-for camera in client.get_cameras():
-    print(camera)
-    url = client.get_camera_preview_url(camera['cameraIndexCode'], protocol='rtmp')
-    print('摄像头{}的直播地址为{}'.format(camera['cameraIndexCode'], url.get('url')))
-#
-#     playback_urls = client.get_cameras_playback_urls(
-#         camera['cameraIndexCode'],
-#         '2021-04-15T16:30:00.000+08:00', '2021-04-14T17:30:00.000+08:00'
-#         # HikVisionClient.iso_format(event_start_time),
-#         # HikVisionClient.iso_format(event_end_time),
-#     )
-#     print(playback_urls)
-
 
 def main():
+    start_time = datetime(2021, 4, 23, 16, 10)
+    end_time = datetime(2021, 4, 23, 16, 15)
     playback_urls = client.get_cameras_playback_urls(
-        '282f614f6a4b403f8a0083126ee9b6c1',
-        HikVisionClient.iso_format(event_start_time),
-        HikVisionClient.iso_format(event_end_time)
+        'f8a3c4d9b8ae42118b4db9fcf7895031',
+        HikVisionClient.iso_format(start_time),
+        HikVisionClient.iso_format(end_time),
     )
     for playback_stream in playback_urls:
         log.info(playback_stream)
-        start_time = playback_stream['start_time']
-        print(start_time, event_start_time.astimezone(tz))
-        if start_time < event_start_time.astimezone(tz):
-            start_time = event_start_time.astimezone(tz)
-
-        end_time = playback_stream['end_time']
         part_num = 1
+        part_files_set = set()
         while True:
             complete_duration = (end_time - start_time).total_seconds()
 
@@ -75,10 +49,14 @@ def main():
                                                             new_start_time.strftime('%Y%m%dT%H%M%S'),
                                                             part_num)
                 shutil.move(file_info['file_name'], part_file_name)
+                part_files_set.add(part_file_name)
                 start_time = new_start_time
                 part_num += 1
             else:
+                part_files_set.add(file_info['file_name'])
                 break
+        if len(part_files_set) > 1:
+            concat(sorted(list(part_files_set)), file_info)
 
 
 def stream_record(stream, start_time, end_time):
@@ -96,5 +74,5 @@ def stream_record(stream, start_time, end_time):
     return get_video_duration(file_name)
 
 
-# if __name__ == '__main__':
-#     main()
+if __name__ == '__main__':
+    main()

tests/test_preview.py

@@ -12,6 +12,6 @@ client = HikVisionClient(config.get('KEY'), config.get('SECRET'),
 #
 for camera in client.get_cameras():
     print(camera)
-    url = client.get_camera_preview_url(camera['cameraIndexCode'], protocol='hls', stream_type=1)
+    url = client.get_camera_preview_url(camera['cameraIndexCode'], protocol='rtsp', stream_type=1)
     print('摄像头{}的直播地址为{}'.format(camera['cameraIndexCode'], url.get('url')))