liyuyi@c-top.com.cn 3 年之前
父節點
當前提交
64ac8db30c

+ 1 - 1
.idea/video_to_word.iml

@@ -2,7 +2,7 @@
 <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="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">

+ 1 - 1
common_func.py

@@ -18,7 +18,7 @@ def get_db_engine(db_info):
                   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
 
 

+ 14 - 3
config/config.yaml

@@ -1,7 +1,13 @@
 source_name_map:
-  1: '内部'
-  2: '头条巨量'
-  3: '快手开眼'
+  1:
+      name: '内部'
+      table: 'ctop_ai_video_info_from_huichuang'
+  2:
+      name: '头条巨量'
+      table: 'ctop_ai_video_info_from_ocean_engine'
+  3:
+      name: '快手开眼'
+      table: 'ctop_ai_material_info_from_kuaishou_kaiyan'
 
 
 # jeecg-boot 生产数据库
@@ -11,6 +17,7 @@ jeecg_boot_product_db:
   password: hcst@2021
   port: 3390
   database: jeecg-boot
+  charset: utf8mb4
 
 # jeecg-boot 测试数据库
 jeecg_boot_dev_db:
@@ -19,6 +26,7 @@ jeecg_boot_dev_db:
   password: hcst@2021
   port: 3390
   database: jeecg-boot
+  charset: utf8mb4
 
 # db_ai_word 测试数据库
 ai_word_dev_db:
@@ -27,6 +35,7 @@ ai_word_dev_db:
   password: hcst@2020
   port: 3306
   database: db_ai_word
+  charset: utf8mb4
 
 
 # db_ai_word 生产数据库
@@ -36,6 +45,7 @@ ai_word_product_db:
   password: hcst@2020
   port: 3306
   database: db_ai_word
+  charset: utf8mb4
 
 
 # 本地数据库 mysql
@@ -45,6 +55,7 @@ local_db:
   password: root@123
   port: 3306
   database: mysql
+  charset: utf8mb4
 
 # 每次写入数据库的行数
 chunk_size: 200

+ 5 - 4
config/url.py

@@ -1,13 +1,14 @@
-import os
-os_env = os.getenv('LYY_DEV', 'unknown')
+import uuid
+
+mac = uuid.UUID(int=uuid.getnode()).hex[-12:]
 debug_url = "http://139.186.165.84:31013/"
 product_url = "http://139.186.27.96:31013/"
 
 # 向腾讯云发送语音转脚本的任务请求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 = (debug_url if mac == '5254003fa716' else product_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 = (debug_url if mac == '5254003fa716' else product_url) + 'asr/task/result'
 
 # 巨量引擎获取优质素材信息
 get_material_info_from_ocean_engine_url = "https://cc.oceanengine.com/creative_radar_api/v1/material/list"

+ 8 - 6
database.py

@@ -17,12 +17,13 @@ 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)
 
@@ -33,10 +34,10 @@ engine = create_engine('mysql+mysqlconnector://hcst:' + passowrd + '@139.186.165
 DBSession = sessionmaker(bind=engine)
 
 
-def query(md5, task_status, task_id):
+def query(signature, task_status, task_id):
     condition = (1 == 1)
-    if md5 != None:
-        condition = and_(condition, Task.md5 == md5)
+    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:
@@ -71,6 +72,7 @@ def update(task):
     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()
@@ -80,8 +82,8 @@ def update(task):
 
 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)
+    print(task.signature)
     # update(task)

+ 69 - 0
get_material_and_script/get_material_from_huichuang.py

@@ -0,0 +1,69 @@
+import datetime
+from datetime import timedelta
+
+import pandas as pd
+
+from common_func import mysql_replace_into
+
+
+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 禁用',
+        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}'"
+                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')
+
+        # 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)

+ 6 - 5
get_material_and_script/get_material_from_kuaishou_kaiyan.py

@@ -1,11 +1,9 @@
 import datetime
 import json
 import traceback
-
 import pandas as pd
 import requests
-
-from common_func import NpEncoder, get_db_engine, get_logger
+from common_func import NpEncoder, get_db_engine, get_logger, mysql_replace_into
 from config.url import get_video_info_from_kuaishou_kaiyan
 
 
@@ -71,17 +69,20 @@ class GetMaterialFromKuaishouKaiyan(object):
             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 写入数据库, 表中以 photoId + query_word 作为联合唯一键,写入数据库时如果唯一键重复,则 replace_into
+            # 2-3 写入数据库, 表中以 photoId + 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)
+                                     chunksize=1000,
+                                     method=mysql_replace_into)
             except:
                 self.logger.error(traceback.format_exc())
 

