alg_growth_gh_reply_video_v1.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291
  1. # -*- coding: utf-8 -*-
  2. import pandas as pd
  3. import traceback
  4. import odps
  5. from odps import ODPS
  6. from threading import Timer
  7. from datetime import datetime, timedelta
  8. from db_helper import MysqlHelper
  9. from my_utils import check_table_partition_exits_v2, get_dataframe_from_odps, \
  10. get_odps_df_of_max_partition, get_odps_instance, get_odps_df_of_recent_partitions
  11. from my_utils import request_post, send_msg_to_feishu
  12. from my_config import set_config
  13. import numpy as np
  14. from log import Log
  15. import os
  16. from argparse import ArgumentParser
  17. CONFIG, _ = set_config()
  18. LOGGER = Log()
  19. BASE_GROUP_NAME = 'stg0909-base'
  20. EXPLORE1_GROUP_NAME = 'stg0909-explore1'
  21. EXPLORE2_GROUP_NAME = 'stg0909-explore2'
  22. #TODO: fetch gh_id from external data source
  23. GH_IDS = ('gh_ac43e43b253b', 'gh_93e00e187787', 'gh_77f36c109fb1', 'gh_68e7fdc09fe4')
  24. CDN_IMG_OPERATOR = "?x-oss-process=image/resize,m_fill,w_600,h_480,limit_0/format,jpg/watermark,image_eXNoL3BpYy93YXRlcm1hcmtlci9pY29uX3BsYXlfd2hpdGUucG5nP3gtb3NzLXByb2Nlc3M9aW1hZ2UvcmVzaXplLHdfMTQ0,g_center"
  25. ODS_PROJECT = "loghubods"
  26. EXPLORE_POOL_TABLE = 'alg_growth_video_return_stats_history'
  27. GH_REPLY_STATS_TABLE = 'alg_growth_gh_reply_video_stats'
  28. ODPS_RANK_RESULT_TABLE = 'alg_gh_autoreply_video_rank_data'
  29. RDS_RANK_RESULT_TABLE = 'alg_gh_autoreply_video_rank_data'
  30. STATS_PERIOD_DAYS = 5
  31. SEND_N = 2
  32. def check_data_partition(project, table, data_dt, data_hr=None):
  33. """检查数据是否准备好"""
  34. try:
  35. partition_spec = {'dt': data_dt}
  36. if data_hr:
  37. partition_spec['hour'] = data_hr
  38. part_exist, data_count = check_table_partition_exits_v2(
  39. project, table, partition_spec)
  40. except Exception as e:
  41. data_count = 0
  42. return data_count
  43. def get_last_strategy_result(project, rank_table, dt_version, key):
  44. strategy_df = get_odps_df_of_max_partition(
  45. project, rank_table, { 'ctime': dt_version }
  46. ).to_pandas()
  47. sub_df = strategy_df.query(f'strategy_key == "{key}"')
  48. sub_df = sub_df[['gh_id', 'video_id', 'strategy_key', 'sort']].drop_duplicates()
  49. return sub_df
  50. def process_reply_stats(project, table, period, run_dt):
  51. # 获取多天即转统计数据用于聚合
  52. df = get_odps_df_of_recent_partitions(project, table, period, {'dt': run_dt})
  53. df = df.to_pandas()
  54. df['video_id'] = df['video_id'].astype('int64')
  55. df = df[['gh_id', 'video_id', 'send_count', 'first_visit_uv', 'day0_return']]
  56. # 账号内聚合
  57. df = df.groupby(['video_id', 'gh_id']).agg({
  58. 'send_count': 'sum',
  59. 'first_visit_uv': 'sum',
  60. 'day0_return': 'sum'
  61. }).reset_index()
  62. # 聚合所有数据作为default
  63. default_stats_df = df.groupby('video_id').agg({
  64. 'send_count': 'sum',
  65. 'first_visit_uv': 'sum',
  66. 'day0_return': 'sum'
  67. }).reset_index()
  68. default_stats_df['gh_id'] = 'default'
  69. merged_df = pd.concat([df, default_stats_df]).reset_index(drop=True)
  70. merged_df['score'] = merged_df['day0_return'] / (merged_df['send_count'] + 500)
  71. return merged_df
  72. def rank_for_layer1(run_dt, run_hour, project, table):
  73. # TODO: 加审核&退场
  74. df = get_odps_df_of_max_partition(project, table, {'dt': run_dt})
  75. df = df.to_pandas()
  76. # 确保重跑时可获得一致结果
  77. dt_version = f'{run_dt}{run_hour}'
  78. np.random.seed(int(dt_version)+1)
  79. # TODO: 修改权重计算策略
  80. df['score'] = df['rov']
  81. sampled_df = df.sample(n=SEND_N, weights=df['score'])
  82. sampled_df['sort'] = range(1, len(sampled_df) + 1)
  83. sampled_df['strategy_key'] = EXPLORE1_GROUP_NAME
  84. sampled_df['dt_version'] = dt_version
  85. gh_name_df = pd.DataFrame({'gh_id': GH_IDS + ('default', )})
  86. sampled_df['_tmpkey'] = 1
  87. gh_name_df['_tmpkey'] = 1
  88. extend_df = sampled_df.merge(gh_name_df, on='_tmpkey').drop('_tmpkey', axis=1)
  89. result_df = extend_df[['strategy_key', 'dt_version', 'gh_id', 'sort', 'video_id', 'score']]
  90. return result_df
  91. def rank_for_layer2(run_dt, run_hour, project, stats_table, rank_table):
  92. stats_df = process_reply_stats(project, stats_table, STATS_PERIOD_DAYS, run_dt)
  93. # 确保重跑时可获得一致结果
  94. dt_version = f'{run_dt}{run_hour}'
  95. np.random.seed(int(dt_version)+1)
  96. # TODO: 计算账号间相关性
  97. ## 账号两两组合,取有RoVn数值视频的交集,单个账号内的RoVn(平滑后)组成向量
  98. ## 求向量相关系数或cosine相似度
  99. ## 单个视频的RoVn加权求和
  100. # 当前实现基础版本:只在账号内求二级探索排序分
  101. sampled_dfs = []
  102. # 处理default逻辑(default-explore2)
  103. default_stats_df = stats_df.query('gh_id == "default"')
  104. sampled_df = default_stats_df.sample(n=SEND_N, weights=default_stats_df['score'])
  105. sampled_df['sort'] = range(1, len(sampled_df) + 1)
  106. sampled_dfs.append(sampled_df)
  107. # 基础过滤for账号
  108. df = stats_df.query('day0_return > 100')
  109. # fallback to base if necessary
  110. base_strategy_df = get_last_strategy_result(
  111. project, rank_table, dt_version, BASE_GROUP_NAME)
  112. for gh_id in GH_IDS:
  113. sub_df = df.query(f'gh_id == "{gh_id}"')
  114. if len(sub_df) < SEND_N:
  115. LOGGER.warning(
  116. "gh_id[{}] rows[{}] not enough for layer2, fallback to base"
  117. .format(gh_id, len(sub_df)))
  118. sub_df = base_strategy_df.query(f'gh_id == "{gh_id}"')
  119. sub_df['score'] = sub_df['sort']
  120. sampled_df = sub_df.sample(n=SEND_N, weights=sub_df['score'])
  121. sampled_df['sort'] = range(1, len(sampled_df) + 1)
  122. sampled_dfs.append(sampled_df)
  123. extend_df = pd.concat(sampled_dfs)
  124. extend_df['strategy_key'] = EXPLORE2_GROUP_NAME
  125. extend_df['dt_version'] = dt_version
  126. result_df = extend_df[['strategy_key', 'dt_version', 'gh_id', 'sort', 'video_id', 'score']]
  127. return result_df
  128. def rank_for_base(run_dt, run_hour, project, stats_table, rank_table):
  129. stats_df = process_reply_stats(project, stats_table, STATS_PERIOD_DAYS, run_dt)
  130. #TODO: support to set base manually
  131. dt_version = f'{run_dt}{run_hour}'
  132. # 获取当前base信息, 策略表dt_version(ctime partition)采用当前时间
  133. base_strategy_df = get_last_strategy_result(
  134. project, rank_table, dt_version, BASE_GROUP_NAME)
  135. default_stats_df = stats_df.query('gh_id == "default"')
  136. # 在账号内排序,决定该账号(包括default)的base利用内容
  137. # 排序过程中,确保当前base策略参与排序,因此先关联再过滤
  138. gh_ids_str = ','.join(f'"{x}"' for x in GH_IDS)
  139. stats_df = stats_df.query(f'gh_id in ({gh_ids_str})')
  140. stats_with_strategy_df = stats_df \
  141. .merge(
  142. base_strategy_df,
  143. on=['gh_id', 'video_id'],
  144. how='left') \
  145. .query('strategy_key.notna() or score > 0.1')
  146. # 合并default和分账号数据
  147. grouped_stats_df = pd.concat([default_stats_df, stats_with_strategy_df]).reset_index()
  148. def set_top_n(group, n=2):
  149. group_sorted = group.sort_values(by='score', ascending=False)
  150. top_n = group_sorted.head(n)
  151. top_n['sort'] = range(1, n + 1)
  152. return top_n
  153. ranked_df = grouped_stats_df.groupby('gh_id').apply(set_top_n, SEND_N)
  154. ranked_df = ranked_df.reset_index(drop=True)
  155. #ranked_df['sort'] = grouped_stats_df.groupby('gh_id')['score'].rank(ascending=False)
  156. ranked_df['strategy_key'] = BASE_GROUP_NAME
  157. ranked_df['dt_version'] = dt_version
  158. ranked_df = ranked_df[['strategy_key', 'dt_version', 'gh_id', 'sort', 'video_id', 'score']]
  159. return ranked_df
  160. def check_result_data(df):
  161. for gh_id in GH_IDS + ('default', ):
  162. for key in (EXPLORE1_GROUP_NAME, EXPLORE2_GROUP_NAME, BASE_GROUP_NAME):
  163. sub_df = df.query(f'gh_id == "{gh_id}" and strategy_key == "{key}"')
  164. if len(sub_df) != SEND_N:
  165. raise Exception(f"Result not enough for gh_id[{gh_id}] group[{group}]")
  166. def build_and_transfer_data(run_dt, run_hour, project, **kwargs):
  167. dt_version = f'{run_dt}{run_hour}'
  168. dry_run = kwargs.get('dry_run', False)
  169. layer1_rank = rank_for_layer1(run_dt, run_hour, ODS_PROJECT, EXPLORE_POOL_TABLE)
  170. layer2_rank = rank_for_layer2(run_dt, run_hour, ODS_PROJECT,
  171. GH_REPLY_STATS_TABLE, ODPS_RANK_RESULT_TABLE)
  172. base_rank = rank_for_base(run_dt, run_hour, ODS_PROJECT,
  173. GH_REPLY_STATS_TABLE, ODPS_RANK_RESULT_TABLE)
  174. final_rank_df = pd.concat([layer1_rank, layer2_rank, base_rank]).reset_index(drop=True)
  175. check_result_data(final_rank_df)
  176. odps_instance = get_odps_instance(project)
  177. odps_ranked_df = odps.DataFrame(final_rank_df)
  178. video_df = get_dataframe_from_odps('videoods', 'wx_video')
  179. video_df['cover_url'] = video_df['cover_img_path'] + CDN_IMG_OPERATOR
  180. video_df = video_df['id', 'title', 'cover_url']
  181. final_df = odps_ranked_df.join(video_df, on=('video_id', 'id'))
  182. final_df = final_df.to_pandas()
  183. final_df = final_df[['strategy_key', 'dt_version', 'gh_id', 'sort', 'video_id', 'title', 'cover_url', 'score']]
  184. # reverse sending order
  185. final_df['sort'] = SEND_N + 1 - final_df['sort']
  186. if dry_run:
  187. print(final_df[['strategy_key', 'gh_id', 'sort', 'video_id', 'score', 'title']])
  188. return
  189. # save to ODPS
  190. t = odps_instance.get_table(ODPS_RANK_RESULT_TABLE)
  191. part_spec_dict = {'dt': run_dt, 'hour': run_hour, 'ctime': dt_version}
  192. part_spec =','.join(['{}={}'.format(k, part_spec_dict[k]) for k in part_spec_dict.keys()])
  193. with t.open_writer(partition=part_spec, create_partition=True, overwrite=True) as writer:
  194. writer.write(list(final_df.itertuples(index=False)))
  195. # sync to MySQL
  196. data_to_insert = [tuple(row) for row in final_df.itertuples(index=False)]
  197. data_columns = list(final_df.columns)
  198. mysql = MysqlHelper(CONFIG.MYSQL_CRAWLER_INFO)
  199. mysql.batch_insert(RDS_RANK_RESULT_TABLE, data_to_insert, data_columns)
  200. def main_loop():
  201. argparser = ArgumentParser()
  202. argparser.add_argument('-n', '--dry-run', action='store_true')
  203. args = argparser.parse_args()
  204. try:
  205. now_date = datetime.today()
  206. LOGGER.info(f"开始执行: {datetime.strftime(now_date, '%Y-%m-%d %H:%M')}")
  207. now_hour = now_date.strftime("%H")
  208. last_date = now_date - timedelta(1)
  209. last_dt = last_date.strftime("%Y%m%d")
  210. # 查看当前天级更新的数据是否已准备好
  211. # 当前上游统计表为天级更新,但字段设计为兼容小时级
  212. h_data_count = check_data_partition(ODS_PROJECT, GH_REPLY_STATS_TABLE, last_dt, '00')
  213. if h_data_count > 0:
  214. LOGGER.info('上游数据表查询数据条数={},开始计算'.format(h_data_count))
  215. run_dt = now_date.strftime("%Y%m%d")
  216. LOGGER.info(f'run_dt: {run_dt}, run_hour: {now_hour}')
  217. build_and_transfer_data(run_dt, now_hour, ODS_PROJECT,
  218. dry_run=args.dry_run)
  219. LOGGER.info('数据更新完成')
  220. else:
  221. LOGGER.info("上游数据未就绪,等待60s")
  222. Timer(60, main_loop).start()
  223. return
  224. except Exception as e:
  225. LOGGER.error(f"数据更新失败, exception: {e}, traceback: {traceback.format_exc()}")
  226. if CONFIG.ENV_TEXT == '开发环境':
  227. return
  228. send_msg_to_feishu(
  229. webhook=CONFIG.FEISHU_ROBOT['server_robot'].get('webhook'),
  230. key_word=CONFIG.FEISHU_ROBOT['server_robot'].get('key_word'),
  231. msg_text=f"rov-offline{CONFIG.ENV_TEXT} - 数据更新失败\n"
  232. f"exception: {e}\n"
  233. f"traceback: {traceback.format_exc()}"
  234. )
  235. if __name__ == '__main__':
  236. LOGGER.info("%s 开始执行" % os.path.basename(__file__))
  237. LOGGER.info(f"environment: {CONFIG.ENV_TEXT}")
  238. main_loop()