import datetime import os import sys import pandas as pd import yaml from common_func import mysql_replace_into curr_path = os.path.abspath(os.path.dirname(__file__)) project_root_path = curr_path[:curr_path.find("video_to_word") + len("video_to_word")] sys.path.append(project_root_path) from common_func import get_db_engine, get_logger from get_material_and_script.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 获取素材编码和素材url project_id = 1860892 query_word = '爱奇艺极速版(新)' # `channel_type` int(2) DEFAULT '0' COMMENT '0:自产 1:素造', sql = f"select signature,url video_url, channel_type from ctop_kuaishou_video_get where account_id in " \ f"(select account_id from ctop_user_allocation where project_id = {project_id}) " \ f"and channel_type = 0 and stat_date>='2021-08-01' group by signature" df = pd.read_sql(sql, jeecg_boot_engine) df['query_word'] = query_word df['project_id'] = project_id df['stat_date'] = datetime.date.today().strftime("%Y-%m-%d") df.to_sql(name="ctop_ai_video_info_from_huichuang", con=ai_word_engine, if_exists='append', index=False, chunksize=1000, method=mysql_replace_into) # 0 记录任务执行情况的字段 size = 0 message = "" task_status = 0 inst = None # 2 调用腾讯云的语音转脚本服务,获取脚本 if not df.empty: get_script_ins = GetScriptFromTengXunYunServer(logger=logger, db_engine=ai_word_engine, task_df=df[['signature', 'video_url']], task_ids=None) get_script_ins.submit_task() if get_script_ins.task_ids: get_script_ins.get_result() size = len(get_script_ins.task_ids) else: size = 0 # 3 任务执行情况写入数据库 task_info = {'query_word': query_word, 'stat_date': datetime.date.today().strftime('%Y-%m-%d'), 'source_code': 1, 'size': size} logger.info(f"{task_info}")