Просмотр исходного кода

Merge branch 'local_dev'

# Conflicts:
#	.idea/video_to_word.iml
#	config/url.py
#	get_high_quality_script.py
#	get_video_from_ocean_engine.py
liyuyi@c-top.com.cn 3 лет назад
Родитель
Сommit
bb55a37d1e

+ 5 - 1
.idea/video_to_word.iml

@@ -1,7 +1,11 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<module version="4">
+<module type="PYTHON_MODULE" version="4">
   <component name="NewModuleRootManager">
     <content url="file://$MODULE_DIR$" />
+    <orderEntry type="jdk" jdkName="Remote Python 3.8.11 (sftp://root@139.186.165.84:22/data/Miniconda3/envs/video_to_word/bin/python)" jdkType="Python SDK" />
     <orderEntry type="sourceFolder" forTests="false" />
   </component>
+  <component name="TestRunnerService">
+    <option name="PROJECT_TEST_RUNNER" value="pytest" />
+  </component>
 </module>

Разница между файлами не показана из-за своего большого размера
+ 10 - 8
asr_client.py


+ 35 - 2
common_func.py

@@ -4,17 +4,21 @@ import json
 import numpy as np
 import logging
 from concurrent_log import ConcurrentTimedRotatingFileHandler
+from config.url import get_none_water_mark_url
+import requests
+import pandas as pd
+import base64
 
 
 # 创建数据库连接引擎
 def get_db_engine(db_info):
     db_con_str = 'mysql+pymysql://%s:%s@%s:%d/%s' % \
-                 (db_info['username'],
+                 (db_info['user'],
                   parse.quote_plus(db_info['password']),
                   db_info['host'],
                   db_info['port'],
                   db_info['database'])
-    engine = create_engine(db_con_str, connect_args={'charset': 'utf8'})
+    engine = create_engine(db_con_str, connect_args={'charset': 'utf8mb4'})
     return engine
 
 
@@ -56,3 +60,32 @@ def get_logger(log_file_name, log_name):
     logger.addHandler(log_handler)
     logger.setLevel(logging.DEBUG)  # 日志打印级别
     return logger
+
+
+# 获取头条无水印有时效性的链接
+def get_toutiao_none_water_mark_and_time_efficient_url(vid):
+    """
+    依据 vid,得到无水印,有时效性的链接
+    :return:
+    """
+    request_path = get_none_water_mark_url + str(vid)
+    request = requests.get(request_path)
+    response_data = json.loads(request.text)
+    if response_data.get('message') == 'success' and response_data.get('data') and response_data.get('data').get('video_list'):
+        video_df = pd.DataFrame()
+        for k, v in response_data.get('data').get('video_list').items():
+            single_video_df = pd.DataFrame([v])
+            video_df = video_df.append(single_video_df)
+
+        video_df.sort_values(by='size', ascending=True, inplace=True)  # 获取分辨率最高的素材
+        main_url = video_df['main_url'].values[-1]
+        main_url_decode = base64.b64decode(main_url)
+        main_url_decode = str(main_url_decode, encoding='utf-8')  # bytes to str
+        return main_url_decode
+    else:
+        return None
+
+
+if __name__ == '__main__':
+    r = get_toutiao_none_water_mark_and_time_efficient_url('v02033290000budu3753giguv9qli6bg')
+    print(r)

+ 58 - 29
config/config.yaml

@@ -1,46 +1,75 @@
-projectName:
-  - '淘特'
-  - '支付宝'
-
-
-projectInfoForExportScript:
-  -
-      project_id: (458)
-      project_name: '淘特'
-      channel:
-        - 1 # 汇创思拓
-        - 0 # 巨量引擎
-  -
-      project_id: (67,123,42)
-      project_name: '支付宝'
-      channel:
-        - 1 # 汇创思拓
-        - 0 # 巨量引擎
-
-
-# 生产数据库 jeecg-boot
-productDB:
+source_name_map:
+  1:
+      name: '内部创意'
+      table: 'ctop_ai_video_info_from_huichuang'
+      status: 1
+  2:
+      name: '巨量创意'
+      table: 'ctop_ai_video_info_from_ocean_engine'
+      status: 1
+  3:
+      name: '开眼快创'
+      table: 'ctop_ai_material_info_from_kuaishou_kaiyan'
+      status: 1
+
+mac_ip_config:
+  # 开发环境
+  '5254003fa716':
+      url: "http://139.186.165.84:31013/"
+  # 测试环境
+  '52540003f5dd':
+      url: "http://139.186.27.96:31013/"
+  # 生产环境
+  '525400c98142':
+      url: "http://114.117.193.186:31013/"
+
+
+# jeecg-boot 生产数据库
+jeecg_boot_product_db:
+  host: 139.186.27.96
+  user: data
+  password: hcst@2021
+  port: 3390
+  database: jeecg-boot
+  charset: utf8mb4
+
+# jeecg-boot 测试数据库
+jeecg_boot_dev_db:
   host: 139.186.27.96
-  username: data
+  user: data
   password: hcst@2021
   port: 3390
   database: jeecg-boot
+  charset: utf8mb4
 
-# 开发测试数据库 db_ai_word
-devDB:
+# db_ai_word 测试数据库
+ai_word_dev_db:
   host: 139.186.165.84
-  username: hcst
+  user: hcst
   password: hcst@2020
   port: 3306
   database: db_ai_word
+  charset: utf8mb4
+
+
+# db_ai_word 生产数据库
+ai_word_product_db:
+  host: 139.186.27.96
+  user: data
+  password: hcst@2021
+  port: 3390
+  database: db_ai_word
+  charset: utf8mb4
+
 
 # 本地数据库 mysql
-localDB:
+local_db:
   host: 192.168.1.193
-  username: root
+  user: root
   password: root@123
   port: 3306
   database: mysql
+  charset: utf8mb4
 
 # 每次写入数据库的行数
-chunkSize: 200
+chunk_size: 200

+ 17 - 6
config/url.py

@@ -1,13 +1,18 @@
-import os
-os_env = os.getenv('LYY_DEV', 'unknown')
-debug_url = "http://139.186.165.84:31013/"
-product_url = "http://139.186.27.96:31013/"
+import uuid
+import yaml
+
+
+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)
+    mac_ip_config = config['mac_ip_config']
+
+mac = uuid.UUID(int=uuid.getnode()).hex[-12:]
 
 # 向腾讯云发送语音转脚本的任务请求url
-voice_to_script_task_submit_url = (debug_url if os_env == 'dev' else product_url) + 'asr/task/submit'
+voice_to_script_task_submit_url = mac_ip_config[mac]['url'] + 'asr/task/submit'
 
 # 依据task_id,向腾讯云获取脚本
-voice_to_script_task_result_url = (debug_url if os_env == 'dev' else product_url) + 'asr/task/result'
+voice_to_script_task_result_url = mac_ip_config[mac]['url'] + 'asr/task/result'
 
 # 巨量引擎获取优质素材信息
 get_material_info_from_ocean_engine_url = "https://cc.oceanengine.com/creative_radar_api/v1/material/list"
@@ -15,5 +20,11 @@ get_material_info_from_ocean_engine_url = "https://cc.oceanengine.com/creative_r
 # 依据 vid, 从巨量引擎获取视频信息
 get_video_info_from_ocean_engine_url = "https://cc.oceanengine.com/creative_content_server/api/video/info"
 
+# 快手开眼快创获取优质素材信息
+get_video_info_from_kuaishou_kaiyan = "https://cc.e.kuaishou.com/rest/creativeCentor/inspired/list"
+
+# 头条无水印链接
+get_none_water_mark_url = "http://i.snssdk.com/video/urls/1/toutiao/mp4/"
+
 # 头条素材永久链接
 toutiao_static_video_url = "http://i.snssdk.com/video/code/1/toutiao/"

+ 38 - 17
database.py

@@ -1,38 +1,56 @@
 # 导入:
-from hashlib import md5
-from sqlalchemy.orm import sessionmaker
-from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy import Column, Integer, String,DateTime,Text,create_engine,and_
+import uuid
 from datetime import datetime
 from urllib import parse
 
+import yaml
+from sqlalchemy import Column, Integer, String, DateTime, Text, and_
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+from common_func import get_db_engine
+
 # 创建对象的基类:
 Base = declarative_base()
 passowrd = parse.quote_plus('hcst@2020')
+
+
 # 定义User对象:
 class Task(Base):
     # 表的名字:
     __tablename__ = 'tb_asr_result'
 
     # 表的结构:
-    md5 =  Column(String(100))
+    signature = Column(String(100))
     task_id = Column(Integer, primary_key=True)
     task_status = Column(Integer)
     task_result = Column(Text)
     word_text = Column(Text)
     word_split = Column(Text)
+    word_text_md5 = Column(Text)
     create_time = Column(DateTime, nullable=True, default=datetime.now)
     update_time = Column(DateTime)
 
+
 # 初始化数据库连接:
-engine = create_engine('mysql+mysqlconnector://hcst:'+passowrd+'@139.186.165.84:3306/db_ai_word')
+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 数据库连接引擎,依据开发环境/生产环境 进行切换
+mac = uuid.UUID(int=uuid.getnode()).hex[-12:]
+if mac in ['5254003fa716', '52540003f5dd']:
+    engine = get_db_engine(config['ai_word_dev_db'])
+else:
+    engine = get_db_engine(config['ai_word_product_db'])
+
 # 创建DBSession类型:
 DBSession = sessionmaker(bind=engine)
 
