knowledge/content_indentify/video_identifier.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
视频识别脚本
主要功能：使用 Gemini API 从三个维度分析视频内容
1. ASR (Automatic Speech Recognition) - 语音转文字
2. OCR - 识别视频画面中的文字
3. 关键帧提取与描述 - 提取视频关键帧并进行图像描述
"""

import os
import json
import time
import sys
import uuid
import requests
from typing import Dict, Any, List, Optional
from dotenv import load_dotenv

# 导入自定义模块
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# 导入Google Generative AI
import google.generativeai as genai
from google.generativeai.types import HarmCategory, HarmBlockThreshold

# 缓存目录配置
CACHE_DIR = os.path.join(os.path.dirname(__file__), 'cache')


class VideoIdentifier:
    def __init__(self):
        # 加载环境变量
        load_dotenv()
        
        # 获取API密钥
        self.api_key = os.getenv('GEMINI_API_KEY')
        if not self.api_key:
            raise ValueError("未找到GEMINI_API_KEY环境变量")
        
        # 配置Gemini
        genai.configure(api_key=self.api_key)
        
        # 统一的系统提示词 - 三个维度分析
        self.unified_system_prompt = """你是一个专业的视频内容分析专家。请从以下两个维度分析视频内容，并以JSON格式输出结果：

1. ASR (Automatic Speech Recognition) - 语音转文字：
   - 仅提取视频中的语音内容，转换为文字
   - 保持原始语音的准确性和完整性
   - 不要添加分析、解释或评论

2. 关键帧提取与描述（包含OCR文字识别）：
   - 将视频分解为多个关键时间片段
   - 对每个时间片段进行以下分析：
     * 画面的主要视觉元素和内容
     * 画面的构图和色彩特点
     * 画面中的人物、物体、场景
     * 画面中出现的所有文字内容（OCR识别）
   - 每个时间片段应包含：
     * content: 画面内容的详细描述
     * ocr_content: 该时间段画面中出现的文字内容,仅做文字提取，不要做任何解释或总结

