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 )
            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):
    """主函数，执行视频上传、缓存创建、问题生成"""
    # api_key = 'AIzaSyAa4H7YzIOiqA0QIDuO6kNErFBoWT-BEtc'

    mark, prompt = Material.feishu_list()
    attempt = 0
    max_attempts = 1
    while attempt < max_attempts:
        try:
            # 初始化视频分析类
            analyzer = VideoAnalyzer(api_key )
            for file in genai.list_files():
                file.delete()
            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 = {
                "一、基础信息": {
                    "视觉/音乐/文字": "",
                    "内容选题": "",
                    "视频主题": ""
                },
                "二、主体和场景": {
                    "视频主体": "",
                    "视频场景": []
                },
                "三、情感与风格": {},
                "四、视频传播性与观众": {
                    "片尾引导": {},
                    "传播性判断": "",
                    "观众画像": {}
                },
                "五、音画细节": {
                    "音频细节": {},
                    "视频水印": {},
                    "视频字幕": {},
                    "视频口播": ""
                },
                "六、人物与场景": {
                    "知名人物": {},
                    "人物年龄段": "",
                    "场景描述": []
                },
                "七、时效性与分类": {
                    "时效性": {},
                    "视频一级分类": "",
                    "二级分类": ["品类- 、分数-", "品类- 、分数-", "品类- 、分数-"]
                }
            }
            # 视频分析问题
            # video_analysis_questions = "你是一个专业的视频分析师，我将给你一个视频，请你对视频进行理解，回答下面的问题。\n一、基础信息：\n1视觉/音乐/文字：请从视频中的视觉、音乐、文字这三个维度信息做分析比较，哪个维度的信息是该视频中最重要的，可能成为该视频的要点驱动力？你只要回答视觉/音乐/文字三者其一即可。\n2内容选题：如果需要从视频中提取一个内容选题，你觉得应该是什么？请注意：选题应该体现视频的关键点，亮点，爆点，选题不能超过8个字。\n3视频主题：描述视频的整体主题。\n4视频关键词：使用1到3个关键词对视频进行概括\n二、主体和场景：\n1视频主体：视频中的核心人物或物体，有几个？分别是什么？\n2视频场景：视频属于什么场景？场景可以有多个，每个不超过6个字。\n三、情感与风格：\n1情感倾向：视频传递的情感是积极、消极还是中立或其他？\n2视频风格：判断视频的风格类型（如严肃、轻松、幽默等）。\n四、视频传播性与观众：\n1片尾引导\n•视频片尾是否有引导观众分享？\n•引导时长？\n•引导强度如何？\n2传播性判断：基于中国中老年微信用户，判断该视频的传播性并说明依据。\n3观众画像：\n•推测观众的年龄段，如0～10岁、40～50岁，并描述推测的原因\n•推测观众的性别，男或女，并描述推测的原因\n•推测观众的地域，精确到明确的省份或直辖市（北京、山东等），如普遍使用全部地域则输出全国，或者南方北方；并描述推测的原因\n五、音画细节：\n1音频细节\n•视频中的音频信息，是否有歌曲？\n•视频中的音频信息，歌曲名是什么？\n•视频中的音色音色是怎样的？\n2视频水印\n•是否有产品名的水印？\n•水印是否涉及产品名称是什么？\n3视频字幕\n•是否有字幕？\n•字幕的颜色？\n•字幕的字号？\n•字幕的位置如何？\n4视频口播：提供视频中出现的准确且完整的口播内容。\n六、人物与场景：\n1知名人物\n•视频或音频中是否出现知名人物？\n•视频或音频中是否出现知名人物是谁？\n2人物年龄段：视频中人物的年龄段（如中青年男、中青年女等）。\n3场景描述：视频和声音中涉及的场景描述。\n七、时效性与分类：\n1时效性：\n•适用时效日。\n•适用时效早中晚。\n2视频一级分类：判断视频分别属于下面的哪种一级分类，并输出准确完整的一级品类_分类名称：\n1. 一级品类_音乐\n2. 一级品类_剧情 / 剧情演绎\n3. 一级品类_二次元\n4. 一级品类_游戏\n5. 一级品类_公益\n6. 一级品类_随拍 / 颜值\n7. 一级品类_舞蹈\n8. 一级品类_动物 / 萌宠\n9. 一级品类_三农\n10. 一级品类_科技 / 科技数码\n11. 一级品类_财经\n12. 一级品类_母婴 / 母婴亲子\n13. 一级品类_法律 / 人文社科\n14. 一级品类_科普 / 人文社科\n15. 一级品类_情感 / 情感心理\n16. 一级品类_职场 / 人文社科\n17. 一级品类_教育 / 教育培训\n18. 一级品类_摄影摄像\n19. 一级品类_艺术 / 才艺技能\n20. 一级品类_美食\n21. 一级品类_旅行 / 旅游\n22. 一级品类_地域本地\n23. 一级品类_时尚 / 时尚 / 美妆\n24. 一级品类_文化 / 人文社科\n25. 一级品类_搞笑 / 休闲娱乐\n26. 一级品类_明星 / 名人\n27. 一级品类_综艺\n28. 一级品类_影视综艺\n29. 一级品类_电影\n30. 一级品类_电视剧\n31. 一级品类_汽车\n32. 一级品类_体育 / 运动\n33. 一级品类_医疗健康 / 长寿 / 健身\n34. 一级品类_生活记录 / 生活\n35. 一级品类_生活家居 / 家居家装\n36. 一级品类_时政社会\n37. 一级品类_奇人异象\n38. 一级品类_历史\n39. 一级品类_军事\n40. 一级品类_宗教\n41. 一级品类_短剧\n42. 一级品类_收藏品\n3视频二级分类：判断视频分别属于下面的哪种二级分类（：后是品类的详细说明 不用输出），你可以给到你认为最匹配到1个到3个品类，根据你认为的匹配程度给出分数（1～10分，分数越高越匹配）并按照相关度从高到低排序，请输出准确完整的品类-分类名称（如 品类-祝福音乐）：\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\"品类-对口型表演：对口型表演是一种娱乐形式，表演者模仿他人讲话、唱歌或其他声音内容，观众通过视觉和听觉感知表演者的同步性和表情变化，如1-2个真人在舞台或漫画人物、数字人等形式以幽默夸张方式演绎的对口型口播或演唱表演。这类视频通常具有幽默感或创意性，吸引用户观看和分享。\n示例：模仿经典电影台词、对歌手的现场演唱进行对口型模仿等。\"\n\"品类-动植物实拍：该类别视频通过真实记录动植物的生活方式、行为特点或生态环境，展示自然界的多样性。视频内容可以展示动物的捕猎、交配、迁徙，或植物的生长、开花、结果等过程，通常具有教育意义，也可以提供视觉上的美感。\n示例：记录猎豹捕猎的全过程、热带雨林中植物的生态环境、蜂鸟采花等。\"\n\"品类-动物萌宠：动物萌宠类视频专注展示各种宠物或野生动物的可爱行为和外貌，吸引观众的关注和喜爱。这类视频以轻松、愉快为主，通常包含宠物玩耍、搞怪或与主人互动的场景，目的是提供娱乐和放松。\n示例：猫咪打哈欠、狗狗学会新技能、小动物第一次接触新事物等。\"\n\"品类-动物表演：动物表演类视频展示经过训练的动物进行各种表演，通常包括它们完成特定动作、技巧或戏剧性演出。这类视频中的动物通常是宠物或表演专用动物，展现它们的聪明才智和与人类互动的能力。\n示例：马术表演、狗狗服从训练表演、海豚在水族馆的跳跃表演等。\"\n品类-大型集体艺术\n品类-当代正能量人物：以具体中国当代正能量人物为主要描述对象的内容\n\"品类-传统文化：传统文化类视频介绍和展示中华或其他民族的传统文化元素，包括节庆、礼仪、手工艺、服饰、饮食等。内容通常注重文化的传承和保护，帮助观众了解和尊重传统文化的价值。\n示例：中国春节习俗、传统手工艺品制作、茶道文化介绍等。\"\n品类-吃播探店\n\"品类-长寿知识：长寿知识视频介绍与健康长寿相关的科学知识、生活方式、饮食习惯等内容，帮助观众了解如何通过合理的生活习惯延长寿命。这类视频通常基于健康医学、营养学等领域的研究，内容涉及运动、饮食、心态管理等。\n示例：如何通过饮食保持健康、长寿的运动方式、抗衰老的生活习惯等。\"\n品类-本地生活\n品类-K12教育\n品类-(老)电影切片：中国老电影、电视剧内容片段剪辑影像、视频；"

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