-def query(md5,task_status,task_id):
-    condition = (1==1)
-    if md5 != None:
-        condition = and_(condition, Task.md5 == md5)
+
+def query(signature, task_status, task_id):
+    condition = (1 == 1)
+    if signature != None:
+        condition = and_(condition, Task.signature == signature)
     if task_status != None:
         condition = and_(condition, Task.task_status == task_status)
     if task_id != None:
@@ -42,11 +60,11 @@ def query(md5,task_status,task_id):
     # 创建Query查询,filter是where条件,最后调用one()返回唯一行,如果调用all()则返回所有行:
     task = session.query(Task).filter(condition).order_by(Task.create_time.desc()).all()
     # 关闭Session:
-    print(task)
     session.close()
     return task
 
-def insert(task:Task):
+
+def insert(task: Task):
     # 创建session对象:
     session = DBSession()
     # 添加到session:
@@ -58,23 +76,26 @@ def insert(task:Task):
     session.close()
     return task
 
+
 def update(task):
     # 创建session对象:
     session = DBSession()
-    taskdb = session.query(Task).filter(Task.task_id==task.task_id).one()
+    taskdb = session.query(Task).filter(Task.task_id == task.task_id).one()
     taskdb.task_status = task.task_status
     taskdb.word_text = task.word_text
     taskdb.word_split = task.word_split
+    taskdb.word_text_md5 = task.word_text_md5
     taskdb.task_result = task.task_result
     # 提交即保存到数据库:
     session.commit()
     # 关闭session:
     session.close()
 
+
 if __name__ == '__main__':
     print('test query')
-    # task = Task(task_id = '3',task_status=1,md5='dx')
+    # task = Task(task_id = '3',task_status=1,signature='dx')
     # task = insert(task)
-    task = query(None,None,1)[0]
-    print(task.md5)
-    #update(task)
+    task = query(None, None, 1)[0]
+    print(task.signature)
+    # update(task)

+ 0 - 79
get_data.py

@@ -1,79 +0,0 @@
-from concurrent.futures import ThreadPoolExecutor
-import requests
-import datetime
-import json
-import pandas as pd
-import uuid
-import traceback
-from sqlalchemy import create_engine
-from urllib import parse
-from urllib.parse import urlencode
-
-if __name__ == '__main__':
-    online_db_con_str = "mysql+pymysql://%s:%s@%s:%d/%s" % ("readonly", parse.quote_plus("hcst@2021"), "139.186.27.96", 3390, "jeecg-boot")
-    online_engine = create_engine(online_db_con_str, connect_args={'charset': 'utf8'})
-
-    test_db_con_str = "mysql+pymysql://%s:%s@%s:%d/%s" % ("hcst", parse.quote_plus("hcst@2020"), "139.186.165.84", 3306, "db_ai_word")
-    test_engine = create_engine(test_db_con_str, connect_args={'charset': 'utf8'})
-
-    task_submit_url = 'http://139.186.165.84:31013/asr/task/submit'
-    task_result_url = 'http://139.186.165.84:31013/asr/task/result'
-
-    # 1、获取素材信息
-    # get_material_info_sql = """
-    # select t2.signature, t2.url
-    # from
-    # (select signature, account_id from ctop_kuaishou_report_daily_material
-    #        where account_id in (select account_id from ctop_user_allocation where project_id = 458)
-    #        group by signature
-    #        having sum(activation) <= 10 and sum(activation) >= 1
-    #        ) t1
-    # left join
-    # ctop_kuaishou_video_get t2
-    # on t1.signature = t2.signature and t1.account_id = t2.account_id
-    # where t2.signature is not null
-    # """
-    # material_info_df = pd.read_sql(get_material_info_sql, online_engine)
-    # material_info_df = material_info_df[(~material_info_df.signature.isnull()) & (~material_info_df.url.isnull())]
-    # N = 3000 if len(material_info_df) >= 3000 else len(material_info_df)
-    # material_info_df = material_info_df.sample(n=N, random_state=2077)
-
-    # material_info_df = pd.read_csv('merge_df_taote.csv')
-    # material_info_df = pd.read_csv('merge_df_zhifubao.csv')
-    # material_info_df = material_info_df[(~material_info_df.signature.isnull()) & (~material_info_df.url.isnull())]
-
-    # 2 获取已经提交的任务
-    # submit_task_sql = """select task_id, md5 signature from tb_asr_result """
-    # submit_task_df = pd.read_sql(submit_task_sql, test_engine)
-
-    # 3 需要提交的任务
-    # to_submit_task_df = material_info_df[~material_info_df.signature.isin(submit_task_df.signature.values)]
-
-    # 4 任务提交
-    # for index, row in to_submit_task_df.iterrows():
-    #     material_md5 = row['signature']
-    #     material_url = row['url']
-    #     print(material_md5, material_url)
-    #     request_data = {"md5": material_md5, "url": material_url}
-    #     # 表单形式的入参,不是 json 块入参,请求方式不一样
-    #     request_full_path = task_submit_url + '?' + urlencode(request_data)
-    #     request = requests.post(request_full_path)
-    #     try:
-    #         result = json.loads(request.text)
-    #         print(result)
-    #     except:
-    #         print("error", request.text)
-
-    # 5、任务获取
-    get_result_sql = """select task_id from tb_asr_result where  word_text is null"""
-    get_result_df = pd.read_sql(get_result_sql, test_engine)
-    for index, row in get_result_df.iterrows():
-        task_id = row['task_id']
-        request_data = {'task_id': task_id}
-        request_full_path = task_result_url + '?' + urlencode(request_data)
-        request = requests.post(request_full_path)
-        try:
-            result = json.loads(request.text)
-            print(task_id, result['status'])
-        except:
-            print("error", task_id, request.text)

+ 1 - 6
get_high_quality_script.py

@@ -5,7 +5,6 @@ import yaml
 from common_func import get_db_engine, get_logger
 from get_script_from_tengxunyun import GetScriptFromTengXunYunServer
 import traceback
-from config.url import toutiao_static_video_url
 
 if __name__ == '__main__':
     # 0 创建日志对象
