Merge branch 'feature/zhangbo_flow_recall' of algorithm/rov-offline into master

alg_recsys_rank_item_realtime_1h.py

@@ -80,7 +80,6 @@ def get_sql(date, previous_date_str, project):
     FROM    loghubods.video_each_hour_update_no_province_apptype
     WHERE   dt <= '{}23'
     AND     dt >= '{}00'
-    and apptype not in ('4', '21')
     GROUP BY videoid
     ,dt
     )

alg_recsys_rank_item_realtime_1hrootall.py

@@ -0,0 +1,144 @@
+# -*- coding: utf-8 -*-
+import traceback
+import datetime
+from odps import ODPS
+from threading import Timer
+from utils import RedisHelper, get_data_from_odps, send_msg_to_feishu
+from config import set_config
+from log import Log
+from alg_recsys_recall_4h_region_trend import records_process_for_list
+import json
+from datetime import datetime, timedelta
+import sys
+from utils import execute_sql_from_odps
+
+
+config_, _ = set_config()
+log_ = Log()
+redis_helper = RedisHelper()
+
+REDIS_PREFIX = "item_rt_fea_1hrootall_"
+
+def process_and_store(row):
+    video_id, json_str = row
+    key = REDIS_PREFIX + str(video_id)
+    expire_time = 6 * 3600
+    redis_helper.set_data_to_redis(key, json_str, expire_time)
+    # log_.info("video写入数据key={},value={}".format(key, json_str))
+
+def check_data(project, table, partition) -> int:
+    """检查数据是否准备好,输出数据条数"""
+    odps = ODPS(
+        access_id=config_.ODPS_CONFIG['ACCESSID'],
+        secret_access_key=config_.ODPS_CONFIG['ACCESSKEY'],
+        project=project,
+        endpoint=config_.ODPS_CONFIG['ENDPOINT'],
+        connect_timeout=3000,
+        read_timeout=500000,
+        pool_maxsize=1000,
+        pool_connections=1000
+    )
+    try:
+        t = odps.get_table(name=table)
+        log_.info(f"检查分区是否存在-【 dt={partition} 】")
+        check_res = t.exist_partition(partition_spec=f'dt={partition}')
+        if check_res:
+            sql = f'select * from {project}.{table} where dt = {partition}'
+            log_.info(sql)
+            with odps.execute_sql(sql=sql).open_reader() as reader:
+                data_count = reader.count
+        else:
+            log_.info("表{}分区{}不存在".format(table, partition))
+            data_count = 0
+    except Exception as e:
+        log_.error("table:{},partition:{} no data. return data_count=0:{}".format(table, partition, e))
+        data_count = 0
+    return data_count
+
+def get_sql(partition, project):
+    sql = '''
+    SELECT  videoid
+            ,CONCAT_WS(',',COLLECT_LIST(CONCAT(time,":",exp))) AS exp
+            ,CONCAT_WS(',',COLLECT_LIST(CONCAT(time,":",share))) AS share
+            ,CONCAT_WS(',',COLLECT_LIST(CONCAT(time,":",return))) AS return
+    FROM    (
+                SELECT  videoid
+                        ,time
+                        ,SUM(COALESCE(exp,0)) AS exp
+                        ,SUM(COALESCE(share,0)) AS share
+                        ,SUM(COALESCE(return,0)) AS return
+                FROM    loghubods.alg_recsys_rank_root_info_v1
+                WHERE   dt = '{}'
+                and apptype not in ('12')
+                GROUP BY videoid
+                         ,time
+            ) 
+    GROUP BY videoid
+    '''.format(partition)
+    print("sql:" + sql)
+    records = execute_sql_from_odps(project=project, sql=sql)
+    video_list = []
+    with records.open_reader() as reader:
+        for record in reader:
+            video_id = record['videoid']
+            m = dict()
+            try:
+                m["exp"] = record['exp']
+            except Exception as e:
+                log_.error(e)
+            try:
+                m["share"] = record['share']
+            except Exception as e:
+                log_.error(e)
+            try:
+                m["return"] = record['return']
+            except Exception as e:
+                log_.error(e)
+
+            json_str = json.dumps(m)
+            video_list.append([video_id, json_str])
+    return video_list
+
+
+def h_timer_check():
+    try:
+        date = sys.argv[1]
+        hour = sys.argv[2]
+    except Exception as e:
+        now_date = datetime.today()
+        date = datetime.strftime(now_date, '%Y%m%d')
+        hour = datetime.now().hour
+        log_.info("没有读取到参数，采用系统时间：{}".format(e))
+    log_.info("使用时间参数日期：{}，小时：{}".format(date, str(hour)))
+    # 1 判断上游数据表是否生产完成
+    project = "loghubods"
+    table = "alg_recsys_rank_root_info_v1"
+    partition = str(date) + str(hour)
+    table_data_cnt = check_data(project, table, partition)
+    if table_data_cnt == 0:
+        log_.info("上游数据{}未就绪{}，等待...".format(table, partition))
+        Timer(60, h_timer_check).start()
+    else:
+        log_.info("上游数据就绪，count={}，开始读取数据表".format(table_data_cnt))
+        # 2 读取数据表 处理特征
+        # previous_date_str = (datetime.strptime(date, "%Y%m%d") - timedelta(days=1)).strftime("%Y%m%d")
+        video_list = get_sql(partition, project)
+        # 3 写入redis
+        log_.info("video的数据量:{}".format(len(video_list)))
+        records_process_for_list(video_list, process_and_store, max_size=50, num_workers=8)
+
+        redis_helper.set_data_to_redis(REDIS_PREFIX + "partition", partition, 6 * 3600)
+
+
+
+
+if __name__ == '__main__':
+    log_.info("开始执行:" + datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
+    h_timer_check()
+    log_.info("完成执行:" + datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
+
+
+
+
+# cd /root/zhangbo/rov-offline
+# python alg_recsys_rank_item_realtime_1hrootall.py 20240408 14

alg_recsys_rank_item_realtime_1hrootall_task.sh

@@ -0,0 +1,20 @@
+source /etc/profile
+echo $ROV_OFFLINE_ENV
+if [ ! -d "my_logs_feature" ]; then
+    mkdir my_logs_feature
+fi
+cur_time="`date +%Y%m%d`"
+cur_h="`date +%H`"
+echo "开始执行时间:{$(date "+%Y-%m-%d %H:%M:%S")}"
+
+if [[ $ROV_OFFLINE_ENV == 'test' ]]; then
+  /root/anaconda3/bin/python alg_recsys_rank_item_realtime_1hrootall.py $cur_time $cur_h
+  echo "结束执行时间:{$(date "+%Y-%m-%d %H:%M:%S")}"
+  echo "all done"
+elif [[ $ROV_OFFLINE_ENV == 'pro' ]]; then
+  /root/anaconda3/bin/python alg_recsys_rank_item_realtime_1hrootall.py $cur_time $cur_h
+  echo "结束执行时间:{$(date "+%Y-%m-%d %H:%M:%S")}"
+  echo "all done"
+fi
+
+#sh alg_recsys_rank_item_realtime_1hrootall_task.sh

config.py

@@ -206,14 +206,14 @@ class BaseConfig(object):
         'videos5': {APP_TYPE['LONG_VIDEO']: 1},  # [内容精选]
         'data66': {
             APP_TYPE['VLOG']: 0.3,
-            # APP_TYPE['LOVE_LIVE']: 0.2,
+            APP_TYPE['LOVE_LIVE']: 0.2,
             APP_TYPE['LONG_VIDEO']: 0.2,
             APP_TYPE['SHORT_VIDEO']: 0.05,
             APP_TYPE['WAN_NENG_VIDEO']: 0.05,
             # APP_TYPE['LAO_HAO_KAN_VIDEO']: 1,
             # APP_TYPE['ZUI_JING_QI']: 1,
             APP_TYPE['APP']: 0.05,
-            # APP_TYPE['PIAO_QUAN_VIDEO_PLUS']: 0.05,
+            APP_TYPE['PIAO_QUAN_VIDEO_PLUS']: 0.05,
             APP_TYPE['JOURNEY']: 0.05,
             APP_TYPE['BLESSING_YEAR']: 0.04,
             APP_TYPE['PIAO_QUAN_BLESSING']: 0.01

flowpool_data_update_with_level.py

@@ -325,3 +325,4 @@ if __name__ == '__main__':
     log_.info('flow pool predict end...')
     log_.info(f"expend time = {(time.time() - st_time) * 1000}ms")
 
+# python flowpool_data_update_with_level.py    测试环境必须手动执行python 才能有数据

flowpool_data_update_with_level_tomson.py

@@ -1,327 +0,0 @@
-import datetime
-import random
-import time
-import os
-import traceback
-import random
-import json
-
-from config import set_config
-from utils import request_post, filter_video_status, send_msg_to_feishu, filter_video_status_app, \
-    filter_political_videos
-from log import Log
-from db_helper import RedisHelper
-from odps import ODPS
-
-config_, _ = set_config()
-log_ = Log()
-
-
-def get_videos_from_flow_pool(app_type, size=1000):
-    """
-    从流量池获取视频，循环获取，直到返回数据为None结束
-    :param app_type: 产品标识 type-int
-    :param size: 每次获取视频数量，type-int，默认1000
-    :return: videos  [{'videoId': 1111, 'flowPool': ''}, ...]
-    """
-    # 获取批次标识，利用首次获取数据时间戳为标记
-    batch_flag = int(time.time())
-    request_data = {'appType': app_type, 'batchFlag': batch_flag, 'size': size}
-    videos = []
-    retry = 0
-    print("zhangbo1")
-    while True:
-        print("zhangbo2")
-        print(config_.GET_VIDEOS_FROM_POOL_URL)
-        result = request_post(request_url=config_.GET_VIDEOS_FROM_POOL_URL, request_data=request_data)
-        if result is None:
-            if retry > 2:
-                break
-            retry += 1
-            continue
-        if result['code'] != 0:
-            log_.info('batch_flag: {}, 获取流量池视频失败'.format(batch_flag))
-            if retry > 2:
-                break
-            retry += 1
-            continue
-        if not result['data']:
-            if retry > 2:
-                break
-            retry += 1
-            continue
-        videos.extend(result['data'])
-    return videos
-
-
-def get_videos_remain_view_count(video_info_list):
-    """
-    获取视频在流量池中的剩余可分发数，并存入对应的redis中
-    :param app_type: 产品标识 type-int
-    :param video_info_list: 视频信息 (视频id, 流量池标记) type-list，[(video_id, flow_pool), ...]
-    :return: data type-list，[(video_id, flow_pool, view_count), ...]
-    """
-    redis_helper = RedisHelper()
-    if not video_info_list:
-        return
-
-    # 每次请求10个
-    for i in range(len(video_info_list)//10 + 1):
-        remain_st_time = time.time()
-        videos = [{'videoId': info[0], 'flowPool': info[1]} for info in video_info_list[i*10:(i+1)*10]]
-        request_data = {'videos': videos}
-        result = request_post(request_url=config_.GET_REMAIN_VIEW_COUNT_URL,
-                              request_data=request_data, timeout=(0.5, 3))
-        log_.info(f"i = {i}, expend time = {(time.time()-remain_st_time)*1000}")
-        if result is None:
-            continue
-        if result['code'] != 0:
-            log_.error('获取视频在流量池中的剩余可分发数失败')
-            continue
-        for item in result['data']:
-            if item['distributeCount'] is None:
-                continue
-            distribute_count = int(item['distributeCount'])
-            if distribute_count > 0:
-                # 将分发数更新到本地记录
-                key_name = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{item['videoId']}:{item['flowPool']}"
-                redis_helper.set_data_to_redis(key_name=key_name, value=distribute_count, expire_time=15 * 60)
-            else:
-                # 将本地记录删除
-                key_name = f"{config_.LOCAL_DISTRIBUTE_COUNT_PREFIX}{item['videoId']}:{item['flowPool']}"
-                redis_helper.del_keys(key_name=key_name)
-                # 从流量召回池移除
-                value = '{}-{}'.format(item['videoId'], item['flowPool'])
-                for type_name in config_.APP_TYPE:
-                    for level in range(1, 7):
-                        flow_pool_key = \
-                            f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{config_.APP_TYPE.get(type_name)}:{level}"
-                        redis_helper.remove_value_from_set(key_name=flow_pool_key, values=(value, ))
-                    quick_flow_pool_key = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{config_.APP_TYPE.get(type_name)}" \
-                                          f":{config_.QUICK_FLOW_POOL_ID}"
-                    redis_helper.remove_value_from_set(key_name=quick_flow_pool_key, values=(value, ))
-
-
-def get_flow_pool_recommend_config(flow_pool_id):
-    """获取流量池推荐分发配置"""
-    result = request_post(request_url=config_.GET_FLOW_POOL_RECOMMEND_CONFIG_URL)
-    if result is None:
-        return None
-    if result['code'] != 0:
-        return None
-    flow_pool_distribute_config = result['data'].get('flowPoolDistributeConfig')
-    if flow_pool_distribute_config:
-        if int(eval(flow_pool_distribute_config).get('flowPoolId')) == flow_pool_id:
-            return eval(eval(flow_pool_distribute_config).get('distributeRate'))
-        else:
-            return None
-    else:
-        return None
-
-
-def online_flow_pool_data_to_redis(app_type, video_ids_set, video_info_data):
-    redis_helper = RedisHelper()
-    for tag, values in video_ids_set.items():
-        if tag == 'quick_flow_pool':
-            key_name_video_ids = \
-                f"{config_.QUICK_FLOWPOOL_VIDEO_ID_KEY_NAME_PREFIX}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
-            key_prefix_video_info = \
-                f"{config_.QUICK_FLOWPOOL_VIDEO_INFO_KEY_NAME_PREFIX}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
-        else:
-            key_name_video_ids = f"{config_.FLOWPOOL_VIDEO_ID_KEY_NAME_PREFIX}{app_type}"
-            key_prefix_video_info = f"{config_.FLOWPOOL_VIDEO_INFO_KEY_NAME_PREFIX}{app_type}"
-        # 如果key已存在，删除key
-        if redis_helper.key_exists(key_name=key_name_video_ids):
-            redis_helper.del_keys(key_name=key_name_video_ids)
-        # 写入redis
-        if len(values) > 0:
-            redis_helper.add_data_with_set(key_name=key_name_video_ids, values=values, expire_time=3600)
-            info_values = video_info_data[tag]
-            if len(info_values) > 0:
-                for video_id, info_value in info_values.items():
-                    key_name_video_info = f"{key_prefix_video_info}:{video_id}"
-                    # 如果key已存在，删除key
-                    if redis_helper.key_exists(key_name=key_name_video_info):
-                        redis_helper.del_keys(key_name=key_name_video_info)
-                    if len(info_value) > 0:
-                        redis_helper.add_data_with_set(key_name=key_name_video_info, values=info_value,
-                                                       expire_time=3600)
-
-
-def get_flow_pool_data(app_type, video_info_list, flow_pool_id_list):
-    """
-    获取流量池可分发视频，并将结果上传Redis
-    :param app_type: 产品标识 type-int
-    :return: None
-    """
-    try:
-        # 从流量池获取数据
-        videos = get_videos_from_flow_pool(app_type=app_type)
-        log_.info(f"app_type: {app_type}")
-        # log_.info(f"videos: {videos}")
-        if len(videos) <= 0:
-            log_.info('流量池中无需分发的视频')
-            return video_info_list
-        # video_id 与 flow_pool, level 进行mapping
-        video_ids = set()
-        log_.info('流量池中视频数：{}'.format(len(videos)))
-        mapping = {}
-        for video in videos:
-            flow_pool_id = video['flowPoolId'] # 召回使用的切分ID是在这里做的。流量池中的视频是不区分ID的，也不区分层。
-            if int(flow_pool_id) not in flow_pool_id_list:
-                continue
-            # print(f"flow_pool_id: {flow_pool_id}")
-            video_id = video['videoId']
-            video_ids.add(video_id)
-            item_info = {'flowPool': video['flowPool'], 'level': video['level']}
-            if video_id in mapping:
-                mapping[video_id].append(item_info)
-            else:
-                mapping[video_id] = [item_info]
-        log_.info(f"需更新流量池视频数: {len(video_ids)}")
-
-        # 对视频状态进行过滤
-        if app_type == config_.APP_TYPE['APP']:
-            filtered_videos = filter_video_status_app(list(video_ids))
-        else:
-            filtered_videos = filter_video_status(list(video_ids))
-        log_.info('filter videos status finished, filtered_videos nums={}'.format(len(filtered_videos)))
-
-        # 涉政视频过滤
-        if app_type not in config_.POLITICAL_RECOMMEND_APP_TYPE_LIST:
-            filtered_videos = filter_political_videos(video_ids=filtered_videos)
-
-        if not filtered_videos:
-            log_.info('流量池中视频状态不符合分发')
-            return video_info_list
-
-        # 上传数据到redis
-        quick_flow_pool_redis_data = set()
-        # 普通流量池视频按照层级存储
-        redis_data = dict()
-
-        for video_id in filtered_videos:
-            for item in mapping.get(video_id):
-                flow_pool = item['flowPool']
-                level = item['level']
-                # 判断是否为快速曝光流量池视频
-                value = '{}-{}'.format(video_id, flow_pool)
-                flow_pool_id = int(flow_pool.split('#')[0])  # flowPool: 流量池ID#分级ID#级别Level#生命周期ID
-                if flow_pool_id == config_.QUICK_FLOW_POOL_ID:
-                    quick_flow_pool_redis_data.add(value)
-                else:
-                    if level not in redis_data:
-                        redis_data[level] = set()
-                    redis_data[level].add(value)
-
-                video_info = (video_id, flow_pool)
-                if video_info not in video_info_list:
-                    video_info_list.append(video_info)
-
-        # 快速曝光流量池视频写入redis
-        redis_helper = RedisHelper()
-        quick_flow_pool_key_name = f"{config_.QUICK_FLOWPOOL_KEY_NAME_PREFIX_SET}{app_type}:{config_.QUICK_FLOW_POOL_ID}"
-        # 如果key已存在，删除key
-        if redis_helper.key_exists(quick_flow_pool_key_name):
-            redis_helper.del_keys(quick_flow_pool_key_name)
-        # 写入redis
-        if quick_flow_pool_redis_data:
-            log_.info(f"quick_flow_pool_redis_data = {quick_flow_pool_redis_data}")
-            redis_helper.add_data_with_set(key_name=quick_flow_pool_key_name, values=quick_flow_pool_redis_data,
-                                           expire_time=24 * 3600)
-            # 快速流量池分发概率存入redis
-            distribute_rate_key_name = f"{config_.QUICK_FLOWPOOL_DISTRIBUTE_RATE_KEY_NAME_PREFIX}{config_.QUICK_FLOW_POOL_ID}"
-            distribute_rate = get_flow_pool_recommend_config(flow_pool_id=config_.QUICK_FLOW_POOL_ID)
-            if distribute_rate is not None:
-                redis_helper.set_data_to_redis(key_name=distribute_rate_key_name, value=distribute_rate,
-                                               expire_time=15 * 60)
-
-        # 普通流量池视频写入redis - 分层存储
-        level_list = []
-        for level, videos in redis_data.items():
-            log_.info(f"level: {level}, videos_count: {len(videos)}")
-            level_list.append(level)
-            flow_pool_key_name = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{app_type}:{level}"
-            # 如果key已存在，删除key
-            if redis_helper.key_exists(flow_pool_key_name):
-                redis_helper.del_keys(flow_pool_key_name)
-            # 写入redis
-            if videos:
-                redis_helper.add_data_with_set(key_name=flow_pool_key_name, values=videos, expire_time=24 * 3600)
-
-        # 删除此时不存在的level key
-        for i in range(1, 7):
-            if i not in level_list:
-                flow_pool_key_name = f"{config_.FLOWPOOL_KEY_NAME_PREFIX_SET_LEVEL}{app_type}:{i}"
-                # 如果key已存在，删除key
-                if redis_helper.key_exists(flow_pool_key_name):
-                    redis_helper.del_keys(flow_pool_key_name)
-
-        log_.info('data to redis finished!')
-
-        return video_info_list
-
-    except Exception as e:
-        log_.error('流量池更新失败, appType: {} exception: {}, traceback: {}'.format(
-            app_type, e, traceback.format_exc()))
-        send_msg_to_feishu(
-            webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'),
-            key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'),
-            msg_text='rov-offline{} - 流量池更新失败, appType: {}, exception: {}'.format(config_.ENV_TEXT, app_type, e)
-        )
-        return video_info_list
-
-
-def get_data_from_odps(project, sql):
-    """检查数据是否准备好"""
-    odps = ODPS(
-        access_id=config_.ODPS_CONFIG['ACCESSID'],
-        secret_access_key=config_.ODPS_CONFIG['ACCESSKEY'],
-        project=project,
-        endpoint=config_.ODPS_CONFIG['ENDPOINT'],
-        connect_timeout=3000,
-        read_timeout=500000,
-        pool_maxsize=1000,
-        pool_connections=1000
-    )
-
-    try:
-        with odps.execute_sql(sql=sql).open_reader() as reader:
-            data_df = reader.to_pandas()
-    except Exception as e:
-        data_df = None
-    return data_df
-
-
-if __name__ == '__main__':
-    st_time = time.time()
-    # 为避免第一个app_type获取数据不全，等待1min
-    time.sleep(60)
-    app_type_list = [config_.APP_TYPE['LAO_HAO_KAN_VIDEO'], config_.APP_TYPE['ZUI_JING_QI']]
-    log_.info('flow pool predict start...')
-    # 获取对应流量池id列表
-    redis_helper = RedisHelper()
-    flow_pool_abtest_config = redis_helper.get_data_from_redis(key_name=config_.FLOWPOOL_ABTEST_KEY_NAME)
-    if flow_pool_abtest_config is not None:
-        flow_pool_abtest_config = json.loads(flow_pool_abtest_config)
-    else:
-        flow_pool_abtest_config = {}
-    flow_pool_id_list = flow_pool_abtest_config.get('control_group', [])
-    video_info_list = []
-    for app_name, app_type in config_.APP_TYPE.items():
-        log_.info('{} predict start...'.format(app_name))
-        if app_type == config_.APP_TYPE['LAO_HAO_KAN_VIDEO']:
-            continue
-        elif app_type == config_.APP_TYPE['ZUI_JING_QI']:
-            continue
-        else:
-            video_info_list = get_flow_pool_data(app_type=app_type, video_info_list=video_info_list,
-                                                 flow_pool_id_list=flow_pool_id_list)
-        log_.info('{} predict end...'.format(app_name))
-
-    # 更新剩余分发数
-    log_.info(f"video_info_list count = {len(video_info_list)}")
-    get_videos_remain_view_count(video_info_list)
-    log_.info('flow pool predict end...')
-    log_.info(f"expend time = {(time.time() - st_time) * 1000}ms")
-