feat: [tasks] 定时任务删除过期视频

isc_video_record/db/mysql.py

@@ -202,3 +202,28 @@ def get_events_by_time(cursor, conn, db_table, camera_code, start_time, end_time
     cursor.execute(sql, [camera_code, start_time, end_time])
     return cursor.fetchall()
 
+
+@query(cursor_dict=True)
+def get_expirted_video_info(cursor, conn, db_table, expired_time):
+    sql = '''
+        select
+            id video_id, video_url,
+            device_code, start_time, end_time,
+            '{0}' as db_table
+        from {0}
+        where expired_time < "{1}"
+    '''.format(db_table, expired_time)
+    cursor.execute(sql)
+    return cursor.fetchall()
+
+
+@query()
+def deleted_video_by_id(cursor, conn, db_table, video_id):
+    if isinstance(video_id, list):
+        _filter = 'where id in ({})'.format(','.join(video_id))
+    else:
+        _filter = 'where id = {}'.format(video_id)
+
+    sql = 'delete from {} {}'.format(db_table, _filter)
+    cursor.execute(sql)
+    conn.commit()

isc_video_record/tasks.py

@@ -7,13 +7,14 @@ from apscheduler.schedulers.blocking import BlockingScheduler
 from datetime import datetime, timedelta
 
 from intelab_python_sdk.logger import log
+from oss2.exceptions import NoSuchKey
 
 from isc_video_record.db import redis_connect, mysql, influxdb
 from isc_video_record.utils.alarm_utils import send_markdown
 from isc_video_record.utils.record_utils import get_time_str
 from isc_video_record.const import LAST_CHECK_TIME_KEY, PROCESSING_CAMERA_KEY
 from isc_video_record.utils.excel_utils import gen_excel
-from isc_video_record.utils.aliyun_oss import oss_upload_file
+from isc_video_record.utils.aliyun_oss import oss_upload_file, oss_delete_file
 from isc_video_record.db.mysql import get_camera_info
 from isc_video_record.utils.api_helper import IntelabApiHelper
 
@@ -56,19 +57,46 @@ daily_text = """ ## {}日报,  共{}路摄像头
 
 class Tasks:
 
-    def __init__(self):
-        # TODO 定期删除过期视频文件
-        pass
-
     def start(self):
 
         schedulers = BlockingScheduler({
             'apscheduler.timezone': TIMEZONE})
         schedulers.add_job(self.run_moniter_online, 'interval', minutes=3, seconds=10)
         schedulers.add_job(self.run_moniter_hour, 'cron', minute='0,30')
+        schedulers.add_job(self.clean_expired, 'cron', hour=4, minute=30)
         schedulers.add_job(self.run_moniter_daily, 'cron', hour=9, minute=30)
         schedulers.start()
 
+    @staticmethod
+    def clean_expired():
+        video_duration = 0
+        expired_time = datetime.now()
+
+        for db_index in range(2):
+            db = 'common_video_{}'.format(db_index)
+            for table_index in range(8):
+                deleted_ids = []
+                db_table = '{}.video_data_motion_{}'.format(db, table_index)
+                videos = mysql.get_expirted_video_info(db_table, expired_time)
+                for video in videos:
+                    try:
+                        if video.get('video_url'):
+                            res = oss_delete_file(video['video_url'])
+                            log.info('delete %s, result: %s', video['video_url'], res)
+                    except NoSuchKey:
+                        log.warning('no such key: %s', video['video_url'])
+                    except Exception as e:
+                        log.exception(e)
+                        continue
+
+                    deleted_ids.append(str(video['video_id']))
+                    video_duration += (video['end_time'] - video['start_time']).total_seconds()
+
+                if deleted_ids:
+                    mysql.deleted_video_by_id(db_table, deleted_ids)
+
+        log.info('删除失效日期在%s之前的视频文件共%s时长', expired_time, get_time_str(video_duration))
+
     @staticmethod
     def run_moniter_online():
         measure_points = []
@@ -268,5 +296,6 @@ if __name__ == '__main__':
     log_init(__name__, False)
     t = Tasks()
     # t.start()
-    t.run_moniter_hour()
+    # t.run_moniter_hour()
     # t.run_moniter_online()
+    t.clean_expired()

isc_video_record/utils/aliyun_oss.py

 import os
+import re
 import oss2
 
 import urllib.request
@@ -41,7 +42,15 @@ def oss_upload_file(origin_file, local_file):
 def oss_delete_file(origin_file):
     """ 删除单个文件
     """
-    res = oss_batch_delete_files([origin_file])
+
+    region = None
+    bucket_name = config.get('bucket_name')
+    bucket_info = re.findall(
+        r'https?://([\w\-]+)\.oss-cn-(\w+)\.aliyuncs.com/',
+        origin_file)
+    if len(bucket_info) > 0 and bucket_info[0][0] != config['bucket_name']:
+        bucket_name, region = bucket_info[0]
+    res = oss_batch_delete_files([origin_file], bucket_name, region)
     return True if len(res) > 0 else False
 
 
@@ -80,6 +89,5 @@ if __name__ == '__main__':
     # print(oss_upload_file('test-2.mp4', '/home/wen/Videos/3_C90842327_2020_06_04_13_37_14.mp4'))
     # print(oss_download_file('https://test-qzwjtest.oss-cn-hangzhou.aliyuncs.com/test-2.mp4', 't.mp4'))
     # print(oss_delete_file('https://test-qzwjtest.oss-cn-hangzhou.aliyuncs.com/test-2.mp4'))
-    oss_download_file('D00268229_2020-10-23_14-07-13.mp4', '/tmp/v3/videos/D00268229_2020-10-23_14-07-13.mp4')
-
-
+    # oss_download_file('D00268229_2020-10-23_14-07-13.mp4', '/tmp/v3/videos/D00268229_2020-10-23_14-07-13.mp4')
+    print(oss_delete_file('https://prod-jiandu-shanghai.oss-cn-shanghai.aliyuncs.com/isc_record/ISC_D86639983_20210509T143611_20210509T143859.mp4'))