@@ -106,12 +105,8 @@ if __name__ == '__main__':
             if 0 in channel:
                 logger.info("项目:%s, 巨量引擎外部素材,开始执行!" % project_name)
                 sql = """select signature, video_url  from ctop_ai_video_info_from_ocean_engine 
-                where query_word = '%s' and stat_date >= '%s'""" % (project_name, one_week_date)
+                where project_name = '%s' and stat_date >= '%s'""" % (project_name, one_week_date)
                 df = pd.read_sql(sql, test_read_engine)
-
-                # 把 video_url 替换为永久的链接, 导出excel用
-                df['video_url'] = toutiao_static_video_url + df['signature']
-
                 logger.info("项目:%s, 巨量引擎优质素材个数为 %s!" % (project_name, len(df)))
 
                 sql = """select md5 signature,task_result from tb_asr_result where word_text is not null and md5 in %s""" \

+ 0 - 0
get_material_and_script/__init__.py


+ 81 - 0
get_material_and_script/get_material_from_huichuang.py

@@ -0,0 +1,81 @@
+import datetime
+from datetime import timedelta
+import pandas as pd
+from common_func import mysql_replace_into
+from get_material_and_script.get_script_from_tengxunyun import GetScriptFromTengXunYunServer
+
+
+class GetMaterialFromHuiChuang(object):
+    def __init__(self, query_word, logger, jeecg_boot_db_engine, ai_word_db_engine):
+        self.query_word = query_word
+        self.logger = logger
+        self.jeecg_boot_db_engine = jeecg_boot_db_engine
+        self.ai_word_db_engine = ai_word_db_engine
+        self.high_material_df = pd.DataFrame()
+        self.video_url_df = pd.DataFrame()
+        self.video_df = pd.DataFrame()
+
+    def get_video_basic_info(self):
+        # 1 依据查询词获取 account_id (只获取快手的素材)
+        # `media_id`  '平台类型 1 头条 2快手',
+        # `account_status`  '0 启动 1 禁用',
+        self.logger.info(f"查询词:{self.query_word}, 获取公司内部优质素材,开始执行... ")
+        sql = f"select distinct(account_id) account_id from ctop_user_allocation where project_name like '%%{self.query_word}%%' " \
+              f"and media_id = 2"
+        account_id_lst = pd.read_sql(sql, self.jeecg_boot_db_engine).account_id.values
+
+        # 2 依据 account_id 获取素材
+        if account_id_lst.size > 0:
+            # 2-1 获取近7天的高质量素材
+            days = 7
+            start_date = (datetime.date.today() + timedelta(days=-days)).strftime('%Y-%m-%d')
+            full_date_df = pd.DataFrame()
+            for date in pd.date_range(start=start_date, freq='D', periods=days):
+                stat_date = date.strftime('%Y-%m-%d')
+                sql = f"select signature, activation , charge " \
+                      f"from ctop_kuaishou_report_daily_material  " \
+                      f"where account_id in {tuple(list(account_id_lst) * 2) if account_id_lst.size == 1 else tuple(account_id_lst)} " \
+                      f"and stat_date = '{stat_date}'" \
+                      f"and signature is not null"
+                one_date_df = pd.read_sql(sql, self.jeecg_boot_db_engine)
+                full_date_df = full_date_df.append(one_date_df)
+
+            # 2-2 获取高质量素材(近一周累计激活个数>=50)
+            g = full_date_df.groupby('signature').agg({'activation': sum, 'charge': sum})
+            g.reset_index(drop=False, inplace=True)
+            self.high_material_df = g[g['activation'] >= 50]
+
+        # 3 获取高质量素材的 video_url
+        if not self.high_material_df.empty:
+            signature_lst = self.high_material_df.signature.values
+            sql = f"select url video_url, signature from ctop_kuaishou_video_get " \
+                  f"where signature in {tuple(list(signature_lst) * 2) if signature_lst.size == 1 else tuple(signature_lst)}" \
+                  f"group by signature"
+            self.video_url_df = pd.read_sql(sql, self.jeecg_boot_db_engine)
+
+        # 4 dataframe merge
+        if not self.high_material_df.empty and not self.video_url_df.empty:
+            self.video_df = self.high_material_df.merge(self.video_url_df, on='signature', how='inner')
+
+        self.logger.info(f"查询词:{self.query_word}, 获取公司内部优质素材,共获取到{len(self.video_df)}个")
+
+        # 5 获取到的信息存入数据库
+        if not self.video_df.empty:
+            self.video_df['query_word'] = self.query_word
+            self.video_df['stat_date'] = datetime.date.today().strftime("%Y-%m-%d")
+
+            self.video_df.to_sql(name="ctop_ai_video_info_from_huichuang",
+                                 con=self.ai_word_db_engine,
+                                 if_exists='append',
+                                 index=False,
+                                 chunksize=1000,
+                                 method=mysql_replace_into)
+
+    def get_script_from_teng_xun_yun(self):
+        get_script_ins = GetScriptFromTengXunYunServer(logger=self.logger,
+                                                       db_engine=self.ai_word_db_engine,
+                                                       task_df=self.video_df[['signature', 'video_url']],
+                                                       task_ids=None)
+        get_script_ins.submit_task()
+        if get_script_ins.task_ids:
+            get_script_ins.get_result()

+ 118 - 0
get_material_and_script/get_material_from_kuaishou_kaiyan.py

@@ -0,0 +1,118 @@
+import datetime
+import json
+import traceback
+
+import pandas as pd
+import requests
+
+from common_func import NpEncoder, get_db_engine, get_logger, mysql_replace_into
+from config.url import get_video_info_from_kuaishou_kaiyan
+from get_material_and_script.get_script_from_tengxunyun import GetScriptFromTengXunYunServer
+
+
+class GetMaterialFromKuaishouKaiyan(object):
+    def __init__(self, query_word, logger, db_engine):
+        self.query_word = query_word  # 查询词
+        self.logger = logger
+        self.db_engine = db_engine
+        self.video_df = pd.DataFrame()
+
+    def get_video_basic_info(self):
+        """
+        依据关键词调用快手开眼,获取视频列表,得到视频基本信息如:视频时长、宽度、高度、标题、url链接等
+        """
+        # 1 请求快手开眼快创接口获取视频基本信息
+        page_num = 1
+        page_size = 10
+        total_size = 0
+        self.logger.info(f"查询词:{self.query_word}, 调用快手开眼快创视频接口,开始执行...")
+        while True:
+            # 分页获取
+            self.logger.info(f"查询词:{self.query_word}, 调用快手开眼快创视频接口,分页获取第 {page_num} 页")
+            request_data = {"inspiredSortTypeId": 0,
+                            "platformSourceId": 0,
+                            "formatId": 0,
+                            "pageNum": page_num,
+                            "pageSize": page_size,
+                            "queryCond": self.query_word,
+                            "isFromWeb": False,
+                            "accountType": 0,
+                            "campaignType": 0,
+                            "firstIndustryIds": [],
+                            "secondIndustryIds": []}
+            try:
+                request = requests.post(url=get_video_info_from_kuaishou_kaiyan,
+                                        headers={'Content-Type': 'application/json'},
+                                        data=json.dumps(request_data, cls=NpEncoder)
+                                        )
+                response_data = json.loads(request.text)
+                if response_data.get('msg') == 'OK' and response_data.get('data').get('inspiredAds'):
+                    video_page_df = pd.DataFrame(response_data.get('data').get('inspiredAds'))
+                    # 从 mainMvUrls 字段提取视频的播放链接
+                    video_page_df['video_url'] = video_page_df['mainMvUrls'].apply(lambda x: x[0]['url'])
+                    self.video_df = self.video_df.append(video_page_df)
+
+                    total_size = response_data.get('data').get('totalSize')
+                if page_num * page_size > total_size:
+                    break
+                else:
+                    page_num += 1
+            except:
+                self.logger.error(f"查询词:{self.query_word}, 调用快手开眼快创视频接口,获取第{page_num}页时发生异常信息:{traceback.format_exc()}")
+
+        self.logger.info(f"查询词:{self.query_word}, 调用快手开眼快创视频接口,结束执行,共 {total_size} 个视频!")
+
+        # 2 获取到的信息存入数据库
+        # 2-1 数据字段类型的处理,方便入库
+        # mainMvUrls: list to  str
+        # coverThumbnailUrls:  list to str
+        # headUrls:  list to str
+        if not self.video_df.empty and not self.video_df[~self.video_df.photoId.isnull()].empty:
+            self.video_df = self.video_df[self.video_df.photoId.isnull()]
+            self.video_df[['mainMvUrls', 'coverThumbnailUrls', 'headUrls', 'photoId']] = \
+                self.video_df[['mainMvUrls', 'coverThumbnailUrls', 'headUrls', 'photoId']].astype(str)
+
+            self.video_df.rename(columns={'photoId': 'signature'}, inplace=True)
+
+            # 2-2 添加查询词和日期
+            self.video_df['query_word'] = self.query_word
+            self.video_df['stat_date'] = datetime.datetime.today().strftime('%Y-%m-%d')
+
+            # 2-3 写入数据库, 表中以 signature + query_word + stat_date 作为联合唯一键,写入数据库时如果唯一键重复,则 replace_into
+            try:
+                self.video_df.to_sql(name="ctop_ai_material_info_from_kuaishou_kaiyan",
+                                     con=self.db_engine,
+                                     if_exists='append',
+                                     index=False,
+                                     chunksize=1000,
+                                     method=mysql_replace_into)
+            except:
+                self.logger.error(traceback.format_exc())
+
+    def get_script_from_teng_xun_yun(self):
+        get_script_ins = GetScriptFromTengXunYunServer(logger=self.logger,
+                                                       db_engine=self.db_engine,
+                                                       task_df=self.video_df[['signature', 'video_url']],
+                                                       task_ids=None)
+        get_script_ins.submit_task()
+        if get_script_ins.task_ids:
+            get_script_ins.get_result()
+
+
+if __name__ == '__main__':
+    db_info = {'user': 'hcst',
+               'password': 'hcst@2020',
+               'host': '139.186.165.84',
+               'port': 3306,
+               'database': 'db_ai_word'}
+    db_engine = get_db_engine(db_info)
+
+    logger = get_logger(log_file_name="/data/pythonProject/video_to_word/logs/get_video_from_kuaishou_kaiyan.log",
+                        log_name='get_video_from_kuaishou_kaiyan_logger')
+    logger.info("get_video_from_kuaishou_kaiyan started! id of logger is: %s" % id(logger))
+
+    inst = GetMaterialFromKuaishouKaiyan(query_word='电商', logger=logger, db_engine=db_engine)
+    inst.get_video_basic_info()
+
+# 2147483647
+# 57795440082

+ 169 - 0
get_material_and_script/get_material_from_ocean_engine.py

@@ -0,0 +1,169 @@
+import datetime
+import json
+import traceback
+from urllib.parse import urlencode
+
+import pandas as pd
+import requests
+
+from common_func import mysql_replace_into, NpEncoder, get_db_engine, get_logger
+from config.url import get_material_info_from_ocean_engine_url, get_video_info_from_ocean_engine_url
+from get_material_and_script.get_script_from_tengxunyun import GetScriptFromTengXunYunServer
+
+
+class GetMaterialFromOceanEngine(object):
+    def __init__(self, query_word, period_type, logger, db_engine):
+        self.query_word = query_word  # 查询词
+        self.period_type = period_type  # 查询周期:近3天、近7天等
+        self.logger = logger
+        self.db_engine = db_engine
+        self.signature_lst = None
+        self.video_df = pd.DataFrame()
+
+    def get_material_basic_info(self):
+        """
+        依据关键词调用巨量引擎接口,获取物料列表,得到物料基本信息如:物料的最佳标题、行业标签、video_id等
+        """
+        # 1 请求巨量引擎接口获取物料基本信息
+        self.logger.info(f"查询词:{self.query_word}, 调用头条巨量引擎物料接口,开始执行... ")
+        material_df = pd.DataFrame()  # 初始化返回的结果
+        has_more = True  # 是否还存在分页数据, 初始化为 True
+        limit = 10  # 每页获取10条
+        page = 1  # 第几页
+        while has_more:
+            request_data = {'list_type': 1,
+                            'material_type': 3,
+                            'order_by': 'click_show_rate',
+                            'period_type': self.period_type,
+                            'aggr_app_code': 4,
+                            'aggr_category_list': '[]',
+                            'video_type': '[]',
+                            'keywords': self.query_word,
+                            'landing_type': '[]',
+                            'limit': limit,
+                            'page': page,
+                            'video_duration_type': 5}
+
+            try:
+                self.logger.info(f"查询词:{self.query_word},调用头条巨量引擎物料接口, 分页获取第 {page} 页")
+                request_path = get_material_info_from_ocean_engine_url + '?' + urlencode(request_data)
+                request = requests.get(request_path)
+                result = json.loads(request.text)
+                material_page_df = pd.DataFrame(result['data']['materials'])
+                material_df = material_df.append(material_page_df)
+                if result.get('code') == 0 and result.get('data').get('has_more') is True:
+                    page += 1
+                else:
+                    has_more = False
+            except:
+                self.logger.error(f"查询词:{self.query_word}, 调用头条巨量引擎物料接口,分页获取时发生异常信息: {traceback.format_exc()}")
+
+        self.logger.info(f"查询词:{self.query_word}, 调用头条巨量引擎物料接口, 执行结束,共 {len(material_df)} 个物料信息!")
+
+        # 2 获取到的信息存入数据库
+        # 2-1 数据字段类型的处理,方便入库
+        # metrics: dict to  str
+        # title:  list to str
+        # video_type:  list to str
+        # watermarks:  list to str
+        if not material_df.empty:
+            material_df[['metrics', 'title', 'video_type', 'watermarks']] = \
+                material_df[['metrics', 'title', 'video_type', 'watermarks']].astype(str)
+            material_df.rename(columns={'vid': 'signature'}, inplace=True)
+
+            # 2-1 添加查询词和日期
+            material_df['query_word'] = self.query_word
+            material_df['stat_date'] = datetime.datetime.today().strftime('%Y-%m-%d')
+
+            # 2-3 写入数据库, 表中以 signature + query_word + stat_date 作为联合唯一键,写入数据库时如果唯一键重复,则 replace_into
+            material_df.to_sql(name="ctop_ai_material_info_from_ocean_engine",
+                               con=self.db_engine,
+                               if_exists='append',
+                               index=False,
+                               chunksize=1000,
+                               method=mysql_replace_into)
+
+            # 3 得到素材的video_id
+            self.signature_lst = list(material_df['signature'].values)
+
+    def get_video_basic_info(self):
+        """
+        依据素材签名调用巨量引擎的接口,获取视频的基本信息,如视频时长、宽度、高度、url链接等
+        """
+        # 1 请求巨量引擎接口获取视频基本信息
+        # 分次请求:每次请求的视频个数(为了提高数据获取的完整性,每次只请求10条数据)
+        cnt_per_request = 10
+        # 总的视频个数
+        total_cnt = len(self.signature_lst)
+        self.logger.info(f"查询词:{self.query_word}, 调用头条巨量引擎视频接口,共{total_cnt}个开始执行....")
+        for i in range(0, total_cnt, cnt_per_request):
+            if i + cnt_per_request < total_cnt:
+                query_ids = self.signature_lst[i: i + cnt_per_request]
+                self.logger.info(f"查询词:{self.query_word}, 调用头条巨量引擎视频接口,分页获取第{i+1}个到{i+cnt_per_request}个视频信息")
+            else:
+                query_ids = self.signature_lst[i:]
+                self.logger.info(f"查询词:{self.query_word}, 调用头条巨量引擎视频接口,分页获取第{i+1}个到{total_cnt}个视频信息")
+            try:
+                request_data = {"query_ids": query_ids, "water_mark": "creative_center"}
+                request = requests.post(url=get_video_info_from_ocean_engine_url,
+                                        headers={'Content-Type': 'application/json'},
+                                        data=json.dumps(request_data, cls=NpEncoder))
+                response_data = json.loads(request.text)
+
+                if response_data.get('code') == 0 and response_data.get('data'):
+                    for key, value in response_data['data'].items():
+                        single_dict = value
+                        single_dict['signature'] = key
+                        single_df = pd.DataFrame([single_dict])
+                        self.video_df = self.video_df.append(single_df)
+            except:
+                self.logger.error(f"查询词:{self.query_word}, 调用头条巨量引擎视频接口,分页获取{query_ids},出现异常信息:{traceback.format_exc()}")
+
+        self.logger.info(f"查询词:{self.query_word}, 调用头条巨量引擎视频接口,共{total_cnt}个执行完成!")
+
+        # 2 获取到的信息存入数据库
+        # 2-1  数据类型的处理,方便入库
+        # play_info: list to  str
+        if not self.video_df.empty and not self.video_df[~self.video_df.signature.isnull()].empty:
+            self.video_df = self.video_df[~self.video_df.signature.isnull()]
+            self.video_df['play_info'] = self.video_df['play_info'].astype(str)
+            self.video_df.drop(labels='video_id', axis=1, inplace=True)
+
+            # 2-2 添加查询词和日期
+            self.video_df['query_word'] = self.query_word
+            self.video_df['stat_date'] = datetime.datetime.today().strftime('%Y-%m-%d')
+
+            # 2-3 写入数据库, 表中以 signature + query_word + stat_date 作为联合唯一键,写入数据库时如果唯一键重复,则 replace_into
+            self.video_df.to_sql(name="ctop_ai_video_info_from_ocean_engine",
+                                 con=self.db_engine,
+                                 if_exists='append',
+                                 index=False,
+                                 chunksize=1000,
+                                 method=mysql_replace_into)
+
+    def get_script_from_teng_xun_yun(self):
+        get_script_ins = GetScriptFromTengXunYunServer(logger=self.logger,
+                                                       db_engine=self.db_engine,
+                                                       task_df=self.video_df[['signature', 'video_url']],
+                                                       task_ids=None)
+        get_script_ins.submit_task()
+        if get_script_ins.task_ids:
+            get_script_ins.get_result()
+
+
+if __name__ == '__main__':
+    db_info = {'user': 'hcst',
+               'password': 'hcst@2020',
+               'host': '139.186.165.84',
+               'port': 3306,
+               'database': 'db_ai_word'}
+    db_engine = get_db_engine(db_info)
+
+    logger = get_logger(log_file_name="/data/pythonProject/video_to_word/logs/get_video_from_ocean_engine.log",
+                        log_name='get_video_from_ocean_engine_logger')
+    logger.info("get_video_from_ocean_engine started! id of logger is: %s" % id(logger))
+
+    inst = GetMaterialFromOceanEngine(query_word='淘特', period_type=7, logger=logger, db_engine=db_engine)
+    inst.get_material_basic_info()
+    if inst.signature_lst:
+        inst.get_video_basic_info()

+ 30 - 20
get_script_from_tengxunyun.py

@@ -1,8 +1,10 @@
-import pandas as pd
-from urllib.parse import urlencode
-import requests
 import json
 import time
+from urllib.parse import urlencode
+
+import pandas as pd
+import requests
+
 from config.url import voice_to_script_task_submit_url, voice_to_script_task_result_url
 
 
@@ -18,30 +20,31 @@ class GetScriptFromTengXunYunServer(object):
         向腾讯云提交语音转脚本的任务
         task_df: DataFrame columns 包含 signature 和 video_url
         """
