|
@@ -0,0 +1,183 @@
|
|
|
+import datetime
|
|
|
+from datetime import timedelta
|
|
|
+
|
|
|
+import pandas as pd
|
|
|
+import yaml
|
|
|
+
|
|
|
+from common_func import get_db_engine, get_logger
|
|
|
+from config.url import toutiao_static_video_url
|
|
|
+from get_material_and_script.get_material_from_kuaishou_kaiyan import GetMaterialFromKuaishouKaiyan
|
|
|
+from get_material_and_script.get_material_from_ocean_engine import GetMaterialFromOceanEngine
|
|
|
+from get_script_from_tengxunyun import GetScriptFromTengXunYunServer
|
|
|
+
|
|
|
+if __name__ == '__main__':
|
|
|
+ # 创建日志对象
|
|
|
+ logger = get_logger(log_file_name="/data/pythonProject/video-to-word/logs/get_high_quality_material.log",
|
|
|
+ log_name='get_high_quality_material_logger')
|
|
|
+ logger.info("get_high_quality_material_logger started! id of logger is: %s" % id(logger))
|
|
|
+
|
|
|
+ # 1 读取配置文件
|
|
|
+ with open('/data/pythonProject/video-to-word/config/config.yaml', mode='r', encoding='utf-8') as f:
|
|
|
+ config = yaml.load(f.read(), Loader=yaml.FullLoader)
|
|
|
+
|
|
|
+ # 1-1 数据库连接引擎,依据开发环境/生产环境 进行切换
|
|
|
+ # 读数据库引擎使用生产数据库,写数据库引擎依据系统环境进行切换(测试数据库/生产数据库)
|
|
|
+ # 注意:该项目的读和写 都使用测试数据库
|
|
|
+ # TODO 等数据库迁移,服务上线后需要依据环境进行切换
|
|
|
+ ai_word_engine = get_db_engine(config['ai_word_dev_db'])
|
|
|
+ jeecg_boot_engine = get_db_engine(config['jeecg_boot_product_db'])
|
|
|
+
|
|
|
+ # 1-2 分批写入数据库的行数
|
|
|
+ chunk_size = config['chunk_size']
|
|
|
+
|
|
|
+ # 1-3 渠道编码&名称
|
|
|
+ source_name = config['source_name_map']
|
|
|
+
|
|
|
+ # 2 读取查询表得到关键词和渠道
|
|
|
+ sql = """
|
|
|
+ select query_word,
|
|
|
+ query_time_range,
|
|
|
+ is_new,
|
|
|
+ source_code_lst
|
|
|
+ from ctop_ai_query_word where status = 1
|
|
|
+ """
|
|
|
+ df = pd.read_sql(sql, ai_word_engine)
|
|
|
+ for index, row in df.iterrows():
|
|
|
+ query_word = row['query_word']
|
|
|
+ query_time_range = row['query_time_range']
|
|
|
+ is_new_project_query_word = row['is_new']
|
|
|
+ source_code_lst = eval(row['source_code_lst'])
|
|
|
+
|
|
|
+ # 2-1 获取内外部优质素材
|
|
|
+ full_high_material_df = pd.DataFrame()
|
|
|
+ writer = pd.ExcelWriter('/data/pythonProject/video-to-word/data/%s_内外部优质素材_%s.xlsx' %
|
|
|
+ (query_word, datetime.date.today().strftime('%Y-%m-%d')))
|
|
|
+ for source_code in source_code_lst:
|
|
|
+ if source_code == 1:
|
|
|
+ # 获取公司内部投放素材
|
|
|
+ sql = """
|
|
|
+ select distinct(project_id) project_id from ctop_user_allocation where project_name like '%%{query_word}%%'""".format(
|
|
|
+ query_word=query_word)
|
|
|
+ project_id_lst = pd.read_sql(sql, jeecg_boot_engine).project_id.values
|
|
|
+ project_id_lst = tuple(project_id_lst) if len(project_id_lst) > 1 else tuple(project_id_lst * 2)
|
|
|
+ if project_id_lst:
|
|
|
+ if is_new_project_query_word == 1:
|
|
|
+ # 新增关键词,查询内部高质量素材的时间范围为3个月
|
|
|
+ start_date = (datetime.date.today() + timedelta(days=-30 * 3)).strftime('%Y-%m-%d')
|
|
|
+ full_date_df = pd.DataFrame()
|
|
|
+ # TODO 测试 periods=3 非测试需要 periods=30 * 3
|
|
|
+ for date in pd.date_range(start=start_date, freq='D', periods=30 * 3):
|
|
|
+ stat_date = date.strftime('%Y-%m-%d')
|
|
|
+ sql = """select signature, sum(activation) activation , sum(charge) charge, stat_date
|
|
|
+ from ctop_kuaishou_report_daily_material
|
|
|
+ where account_id in (select account_id from ctop_user_allocation where project_id in %s)
|
|
|
+ and stat_date = '%s'
|
|
|
+ group by signature,stat_date""" % (project_id_lst, stat_date)
|
|
|
+ one_date_df = pd.read_sql(sql, jeecg_boot_engine)
|
|
|
+ full_date_df = full_date_df.append(one_date_df)
|
|
|
+ # 过滤高质量素材(3个月内累计激活个数>=100)
|
|
|
+ g = full_date_df.groupby('signature').agg({'activation': sum, 'charge': sum})
|
|
|
+ g.reset_index(drop=False, inplace=True)
|
|
|
+ high_material_df = g[g['activation'] >= 100]
|
|
|
+ logger.info("新增查询词:%s, 内部高质量素材个数为 %s!" % (query_word, len(high_material_df)))
|
|
|
+ else:
|
|
|
+ # 历史已有关键词,查询内部高质量素材的的时间范围为7天
|
|
|
+ start_date = (datetime.date.today() + timedelta(days=-7)).strftime('%Y-%m-%d')
|
|
|
+ full_date_df = pd.DataFrame()
|
|
|
+ for date in pd.date_range(start=start_date, freq='D', periods=7):
|
|
|
+ stat_date = date.strftime('%Y-%m-%d')
|
|
|
+ sql = """select signature, sum(activation) activation , sum(charge) charge, stat_date
|
|
|
+ from ctop_kuaishou_report_daily_material
|
|
|
+ where account_id in (select account_id from ctop_user_allocation where project_id in %s)
|
|
|
+ and stat_date = '%s'
|
|
|
+ group by signature,stat_date""" % (project_id_lst, stat_date)
|
|
|
+ one_date_df = pd.read_sql(sql, jeecg_boot_engine)
|
|
|
+ full_date_df = full_date_df.append(one_date_df)
|
|
|
+
|
|
|
+ # 获取高质量素材(近一周累计激活个数>=50)
|
|
|
+ g = full_date_df.groupby('signature').agg({'activation': sum, 'charge': sum})
|
|
|
+ g.reset_index(drop=False, inplace=True)
|
|
|
+ high_material_df = g[g['activation'] >= 50]
|
|
|
+ logger.info("历史查询词:%s, 内部高质量素材个数为 %s!" % (query_word, len(high_material_df)))
|
|
|
+ else:
|
|
|
+ logger.info("查询词:%s, 内部没有对应的项目!" % query_word)
|
|
|
+
|
|
|
+ if not high_material_df.empty:
|
|
|
+ # 获取素材的url
|
|
|
+ sql = """select url video_url, signature from ctop_kuaishou_video_get
|
|
|
+ where account_id in (select account_id from ctop_user_allocation where project_id in %s)
|
|
|
+ and signature in %s
|
|
|
+ group by signature
|
|
|
+ """ % (project_id_lst, tuple(high_material_df.signature.values),)
|
|
|
+ url_df = pd.read_sql(sql, jeecg_boot_engine)
|
|
|
+ high_material_df = high_material_df.merge(url_df, on='signature', how='inner')
|
|
|
+ high_material_df.loc[:, 'source_code'] = source_code # 用于后续导出excel的sheet_name
|
|
|
+
|
|
|
+ # 当前渠道的优质素材拼接到 full_high_material_df,用于后续统一获取脚本和导出excel
|
|
|
+ full_high_material_df = full_high_material_df.append(high_material_df)
|
|
|
+
|
|
|
+ if source_code == 3:
|
|
|
+ # 获取快手开眼快创的优质素材
|
|
|
+ inst = GetMaterialFromKuaishouKaiyan(query_word=query_word, logger=logger, db_engine=ai_word_engine)
|
|
|
+ inst.get_video_basic_info()
|
|
|
+ if not inst.video_df.empty:
|
|
|
+ high_material_df = inst.video_df[['photoId', 'video_url']]
|
|
|
+ high_material_df.rename(columns={'photoId': 'signature'}, inplace=True)
|
|
|
+ high_material_df.loc[:, 'source_code'] = source_code
|
|
|
+
|
|
|
+ full_high_material_df = full_high_material_df.append(inst.video_df)
|
|
|
+ else:
|
|
|
+ # TODO 添加日志
|
|
|
+ pass
|
|
|
+
|
|
|
+ if source_code == 2:
|
|
|
+ # 获取头条巨量引擎的优质素材,需要使用参数 query_time_range
|
|
|
+ inst = GetMaterialFromOceanEngine(query_word=query_word,
|
|
|
+ period_type=query_time_range,
|
|
|
+ logger=logger,
|
|
|
+ db_engine=ai_word_engine)
|
|
|
+ inst.get_material_basic_info()
|
|
|
+ inst.get_video_basic_info()
|
|
|
+ if not inst.video_df.empty:
|
|
|
+ high_material_df = inst.video_df[['signature', 'video_url']]
|
|
|
+ high_material_df.loc[:, 'source_code'] = source_code
|
|
|
+
|
|
|
+ # 当前渠道的优质素材拼接到 full_high_material_df,用于后续统一获取脚本和导出excel
|
|
|
+ full_high_material_df = full_high_material_df.append(high_material_df)
|
|
|
+ else:
|
|
|
+ # TODO 添加日志
|
|
|
+ pass
|
|
|
+
|
|
|
+ # 2-2 调用腾讯云的语音转脚本服务
|
|
|
+ full_high_material_df = full_high_material_df[~full_high_material_df.signature.isnull()]
|
|
|
+ get_script_ins = GetScriptFromTengXunYunServer(logger=logger,
|
|
|
+ db_engine=ai_word_engine,
|
|
|
+ task_df=full_high_material_df[['signature', 'video_url']],
|
|
|
+ task_ids=None)
|
|
|
+ get_script_ins.submit_task()
|
|
|
+ if get_script_ins.task_ids:
|
|
|
+ get_script_ins.get_result()
|
|
|
+
|
|
|
+ # 2-3 从数据库获取脚本并导出excel文件
|
|
|
+ sql = """select md5 signature,task_result from tb_asr_result where word_text is not null and md5 in %s""" \
|
|
|
+ % (tuple(full_high_material_df.signature.values),)
|
|
|
+ script_df = pd.read_sql(sql, ai_word_engine)
|
|
|
+ script_df.drop_duplicates('signature', keep='first', inplace=True)
|
|
|
+ script_df['script'] = script_df['task_result'].apply(lambda x: eval(x)['Data']['Result'].split(']')[1].strip())
|
|
|
+
|
|
|
+ charge_script_df = script_df.merge(full_high_material_df, on='signature', how='inner')
|
|
|
+ charge_script_df.rename(columns={'charge': '消耗', 'script': '脚本'}, inplace=True)
|
|
|
+
|
|
|
+ # 头条的视频链接需要替换为永久链接
|
|
|
+ charge_script_df['video_url'] = charge_script_df.apply(
|
|
|
+ lambda item: toutiao_static_video_url + item['signature'] if item.get('source_code') == 2 else item['video_url'], axis=1)
|
|
|
+
|
|
|
+ for source_code in charge_script_df.source_code.unique():
|
|
|
+ charge_script_df[charge_script_df.source_code == source_code][['脚本', 'video_url', '消耗']].to_excel(writer,
|
|
|
+ sheet_name='%s_%s_优质素材' % (
|
|
|
+ query_word,
|
|
|
+ source_name[source_code]),
|
|
|
+ index=False,
|
|
|
+ header=True)
|
|
|
+ writer.save()
|
|
|
+ logger.info("查询词: %s,文件导出完成!" % query_word)
|