liyuyi@c-top.com.cn 4 lat temu
rodzic
commit
5f502cf47e
6 zmienionych plików z 342 dodań i 51 usunięć
  1. 45 0
      common_func.py
  2. 30 0
      config/config.yaml
  3. 11 0
      config/url.py
  4. 37 30
      get_data.py
  5. 219 0
      get_video_from_ocean_engine.py
  6. 0 21
      requirements.txt

+ 45 - 0
common_func.py

@@ -0,0 +1,45 @@
+from sqlalchemy import create_engine
+from urllib import parse
+import json
+import numpy as np
+
+
+# 创建数据库连接引擎
+def get_db_engine(db_info):
+    db_con_str = 'mysql+pymysql://%s:%s@%s:%d/%s' % \
+                 (db_info['username'],
+                  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'})
+    return engine
+
+
+# insert into 替换为 replace into 存在重复唯一键时,先删除此行数据,然后插入新的数据
+def mysql_replace_into(table, conn, keys, data_iter):
+    from sqlalchemy.dialects.mysql import insert
+    from sqlalchemy.ext.compiler import compiles
+    from sqlalchemy.sql.expression import Insert
+
+    @compiles(Insert)
+    def replace_string(insert, compiler, **kw):
+        s = compiler.visit_insert(insert, **kw)
+        s = s.replace("INSERT INTO", "REPLACE INTO")
+        return s
+
+    data = [dict(zip(keys, row)) for row in data_iter]
+
+    conn.execute(table.table.insert(replace_string=""), data)
+
+
+class NpEncoder(json.JSONEncoder):
+    def default(self, obj):
+        if isinstance(obj, np.integer):
+            return int(obj)
+        elif isinstance(obj, np.floating):
+            return float(obj)
+        elif isinstance(obj, np.ndarray):
+            return obj.tolist()
+        else:
+            return super(NpEncoder, self).default(obj)

+ 30 - 0
config/config.yaml

@@ -0,0 +1,30 @@
+projectName:
+  - '淘特'
+  - '支付宝'
+
+# 生产数据库
+productDB:
+  host: 139.186.27.96
+  username: data
+  password: hcst@2021
+  port: 3390
+  database: jeecg-boot
+
+# 开发测试数据库
+devDB:
+  host: 139.186.165.84
+  username: hcst
+  password: hcst@2020
+  port: 3306
+  database: db_ai_word
+
+# 本地数据库
+localDB:
+  host: 192.168.1.193
+  username: root
+  password: root@123
+  port: 3306
+  database: mysql
+
+# 每次写入数据库的行数
+chunkSize: 200

+ 11 - 0
config/url.py

@@ -0,0 +1,11 @@
+# 向腾讯云发送语音转脚本的任务请求url
+voice_to_script_task_submit_url = 'http://139.186.165.84:31012/asr/task/submit'
+
+# 依据task_id,向腾讯云获取脚本
+voice_to_script_task_result_url = 'http://139.186.165.84:31012/asr/task/result'
+
+# 巨量引擎获取优质素材信息
+get_material_info_from_ocean_engine_url = "https://cc.oceanengine.com/creative_radar_api/v1/material/list"
+
+# 依据 vid, 从巨量引擎获取视频信息
+get_video_info_from_ocean_engine_url = "https://cc.oceanengine.com/creative_content_server/api/video/info"

+ 37 - 30
get_data.py

@@ -20,42 +20,49 @@ if __name__ == '__main__':
     task_result_url = 'http://139.186.165.84:31012/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) >= 50) 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())]
+    # 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)
+    # 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)]
+    # 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)
+    # 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"""

+ 219 - 0
get_video_from_ocean_engine.py

@@ -0,0 +1,219 @@
+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
+import datetime
+from config.url import voice_to_script_task_submit_url, voice_to_script_task_result_url, get_material_info_from_ocean_engine_url, \
+    get_video_info_from_ocean_engine_url
+import time
+
+
+def get_material_info(project_name, period_type):
+    material_df = pd.DataFrame()
+    has_more = True  # 是否还存在分页数据, 初始化为 True
+    limit = 10  # 每页获取30条
+    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}
+
+        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
+
+    # 数据类型的处理,方便入库
+    # 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['project_name'] = 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)
+    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]
+        else:
+            query_ids = vid[i:]
+
+        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)
+
+    # 数据类型的处理,方便入库
+    # 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['project_name'] = 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
+
+
+def submit_script_task(df):
+    """
+    向腾讯云提交语音转脚本的任务
+    :param df: DataFrame columns 包含 signature 和 url
+    :return: task_ids
+    """
+    # 1 获取已经被提交过的任务
+    sql = """select md5 signature from tb_asr_result """
+    submitted_task_df = pd.read_sql(sql, read_engine)
+
+    # 2 需要提交的任务,去掉历史被提交过的任务,防止重复提交浪费服务时长
+    to_submit_task_df = df[~df.signature.isin(submitted_task_df.signature.values)]
+
+    # 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_full_path = voice_to_script_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)
+
+    # 4 获取素材对应的发送请求的 task_id
+    sql = """
+    select task_id from tb_asr_result where md5 in %s
+    """ % (tuple(df.signature.values),)
+    task_id_df = pd.read_sql(sql, read_engine)
+    task_ids = task_id_df['task_id'].values
+    return task_ids
+
+
+def get_result_from_tx(task_id_lst):
+    """
+    从腾讯云获取脚本
+    每隔5分钟获取一次,直到没有 执行中或者等待执行 的任务为止
+
+    Status Integer 任务状态码,0:任务等待,1:任务执行中,2:任务成功,3:任务失败。
+    StatusStr String 任务状态,waiting:任务等待,doing:任务执行中,success:任务成功,failed:任务失败。
+    ErrorMsg String 失败原因说明。
+    """
+    while True:
+        print("sleep 5 mins")
+        time.sleep(60 * 5)
+
+        sql = """select task_id, task_status from tb_asr_result where task_id in %s and task_status in (0,1)""" % (tuple(task_id_lst),)
+        task_status_df = pd.read_sql(sql, read_engine)
+        if task_status_df.empty:
+            break
+
+        for task_id in task_status_df.task_id.values:
+            request_data = {'task_id': task_id}
+            request_full_path = voice_to_script_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)
+
+
+if __name__ == '__main__':
+    # 1 读取配置文件
+    with open('config/config.yaml', mode='r', encoding='utf-8') as f:
+        config = yaml.load(f.read(), Loader=yaml.FullLoader)
+
+    # 1-1 数据库连接引擎,依据开发环境/生产环境 进行切换
+    # 读数据库引擎使用生产数据库,写数据库引擎依据系统环境进行切换(测试数据库/生产数据库)
+    # 注意: 该项目的读和写 都使用测试数据库
+    if os.getenv('LYY_DEV', 'unknown') == 'dev':
+        write_engine = get_db_engine(config['devDB'])
+    else:
+        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:
+        # 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']]
+        task_ids = submit_script_task(task_df)
+
+        # 2-4 向腾讯云获取已提交任务的脚本
+        get_result_from_tx(task_ids)

+ 0 - 21
requirements.txt

@@ -1,21 +0,0 @@
-asgiref==3.4.1
-certifi==2021.5.30
-charset-normalizer==2.0.4
-click==8.0.1
-colorama==0.4.4
-fastapi==0.68.1
-greenlet==1.1.1
-h11==0.12.0
-idna==3.2
-mysql==0.0.3
-mysql-connector==2.2.9
-mysqlclient==2.0.3
-pydantic==1.8.2
-requests==2.26.0
-SQLAlchemy==1.4.23
-starlette==0.14.2
-tencentcloud-sdk-python==3.0.484
-typing-extensions==3.10.0.2
-urllib3==1.26.6
-uvicorn==0.15.0
-wincertstore==0.2