google_ai_studio/google_ai/generativeai_video.py

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

import uuid

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'):

        """从给定的视频链接下载视频并保存到指定路径"""
        try:
            # 发送 GET 请求获取视频内容
            random_filename = f"{uuid.uuid4()}.mp4"
            save_path = os.path.join(save_directory, random_filename)
            async with aiohttp.ClientSession() as session:
                async with session.get( video_url ) as response:
                    response.raise_for_status()  # 检查请求是否成功
                    with open( save_path, 'wb' ) as video_file:
                        while True:
                            chunk = await response.content.read( 1024 * 1024 )  # 每次读取 1MB
                            if not chunk:
                                break
                            video_file.write( chunk )

                print( f"视频已成功下载并保存到: {save_path}" )
                return save_path
        except requests.exceptions.RequestException as e:
            print( f"下载视频时出现错误: {e}" )
            return None

    async def upload_video(self, save_path, mime_type = None):
        """上传视频文件并获取视频文件对象"""
        self.video_file = genai.upload_file(save_path, mime_type=mime_type)
        await self._wait_for_processing()

    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):
        generation_config = {
            "temperature": 1,
            "top_p": 0.95,
            "top_k": 64,
            "max_output_tokens": 8192,
            "response_mime_type": "application/json"
        }
        """创建缓存内容，并返回生成模型"""
        # 创建生成模型，使用 gemini-1.5-flash 模型
        model = genai.GenerativeModel(
            model_name="gemini-1.5-flash",
            generation_config=generation_config,

        )

        return model
    async def analyze_video(self, model, questions, sample_data):
            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

    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:
                if os.path.exists( save_path ):
                    os.remove( save_path )
                    print( f"文件已删除: {save_path}" )
                print("视频下载失败")
                return "视频下载失败"
            # 上传并处理视频
            await analyzer.upload_video(save_path)
            # 创建缓存模型
            model =await analyzer.create_cache()
            sample_data = {
                "一、基础信息": {
                    "视觉/音乐/文字": "",
                    "内容选题": "",
                    "视频主题": ""
                },
                "二、主体和场景": {
                    "视频主体": "",
                    "视频场景": []
                },
                "三、情感与风格": {},
                "四、视频传播性与观众": {
                    "片尾引导": {},
                    "传播性判断": "",
                    "观众画像": {}
                },
                "五、音画细节": {
                    "音频细节": {},
                    "视频水印": {},
                    "视频字幕": {},
                    "视频口播": ""
                },
                "六、人物与场景": {
                    "知名人物": {},
                    "人物年龄段": "",
                    "场景描述": []
                },
                "七、时效性与分类": {
                    "时效性": {},
                    "视频一级分类": "",
                    "二级分类": ["品类- 、分数-", "品类- 、分数-", "品类- 、分数-"]
                }
            }
            response =await analyzer.analyze_video( model, prompt, sample_data )
            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( "达到最大重试次数，处理失败" )
                return f"视频分析处理失败:{e}"


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))