+ 19 - 18
get_material_and_script/get_material_from_ocean_engine.py

@@ -65,24 +65,25 @@ class GetMaterialFromOceanEngine(object):
         # 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)
-
-        # 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 作为联合唯一键,写入数据库时如果唯一键重复,则 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)
+        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):
         """

+ 16 - 7
get_material_and_script/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,9 +20,8 @@ class GetScriptFromTengXunYunServer(object):
         向腾讯云提交语音转脚本的任务
         task_df: DataFrame columns 包含 signature 和 video_url
         """
-
         # 1 获取已经被提交过的任务
-        sql = """select md5 signature from tb_asr_result """
+        sql = """select signature from tb_asr_result """
         submitted_task_df = pd.read_sql(sql, self.db_engine)
 
         # 2 需要提交的任务,去掉历史被提交过的任务,防止重复提交浪费服务时长
@@ -31,7 +32,7 @@ class GetScriptFromTengXunYunServer(object):
         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:
@@ -41,7 +42,11 @@ class GetScriptFromTengXunYunServer(object):
                 self.logger.error("素材:%s, 向腾讯云提交语音转脚本任务的返回信息为:%s " % (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),)
+        if len(self.task_df.signature.values) == 1:
+            signature_for_sql = list(self.task_df.signature.values) * 2
+        else:
+            signature_for_sql = list(self.task_df.signature.values)
+        sql = """select task_id from tb_asr_result where signature in %s""" % (tuple(signature_for_sql),)
         task_id_df = pd.read_sql(sql, self.db_engine)
         self.task_ids = list(task_id_df['task_id'].values)
 
@@ -56,6 +61,10 @@ class GetScriptFromTengXunYunServer(object):
         self.logger.info("从腾讯云获取脚本的个数为%s" % len(self.task_ids))
         retry_upper_cnt = 10
         retry_cnt = 1
+
+        if len(self.task_ids) == 1:
+            self.task_ids = self.task_ids * 2
+
         while retry_cnt <= retry_upper_cnt:
             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),)

+ 100 - 30
main.py

@@ -1,13 +1,28 @@
+import datetime
 from concurrent.futures import ThreadPoolExecutor
-from typing import Optional
-
+from io import BytesIO
+from typing import Optional, List
+from urllib.parse import quote
+import hashlib
+import pandas as pd
 import uvicorn
+import yaml
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel, Field
 
 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
 