-
         # 1 获取已经被提交过的任务
-        sql = """select md5 signature from tb_asr_result """
+        sql = f"select signature from tb_asr_result"
         submitted_task_df = pd.read_sql(sql, self.db_engine)
 
         # 2 需要提交的任务,去掉历史被提交过的任务,防止重复提交浪费服务时长
         to_submit_task_df = self.task_df[~self.task_df.signature.isin(submitted_task_df.signature.values)]
-        self.logger.info("向腾讯云提交语音转脚本的任务个数为 %s" % len(to_submit_task_df))
+        self.logger.info(f"向腾讯云提交语音转脚本的任务个数为{len(to_submit_task_df)}")
 
         # 3 发送请求,提交任务
         for index, row in to_submit_task_df.iterrows():
             material_md5 = row['signature']
             material_url = row['video_url']
-            request_data = {"md5": material_md5, "url": material_url}
+            request_data = {"signature": material_md5, "url": material_url}
             request_full_path = voice_to_script_task_submit_url + '?' + urlencode(request_data)
             request = requests.post(request_full_path)
             try:
                 result = json.loads(request.text)
-                self.logger.info("素材:%s, 向腾讯云提交语音转脚本任务的返回信息为:%s " % (material_md5, result))
+                self.logger.info(f"素材:{material_md5}, 向腾讯云提交语音转脚本任务的返回信息为{result}")
             except:
-                self.logger.error("素材:%s, 向腾讯云提交语音转脚本任务的返回信息为:%s " % (material_md5, request.text))
+                self.logger.error(f"素材:{material_md5}, 向腾讯云提交语音转脚本任务的返回信息为为{ request.text}")
 
-        # 4 获取素材对应的发送请求的 task_id
-        sql = """select task_id from tb_asr_result where md5 in %s""" % (tuple(self.task_df.signature.values),)
+        # 4 获取素材对应的 task_id
+        signature_lst = list(self.task_df.signature.values)
+        sql = f"select task_id from tb_asr_result where signature " \
+              f"in {tuple(signature_lst*2) if len(signature_lst) == 1 else tuple(signature_lst)}"
         task_id_df = pd.read_sql(sql, self.db_engine)
         self.task_ids = list(task_id_df['task_id'].values)
 
