直播回放上传至OSS

libs/ali_oss.py

@@ -4,10 +4,10 @@ class AliOss:
 
     AccessKeyId = 'LYu38mijwUW440A4'
     AccessKeySecret = '5yQum5XXdqlP9lx8YQojwyrciB0hu4'
-    BucketName = 'kx-bigdata'
+    BucketName = 'shop-live'
     EndPoint = 'http://oss-cn-beijing.aliyuncs.com/'
 
-    BaseUrl = 'https://kx-bigdata.oss-cn-beijing.aliyuncs.com/live_video/'
+    BaseUrl = 'shop-live.oss-cn-beijing.aliyuncs.com/shop_live/'
 
     @classmethod
     def get_auth_bucket(cls):

libs/mysql_kwai_show_video.py

@@ -1,75 +0,0 @@
-import time
-from libs.db_mysql import DbMysql
-
-
-class MysqlKwaiShowVideo(DbMysql):
-
-    table = 'kwai_show_video'
-
-    def set_video_status(self, id, status):
-        sql = 'UPDATE ' + self.table + ' SET `status`=%s WHERE `id`=%s'
-        try:
-            # 批量插入数据
-            result = self.cursor.execute(sql, [status, id])
-            return result
-        finally:
-            # 提交到数据库执行
-            self.db.commit()
-            # 最终关闭数据库连接
-            self.cursor.close()
-            self.db.close()
-
-    def get_video_by_status(self, status):
-        sql = 'SELECT id,live_id FROM ' + self.table + ' WHERE `status`=%s LIMIT 1'
-        try:
-            # 批量插入数据
-            self.cursor.execute(sql, [status])
-            return self.cursor.fetchone()
-        finally:
-            # 提交到数据库执行
-            self.db.commit()
-            # 最终关闭数据库连接
-            self.cursor.close()
-            self.db.close()
-
-    def get_downloaded_video(self):
-        last_download_at = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(time.time()) - 3600))
-        sql = 'SELECT id,live_id FROM ' + self.table + ' WHERE `status`=2 AND `download_at`<\'' + last_download_at + '\' LIMIT 1'
-        try:
-            # 查询数据
-            self.cursor.execute(sql)
-            return self.cursor.fetchone()
-        finally:
-            # 提交到数据库执行
-            self.db.commit()
-            # 最终关闭数据库连接
-            self.cursor.close()
-            self.db.close()
-
-    def set_video_tmp_url(self, id, url):
-        current_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
-        sql = 'UPDATE ' + self.table + ' SET `status`=2,video_url=%s,download_at=%s WHERE `id`=%s'
-        try:
-            # 批量插入数据
-            result = self.cursor.execute(sql, [url, current_time, id])
-            return result
-        finally:
-            # 提交到数据库执行
-            self.db.commit()
-            # 最终关闭数据库连接
-            self.cursor.close()
-            self.db.close()
-
-    def set_video_url(self, id, url):
-        current_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
-        sql = 'UPDATE ' + self.table + ' SET `status`=5,video_url=%s,uploaded_at=%s WHERE `id`=%s'
-        try:
-            # 批量插入数据
-            result = self.cursor.execute(sql, [url, current_time, id])
-            return result
-        finally:
-            # 提交到数据库执行
-            self.db.commit()
-            # 最终关闭数据库连接
-            self.cursor.close()
-            self.db.close()

libs/mysql_live_replay.py

@@ -30,3 +30,16 @@ class MysqlLiveReplay(DbMysql):
             self.db.commit()  # 提交
             self.cursor.close()  # 关闭cursor
             self.db.close()  # 关闭db连接
+
+    # 获取需要上传的直播信息
+    def get_replay_upload_info(self):
+        last_download_at = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(time.time()) - 3600))
+        sql = 'SELECT id,live_id FROM ' + self.table + ' WHERE `status`=2 AND `download_at`<\'' + last_download_at + '\' LIMIT 1'
+        try:
+            # 执行SQL
+            self.cursor.execute(sql)
+            return self.cursor.fetchone()
+        finally:
+            self.db.commit()  # 提交
+            self.cursor.close()  # 关闭cursor
+            self.db.close()  # 关闭db连接

live_replay_upload.py

