1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283 |
- 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}")
|