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