add bottom videos

bottom_videos.py

@@ -0,0 +1,65 @@
+import datetime
+import traceback
+import os
+
+from utils import execute_sql_from_odps
+from db_helper import RedisHelper
+from config import set_config
+from log import Log
+
+config_ = set_config()
+log_ = Log()
+
+
+def update_bottom_videos():
+    """更新兜底视频"""
+    try:
+        # 获取昨日播放量top1000的视频
+        now_date = datetime.datetime.today()
+        delta_date = now_date - datetime.timedelta(days=1)
+
+        sql = "SELECT video_playcount.videoid, video_playcount.play_count " \
+              "FROM (" \
+              "SELECT videoid, COUNT(*) play_count " \
+              "FROM loghubods.video_action_log_applet " \
+              "WHERE dt = {} " \
+              "AND business = 'videoPlay' " \
+              "GROUP BY videoid" \
+              ") video_playcount INNER " \
+              "JOIN ( " \
+              "SELECT id " \
+              "FROM videoods.wx_video " \
+              "WHERE transcode_status = 3 " \
+              "AND STATUS = 1 " \
+              "AND recommend_status IN ( - 6, 1)" \
+              ") video_status " \
+              "ON video_playcount.videoid = video_status.id " \
+              "ORDER BY video_playcount.play_count DESC " \
+              "LIMIT 1000;".format(delta_date.strftime('%Y%m%d'))
+
+        records = execute_sql_from_odps(project='loghubods', sql=sql)
+
+        # 视频按照昨日播放量写入redis
+        videos = {}
+        with records.open_reader() as reader:
+            for record in reader:
+                video_id = record['videoid']
+                videos[video_id] = record['play_count']
+        print(videos)
+
+        redis_helper = RedisHelper()
+        redis_helper.add_data_with_zset(key_name=config_.BOTTOM_KEY_NAME, data=videos)
+        # 移除bottom key的过期时间，将其转换为永久状态
+        redis_helper.persist_key(key_name=config_.BOTTOM_KEY_NAME)
+
+        log_.info('{} update bottom videos success!'.format(now_date))
+    except Exception as e:
+        log_.error(traceback.format_exc())
+
+
+if __name__ == '__main__':
+    update_bottom_videos()
+    # 将日志上传到oss
+    log_cmd = "ossutil cp -r -f {} oss://{}/{}".format(log_.logname, config_.BUCKET_NAME,
+                                                       config_.OSS_FOLDER_LOGS + 'bottom_videos/')
+    os.system(log_cmd)

config.py

@@ -40,6 +40,9 @@ class BaseConfig(object):
     # 流量池离线模型结果存放 redis key前缀，完整格式 com.weiqu.video.flowpool.hot.item.score.{appType}
     FLOWPOOL_KEY_NAME_PREFIX = 'com.weiqu.video.flowpool.hot.item.score.'
 
+    # 兜底视频redis存储key
+    BOTTOM_KEY_NAME = 'com.weiqu.video.bottom'
+
 
 class DevelopmentConfig(BaseConfig):
     """开发环境配置"""

db_helper.py

@@ -197,6 +197,15 @@ class RedisHelper(object):
         conn = self.connect()
         conn.srem(key_name, *values)
 
+    def persist_key(self, key_name):
+        """
+        移除key的过期时间，将其转换为永久状态
+        :param key_name: key
+        :return:
+        """
+        conn = self.connect()
+        conn.persist(key_name)
+
 
 class HologresHelper(object):
     def __init__(self):

rov_train.py

@@ -164,11 +164,11 @@ def predict():
     redis_helper.add_data_with_zset(key_name=key_name, data=redis_data)
     log_.info('data to redis finished!')
     # 通知后端更新数据
-    result = request_post(request_url=config_.NOTIFY_BACKEND_UPDATE_ROV_SCORE_URL, request_data={'videos': json_data})
-    if result['code'] == 0:
-        log_.info('notify backend success!')
-    else:
-        log_.error('notify backend fail!')
+    # result = request_post(request_url=config_.NOTIFY_BACKEND_UPDATE_ROV_SCORE_URL, request_data={'videos': json_data})
+    # if result['code'] == 0:
+    #     log_.info('notify backend success!')
+    # else:
+    #     log_.error('notify backend fail!')
 
 
 def predict_test():

utils.py

@@ -10,6 +10,22 @@ from db_helper import HologresHelper
 config_ = set_config()
 
 
+def execute_sql_from_odps(project, sql, connect_timeout=3000, read_timeout=500000,
+                       pool_maxsize=1000, pool_connections=1000):
+    odps = ODPS(
+        access_id='LTAI4FtW5ZzxMvdw35aNkmcp',
+        secret_access_key='0VKnydcaHK3ITjylbgUsLubX6rwiwc',
+        project=project,
+        endpoint='http://service.cn.maxcompute.aliyun.com/api',
+        connect_timeout=connect_timeout,
+        read_timeout=read_timeout,
+        pool_maxsize=pool_maxsize,
+        pool_connections=pool_connections
+    )
+    records = odps.execute_sql(sql=sql)
+    return records
+
+
 def get_data_from_odps(date, project, table, connect_timeout=3000, read_timeout=500000,
                        pool_maxsize=1000, pool_connections=1000):
     """
@@ -97,7 +113,7 @@ def filter_video_status(video_ids):
     对视频状态进行过滤
     :param video_ids: 视频id列表 type-list
     :return: filtered_videos
-        """
+    """
     if len(video_ids) == 1:
         sql = "set hg_experimental_enable_shard_pruning=off; " \
               "SELECT video_id " \