aigc-server/gpt_process.py

import time
import requests
import traceback
from utils import download_video
from audio_process import get_wav
from xunfei_asr import call_asr
from config import set_config
from log import Log

config_ = set_config()
log_ = Log()


def request_gpt(prompt):
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {config_.GPT_OPENAI_API_KEY}',
    }
    proxies = config_.PROXIES

    json_data = {
        'model': 'gpt-3.5-turbo',
        'messages': [
            {
                'role': 'user',
                'content': f'{prompt}',
            },
        ],
    }
    # raise ConnectionResetError
    response = requests.post(url=config_.GPT_HOST, headers=headers, json=json_data, proxies=proxies)
    # print(response.json())
    # print(response.json()['choices'][0]['message']['content'])
    # print('\n')
    result_content = response.json()['choices'][0]['message']['content']
    return result_content


def title_generate(video_id, video_path):
    """
    视频生成标题
    :param video_id: videoId
    :param video_path: videoPath
    :return:
    """
    # 1. 下载视频
    video_file_path = download_video(video_path=video_path, video_id=video_id, download_folder='videos')
    log_.info(f"video_file_path = {video_file_path}")
    # 2. 获取视频中的音频
    audio_path = get_wav(video_path=video_file_path)
    log_.info(f"audio_path = {audio_path}")
    # 3. asr
    dialogue_path, asr_res = call_asr(audio_path=audio_path)
    log_.info({
        'asrResult': {'videoId': video_id, 'asrRes': asr_res}
    })
    # 4. gpt产出结果
    prompt = f"{config_.GPT_PROMPT['title']['prompt2']}{asr_res.strip()}"
    gpt_res = request_gpt(prompt=prompt)
    return gpt_res
    # except ConnectionResetError:
    #     log_.info(video_id)
    # except Exception as e:
    #     log_.info(traceback.format_exc())
    # else:
    #     print(gpt_res)


    # print(gpt_res)
    # log_.info(f"gpt_res = {gpt_res}")



if __name__ == '__main__':
    title_generate(video_id='001', video_path='')