请严格按照以下JSON格式输出，使用中文输出，不要添加任何其他文字：
{
    "asr_content": "提取的语音文字内容",
    "iframe_details": [
        {
            "time_start": "开始时间（秒）",
            "time_end": "结束时间（秒）",
            "content": "该时间段画面内容的详细描述",
            "ocr_content": "该时间段画面中出现的文字内容"
        }
    ]
}"""
    
    def download_video(self, video_url: str) -> Optional[str]:
        """下载视频到本地缓存"""
        file_path = os.path.join(CACHE_DIR, f'{str(uuid.uuid4())}.mp4')
        try:
            # 确保缓存目录存在
            try:
                os.makedirs(CACHE_DIR, exist_ok=True)
            except Exception as e:
                print(f'创建缓存目录失败: {e}')
                return None
            
            # 尝试下载视频
            for attempt in range(3):
                try:
                    response = requests.get(url=video_url, timeout=60)
                    if response.status_code == 200:
                        try:
                            with open(file_path, 'wb') as f:
                                f.write(response.content)
                            print(f'视频下载成功: {video_url} -> {file_path}')
                            return file_path
                        except Exception as e:
                            print(f'视频保存失败: {e}')
                            return None
                    else:
                        print(f'视频下载失败，状态码: {response.status_code}')
                except Exception as e:
                    print(f'下载尝试 {attempt + 1} 失败: {e}')
                    if attempt < 2:  # 不是最后一次尝试
                        time.sleep(1)
                        continue
                    else:
                        print(f'所有下载尝试都失败了')
                        return None
                        
        except Exception as e:
            print(f'下载过程异常: {e}')
            return None
        
        return None
    
    def upload_video_to_gemini(self, video_path: str) -> Optional[Any]:
        """上传视频到Gemini进行分析"""
        max_retries = 3
        retry_delay = 5
        
        for attempt in range(max_retries):
            try:
                print(f"  开始上传视频到Gemini... (尝试 {attempt + 1}/{max_retries})")
                print(f"    文件路径: {video_path}")
                
                # 1. 文件检查
                if not os.path.exists(video_path):
                    print(f"    错误: 文件不存在")
                    return None
                
                file_size = os.path.getsize(video_path)
                print(f"    文件大小: {file_size / (1024*1024):.2f} MB")
                
                if file_size == 0:
                    print(f"    错误: 文件大小为0")
                    return None
                
                # 2. 文件权限检查
                try:
                    with open(video_path, 'rb') as f:
                        # 尝试读取文件开头，检查是否可读
                        f.read(1024)
                    print(f"    文件权限: 可读")
                except Exception as e:
                    print(f"    错误: 文件无法读取 - {e}")
                    return None
                
                # 3. 网络连接检查
                try:
                    print(f"    检查网络连接...")
                    # 测试基本网络连接
                    test_response = requests.get("https://generativelanguage.googleapis.com", timeout=10)
                    print(f"    网络连接: 正常 (状态码: {test_response.status_code})")
                except Exception as e:
                    print(f"    警告: 网络连接测试失败 - {e}")
                    print(f"    继续尝试上传...")
                
                # 4. 尝试上传文件
                print(f"    开始上传文件...")
                try:
                    video_file = genai.upload_file(path=video_path, mime_type='video/mp4')
                    print(f"    文件上传请求已发送，文件ID: {video_file.name}")
                except Exception as e:
                    print(f"    错误: 文件上传请求失败 - {e}")
                    print(f"    错误类型: {type(e).__name__}")
                    print(f"    错误详情: {str(e)}")
                    
                    # 如果是网络相关错误，尝试重试
                    if any(keyword in str(e).lower() for keyword in ['broken pipe', 'connection', 'timeout', 'network']):
                        if attempt < max_retries - 1:
                            print(f"    网络错误，等待 {retry_delay} 秒后重试...")
                            time.sleep(retry_delay)
                            retry_delay *= 2  # 指数退避
                            continue
                        else:
                            print(f"    所有重试都失败了")
                            return None
                    else:
                        # 非网络错误，直接返回
                        return None
                
                # 5. 等待文件处理完成
                print(f"    等待文件处理完成...")
                max_wait_time = 120  # 最大等待2分钟
                wait_count = 0
                
                while video_file.state.name == 'PROCESSING' and wait_count < max_wait_time:
                    time.sleep(2)  # 每2秒检查一次
                    wait_count += 2
                    
                    try:
                        # 获取最新状态
                        video_file = genai.get_file(name=video_file.name)
                        current_state = video_file.state.name
                        print(f"      状态: {current_state} ({wait_count}秒)")
                        
                        # 检查是否有错误状态
                        if current_state in ['FAILED', 'ERROR', 'INVALID']:
                            print(f"    错误: 文件处理失败，状态: {current_state}")
                            if hasattr(video_file, 'error'):
                                print(f"    错误详情: {video_file.error}")
                            
                            # 如果是处理失败，尝试重试
                            if attempt < max_retries - 1:
                                print(f"    文件处理失败，等待 {retry_delay} 秒后重试...")
                                time.sleep(retry_delay)
                                retry_delay *= 2
                                break  # 跳出等待循环，进行重试
                            else:
                                return None
                                
                    except Exception as e:
                        print(f"      警告: 获取文件状态失败 - {e}")
                        if wait_count > 60:  # 超过1分钟后，尝试继续
                            print(f"      继续等待...")
                            continue
                        else:
                            print(f"    错误: 无法获取文件状态")
                            return None
                
                # 6. 检查最终状态
                if video_file.state.name == 'ACTIVE':
                    print(f'    视频上传成功: {video_file.name}')
                    print(f"    最终状态: {video_file.state.name}")
                    return video_file
                else:
                    print(f'    错误: 视频文件上传失败')
                    print(f"    最终状态: {video_file.state.name}")
                    print(f"    等待时间: {wait_count}秒")
                    
                    # 尝试获取更多错误信息
                    try:
                        file_info = genai.get_file(name=video_file.name)
                        print(f"    文件信息: {file_info}")
                    except Exception as e:
                        print(f"    无法获取文件详细信息: {e}")
                    
                    # 如果不是最后一次尝试，进行重试
                    if attempt < max_retries - 1:
                        print(f"    上传失败，等待 {retry_delay} 秒后重试...")
                        time.sleep(retry_delay)
                        retry_delay *= 2
                        continue
                    else:
                        return None
                        
            except Exception as e:
                error_type = type(e).__name__
                error_msg = str(e)
                
                print(f'    错误: 视频上传到Gemini失败')
                print(f"    错误类型: {error_type}")
                print(f"    错误信息: {error_msg}")
                
                # 针对特定错误的处理建议
                if "Broken pipe" in error_msg:
                    print(f"    诊断: Broken pipe 错误通常表示:")
                    print(f"      - 网络连接不稳定")
                    print(f"      - 服务器连接中断")
                    print(f"      - 防火墙或代理问题")
                    print(f"    建议:")
                    print(f"      - 检查网络连接")
                    print(f"      - 尝试使用VPN或更换网络")
                    print(f"      - 检查防火墙设置")
                elif "Connection" in error_msg:
                    print(f"    诊断: 连接错误")
                    print(f"    建议: 检查网络连接和API密钥")
                elif "Timeout" in error_msg:
                    print(f"    诊断: 超时错误")
                    print(f"    建议: 网络较慢，可以增加超时时间")
                elif "Permission" in error_msg:
                    print(f"    诊断: 权限错误")
                    print(f"    建议: 检查API密钥和权限设置")
                
                # 如果是网络相关错误，尝试重试
                if any(keyword in error_msg.lower() for keyword in ['broken pipe', 'connection', 'timeout', 'network']):
                    if attempt < max_retries - 1:
                        print(f"    网络错误，等待 {retry_delay} 秒后重试...")
                        time.sleep(retry_delay)
                        retry_delay *= 2
                        continue
                    else:
                        print(f"    所有重试都失败了")
                        return None
                else:
                    # 非网络错误，直接返回
                    return None
        
        return None
    
    def analyze_video_with_gemini(self, video_file: Any, video_info: Dict[str, Any]) -> Dict[str, Any]:
        """使用Gemini API分析视频内容"""
        
        try:
            # 创建Gemini模型
            model = genai.GenerativeModel(
                model_name='gemini-2.0-flash',
                generation_config=genai.GenerationConfig(
                    response_mime_type='application/json',
                    temperature=0.3,
                    max_output_tokens=20480
                ),
                safety_settings={
                    HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_NONE,
                    HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_NONE,
                }
            )
            
            # 生成内容
            response = model.generate_content(
                contents=[video_file, self.unified_system_prompt],
                request_options={'timeout': 300}
            )
            
            # 检查错误
            if hasattr(response, '_error') and response._error:
                raise Exception(f"生成错误: {response._error}")
            
            # 解析JSON响应
            try:
                result = json.loads(response.text.strip())
                print(f"[视频分析] 响应: {result}")
                
                if not isinstance(result, dict):
                    raise ValueError("响应格式错误：非字典结构")
                
                # 确保包含所有必需字段
                required_fields = ['asr_content', 'iframe_details']
                for field in required_fields:
                    if field not in result:
                        if field == 'iframe_details':
                            result[field] = [{
                                'time_start': 0,
                                'time_end': 0,
                                'content': f'{field}分析失败',
                                'ocr_content': f'{field}分析失败'
                            }]
                        else:
                            result[field] = f"{field}分析失败"
                
                return result
                
            except json.JSONDecodeError as e:
                print(f"JSON解析失败: {e}")
                return {
                    'asr_content': 'ASR分析失败：JSON解析错误',
                    'iframe_details': [{
                        'time_start': 0,
                        'time_end': 0,
                        'content': '关键帧分析失败：JSON解析错误',
                        'ocr_content': '关键帧分析失败：JSON解析错误'
                    }]
                }
                
            else:
                return {
                    'asr_content': 'ASR分析失败：API无响应',
                    'iframe_details': [{
                        'time_start': 0,
                        'time_end': 0,
                        'content': '关键帧分析失败：API无响应',
                        'ocr_content': '关键帧分析失败：API无响应'
                    }]
                }
                
        except Exception as e:
            return {
                'asr_content': f'ASR分析失败: {str(e)}',
                'iframe_details': [{
                    'time_start': 0,
                    'time_end': 0,
                    'content': f'关键帧分析失败: {str(e)}',
                    'ocr_content': f'关键帧分析失败: {str(e)}'
                }]
            }
    
    def extract_video_urls(self, formatted_content: Dict[str, Any]) -> List[Dict[str, Any]]:
        """提取视频URL列表"""
        video_data = []
        video_url_list = formatted_content.get('video_url_list', [])
        
        for video_item in video_url_list:
            if isinstance(video_item, dict) and 'video_url' in video_item:
                video_data.append({
                    'url': video_item['video_url'],
                    'duration': video_item.get('video_duration', 0)
                })
        
        return video_data
    
    def process_video_single(self, video_info: Dict[str, Any]) -> Dict[str, Any]:
        """处理单个视频的完整流程"""
        print(f"开始处理视频: {video_info['url'][:50]}...")
        
        video_path = None
        video_file = None
        try:
            # 1. 下载视频
            print("  1. 下载视频...")
            video_path = self.download_video(video_info['url'])
            if not video_path:
                print("  视频下载失败")
                return {
                    'url': video_info['url'],
                    'duration': video_info['duration'],
                    'asr_content': '视频下载失败',
                    'iframe_details': [{
                        'time_start': 0,
                        'time_end': 0,
                        'content': '视频下载失败',
                        'ocr_content': '视频下载失败'
                    }]
                }
            
            # 2. 上传到Gemini
            print("  2. 上传视频到Gemini...")
            video_file = self.upload_video_to_gemini(video_path)
            if not video_file:
                print("  视频上传到Gemini失败")
                return {
                    'url': video_info['url'],
                    'duration': video_info['duration'],
                    'asr_content': '视频上传失败',
                    'iframe_details': [{
                        'time_start': 0,
                        'time_end': 0,
                        'content': '视频上传失败',
                        'ocr_content': '视频上传失败'
                    }]
                }
            
            # 3. 使用Gemini分析
            print("  3. 使用Gemini分析视频内容...")
            analysis_result = self.analyze_video_with_gemini(video_file, video_info)
            
            # 4. 组合结果
            final_result = {
                'url': video_info['url'],
                'duration': video_info['duration'],
                'asr_content': analysis_result.get('asr_content', 'ASR分析失败'),
                'iframe_details': analysis_result.get('iframe_details', '关键帧分析失败'),
            }
            
            print("  视频分析完成")
            return final_result
            
        except Exception as e:
            print(f"  视频处理异常: {e}")
            return {
                'url': video_info['url'],
                'duration': video_info['duration'],
                'asr_content': f'处理异常: {str(e)}',
                'iframe_details': [{
                    'time_start': 0,
                    'time_end': 0,
                    'content': f'处理异常: {str(e)}',
                    'ocr_content': f'处理异常: {str(e)}'
                }],
                'analysis_timestamp': int(time.time() * 1000)
            }
        finally:
            # 清理临时文件
            if video_path and os.path.exists(video_path):
                try:
                    os.remove(video_path)
                    print(f"  临时文件已清理: {video_path}")
                except Exception as e:
                    print(f"  清理临时文件失败: {e}")
            
            # 清理Gemini文件
            if video_file and hasattr(video_file, 'name'):
                try:
                    genai.delete_file(name=video_file.name)
                    print(f"  Gemini文件已清理: {video_file.name}")
                except Exception as e:
                    print(f"  清理Gemini文件失败: {e}")
    
    def process_videos(self, formatted_content: Dict[str, Any]) -> List[Dict[str, Any]]:
        """处理视频识别的主函数"""
        print("开始视频识别处理...")
        
        # 提取视频URL
        video_data = self.extract_video_urls(formatted_content)
        print(f"提取到 {len(video_data)} 个视频")
        
        if not video_data:
            print("没有视频需要分析")
            return []
        
        # 逐个处理视频
        results = []
        for i, video_info in enumerate(video_data):
            print(f"\n处理视频 {i+1}/{len(video_data)}")
            result = self.process_video_single(video_info)
            results.append(result)
            
            # 添加延迟避免API限制
            if i < len(video_data) - 1:  # 不是最后一个视频
                time.sleep(2)
        
        if results:
            print(f"\n视频识别完成，共分析 {len(results)} 个视频")
            print("分析维度：ASR、关键帧提取")
        else:
            print("视频识别失败")
        
        return results


def main():
    """测试函数"""
    # 模拟数据
    test_content = {
        "video_url_list": [
            {
                "video_url": "https://vd9.bdstatic.com/mda-rf03dz9qrusbwrrb/mb/720p/mv_cae264_backtrack_720p_normal/1748751326307005666/mda-rf03dz9qrusbwrrb.mp4?v_from_s=hkapp-haokan-hbe&auth_key=1755078490-0-0-94814ae256d196c133940bc5fa7054ea&bcevod_channel=searchbox_feed&cr=2&cd=0&pd=1&pt=3&logid=2890204804&vid=12887026108358975692&klogid=2890204804&abtest=",
                "video_duration": 187
            }
        ]
    }
    
    identifier = VideoIdentifier()
    result = identifier.process_videos(test_content)
    
    print(f"识别结果: {json.dumps(result, ensure_ascii=False, indent=2)}")


if __name__ == '__main__':
    main()