import os import time import uuid from typing import Optional import functools import google.generativeai as genai import orjson import requests from google.generativeai.types import (HarmBlockThreshold, HarmCategory) from google.api_core import exceptions as google_exceptions from requests.exceptions import RequestException, Timeout, ConnectionError from loguru import logger from utils.coze_hook import CozeHook from utils.google_ai_prompt import VIDEO_TOPIC_ANALYSIS_PROMPT, VIDEO_SEGMENT_ANALYSIS_PROMPT, VIDEO_ANALYSIS_PROMPT # from utils.feishu_data import Material CACHE_DIR = os.path.join(os.getcwd(), 'video_cache') # CACHE_DIR = '/Users/z/Downloads/' # PROXY_ADDR = 'http://localhost:1081' # os.environ['http_proxy'] = PROXY_ADDR # os.environ['https_proxy'] = PROXY_ADDR def retry_on_error(max_retries: int = 3, backoff_factor: float = 1.0): """ 装饰器：在特定错误时重试 """ def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): last_exception = None for attempt in range(max_retries + 1): try: return func(*args, **kwargs) except Exception as e: last_exception = e # 判断是否应该重试 should_retry = False if isinstance(e, google_exceptions.GoogleAPIError): # 对于429(频率限制)、500(服务器错误)、503(服务不可用)进行重试 should_retry = e.code in [429, 500, 503] elif isinstance(e, (Timeout, ConnectionError)): # 网络超时和连接错误进行重试 should_retry = True elif "overloaded" in str(e).lower() or "timeout" in str(e).lower(): # 服务器过载或超时进行重试 should_retry = True if should_retry and attempt < max_retries: wait_time = backoff_factor * (2 ** attempt) logger.warning(f"[重试] 第{attempt + 1}次尝试失败，{wait_time}秒后重试: {str(e)}") time.sleep(wait_time) continue else: # 不应该重试或已达到最大重试次数 break # 重试失败，抛出最后一次的异常 raise last_exception return wrapper return decorator def handle_genai_error(error: Exception) -> str: """ 统一处理Google GenerativeAI相关的错误 返回用户友好的错误信息 """ error_type = type(error).__name__ error_msg = str(error) # Google API 相关错误 if isinstance(error, google_exceptions.GoogleAPIError): if error.code == 400: return f"请求参数错误: {error_msg}" elif error.code == 401: return f"API密钥无效或已过期: {error_msg}" elif error.code == 403: return f"权限不足或服务不可用: {error_msg}" elif error.code == 404: return f"模型或资源不存在: {error_msg}" elif error.code == 429: return f"请求频率超限，请稍后重试: {error_msg}" elif error.code == 500: return f"服务器内部错误: {error_msg}" elif error.code == 503: return f"服务暂时不可用: {error_msg}" else: return f"Google API错误 ({error.code}): {error_msg}" # 网络相关错误 elif isinstance(error, (RequestException, Timeout, ConnectionError)): return f"网络连接错误: {error_msg}" # 通用错误处理 elif "API_KEY" in error_msg.upper() or "PERMISSION" in error_msg.upper(): return f"API密钥错误或权限不足: {error_msg}" elif "MODEL" in error_msg.upper() and ("NOT_FOUND" in error_msg.upper() or "UNAVAILABLE" in error_msg.upper()): return f"模型不可用或不存在: {error_msg}" elif "QUOTA" in error_msg.upper() or "LIMIT" in error_msg.upper(): return f"配额超限或请求限制: {error_msg}" elif "TIMEOUT" in error_msg.upper(): return f"请求超时: {error_msg}" elif "OVERLOADED" in error_msg.upper(): return f"服务器负载过高，请稍后重试: {error_msg}" else: return f"创建GenerativeModel失败 ({error_type}): {error_msg}" def load_prompts(): """从prompt.py加载Prompt""" try: print("\n[初始化] 从prompt.py加载Prompt") prompts = [ # { # "name": "视频选题与要点理解", # "content": VIDEO_TOPIC_ANALYSIS_PROMPT # }, # { # "name": "视频分段与时间点分析", # "content": VIDEO_SEGMENT_ANALYSIS_PROMPT # } { "name": "视频内容分析", "content": VIDEO_ANALYSIS_PROMPT } ] print(f"[成功] 加载 {len(prompts)} 个Prompt") return prompts except Exception as e: raise Exception(f"加载Prompt失败: {str(e)}") class GoogleAI(object): @classmethod def download_video(cls, video_link: 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: error_info = { "error_type": type(e).__name__, "error_message": str(e), "cache_dir": CACHE_DIR, "current_dir": os.getcwd(), "dir_exists": os.path.exists(CACHE_DIR), "dir_permissions": oct(os.stat(os.path.dirname(CACHE_DIR)).st_mode)[-3:] if os.path.exists(os.path.dirname(CACHE_DIR)) else "N/A" } error_json = orjson.dumps(error_info, option=orjson.OPT_INDENT_2).decode('utf-8') logger.error(f'[内容分析] 创建缓存目录失败: {error_json}') return None for _ in range(3): try: response = requests.get(url=video_link, timeout=60) print(f"response content: {file_path}") if response.status_code == 200: try: with open(file_path, 'wb') as f: f.write(response.content) logger.info(f'[内容分析] 视频链接: {video_link}, 存储地址: {file_path}') except Exception as e: error_info = { "error_type": type(e).__name__, "error_message": str(e), "file_path": file_path, "content_length": len(response.content) if response.content else 0 } error_json = orjson.dumps(error_info, option=orjson.OPT_INDENT_2).decode('utf-8') logger.error(f'[内容分析] 视频保存失败: {error_json}') return None return file_path except Exception: time.sleep(1) continue except Exception: logger.error(f'[内容分析] 创建缓存目录失败') return None @classmethod @retry_on_error(max_retries=2, backoff_factor=1.5) def _analyze_content(cls, video, prompt): # logger.info(f"[视频分析] 开始分析, 视频: {video}, 提示: {prompt}") """增强版内容分析""" # 添加模型创建的错误处理 try: 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, } ) except Exception as e: error_msg = handle_genai_error(e) logger.error(f"[视频分析] {error_msg}") raise Exception(error_msg) try: response = model.generate_content( contents=[video, prompt], request_options={'timeout': 300} ) if hasattr(response, '_error') and response._error: raise Exception(f"生成错误: {response._error}") result = orjson.loads(response.text.strip()) logger.info(f"[视频分析] 响应: {result}") print(f"[视频分析] 响应: {result}") if not isinstance(result, dict): raise ValueError("响应格式错误：非字典结构") return result except orjson.JSONDecodeError as e: error_msg = f"响应解析失败，非JSON格式: {str(e)}" logger.error(f"[视频分析] {error_msg}") raise Exception(error_msg) except Exception as e: # 如果是Google API相关错误，使用统一错误处理 if isinstance(e, (google_exceptions.GoogleAPIError, RequestException, Timeout, ConnectionError)): error_msg = handle_genai_error(e) logger.error(f"[视频分析] {error_msg}") raise Exception(error_msg) else: error_msg = f"分析失败: {str(e)}" logger.error(f"[视频分析] {error_msg}") raise Exception(error_msg) @classmethod def run(cls, api_key, video_url): print(f"api_key:{api_key},video_url:{video_url}") video_path = None try: genai.configure(api_key=api_key) video_path = cls.download_video(video_link=video_url) if not video_path: logger.error(f'[内容分析] 视频下载失败, 跳过任务') os.remove(video_path) logger.info(f"[内容分析] 文件已删除: {video_path}") return "[异常] 视频下载失败","" video = genai.upload_file(path=video_path, mime_type='video/mp4') while video.state.name == 'PROCESSING': time.sleep(1) video = genai.get_file(name=video.name) if video.state.name != 'ACTIVE': genai.delete_file(name=video.name) os.remove(video_path) return "[异常] 上传视频失败", "" logger.info(f"[内容分析] 文件下载完成: {video_path}") prompts = load_prompts() analysis_data = {} for prompt in prompts[:3]: # print(f"[分析] 正在执行: {prompt['name']}") try: result = cls._analyze_content(video, prompt['content']) # 提取 result 中的 "内容分段" 和 "视频简介" analysis_data['视频选题与要点理解'] = { "视频简介": result.get('视频简介', ''), "视频内容类型": result.get('视频内容类型', ''), "段落类型相似度": result.get('段落类型相似度', 1) } analysis_data['视频分段与时间点分析'] = { "内容分段": result.get('内容分段', []) } except Exception as e: analysis_data[prompt['name']] = { "error": str(e), "error_type": type(e).__name__ } # print(f"[分析] 所有分析完成, 结果: {analysis_data}") coze_hook = CozeHook() demand_list = coze_hook.run(analysis_data["视频选题与要点理解"], analysis_data["视频分段与时间点分析"]) # print(f"[分析] 所有分析完成, 结果: {demand_list}") genai.delete_file(name=video.name) os.remove(video_path) return analysis_data, demand_list except Exception as e: logger.error(f"[内容分析] 处理异常,异常信息{e}") os.remove(video_path) return f"[异常] {e}","" @classmethod def _analyze_content_with_api(cls, video_url): """使用API分析视频内容""" try: # 检查视频URL是否有效 if not video_url or not video_url.startswith('http'): raise Exception("无效的视频URL") # 获取视频文件以确定正确的MIME类型 try: response = requests.head(video_url, timeout=10) content_type = response.headers.get('content-type', '') if not content_type or 'video' not in content_type.lower(): # 如果无法从HEAD请求获取正确的content-type，尝试GET请求 response = requests.get(video_url, stream=True, timeout=10) content_type = response.headers.get('content-type', '') if not content_type or 'video' not in content_type.lower(): content_type = 'video/mp4' # 默认使用mp4 except Exception as e: logger.warning(f"[内容分析] 获取视频MIME类型失败: {str(e)}, 使用默认类型video/mp4") content_type = 'video/mp4' # 使用API分析视频内容 response = requests.post( 'http://ai-api.piaoquantv.com/aigc-server/gemini/generateContent', json={ "mediaUrl": video_url, "type": 2, "prompt": VIDEO_ANALYSIS_PROMPT, "model": "gemini-2.0-flash", "temperature": "0.3", "mimeType": content_type # 添加正确的MIME类型 }, timeout=300 ) response.raise_for_status() result = response.json() # print(f"[内容分析] API原始响应: {result}") if not result: raise Exception("API返回结果为空") if result.get('code') != 0: error_msg = result.get('msg', '未知错误') if 'data' in error_msg and 'error' in error_msg: try: error_data = orjson.loads(error_msg) if isinstance(error_data, dict) and 'error' in error_data: error_msg = f"API错误: {error_data['error'].get('message', error_msg)}" except: pass raise Exception(f"API返回错误: {error_msg}") if not result.get('data') or not result['data'].get('result'): raise Exception("API返回数据格式错误: 缺少result字段") try: # 解析返回的JSON字符串 analysis_result = orjson.loads(result['data']['result']) if not isinstance(analysis_result, dict): raise ValueError("API返回的result不是有效的JSON对象") # 构建analysis_data analysis_data = { '视频选题与要点理解': { "视频简介": analysis_result.get('视频简介', ''), "视频内容类型": analysis_result.get('视频内容类型', ''), "段落类型相似度": analysis_result.get('段落类型相似度', 1) }, '视频分段与时间点分析': { "内容分段": analysis_result.get('内容分段', []) } } # 使用coze_hook处理数据 coze_hook = CozeHook() demand_list = coze_hook.run( analysis_data["视频选题与要点理解"], analysis_data["视频分段与时间点分析"] ) if not demand_list: raise Exception("CozeHook处理结果为空") # print(f"[内容分析] API分析完成, 结果: {analysis_data}, {demand_list}") return analysis_data, demand_list except orjson.JSONDecodeError as e: raise Exception(f"解析API返回的JSON失败: {str(e)}") except Exception as e: raise Exception(f"处理API返回数据时出错: {str(e)}") except requests.exceptions.RequestException as e: error_msg = f"API请求失败: {str(e)}" logger.error(f"[内容分析] {error_msg}") return f"[异常] {error_msg}", None except Exception as e: error_msg = f"API分析失败: {str(e)}" logger.error(f"[内容分析] {error_msg}") return f"[异常] {error_msg}", None if __name__ == '__main__': # 使用示例：展示错误处理 try: ai = GoogleAI() result = ai.run("AIzaSyDs7rd3qWV2ElnP4xtY_b0EiLUdt3yviRs", "http://rescdn.yishihui.com/jq_oss/video/2025012215472528213") print(f"分析成功: {result}") except Exception as e: error_msg = str(e) print(f"分析失败: {error_msg}") # 根据错误类型进行不同的处理 if "API密钥" in error_msg: print("请检查API密钥是否正确配置") elif "权限不足" in error_msg: print("请检查API密钥权限或服务可用性") elif "配额超限" in error_msg: print("请检查API配额或稍后重试") elif "网络" in error_msg: print("请检查网络连接") elif "服务器负载" in error_msg: print("服务器繁忙，请稍后重试") else: print("未知错误，请联系技术支持") # ai._analyze_content_with_api("http://rescdn.yishihui.com/longvideo/crawler_local/video/prod/20241206/5f98b0e4464d02d6c75907302793902d12277")