@@ -53,16 +56,17 @@ class GetScriptFromTengXunYunServer(object):
         StatusStr String 任务状态,waiting:任务等待,doing:任务执行中,success:任务成功,failed:任务失败。
         ErrorMsg String 失败原因说明。
         """
-        self.logger.info("从腾讯云获取脚本的个数为%s" % len(self.task_ids))
-        while True:
-            sql = """select task_id, task_status from tb_asr_result where task_id in %s and task_status in (0,1)""" % \
-                  (tuple(self.task_ids),)
+        self.logger.info(f"需要获取的脚本个数: {len(self.task_ids)}")
+        # 从腾讯云获取脚本的最大重试次数
+        retry_upper_cnt = 10
+        retry_cnt = 1
+        while retry_cnt <= retry_upper_cnt:
+            sql = f"select task_id, task_status from tb_asr_result where task_id " \
+                  f"in {tuple(self.task_ids * 2) if len(self.task_ids)==1 else tuple(self.task_ids)}" \
+                  f" and task_status in (0,1)"
             task_status_df = pd.read_sql(sql, self.db_engine)
             if task_status_df.empty:
                 break
-            else:
-                self.logger.info("从腾讯云获取脚本, 休眠一分钟,等待腾讯云任务计算完成。")
-                time.sleep(60 * 1)
 
             for task_id in task_status_df.task_id.values:
                 request_data = {'task_id': task_id}
@@ -70,6 +74,12 @@ class GetScriptFromTengXunYunServer(object):
                 request = requests.post(request_full_path)
                 try:
                     result = json.loads(request.text)
-                    self.logger.info("task_id:%s, 从腾讯云获取脚本返回信息为:%s " % (task_id, result))
+                    self.logger.info(f"task_id:{task_id}, 从腾讯云获取脚本返回信息为:{result}")
                 except:
-                    self.logger.error("task_id:%s, 从腾讯云获取脚本返回信息为:%s " % (task_id, request.text))
+                    self.logger.error(f"task_id:{task_id}, 从腾讯云获取脚本返回信息为:{request.text} ")
+
+            if retry_cnt >= 2:
+                self.logger.info("从腾讯云获取脚本, 休眠3s,等待腾讯云任务计算完成。")
+                time.sleep(3 * 1)
+
+            retry_cnt += 1

+ 0 - 178
get_video_from_ocean_engine.py

@@ -1,178 +0,0 @@
-import json
-import os
-from urllib.parse import urlencode
-import pandas as pd
-import requests
-import yaml
-from common_func import get_db_engine, mysql_replace_into, NpEncoder, get_logger
-import datetime
-from config.url import get_material_info_from_ocean_engine_url, get_video_info_from_ocean_engine_url
-import traceback
-from get_script_from_tengxunyun import GetScriptFromTengXunYunServer
-
-
-def get_material_info(project_name, period_type):
-    material_df = pd.DataFrame()  # 初始化返回的结果
-    has_more = True  # 是否还存在分页数据, 初始化为 True
-    limit = 10  # 每页获取10条
-    page = 1  # 第几页
-    while has_more:
-        request_data = {'list_type': 1,
-                        'material_type': 3,
-                        'order_by': 'click_show_rate',
-                        'period_type': period_type,
-                        'aggr_app_code': 4,
-                        'aggr_category_list': '[]',
-                        'video_type': '[]',
-                        'keywords': project_name,
-                        'landing_type': '[]',
-                        'limit': limit,
-                        'page': page,
-                        'video_duration_type': 5}
-
-        try:
-            request_path = get_material_info_from_ocean_engine_url + '?' + urlencode(request_data)
-            request = requests.get(request_path)
-            result = json.loads(request.text)
-            material_page_df = pd.DataFrame(result['data']['materials'])
-            material_df = material_df.append(material_page_df)
-            if result.get('code') == 0 and result.get('data').get('has_more') is True:
-                page += 1
-                logger.info("project_name:%s, get_material_info_from_ocean_engine 分页获取第 %s 页" % (project_name, page))
-            else:
-                has_more = False
-        except:
-            logger.error("project_name:%s, get_material_info_from_ocean_engine 分页获取时发生异常信息: %s" %
-                         (project_name, traceback.format_exc()))
-
-    logger.info("project_name:%s, get_material_info_from_ocean_engine 分页获取获取完成,共 %s 个物料信息!" %
-                (project_name, len(material_df)))
-
-    # 数据类型的处理,方便入库
-    # metrics dict to  str
-    # title list to str
-    # video_type list to str
-    # watermarks list to str
-    material_df[['metrics', 'title', 'video_type', 'watermarks']] = \
-        material_df[['metrics', 'title', 'video_type', 'watermarks']].astype(str)
-    material_df.rename(columns={'vid': 'signature'}, inplace=True)
-
-    # 添加项目名称和日期
-    material_df['query_word'] = project_name
-    material_df['stat_date'] = datetime.datetime.today().strftime('%Y-%m-%d')
-
-    # 写入数据库
-    material_df.to_sql(name="ctop_ai_material_info_from_ocean_engine",
-                       con=write_engine,
-                       if_exists='append',
-                       index=False,
-                       chunksize=chunk_size,
-                       method=mysql_replace_into)
-
-    return material_df
-
-
-def get_video_info(vid, project_name):
-    """
-    为了提高数据获取的完整性,每次只请求10条数据
-    :param vid:
-    :param project_name:
-    :return:
-    """
-    video_df = pd.DataFrame()
-    # 每次请求的视频个数
-    cnt_per_request = 10
-    # 总的视频个数
-    total_cnt = len(vid)
-    logger.info("project_name:%s, get_video_info_from_ocean_engine,共 %s 个视频需要请求接口获取video_url" % (project_name, total_cnt))
-    for i in range(0, total_cnt, cnt_per_request):
-        if i + cnt_per_request < total_cnt:
-            query_ids = vid[i: i + cnt_per_request]
-            logger.info("project_name:%s, get_video_info_from_ocean_engine 分页获取第 %s 个 到 %s 个视频信息" %
-                        (project_name, i, i + cnt_per_request))
-        else:
-            query_ids = vid[i:]
-            logger.info("project_name:%s, get_video_info_from_ocean_engine 分页获取第 %s 个 到 %s 个视频信息" %
-                        (project_name, i, total_cnt))
-
-        try:
-            request_data = {"query_ids": query_ids, "water_mark": "creative_center"}
-            request = requests.post(url=get_video_info_from_ocean_engine_url,
-                                    headers={'Content-Type': 'application/json'},
-                                    data=json.dumps(request_data, cls=NpEncoder)
-                                    )
-            response_data = json.loads(request.text)
-
-            if response_data.get('code') == 0 and response_data.get('data'):
-                for key, value in response_data['data'].items():
-                    single_dict = value
-                    single_dict['signature'] = key
-                    single_df = pd.DataFrame([single_dict])
-                    video_df = video_df.append(single_df)
-        except:
-            logger.error("project_name:%s, get_video_info_from_ocean_engine 分页获取 %s,出现异常信息:%s" %
-                         (project_name, query_ids, traceback.format_exc()))
-
-    # 数据类型的处理,方便入库
-    # play_info list to  str
-    video_df['play_info'] = video_df['play_info'].astype(str)
-    video_df.drop(labels='video_id', axis=1, inplace=True)
-
-    # 添加项目名称和日期
-    video_df['query_word'] = project_name
-    video_df['stat_date'] = datetime.datetime.today().strftime('%Y-%m-%d')
-
-    # 写入数据库
-    video_df.to_sql(name="ctop_ai_video_info_from_ocean_engine",
-                    con=write_engine,
-                    if_exists='append',
-                    index=False,
-                    chunksize=chunk_size,
-                    method=mysql_replace_into)
-
-    return video_df
-
-
-if __name__ == '__main__':
-    # 创建日志对象
-    logger = get_logger(log_file_name="/data/pythonProject/video-to-word/logs/get_video_from_ocean_engine.log",
-                        log_name='get_video_from_ocean_engine_logger')
-    logger.info("get_video_from_ocean_engine 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 数据库连接引擎,依据开发环境/生产环境 进行切换
-    # 读数据库引擎使用生产数据库,写数据库引擎依据系统环境进行切换(测试数据库/生产数据库)
-    # 注意:该项目的读和写 都使用测试数据库
-    write_engine = get_db_engine(config['devDB'])
-    read_engine = get_db_engine(config['devDB'])
-
-    # 1-2 分批写入数据库的行数
-    chunk_size = config['chunkSize']
-
-    # 1-3 读取项目列表
-    project_name_lst = config['projectName']
-
-    # 2  分项目获取巨量引擎数据
-    for project in project_name_lst:
-        try:
-            logger.info("****************************** %s 项目开始执行 ******************************" % project)
-            # 2-1 获取物料列表并写入数据库
-            material_info_df = get_material_info(project, 7)
-
-            # 2-2 根据 signature 获取 url并写入数据库
-            vid_lst = material_info_df['signature'].values
-            video_info_df = get_video_info(vid_lst, project)
-
-            # 2-3 向腾讯云提交语音转脚本的任务
-            task_df = video_info_df[['signature', 'video_url']]
-            get_script_ins = GetScriptFromTengXunYunServer(logger, read_engine, task_df, task_ids=None)
-            get_script_ins.submit_task()
-            if get_script_ins.task_ids:
-                get_script_ins.get_result()
-
-            logger.info("****************************** %s 项目执行完成 ******************************" % project)
-        except:
-            logger.error("project_name: %s, 发生异常信息: %s" % (project, traceback.format_exc()))

+ 328 - 43
main.py

@@ -1,20 +1,49 @@
-from typing import Optional
-from fastapi import FastAPI
+import hashlib
+import traceback
+import uuid
 from concurrent.futures import ThreadPoolExecutor
-from sqlalchemy.sql.elements import Null
-from sqlalchemy.sql.expression import null
 from datetime import date
-import requests
-import uuid
+from datetime import timedelta
+from io import BytesIO
+from typing import Optional, List
+from urllib.parse import quote
 import os
+import sys
+import pandas as pd
 import uvicorn
+import yaml
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import StreamingResponse
+from loguru import logger
+from pydantic import BaseModel, Field
+
+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 asr_client import send_asr_request, send_task_request
+from common_func import get_db_engine
+from config.url import toutiao_static_video_url
 from database import insert, update, query, Task
-from datetime import datetime
-from fastapi.middleware.cors import CORSMiddleware
-import hashlib
-import traceback
-import json
+from time_task.get_material_and_script_by_query_word import get_material_and_script
+
+logger.remove()  # 删去 import logger之后自动产生的handler,不删除的话会出现重复输出的现象
+logger.add("/data/pythonProject/video_to_word/logs/main_server.{time:YYYY-MM-DD}.log",
+           rotation="00:00",
+           format="{time:YYYY-MM-DD HH:mm:ss,SSS} [{process}] [{thread}] {level} {file} {line} - {message}",
+           level="INFO")
+
+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)
+    source_name_map = config['source_name_map']
+
+    # 数据库连接引擎,依据开发、测试环境/生产环境 进行切换
+    mac = uuid.UUID(int=uuid.getnode()).hex[-12:]
+    if mac in ['5254003fa716', '52540003f5dd']:
+        ai_word_engine = get_db_engine(config['ai_word_dev_db'])
+    else:
+        ai_word_engine = get_db_engine(config['ai_word_product_db'])
 
 threadPool = ThreadPoolExecutor(max_workers=4)
 app = FastAPI()
@@ -41,24 +70,24 @@ app.add_middleware(
 
 
 class QueryItem():
-    md5: Optional[str] = None
+    signature: Optional[str] = None
     url: Optional[str] = None
 
 
-@app.get('/')
+@app.get('/', tags=['back-end task'])
 def index():
     return {'message': '你已经正确创建 FastApi 服务!'}
 
 
-@app.post('/asr/task/submit')
-def task_submit(md5: str, url: str):
+@app.post('/asr/task/submit', tags=['back-end task'])
+def task_submit(signature: str, url: str):
     json = send_asr_request(url)
-    task = Task(md5=md5, task_id=json.Data.TaskId, task_result=json.to_json_string(), task_status=1)
+    task = Task(signature=signature, task_id=json.Data.TaskId, task_result=json.to_json_string(), task_status=1)
     insert(task)
     return {'code': 0, 'taskId': json.Data.TaskId}
 
 
-@app.post('/asr/task/result')
+@app.post('/asr/task/result', tags=['back-end task'])
 def task_submit(task_id: int):
     json = send_task_request(task_id)
     task = query(None, None, task_id)[0]
@@ -68,6 +97,7 @@ def task_submit(task_id: int):
         if json.Data.Status == 2:
             task.word_text = json.Data.ResultDetail[0].FinalSentence
             task.word_split = json.Data.ResultDetail[0].SliceSentence
+            task.word_text_md5 = hashlib.md5(task.word_text.encode('utf-8')).hexdigest()
     except:
         # 提取原始文本内容和分词内容发生异常,把 task_status 置为 -1
         task.task_status = -1
@@ -75,37 +105,292 @@ def task_submit(task_id: int):
     return {'code': 0, 'status': json.Data.StatusStr}
 
 
-@app.post('/asr/task/list')
+@app.post('/asr/task/list', tags=['back-end task'])
 def task_submit(task_status: int):
     task = query(None, task_status, None)
     return {'code': 0, 'data': task}
 
 
-# @app.post('/asr/task/result')
-# def retry(item:Item):
-#     #插入数据库
-#     task = query(item.taskId,None,None,None)[0]
-#     item.inputVideoUrl = task.input_video_url
-#     item.inputImageUrl = task.input_image_url
-#     threadPool.submit(videoSwap,item).add_done_callback(swapFinish)
-#     return {'code':0,'data':{'taskId': task.id}}
-
-# @app.post('/jeecg-boot/task/single')
-# def single(item:Item):
-#     #插入数据库
-#     old_task = query(None,item.videoMd5,item.imageMd5,None)
-#     if len(old_task) > 0:
-#         return {'code':-1,'data':old_task}
-#     uid = str(uuid.uuid4())
-#     suid = ''.join(uid.split('-'))
-#     video_input = item.inputVideoUrl
-#     image_input = item.inputImageUrl
-#     task = Task(input_video_url = video_input,input_image_url=image_input,status='waiting',input_video_md5=item.videoMd5,input_image_md5=item.imageMd5,create_by=item.createBy)
-#     task = insert(task)
-#     item.taskId = task.id
-#     threadPool.submit(videoSwap,item).add_done_callback(swapFinish)
-
-#     return {'code':0,'data':{'taskId': task.id}}
+class BaseResponse(BaseModel):
+    message: str = Field(..., description='消息')
+    success: bool = Field(..., description='true or false')
+    code: int = Field(..., description='')
+
+
+class TaskDetail(BaseModel):
+    source_name: str = Field('内部创意', description='数据来源名称')
+    query_word: str = Field('红包', description='关键词')
+    stat_date: str = Field('2021-11-11', description='日期')
+    script_num: int = Field(180, description='脚本数量')
+    task_status: str = Field('执行成功', description='状态')
+    number: int = Field(0, description='序号')
+
+
+class ConfigDetail(BaseModel):
+    config_id: str = Field(..., description="脚本配置id")
+    query_word_lst: List[str] = Field(..., description="关键词")
+    create_time: str = Field(..., description="创建时间")
+    operator: str = Field(..., description="创建人")
+    number: int = Field(..., description="序号")
+
+
+class TaskResponse(BaseResponse):
+    total_num: int = Field(0, description="总个数")
+    page_num: int = Field(1, description="第几页")
+    page_size: int = Field(10, description="每页个数")
+    config_id: str = Field('', description="脚本配置id")
+    result: List[TaskDetail] = Field(..., description="结果详情")
+
+
+class ConfigResponse(BaseResponse):
+    total_num: int = Field(0, description="总个数")
+    page_num: int = Field(1, description="第几页")
+    page_size: int = Field(10, description="每页个数")
+    result: List[ConfigDetail] = Field(..., description="结果详情")
+
+
+class QueryWordItem(BaseModel):
+    query_word: str = Field("红包", description="查询词", min_length=1)
+    stat_date: str = Field("2021-11-16", description="日期", min_length=10, max_length=10)
+    source_code: int = Field(2, description="数据来源编码{1:'内部创意', 2:'巨量创意', 3:'开眼快创'}")
+
+
+class ScriptConfigLst(BaseModel):
+    start_date: Optional[date] = Field(date.today() + timedelta(days=-6), description="开始日期-用于查询")
+    end_date: Optional[date] = Field(date.today(), description="结束日期-用于查询")
+    search_word: Optional[str] = Field('', description="关键词-用于查询")
+    page_num: int = Field(1, description="第几页")
+    page_size: int = Field(10, description="每页的大小")
+
+
+class QueryWordTaskInfoLst(BaseModel):
+    start_date: Optional[date] = Field(date.today() + timedelta(days=-30), description="开始日期-用于查询")
+    end_date: Optional[date] = Field(date.today(), description="结束日期-用于查询")
+    search_word: Optional[str] = Field('', description="关键词-用于查询")
+    page_num: int = Field(1, description="第几页")
+    page_size: int = Field(10, description="每页的大小")
+    config_id: Optional[str] = Field('', description="脚本配置id")
+    source_code: Optional[List[int]] = Field([0], description="数据来源编码{1:'内部创意', 2:'巨量创意', 3:'开眼快创'}")
+
+
+class AddScriptConfig(BaseModel):
+    query_word_lst: List[str] = Field(..., description="关键词组")
+    operator: str = Field(..., description="操作者")
+
+    class Config:
+        schema_extra = {
+            "example": {
+                "query_word_lst": ["红包", "淘特"],
+                "operator": "龙猫"
+            }
+        }
+
+
+@logger.catch
+@app.post('/export_script_file/', tags=['front-end interactive'],
+          description="导出文件",
+          summary='导出文件'
+          )
+def export_script_file(item: List[QueryWordItem]):
+    try:
+        video_df = pd.DataFrame()
+        # 1 从数据库获取视频数据
+        # 如果同一个素材有多个查询词,则合并打上这多个查询词
+        for obj in item:
+            query_word = obj.query_word
+            stat_date = obj.stat_date
+            source_code = obj.source_code
+            sql = f"select signature, video_url, query_word, stat_date, {source_code} source_code from {source_name_map[source_code]['table']} " \
+                  f"where query_word = '{query_word}' " \
+                  f"and stat_date = '{stat_date}'"
+            df = pd.read_sql(sql, ai_word_engine)
+            video_df = video_df.append(df)
+
+        # 按 'signature' + 'query_word' + 'stat_date' 进行去重
+        video_df.drop_duplicates(['signature', 'query_word', 'stat_date', 'source_code'], keep='last', inplace=True)
+
+        video_query_word_df = video_df.groupby('signature').apply(lambda x: pd.Series({'query_word_lst': x['query_word'].unique(),
+                                                                                       'video_url': x['video_url'].values[0],
+                                                                                       'source_code': x['source_code'].values[0]}))
+        video_query_word_df.reset_index(inplace=True, drop=False)
+
+        # 如果来源==2 (头条巨量引擎),把视频链接替换为永久链接
+        video_query_word_df['video_url'] = video_query_word_df.apply(
+            lambda row: toutiao_static_video_url + row['signature'] if row.get('source_code') == 2 else row['video_url'], axis=1)
+
+        # 2 根据第一步的视频数据获取脚本
+        if not video_query_word_df.empty:
+            signature_lst = list(video_query_word_df.signature.values) if len(video_query_word_df.signature.values) > 1 \
+                else list(video_query_word_df.signature.values) * 2
+            sql = f"select signature, word_text from tb_asr_result where signature in {tuple(signature_lst)}" \
+                  f"and word_text is not null"
+            script_df = pd.read_sql(sql, ai_word_engine)
+            out_df = video_query_word_df.merge(script_df, on='signature', how='inner')
+
+        # 3 返回流数据
+        if not out_df.empty:
+            bio = BytesIO()
+            writer = pd.ExcelWriter(bio, engine='xlsxwriter')
+            out_df[['signature', 'query_word_lst', 'word_text', 'video_url']].to_excel(writer, index=False, encoding='utf8mb4')
+            writer.save()
+            bio.seek(0)
+
+            # 组装header
+            now_date = date.today().strftime('%Y-%m-%d')
+            headers = {"content-type": "application/vnd.ms-excel",
+                       "content-disposition": f"attachment;filename={quote('优质素材脚本_')}{now_date}.xlsx"
+                       }
+            logger.info(f"request body: {item}, message: 数据导出成功")
+            return StreamingResponse(bio, media_type='xlsx', headers=headers)
+        else:
+            logger.info(f"request body: {item}, message: 没有获取到对应的数据")
+            return None
+    except:
+        logger.error(f"request body: {item}, message: {traceback.format_exc()}")
+        return None
+
+
+@logger.catch
+@app.post('/get_script_config_lst/', tags=['front-end interactive'], response_model=ConfigResponse,
+          description="脚本配置列表",
+          summary='脚本配置列表'
+          )
+def get_script_config_lst(item: ScriptConfigLst):
+    try:
+        end_date = item.end_date + timedelta(days=1)
+        sql = f"select * from ctop_ai_script_query_word_config " \
+              f"where start_time >= '{item.start_date}' " \
+              f"and start_time < '{end_date}' " \
+              f"and ('{item.search_word}' = '' or query_word like '%%{item.search_word}%%')"
+        org_df = pd.read_sql(sql, ai_word_engine)
+
+        g_df = org_df.groupby('config_id').apply(lambda x: pd.Series({'query_word_lst': list(x['query_word'].unique()),
+                                                                      'operator': x['operator'].min(),
+                                                                      'create_time': str(x['start_time'].min())}))
+        g_df.reset_index(drop=False, inplace=True)
+        g_df.sort_values(by='create_time', ascending=False, inplace=True)
+        g_df['number'] = list(range(1, len(g_df) + 1))
+        total_num = g_df.shape[0]
+        detail = g_df.iloc[(item.page_num - 1) * item.page_size: item.page_num * item.page_size].to_dict('records')
+        response = {'code': 0,
+                    "message": "查询成功",
+                    "success": True,
+                    "result": detail,
+                    "total_num": total_num,
+                    "page_num": item.page_num,
+                    "page_size": item.page_size}
+        logger.info(f"request body: {item}, response body: {response}")
+        return response
+    except:
+        response = {"code": -1,
+                    "message": traceback.format_exc(),
+                    "success": False,
+                    "result": None}
+        logger.error(f"request body: {item}, response body: {response}")
+        return response
+
+
+@logger.catch
+@app.post('/get_query_word_task_info_lst/', tags=['front-end interactive'], response_model=TaskResponse,
+          description="脚本数据导出列表",
+          summary='脚本数据导出列表'
+          )
+def get_query_word_task_info_lst(item: QueryWordTaskInfoLst):
+    try:
+        end_date = item.end_date + timedelta(days=1)
+        source_code_lst = item.source_code * 2 if len(item.source_code) == 1 else item.source_code
+        if item.config_id != '':
+            sql = f"select distinct(query_word) query_word from ctop_ai_script_query_word_config where config_id = {item.config_id}"
+            query_word_lst = list(pd.read_sql(sql, ai_word_engine).query_word.values)
+            if len(query_word_lst) > 0:
+                query_word_lst = query_word_lst * 2 if len(query_word_lst) == 1 else query_word_lst
+                sql = f"select * from ctop_ai_query_word_task_record where query_word in {tuple(query_word_lst)}" \
+                      f"and stat_date >= '{item.start_date}' and stat_date < '{end_date}' " \
+                      f"and ('{item.source_code}' = '[0]' or source_code in {tuple(source_code_lst)}) " \
+                      f"and ('{item.search_word}' = '' or query_word = '{item.search_word}')"
+                df = pd.read_sql(sql, ai_word_engine)
+        else:
+            sql = f"select * from ctop_ai_query_word_task_record where " \
+                  f"stat_date >= '{item.start_date}' and stat_date < '{end_date}' " \
+                  f"and ('{item.source_code}' = '[0]' or source_code in {tuple(source_code_lst)}) " \
+                  f"and ('{item.search_word}' = '' or query_word = '{item.search_word}')"
+            df = pd.read_sql(sql, ai_word_engine)
+
+        df['source_name'] = df['source_code'].apply(lambda x: source_name_map[x]['name'])
+        df = df[['source_name', 'query_word', 'stat_date', 'script_num', 'task_status']]
+        df.sort_values(['stat_date', 'source_name', 'query_word'], ascending=False, inplace=True)
+        df['number'] = list(range(1, len(df) + 1))
+        total_num = df.shape[0]
+        detail = df.iloc[(item.page_num - 1) * item.page_size: item.page_num * item.page_size].to_dict('records')
+
+        response = {'code': 0,
+                    "message": "查询成功",
+                    "success": True,
+                    "result": detail,
+                    "total_num": total_num,
+                    "page_num": item.page_num,
+                    "page_size": item.page_size,
+                    "config_id": item.config_id}
+        logger.info(f"request body: {item}, response body: {response}")
+        return response
+    except:
+        response = {"code": -1,
+                    "message": traceback.format_exc(),
+                    "success": False,
+                    "result": None}
+        logger.error(f"request body: {item}, response body: {response}")
+        return response
+
+
+@logger.catch
+@app.post('/add_script_config/',
+          tags=['front-end interactive'],
+          description="新增脚本配置",
+          summary='新增脚本配置',
+          response_model=BaseResponse)
+def add_script_config(item: AddScriptConfig):
+    try:
+        # 按查询词拆分配置记录
+        config_id = str(uuid.uuid4())
+        config_df = pd.DataFrame(data=item.query_word_lst, columns=['query_word'])
+        config_df['config_id'] = config_id
+        config_df['operator'] = item.operator
+        config_df['operate_type'] = 1
+
+        # 新增配置记录插入到 ctop_ai_script_query_word_config
+        config_df.to_sql(name="ctop_ai_script_query_word_config",
+                         con=ai_word_engine,
+                         if_exists='append',
+                         index=False)
+        logger.info(f"request body: {item}, code:0, message: add_script_config success")
+        return {"code": 0,
+                "message": "add success",
+                "success": True}
+    except:
+        logger.error(f"request body: {item}, code:-1, message: add_script_config fail {traceback.format_exc()}")
+        return {"code": -1,
+                "message": traceback.format_exc(),
+                "success": False}
+
+
+@logger.catch
+@app.post('/get_material_and_script_time_task/',
+          response_model=BaseResponse,
+          tags=['back-end task'],
+          description="获取素材和脚本任务",
+          summary='获取素材和脚本任务')
+def get_material_and_script_time_task():
+    try:
+        get_material_and_script()
+        logger.info(f"{date.today().strftime('%Y-%m-%d')}, 获取素材和脚本任务执行完成.")
+        return {"code": 0,
+                "success": True,
+                "message": f"{date.today().strftime('%Y-%m-%d')},获取素材和脚本任务执行完成."}
+    except:
+        logger.error(f"{date.today().strftime('%Y-%m-%d')}, 获取素材和脚本任务执行发生异常. {traceback.format_exc()}")
+        return {"code": -1,
+                "success": False,
+                "message": f"{date.today().strftime('%Y-%m-%d')},获取素材和脚本任务执行发生异常 .{traceback.format_exc()}"}
 
 
 if __name__ == '__main__':

+ 11 - 11
readme.md

@@ -3,35 +3,35 @@
 1、每周日从巨量引擎获取近一周的优质素材。 (项目名称可以配置)<br>
 (素材数据存为了两张表:ctop_ai_material_info_from_ocean_engine 和 ctop_ai_video_info_from_ocean_engine)<br>
 2、每周一导出指定项目近一周的公司内部高质量&低质量素材,巨量引擎优质素材的脚本。(项目名称可以配置)<br>
-3、每次请求腾讯云的语音转脚本服务,会过根据md5滤掉已经请求过的素材。<br>
+3、每次请求腾讯云的语音转脚本服务,会过根据signature滤掉已经请求过的素材。<br>
 
 **项目启动说明**
 
 1、git clone 最新代码<br>
- git clone http://git.tjyourong.com.cn/liyuyi/video-to-word.git <br>
+ git clone http://git.tjyourong.com.cn/liyuyi/video_to_word.git <br>
 
 2、启动获取腾讯云语音转脚本的服务<br>
-切换到目录: cd /data/pythonProject/video-to-word
-nohup /data/Miniconda3/envs/video-to-word/bin/python -u /data/pythonProject/video-to-word/main.py >/dev/null 2>&1 &
+切换到目录: cd /data/pythonProject/video_to_word
+nohup /data/Miniconda3/envs/video_to_word/bin/python -u /data/pythonProject/video_to_word/main.py >/dev/null 2>&1 &
 
 3、定时任务配置<br>
 3-1 定期从巨量引擎获取优质素材,并转化为脚本。每周日晚上10:30执行一次<br>
-30 20  *  *  7   /data/Miniconda3/envs/video-to-word/bin/python -u  /data/pythonProject/video-to-word/get_video_from_ocean_engine.py  >> /data/pythonProject/video-to-word/logs/get_video_from_ocean_engine.log &
+30 20  *  *  7   /data/Miniconda3/envs/video_to_word/bin/python -u  /data/pythonProject/video_to_word/get_video_from_ocean_engine.py  >> /data/pythonProject/video_to_word/logs/get_video_from_ocean_engine.log &
 
 -- for test<br>
-nohup /data/Miniconda3/envs/video-to-word/bin/python -u  /data/pythonProject/video-to-word/get_video_from_ocean_engine.py  >> /data/pythonProject/video-to-word/logs/get_video_from_ocean_engine.log &
+nohup /data/Miniconda3/envs/video_to_word/bin/python -u  /data/pythonProject/video_to_word/get_video_from_ocean_engine.py  >> /data/pythonProject/video_to_word/logs/get_video_from_ocean_engine.log &
 
 3-2 每周导出内部高质量&低质量,巨量引擎优质素材的脚本。每周一早上08:30 执行一次<br>
-30 8  *  *  1   /data/Miniconda3/envs/video-to-word/bin/python -u  /data/pythonProject/video-to-word/get_high_quality_script.py  >> /data/pythonProject/video-to-word/logs/get_high_quality_script.log &
+30 8  *  *  1   /data/Miniconda3/envs/video_to_word/bin/python -u  /data/pythonProject/video_to_word/get_high_quality_script.py  >> /data/pythonProject/video_to_word/logs/get_high_quality_script.log &
 
 -- for test<br>
-nohup /data/Miniconda3/envs/video-to-word/bin/python -u  /data/pythonProject/video-to-word/get_high_quality_script.py  >> /data/pythonProject/video-to-word/logs/get_high_quality_script.log &
+nohup /data/Miniconda3/envs/video_to_word/bin/python -u  /data/pythonProject/video_to_word/get_high_quality_script.py  >> /data/pythonProject/video_to_word/logs/get_high_quality_script.log &
 
 
 4、查看执行日志<br>
-cd /data/pythonProject/video-to-word/logs<br>
-tail -f /data/pythonProject/video-to-word/logs/get_high_quality_script.log<br>
-tail -f /data/pythonProject/video-to-word/logs/get_video_from_ocean_engine.log<br>
+cd /data/pythonProject/video_to_word/logs<br>
+tail -f /data/pythonProject/video_to_word/logs/get_high_quality_script.log<br>
+tail -f /data/pythonProject/video_to_word/logs/get_video_from_ocean_engine.log<br>
 
 
 配置定时任务:<br>

+ 0 - 0
time_task/__init__.py


+ 128 - 0
time_task/get_material_and_script_by_query_word.py

@@ -0,0 +1,128 @@
+import datetime
+import os
+import sys
+import traceback
+import uuid
+from datetime import date
+import pandas as pd
+import yaml
+from loguru import logger
+
+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, mysql_replace_into
+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_material_and_script.get_material_from_huichuang import GetMaterialFromHuiChuang
+
+
+def get_material_and_script():
+    # 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 数据库连接引擎,依据开发环境/生产环境 进行切换
+    mac = uuid.UUID(int=uuid.getnode()).hex[-12:]
+    if mac in ['5254003fa716', '52540003f5dd']:
+        ai_word_engine = get_db_engine(config['ai_word_dev_db'])
+    else:
+        ai_word_engine = get_db_engine(config['ai_word_product_db'])
+
+    jeecg_boot_engine = get_db_engine(config['jeecg_boot_product_db'])
+
+    # 1-2 分批写入数据库的行数
+    chunk_size = config['chunk_size']
+
+    # 1-3 渠道编码&名称
+    source_name_map = config['source_name_map']
+
+    # 2 读取查询表得到关键词和渠道
+    sql = f"select query_word from ctop_ai_script_query_word_config where end_time = '9999-12-31'"
+    df = pd.read_sql(sql, ai_word_engine)
+    query_word_lst = list(df['query_word'].unique())
+    query_word_time = 3
+    for query_word in query_word_lst:
+        for source_code, value in source_name_map.items():
+            if value['status'] == 1:
+                # 任务开始执行
+                task_info = {"source_code": source_code,
+                             "task_status": "执行中",
+                             "query_word": query_word,
+                             "script_num": None,
+                             "stat_date": datetime.datetime.today().strftime('%Y-%m-%d')}
+                task_info_df = pd.DataFrame([task_info])
+                task_info_df.to_sql(name='ctop_ai_query_word_task_record',
+                                    con=ai_word_engine,
+                                    if_exists='append',
+                                    index=False,
+                                    chunksize=1000,
+                                    method=mysql_replace_into)
+                try:
+                    inst = None
+                    # 获取视频信息
+                    if source_code == 1:
+                        # 获取公司内部的优质素材
+                        inst = GetMaterialFromHuiChuang(query_word=query_word,
+                                                        logger=logger,
+                                                        jeecg_boot_db_engine=jeecg_boot_engine,
+                                                        ai_word_db_engine=ai_word_engine)
+                        inst.get_video_basic_info()
+                    if source_code == 3:
+                        # 获取快手开眼快创的优质素材
+                        inst = GetMaterialFromKuaishouKaiyan(query_word=query_word,
+                                                             logger=logger,
+                                                             db_engine=ai_word_engine)
+                        inst.get_video_basic_info()
+                    if source_code == 2:
+                        # 获取头条巨量引擎的优质素材,需要使用参数 query_time_range
+                        inst = GetMaterialFromOceanEngine(query_word=query_word,
+                                                          period_type=query_word_time,
+                                                          logger=logger,
+                                                          db_engine=ai_word_engine)
+                        inst.get_material_basic_info()
+                        if inst.signature_lst:
+                            inst.get_video_basic_info()
+
+                    # 获取脚本信息
+                    if not inst.video_df.empty:
+                        inst.get_script_from_teng_xun_yun()
+                        sql = f"select signature from tb_asr_result where signature in" \
+                              f" {tuple(inst.video_df.signature.values) if len(inst.video_df) > 1 else tuple(list(inst.video_df.signature.values) * 2)}" \
+                              f"and word_text is not null"
+                        df = pd.read_sql(sql, ai_word_engine)
+                        script_num = len(df)
+                    else:
+                        script_num = 0
+                    # 更新任务执行情况: 执行状态 和 成功的数据量
+                    with ai_word_engine.begin() as conn:
+                        sql = f"update ctop_ai_query_word_task_record set script_num = {script_num}, task_status = '执行完成', message='执行成功'" \
+                              f"where query_word = '{query_word}'" \
+                              f"and stat_date = '{datetime.datetime.today().strftime('%Y-%m-%d')}'" \
+                              f"and source_code = {source_code}"
+                        conn.execute(sql)
+                    logger.info(f"查询词:{query_word} 数据来源: {source_code} 日期: {datetime.datetime.today().strftime('%Y-%m-%d')},"
+                                f"执行完成获取 {script_num} 条数据")
+                except:
+                    with ai_word_engine.begin() as conn:
+                        sql = f"update ctop_ai_query_word_task_record set  task_status = '执行异常' , message ={traceback.format_exc()}" \
+                              f"where query_word = '{query_word}'" \
+                              f"and stat_date = '{datetime.datetime.today().strftime('%Y-%m-%d')}'" \
+                              f"and source_code = {source_code}"
+                        conn.execute(sql)
+                    logger.error(f"查询词:{query_word} 数据来源: {source_code} 日期: {datetime.datetime.today().strftime('%Y-%m-%d')},"
+                                 f"执行异常: {traceback.format_exc()}")
+
+
+if __name__ == '__main__':
+    logger.remove()
+    logger.add("/data/pythonProject/video_to_word/logs/get_high_quality_material.{time:YYYY-MM-DD}.log",
+               rotation="00:00",
+               format="{time:YYYY-MM-DD HH:mm:ss,SSS} [{process}] [{thread}] {level} {file} {line} - {message}",
+               level="INFO")
+    try:
+        get_material_and_script()
+        logger.info(f"{date.today().strftime('%Y-%m-%d')}, 获取素材和脚本任务执行完成.")
+    except:
+        logger.error(f"{date.today().strftime('%Y-%m-%d')}, 获取素材和脚本任务执行发生异常. {traceback.format_exc()}")

+ 83 - 0
time_task/tmp_task.py

@@ -0,0 +1,83 @@
+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}")