+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']
+
+ai_word_engine = get_db_engine(config['ai_word_dev_db'])
+
 threadPool = ThreadPoolExecutor(max_workers=4)
 app = FastAPI()
 origins = [
@@ -33,7 +48,7 @@ app.add_middleware(
 
 
 class QueryItem():
-    md5: Optional[str] = None
+    signature: Optional[str] = None
     url: Optional[str] = None
 
 
@@ -43,9 +58,9 @@ def index():
 
 
 @app.post('/asr/task/submit')
-def task_submit(md5: str, url: str):
+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}
 
@@ -60,6 +75,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
@@ -73,33 +89,87 @@ def task_submit(task_status: int):
     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 QueryWordItem(BaseModel):
+    query_word: str = Field(..., description="查询词", min_length=1)
+    stat_date: str = Field(..., description="日期", min_length=10, max_length=10)
+    source: int = Field(..., description="来源,")
+
+
+@app.post('/export_excel/')
+def export_excel(item: List[QueryWordItem]):
+    video_df = pd.DataFrame()
+    if len(item) == 1:
+        # 单个条目,直接导出
+        pass
+    else:
+        # 1 从数据库获取视频数据
+        # 多个条目,如果同一个素材有多个查询词,则合并打上这多个查询词
+        for obj in item:
+            query_word = obj.query_word
+            stat_date = obj.stat_date
+            source = obj.source
+            sql = f"select signature, video_url, query_word, stat_date, {source} source from {source_name_map[source]['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'], keep='last', inplace=True)
+        g = video_df.groupby('signature')
+
+        query_word_lst_df = g.apply(lambda x: x['query_word'].unique())
+        query_word_lst_df.name = 'query_word_lst'
+
+        url_df = g.apply(lambda x: x['video_url'].values[0])
+        url_df.name = 'video_url'
+
+        source_df = g.apply(lambda x: x['source'].values[0])
+        source_df.name = 'source'
+
+        video_query_word_df = pd.concat([query_word_lst_df, url_df, source_df], axis=1)
+        video_query_word_df.reset_index(inplace=True, drop=False)
+
+        video_query_word_df['video_url'] = video_query_word_df.apply(
+            lambda row: toutiao_static_video_url + row['signature'] if row.get('source') == 2 else row['video_url'], axis=1)
+
+        # 2 根据第一步的视频数据获取脚本
+        if not video_query_word_df.empty:
+            sql = f"select signature, word_text from tb_asr_result where signature in " \
+                  f"{tuple(video_query_word_df.signature.values) if len(video_query_word_df.signature.values) > 1 else tuple(list(video_query_word_df.signature.values) * 2)} " \
+                  f"and task_status = 2"
+            script_df = pd.read_sql(sql, ai_word_engine)
+            out_df = video_query_word_df.merge(script_df, on='signature', how='inner')
+        else:
+            pass
+
+        # 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 = datetime.date.today().strftime('%Y-%m-%d')
+            headers = {"content-type": "application/vnd.ms-excel",
+                       "content-disposition": f"attachment;filename={quote('优质素材脚本_')}{now_date}.xlsx"
+                       }
+
+            return StreamingResponse(bio, media_type='xlsx', headers=headers)
+
+    return None
 
 
 if __name__ == '__main__':
+    # 1 读取配置文件
+
+    # test_items = [{'query_word': '红包', 'stat_date': '2021-10-28', 'source': 2},
+    #               {'query_word': '红包', 'stat_date': '2021-10-28', 'source': 3},
+    #               {'query_word': '赚钱', 'stat_date': '2021-10-28', 'source': 2},
+    #               {'query_word': '赚钱', 'stat_date': '2021-10-28', 'source': 3}]
+    # export_excel(test_items)
+
     uvicorn.run(app='main:app', host="0.0.0.0", port=31013, reload=True, debug=True)
 # gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker #线上启动命令

+ 1 - 1
readme.md

@@ -3,7 +3,7 @@
 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>
 
 **项目启动说明**
 

+ 59 - 142
time_task/get_material_and_script_by_query_word.py

@@ -1,7 +1,6 @@
 import datetime
 import os
 import sys
-from datetime import timedelta
 
 import pandas as pd
 import yaml
@@ -11,14 +10,12 @@ project_root_path = curr_path[:curr_path.find("video_to_word") + len("video_to_w
 sys.path.append(project_root_path)
 
 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_material_and_script.get_script_from_tengxunyun import GetScriptFromTengXunYunServer
-
+from get_material_and_script.get_material_from_huichuang import GetMaterialFromHuiChuang
 
 if __name__ == '__main__':
-    print("test end")
     # 创建日志对象
     logger = get_logger(log_file_name="/data/pythonProject/video_to_word/logs/get_high_quality_material.log",
                         log_name='get_high_quality_material_logger')
@@ -45,147 +42,67 @@ if __name__ == '__main__':
     sql = """
     select  query_word,
             query_time_range,
-            is_new,
-            source_code_lst
-     from ctop_ai_query_word where status = 1
+            source_code
+     from ctop_ai_query_word where status = 0
     """
     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个月
-                        # TODO periods=18 需要修改
-                        start_date = (datetime.date.today() + timedelta(days=-18)).strftime('%Y-%m-%d')
-                        full_date_df = pd.DataFrame()
-                        for date in pd.date_range(start=start_date, freq='D', periods=18):
-                            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 添加日志
-                    logger.info("查询词:%s,快手开眼快创没有对应的素材!" % query_word)
-
-            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()
+        source_code = eval(row['source_code'])
+
+        # 0 记录任务执行情况的字段
+        size = 0
+        message = ""
+        task_status = 0
+        inst = None
+
+        # 1 获取优质素材video_url
+        video_df = pd.DataFrame()
+        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_time_range,
+                                              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:
-                    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)
+
+        # 2 调用腾讯云的语音转脚本服务,获取脚本
+        if (inst is not None) and (not inst.video_df.empty):
+            inst.video_df = inst.video_df[~inst.video_df.signature.isnull()]
+            get_script_ins = GetScriptFromTengXunYunServer(logger=logger,
+                                                           db_engine=ai_word_engine,
+                                                           task_df=inst.video_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': source_code,
+                     'size': size}
+        logger.info(f"{task_info}")
+
+

+ 84 - 0
time_task/tmp_task.py

@@ -0,0 +1,84 @@
+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
+from get_material_and_script.get_material_from_huichuang import GetMaterialFromHuiChuang
+
+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 = 458
+    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 limit 100"
+    df = pd.read_sql(sql, jeecg_boot_engine)
+    df['query_word'] = query_word
+    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.emtpy:
+        inst.video_df = inst.video_df[~inst.video_df.signature.isnull()]
+        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}")