google_ai_studio/google_ai/generativeai_video.py

import asyncio
import os

import aiohttp
import requests
import google.generativeai as genai

import uuid



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,
                    "你是一个专业的视频分析师，负责根据访问的视频文件回答用户的所有问题\\n"+questions +
                    "输出返回格式样例:\n"+ str(sample_data)

                ]
            }
            response = chat_session.send_message( message_content )
            return response

async def main(video_path):
    """主函数，执行视频上传、缓存创建、问题生成"""

    # video_path = '/root/3333.mp4'
    api_key = 'AIzaSyDs4FWRuwrEnQzu1M_Skio6NII6Mp4whAw'

    # 初始化视频分析类
    analyzer = VideoAnalyzer(api_key )
    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("视频下载失败")
    # 上传并处理视频
    await analyzer.upload_video(save_path)

    # 创建缓存模型
    model =await analyzer.create_cache()
    sample_data = {
        "基础信息": {
            "视觉/音乐/文字": "",
            "内容选题": "",
            "视频主题": ""
        },
        "主体和场景": {
            "视频主体": "",
            "视频场景": []
        },
        "情感与风格": {},
        "视频传播性与观众": {
            "片尾引导": {},
            "传播性判断": "",
            "观众画像": {}
        },
        "音画细节": {
            "音频细节": {},
            "视频水印": {},
            "视频字幕": {},
            "视频口播": ""
        },
        "人物与场景": {
            "知名人物": {},
            "人物年龄段": "",
            "场景描述": []
        },
        "时效性与分类": {
            "时效性": {},
            "视频一级分类": "",
            "二级分类": ""
        }
    }

    # 视频分析问题
    video_analysis_questions = "一、基础信息：\n" \
                               "1.视觉/音乐/文字： 请从视频中的视觉、音乐、文字这三个维度信息做分析比较，哪个维度的信息是该视频中最重要的，可能成为该视频的要点驱动力？你只要回答 视觉/音乐/文字 三者其一即可。\n" \
                               "2.内容选题： 如果需要从视频中提取一个内容选题，你觉得应该是什么？请注意：选题应该体现视频的关键点，亮点，爆点，选题不能超过8个字。\n" \
                               "3.视频主题：描述视频的整体主题。\n " \
                               "二、主体和场景：\n" \
                               "1.视频主体：视频中的核心人物或物体，有几个？分别是什么？\n" \
                               "2.视频场景：视频属于什么场景？场景可以有多个，每个不超过6个字\n" \
                               "三、情感与风格：\n" \
                               "1.情感倾向：视频传递的情感是积极、消极还是中立或其他？\n" \
                               "2.视频风格：判断视频的风格类型（如严肃、轻松、幽默等）。\n四、视频传播性与观众：\n" \
                               "1.片尾引导\n" \
                               "视频片尾是否有引导观众分享？\n" \
                               "引导时长？\n" \
                               "引导强度如何？\n" \
                               "2.传播性判断：基于中国中老年微信用户，判断该视频的传播性并说明依据。\n" \
                               "3.观众画像：\n" \
                               "推测观众的年龄\n" \
                               "推测观众的性别\n" \
                               "推测观众的地域\n" \
                               "五、音画细节：\n" \
                               "1.音频细节\n" \
                               "视频中的音频信息，是否有歌曲？\n" \
                               "视频中的音频信息，歌曲名是什么？\n" \
                               "视频中的音色音色是怎样的？\n" \
                               "2.视频水印\n" \
                               "是否有产品名的水印？\n" \
                               "水印是否涉及产品名称是什么？\n" \
                               "3.视频字幕\n" \
                               "是否有字幕？\n" \
                               "字幕的颜色？\n" \
                               "字幕的字号？\n" \
                               "字幕的位置如何？\n" \
                               "4. 视频口播：提供视频中出现的准确且完整的口播内容。\n" \
                               "六、人物与场景：\n" \
                               "1.知名人物\n" \
                               "视频或音频中是否出现知名人物？\n" \
                               "视频或音频中是否出现知名人物是谁？\n" \
                               "2.人物年龄段：视频中人物的年龄段（如中青年男、中青年女等）。\n" \
                               "3.场景描述：视频和声音中涉及的场景描述。\n" \
                               "七、时效性与分类：\n" \
                               "1.时效性：\n" \
                               "适用时效日\n" \
                               "适用时效早中晚\n" \
                               "2.视频一级分类：判断视频分别属于下面的哪种一级分类，并输出准确完整的一级品类_分类名称:\n" \
                               "一级分类范围为：\n" \
                               "一级品类_音乐\n" \
                               "一级品类_剧情 / 剧情演绎\n" \
                               "一级品类_二次元\n" \
                               "一级品类_游戏\n" \
                               "一级品类_公益\n" \
                               "一级品类_随拍 / 颜值\n" \
                               "一级品类_舞蹈\n" \
                               "一级品类_动物 / 萌宠\n" \
                               "一级品类_三农\n" \
                               "一级品类_科技 / 科技数码\n" \
                               "一级品类_财经\n" \
                               "一级品类_母婴 / 母婴亲子\n" \
                               "一级品类_法律 / 人文社科\n" \
                               "一级品类_科普 / 人文社科\n" \
                               "一级品类_情感 / 情感心理\n" \
                               "一级品类_职场 / 人文社科\n" \
                               "一级品类_教育 / 教育培训\n" \
                               "一级品类_摄影摄像\n" \
                               "一级品类_艺术 / 才艺技能\n" \
                               "一级品类_美食\n" \
                               "一级品类_旅行 / 旅游\n" \
                               "一级品类_地域本地\n" \
                               "一级品类_时尚 / 时尚 / 美妆\n" \
                               "一级品类_文化 / 人文社科\n" \
                               "一级品类_搞笑 / 休闲娱乐\n" \
                               "一级品类_明星 / 名人\n" \
                               "一级品类_综艺\n" \
                               "一级品类_影视综艺\n" \
                               "一级品类_电影\n" \
                               "一级品类_影视综艺\n" \
                               "一级品类_电视剧\n" \
                               "一级品类_影视综艺\n" \
                               "一级品类_汽车\n" \
                               "一级品类_体育 / 运动\n" \
                               "一级品类_医疗健康 / 长寿 / 健身\n" \
                               "一级品类_生活记录 / 生活\n" \
                               "一级品类_生活家居 / 家居家装\n" \
                               "一级品类_时政社会\n" \
                               "一级品类_奇人异象\n" \
                               "一级品类_历史\n" \
                               "一级品类_军事\n" \
                               "一级品类_宗教\n" \
                               "一级品类_短剧\n" \
                               "一级品类_收藏品\n" \
                               "3.二级分类： 判断视频分别属于下面的哪种二级分类，并输出准确完整的品类-分类名称\n" \
                               "二级分类范围为：\n" \
                               "品类-祝福音乐\n" \
                               "品类-人生感悟音乐\n" \
                               "品类-民族异域音乐\n" \
                               "品类-亲情音乐\n" \
                               "品类-红歌老歌\n" \
                               "品类-音乐知识\n" \
                               "品类-正能量剧情\n" \
                               "品类-对口型表演\n" \
                               "品类-快闪\n" \
                               "品类-拟真游戏\n" \
                               "品类-麻将\n" \
                               "品类-棋牌\n" \
                               "品类-老年审美美女\n" \
                               "品类-老年审美帅哥\n" \
                               "品类-红歌老歌舞蹈\n" \
                               "品类-广场舞\n" \
                               "品类-舞蹈教程\n" \
                               "品类-宠物日常\n" \
                               "品类-动物表演\n" \
                               "品类-生动物\n" \
                               "品类-农村生活\n" \
                               "品类-农业技术\n" \
                               "品类-老年相关科技\n" \
                               "品类-未来科幻\n" \
                               "品类-国家科技力量\n" \
                               "品类-保险\n" \
                               "品类-理财\n" \
                               "品类-亲子日常\n" \
                               "品类-K12教育\n" \
                               "品类-老年相关法律科普\n" \
                               "品类-知识科普\n" \
                               "品类-生活技巧科普\n" \
                               "品类-怀念时光\n" \
                               "品类-人生忠告\n" \
                               "品类-迷信祝福\n" \
                               "品类-节日祝福\n" \
                               "品类-早中晚好\n" \
                               "品类-退休前\n" \
                               "品类-退休后\n" \
                               "品类-益智解密\n" \
                               "品类-老年教育\n" \
                               "品类-风景实拍\n" \
                               "品类-动植物实拍\n" \
                               "品类-人像模特实拍\n" \
                               "品类-摄影教学\n" \
                               "品类-名画赏析\n" \
                               "品类-杂技柔术\n" \
                               "品类-魔术\n" \
                               "品类-魔术特效\n" \
                               "品类-书法\n" \
                               "品类-绘画\n" \
                               "品类-木工\n" \
                               "品类-口技\n" \
                               "品类-大型集体艺术\n" \
                               "品类-戏曲戏剧\n" \
                               "品类-二人转\n" \
                               "品类-其他才艺\n" \
                               "品类-美食测评\n" \
                               "品类-美食教程\n" \
                               "品类-吃播探店\n" \
                               "品类-旅行记录\n" \
                               "品类-旅行攻略\n" \
                               "品类-省份城市亮点\n" \
                               "品类-本地新闻\n" \
                               "品类-本地生活\n" \
                               "品类-老年时尚\n" \
                               "品类-美妆护肤穿搭\n" \
                               "品类-传统文化\n" \
                               "品类-国际文化\n" \
                               "品类-搞笑瞬间合集\n" \
                               "品类-搞笑段子\n" \
                               "品类-历史名人\n" \
                               "品类-当代正能量人物\n" \
                               "品类-老明星\n" \
                               "品类-老年人上综艺\n" \
                               "品类-老年关心纪录片\n" \
                               "品类-老综艺影像\n" \
                               "品类-电影切片\n" \
                               "品类-电影解说\n" \
                               "品类-电视剧切片\n" \
                               "品类-电视剧解说\n" \
                               "品类-中国队比赛\n" \
                               "品类-老年运动\n" \
                               "品类-健康知识\n" \
                               "品类-长寿知识\n" \
                               "品类-饮食健康\n" \
                               "品类-健身操\n" \
                               "品类-老年生活\n" \
                               "品类-生活小妙招\n" \
                               "品类-园艺花艺\n" \
                               "品类-民生政策\n" \
                               "品类-流行病疫情\n" \
                               "品类-社会风气\n" \
                               "品类-食品安全\n" \
                               "品类-贪污腐败\n" \
                               "品类-人财诈骗\n" \
                               "品类-核污染\n" \
                               "品类-惠民新闻\n" \
                               "品类-天气变化\n" \
                               "品类-国家力量\n" \
                               "品类-国际时政\n" \
                               "品类-他国政策\n" \
                               "品类-惊奇事件\n" \
                               "品类-罕见画面\n" \
                               "品类-中国战争史\n" \
                               "品类-中国党史\n" \
                               "品类-中国历史影像\n" \
                               "品类-国际军事\n" \
                               "品类-国内军事\n" \
                               "品类-国家统一\n" \

    # 分析视频并打印结果
    response =await analyzer.analyze_video( model, video_analysis_questions, 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


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/jq_oss/video/2024092618393680952.mp4'
    asyncio.run(main(video_path))