import datetime
import traceback
import gevent
from db_helper import RedisHelper
from my_utils import send_msg_to_feishu
from my_config import set_config
from log import Log
config_, env = set_config()
log_ = Log()
# initial_param = {'data': 'data1', 'rule': 'rule4'}
# new_param = config_.LHK_RULE_PARAMS
redis_helper = RedisHelper()
def get_religion_videos(now_date, religion_name):
"""获取宗教视频列表"""
key_name_prefix = config_.RELIGION_VIDEOS[religion_name].get('key_name_prefix')
religion_key_name = f"{key_name_prefix}{datetime.datetime.strftime(now_date, '%Y%m%d')}"
if not redis_helper.key_exists(religion_key_name):
redis_dt = datetime.datetime.strftime((now_date - datetime.timedelta(days=1)), '%Y%m%d')
religion_key_name = f"{key_name_prefix}{redis_dt}"
religion_videos = redis_helper.get_all_data_from_zset(key_name=religion_key_name, desc=True, with_scores=True)
if religion_videos is None:
return []
return religion_videos
def merge_process(initial_key_name, new_key_name, now_videos, religion_video_id_list, rank_count):
initial_data = redis_helper.get_all_data_from_zset(key_name=initial_key_name, with_scores=True)
if initial_data is None or len(initial_data) == 0:
return now_videos, religion_video_id_list
initial_video_ids = [int(video_id) for video_id, _ in initial_data]
initial_video_ids = [video_id for video_id in initial_video_ids if video_id not in now_videos]
religion_video_id_list = [video_id for video_id in religion_video_id_list if video_id not in initial_video_ids]
if len(religion_video_id_list) == 0:
new_video_ids = initial_video_ids
else:
new_video_ids = []
for i, video_id in enumerate(initial_video_ids):
new_video_ids.append(video_id)
now_videos.append(video_id)
if i % rank_count == 1 and len(religion_video_id_list) > 0:
new_video_ids.append(religion_video_id_list[0])
now_videos.append(religion_video_id_list[0])
religion_video_id_list = religion_video_id_list[1:]
# 按照排序给定分数
new_result = {}
step = 100 / (len(new_video_ids) * 2)
for i, video_id in enumerate(new_video_ids):
score = 100 - i * step
new_result[int(video_id)] = score
# 写入新的key中
redis_helper.add_data_with_zset(key_name=new_key_name, data=new_result, expire_time=2 * 24 * 3600)
return now_videos, religion_video_id_list
def merge_with_region(now_date, now_h, region, religion_video_id_list,
initial_param, lhk_data_key, lhk_rule_key, rank_count):
initial_data_key = initial_param.get('data')
initial_rule_key = initial_param.get('rule')
now_videos = []
# 地域小时级数据合并
region_h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{region}:{initial_data_key}:{initial_rule_key}:" \
f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
new_region_h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_REGION_BY_H}{region}:{lhk_data_key}:{lhk_rule_key}:" \
f"{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
now_videos, religion_video_id_list = merge_process(initial_key_name=region_h_key_name,
new_key_name=new_region_h_key_name,
now_videos=now_videos,
religion_video_id_list=religion_video_id_list,
rank_count=rank_count)
# 地域24h数据合并
region_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H}{region}:{initial_data_key}:" \
f"{initial_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
new_region_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP1_REGION_24H_H}{region}:{lhk_data_key}:" \
f"{lhk_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
now_videos, religion_video_id_list = merge_process(initial_key_name=region_24h_key_name,
new_key_name=new_region_24h_key_name,
now_videos=now_videos,
religion_video_id_list=religion_video_id_list,
rank_count=rank_count)
# 24h筛选数据合并
dup2_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H}{region}:{initial_data_key}:" \
f"{initial_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
new_dup2_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP2_REGION_24H_H}{region}:{lhk_data_key}:" \
f"{lhk_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
now_videos, religion_video_id_list = merge_process(initial_key_name=dup2_24h_key_name,
new_key_name=new_dup2_24h_key_name,
now_videos=now_videos,
religion_video_id_list=religion_video_id_list,
rank_count=rank_count)
# 24h筛选后剩余数据合并
dup3_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H}{region}:{initial_data_key}:" \
f"{initial_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
new_dup3_24h_key_name = f"{config_.RECALL_KEY_NAME_PREFIX_DUP3_REGION_24H_H}{region}:{lhk_data_key}:" \
f"{lhk_rule_key}:{datetime.datetime.strftime(now_date, '%Y%m%d')}:{now_h}"
now_videos, religion_video_id_list = merge_process(initial_key_name=dup3_24h_key_name,
new_key_name=new_dup3_24h_key_name,
now_videos=now_videos,
religion_video_id_list=religion_video_id_list,
rank_count=rank_count)
log_.info(f"region = {region} update end!")
def merge_videos(now_date, now_h, param, rule_params):
"""将宗教视频插入到默认视频列表中"""
# 获取宗教视频列表
lhk_data_key = param.get('data')
lhk_rule_key = param.get('rule')
religion_name = rule_params.get(lhk_rule_key).get('religion_name')
initial_param = rule_params.get(lhk_rule_key).get('initial_param')
rank_count = rule_params.get(lhk_rule_key).get('rank_count')
religion_videos = get_religion_videos(now_date=now_date, religion_name=religion_name)
religion_video_id_list = [int(video_id) for video_id, _ in religion_videos]
# 列表合并
region_code_list = [code for region, code in config_.REGION_CODE.items()]
task_list = [
gevent.spawn(
merge_with_region,
now_date, now_h, region, religion_video_id_list, initial_param, lhk_data_key, lhk_rule_key, rank_count
)
for region in region_code_list
]
gevent.joinall(task_list)
# 特殊城市视频数据准备
for region, city_list in config_.REGION_CITY_MAPPING.items():
t = [
gevent.spawn(
merge_with_region,
now_date, now_h, city_code, religion_video_id_list, initial_param, lhk_data_key, lhk_rule_key, rank_count
)
for city_code in city_list
]
gevent.joinall(t)
def main():
try:
log_.info(f"laohaokan recommend data update start...")
now_date = datetime.datetime.today()
now_h = datetime.datetime.now().hour
log_.info(f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d%H')}")
lhk_rule_params = config_.LHK_RULE_PARAMS
rule_params = lhk_rule_params.get('rule_params', {})
params_list = lhk_rule_params.get('params_list', [])
for param in params_list:
log_.info(f"param = {param} update start...")
merge_videos(now_date, now_h, param, rule_params)
log_.info(f"param = {param} update end!")
log_.info(f"laohaokan recommend data update end!")
# send_msg_to_feishu(
# webhook=config_.FEISHU_ROBOT['server_robot'].get('webhook'),
# key_word=config_.FEISHU_ROBOT['server_robot'].get('key_word'),
# msg_text=f"rov-offline{config_.ENV_TEXT} - 老好看推荐视频数据更新完成\n"
# f"now_date: {datetime.datetime.strftime(now_date, '%Y%m%d')}\n"
# f"now_h: {now_h}\n"
# f"finished time: {datetime.datetime.strftime(datetime.datetime.now(), '%Y%m%d %H:%M:%S')}"
# )
except Exception as e:
log_.error(f"老好看推荐视频数据更新失败, exception: {e}, traceback: {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=f"rov-offline{config_.ENV_TEXT} - 老好看推荐视频数据更新失败\n"
f"exception: {e}\n"
f"traceback: {traceback.format_exc()}"
)
if __name__ == '__main__':
main()