google_ai_studio/google_ai/generativeai_video.py

import asyncio
import os
import time

import cv2
import aiohttp
import requests
import google.generativeai as genai

import uuid

from google.generativeai.types import HarmCategory, HarmBlockThreshold

from common import Common
from common.feishu_data import Material


class VideoAnalyzer:
    def __init__(self, api_key):
        """初始化类，配置 API 密钥和视频路径"""
        genai.configure(api_key=api_key)
        self.video_file = None

    async def process_and_delete_file(self, file_path):
        """删除文件"""
        try:
            print(f"正在处理文件: {file_path}" )
            os.remove( file_path )
            print( f"文件已删除: {file_path}" )
        except Exception as e:
            print( f"处理或删除文件时发生错误: {str( e )}" )

    async def download_video(self, video_url,  save_directory='/root/google_ai_studio/path'):
    # async def download_video(self, video_url, save_directory='/Users/tzld/Desktop/google_ai_studio/path'):

        """从给定的视频链接下载视频并保存到指定路径"""
        random_filename = f"{uuid.uuid4()}.mp4"
        save_path = os.path.join(save_directory, random_filename)

        max_retries = 3  # 最大重试次数
        for attempt in range(max_retries):
            try:
                # 发送 GET 请求获取视频内容
                headers = {}
                payload = {}
                response = requests.request("GET", video_url, headers=headers, data=payload)
                if response.status_code == 200:
                    # 以二进制写入模式打开文件
                    with open(save_path, "wb") as file:
                        # 将响应内容写入文件
                        file.write(response.content)
                    print(f"视频已成功下载并保存到: {save_path}")
                    return save_path
                else:
                    print(f"请求失败，状态码: {response.status_code}")

            except requests.exceptions.RequestException as e:
                print(f"下载视频时出现错误: {e}")

            # 如果下载失败，等待一段时间后重试
            time.sleep(1)

        print("达到最大重试次数，视频下载失败")
        return None

    async def upload_video(self, save_path, mime_type = None):
        """上传视频文件并获取视频文件对象"""
        try:
            self.video_file = genai.upload_file(save_path, mime_type=mime_type)
            await self._wait_for_processing()
        except Exception as e:
            Common.logger("ai").info(f'上传视频文件并获取视频文件对象失败异常信息{e}')
            self.video_file.delete()
            return f"上传视频文件并获取视频文件对象失败:{e}"

    async def _wait_for_processing(self):
        """等待视频文件处理完成"""
        while self.video_file.state.name == 'PROCESSING':
            print( '等待视频处理完成...' )
            await asyncio.sleep(2)  # 使用异步睡眠代替阻塞睡眠
            self.video_file = genai.get_file( self.video_file.name )
        print( f'视频处理完成: {self.video_file.uri}' )

    async def create_cache(self):
        try:
            generation_config = {
                "response_mime_type": "application/json"
            }
            """创建缓存内容，并返回生成模型"""
            # 创建生成模型，使用 gemini-1.5-flash 模型
            model = genai.GenerativeModel(
                model_name="gemini-1.5-flash",
                generation_config=generation_config,
                safety_settings={
                    HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_NONE,
                }

            )

            return model
        except Exception as e:
            Common.logger("ai").info(f'视频创建缓存内容，并返回生成模型异常信息{e}')
            self.video_file.delete()
            print(f"视频创建缓存内容，并返回生成模型异常信息{e}")
            return f"视频创建缓存内容，并返回生成模型异常信息{e}"

    async def analyze_video(self, model, questions, sample_data):
        try:
            chat_session = model.start_chat(history=[])
            message_content = {
                "parts": [
                    self.video_file,
                    str(questions) +
                    "输出返回格式样例:\n" + str(sample_data)

                ]
            }
            response = chat_session.send_message( message_content )
            self.video_file.delete()
            return response
        except Exception as e:
            Common.logger("ai").info(f'视频处理请求失败:{e}')
            self.video_file.delete()
            print( f"视频处理请求失败:{e}" )
            return f"视频处理请求失败:{e}"

    def video_duration(self, filename):
        cap = cv2.VideoCapture( filename )
        if cap.isOpened():
            rate = cap.get( 5 )
            frame_num = cap.get( 7 )
            duration = frame_num / rate
            return int(duration)
        return 0

async def main(video_path, api_key, prompt, mark):
    """主函数，执行视频上传、缓存创建、问题生成"""
    attempt = 0
    max_attempts = 1
    while attempt < max_attempts:
        try:
            # 初始化视频分析类
            analyzer = VideoAnalyzer(api_key )
            for file in genai.list_files():
                # file.delete()
                print( "  ", file.name )
            duration = analyzer.video_duration( video_path )
            print( f"视频时长为{duration}秒" )
            if int( duration ) >= 600 or int( duration ) == 0:
                return f"视频时长过长/视频时长为:{duration}秒"
            save_path = await analyzer.download_video(video_path)
            if not save_path or save_path == None:
                if os.path.exists( save_path ):
                    os.remove( save_path )
                    print( f"文件已删除: {save_path}" )
                print("视频下载失败")
                return "视频下载失败"
            # 上传并处理视频
            upload_response = await analyzer.upload_video( save_path )
            if upload_response:
                return upload_response, mark
            # 创建缓存模型
            model =await analyzer.create_cache()
            if isinstance( model, str ):
                return model, mark
            print("创建缓存模型成功")
            sample_data = {
                "一、基础信息": {
                    "视觉/音乐/文字": "",
                    "内容选题": "",
                    "视频主题": ""
                },
                "二、主体和场景": {
                    "视频主体": "",
                    "视频场景": []
                },
                "三、情感与风格": {},
                "四、视频传播性与观众": {
                    "片尾引导": {},
                    "传播性判断": "",
                    "观众画像": {}
                },
                "五、音画细节": {
                    "音频细节": {},
                    "视频水印": {},
                    "视频字幕": {},
                    "视频口播": ""
                },
                "六、人物与场景": {
                    "知名人物": {},
                    "人物年龄段": "",
                    "场景描述": []
                },
                "七、时效性与分类": {
                    "时效性": {},
                    "视频一级分类": "",
                    "二级分类": ["品类- 、分数-", "品类- 、分数-", "品类- 、分数-"]
                }
            }
            response =await analyzer.analyze_video( model, prompt, sample_data )
            if isinstance( response, str ):
                return response, mark
            print( response.usage_metadata )
            print(response.text)
            if os.path.exists( save_path ):
                os.remove( save_path )
                print( f"文件已删除: {save_path}" )
            return response.text, mark
        except Exception as e:
            attempt += 1  # 增加尝试次数
            if attempt < max_attempts:
                print(f"重试第 {attempt} 次...")
            else:
                print( "达到最大重试次数，处理失败" )
                Common.logger( "ai" ).info( f'视频分析处理失败异常信息{e}' )
                return f"视频分析处理失败:{e}", None


if __name__ == "__main__":
    proxy_url = 'http://127.0.0.1:1081'
    os.environ["http_proxy"] = proxy_url
    os.environ["https_proxy"] = proxy_url
    # video_path = 'http://temp.yishihui.com/longvideo/transcode/video/vpc/20240926/66510681PACx7zsp2wDBHJlicE.mp4'
    video_path = 'http://temp.yishihui.com/longvideo/transcode/video/vpc/20240605/68754804SJz5E9JNe5hAdSkRwF.mp4'
    api_key = "AIzaSyCor0q5w37Dy6fGxloLlCT7KqyEFU3PWP8"

    asyncio.run(main(video_path,api_key))