@@ -0,0 +1,77 @@
+import time
+import sys
+import os
+
+from log.print_log import PrintLog
+from libs.mysql_live_replay import MysqlLiveReplay
+from libs.ali_oss import AliOss
+
+
+def upload_to_oss(live_id):
+    try:
+        file_name = live_id + '.flv'
+        local_path = sys.path[0] + '/file/' + file_name
+        oss_path = 'shop_live_replay/' + file_name
+
+        PrintLog.print('开始上传')
+        bucket = AliOss.get_auth_bucket()
+        bucket.put_object_from_file(oss_path, local_path, progress_callback=percentage)
+        PrintLog.print(live_id + ' 上传完成')
+
+        # 删除磁盘上的直播回放文件
+        try:
+            os.remove(local_path)
+            PrintLog.print(live_id + ' 删除文件成功')
+        except Exception as e:
+            PrintLog.print('删除文件抛出异常：' + str(e))
+    except Exception as e:
+        PrintLog.print('上传文件抛出异常：' + str(e))
+        raise e
+        pass
+
+
+# 当无法确定待上传的数据长度时，total_bytes的值为None。
+def percentage(consumed_bytes, total_bytes):
+    global start_time
+    current_time = time.time()
+    current_consumed_m_bytes = round(consumed_bytes / 1024 / 1024, 2)
+    rate = int(100 * (float(consumed_bytes) / float(total_bytes)))
+    diff_time = round(current_time - start_time)
+
+    if total_bytes:
+        print('\r{0}% '.format(rate), end='')
+        sys.stdout.flush()
+    if diff_time > 60:
+        start_time = current_time
+        PrintLog.print('\r已上传' + str(current_consumed_m_bytes) + 'M (' + str(rate) + '%)')
+
+
+def upload_replay():
+    PrintLog.print('开始上传直播回放')
+    # 获取需要上传的直播信息
+    replay_info = MysqlLiveReplay().get_replay_upload_info()
+
+    if replay_info is None:
+        time.sleep(60)
+        return
+
+    id, live_id = replay_info
+
+    upload_to_oss(live_id)
+
+
+if __name__ == "__main__":
+    while True:
+        try:
+            hour = int(time.strftime("%H", time.localtime()))
+            if hour > 17:
+                PrintLog.print('当前时间段不进行上传操作')
+                time.sleep(3600)
+                continue
+
+            start_time = time.time()
+            upload_replay()
+        except Exception as e:
+            PrintLog.print('抛出异常' + str(e))
+            time.sleep(1)
+            continue

show_video_download.py

@@ -1,87 +0,0 @@
-#!/usr/bin/python3
-# coding=utf-8
-# -*- coding: utf-8 -*-
-
-import requests
-import time
-import json
-import threading
-from libs.rds_wait_upload_show_video_set import RdsWaitUploadShowVideoSet
-from libs.mysql_kwai_show_video import MysqlKwaiShowVideo
-from log.print_log import PrintLog
-
-
-def download_video(live_video_params):
-    params = json.loads(live_video_params)
-
-    for play_url in params['play_urls']:
-        try:
-            MysqlKwaiShowVideo().set_video_status(params['id'], '1')
-
-            # requests.get返回的是一个可迭代对象（Iterable），此时Python SDK会通过Chunked Encoding方式上传。
-            input = requests.get(play_url, stream=True)
-
-            PrintLog.print(params['live_id'] + ' 开始下载：' + play_url)
-
-            f = open('file/' + params['live_id'] + ".flv", "ab")  # 有则追加，没有则创建
-
-            # 将视频数据分片写入文件
-            for chunk in input.iter_content(chunk_size=512):
-                if chunk:
-                    f.write(chunk)
-
-            f.close()
-
-            PrintLog.print(params['live_id'] + ' 下载完成：' + play_url)
-
-            tmp_url = 'https://ks.wenxingshuju.com/showvideo/' + params['live_id'] + '.flv'
-            res = MysqlKwaiShowVideo().set_video_tmp_url(params['id'], tmp_url)
-            return
-        except Exception as e:
-            PrintLog.print(params['live_id'] + '下载异常：' + play_url + "\r\n" + str(e))
-            MysqlKwaiShowVideo().set_video_status(params['id'], '3')
-            continue
-
-
-def get_active_thread_set():
-    # 待爬取的直播ID集合
-    active_thread_set = set()
-    for active_thread in threading.enumerate():  # 线程列表
-        if active_thread.getName() != 'MainThread':
-            active_thread_set.add(active_thread.getName())
-
-    return active_thread_set
-
-
-if __name__ == "__main__":
-
-    print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' ' + "主方法开始执行")
-
-    rds_set = RdsWaitUploadShowVideoSet()
-
-    while True:
-
-        # 从 待下载直播视频集合中 列表中获取一条
-        live_video_params = rds_set.pop_params()
-
-        # 如果没有，则等一秒，循环
-        if live_video_params is None:
-            time.sleep(1)
-            continue
-
-        try:
-            params = json.loads(live_video_params)
-
-            thread_name = 'show_download_' + params['live_id']
-            # 判断得到的直播间ID在爬取中
-            # 判断线程中是否已经有该任务，有则跳过
-            if thread_name in get_active_thread_set():
-                print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ' ' + " %s 已经在执行" % thread_name)
-                continue
-
-            task = threading.Thread(target=download_video, args=(live_video_params,), name=thread_name)
-            task.start()  # 准备就绪,等待cpu执行
-        except Exception as e:
-            PrintLog.print('抛出异常：' + str(e) + '\n' + live_video_params)
-            time.sleep(1)
-            continue