maxkb/apps/common/handle/impl/media/media_split_handle.py

# -*- coding: utf-8 -*-
"""
音视频处理器 - MaxKB集成层
"""
from typing import List
from common.handle.base_split_handle import BaseSplitHandle
from common.utils.logger import maxkb_logger

class MediaSplitHandle(BaseSplitHandle):
    """
    音视频处理器 - MaxKB集成层
    """
    
    def __init__(self):
        super().__init__()
        self.adapter = None
        
    def support(self, file, get_buffer, **kwargs):
        """检查是否支持该文件类型"""
        file_name = file.name.lower()
        
        # 支持的音频格式
        audio_exts = ('.mp3', '.wav', '.m4a', '.flac', '.aac', '.ogg', '.wma')
        # 支持的视频格式
        video_exts = ('.mp4', '.avi', '.mov', '.mkv', '.webm', '.flv', '.wmv')
        
        return any(file_name.endswith(ext) for ext in audio_exts + video_exts)
    
    def handle(self, file, pattern_list: List, with_filter: bool, limit: int,
               get_buffer, save_image, **kwargs):
        """处理音视频文件 - 使用默认文本"""
        
        maxkb_logger.info(f"MediaSplitHandle.handle called with file: {file.name}")
        maxkb_logger.info(f"Using default text for media processing (no actual audio processing)")
        
        # 获取文件名和类型
        file_name = file.name
        file_ext = file_name.lower().split('.')[-1]
        
        # 判断媒体类型
        audio_exts = {'mp3', 'wav', 'm4a', 'flac', 'aac', 'ogg', 'wma'}
        video_exts = {'mp4', 'avi', 'mov', 'mkv', 'webm', 'flv', 'wmv'}
        
        if file_ext in audio_exts:
            media_type = "音频"
            default_segments = self._get_audio_default_segments(file_name)
        elif file_ext in video_exts:
            media_type = "视频"
            default_segments = self._get_video_default_segments(file_name)
        else:
            media_type = "媒体"
            default_segments = self._get_media_default_segments(file_name)
        
        maxkb_logger.info(f"Processing {media_type} file: {file_name}")
        maxkb_logger.info(f"Generating {len(default_segments)} default segments")
        
        # 转换为MaxKB段落格式
        paragraphs = []
        for i, segment_data in enumerate(default_segments):
            paragraph = {
                'content': segment_data['content'],
                'title': segment_data['title'],
                'metadata': {
                    'start_time': segment_data.get('start_time'),
                    'end_time': segment_data.get('end_time'),
                    'index': i,
                    'is_demo': True,
                    'media_type': media_type,
                    'file_name': file_name
                }
            }
            paragraphs.append(paragraph)
        
        # 应用限制
        if limit > 0:
            paragraphs = paragraphs[:limit]
        
        # 添加处理元数据
        metadata = {
            'media_processing_status': 'success',
            'media_type': media_type,
            'is_demo_content': True,
            'processing_mode': 'default_text'
        }
        
        maxkb_logger.info(f"Successfully created {len(paragraphs)} default paragraphs for {file_name}")
        
        return {
            'name': file.name,
            'content': paragraphs,
            'metadata': metadata
        }
    
    def _get_audio_default_segments(self, file_name: str) -> List[dict]:
        """生成音频文件的默认段落"""
        base_name = file_name.split('.')[0]
        
        return [
            {
                'title': '开场介绍',
                'content': f'这是音频文件 "{base_name}" 的第一段内容演示。本段包含了会议的开场介绍和主要议题的说明。\n\n主要内容：\n- 会议目的和议程说明\n- 参会人员介绍\n- 会议背景和重要性\n- 预期成果和目标设定',
                'start_time': 0,
                'end_time': 180
            },
            {
                'title': '主要内容讨论',
                'content': f'这是音频文件 "{base_name}" 的第二段内容演示。本段详细讨论了项目的进展情况和下一步的工作计划。\n\n主要内容：\n- 项目当前进展汇报\n- 关键问题和挑战分析\n- 解决方案讨论\n- 资源需求和分配',
                'start_time': 180,
                'end_time': 360
            },
            {
                'title': '总结与行动项',
                'content': f'这是音频文件 "{base_name}" 的第三段内容演示。本段总结了会议的主要结论和行动项，明确了责任人和时间节点。\n\n主要内容：\n- 会议要点总结\n- 行动项和责任分配\n- 时间节点和里程碑\n- 后续跟进计划',
                'start_time': 360,
                'end_time': 540
            }
        ]
    
    def _get_video_default_segments(self, file_name: str) -> List[dict]:
        """生成视频文件的默认段落"""
        base_name = file_name.split('.')[0]
        
        return [
            {
                'title': '开场介绍',
                'content': f'这是视频文件 "{base_name}" 的第一段内容演示。本段包含了视频的开场介绍和主要内容概述。\n\n主要内容：\n- 产品/服务介绍\n- 功能特性概述\n- 目标用户群体\n- 使用场景说明',
                'start_time': 0,
                'end_time': 120
            },
            {
                'title': '功能演示',
                'content': f'这是视频文件 "{base_name}" 的第二段内容演示。本段详细展示了产品的功能特性和使用方法。\n\n主要内容：\n- 核心功能演示\n- 操作步骤说明\n- 使用技巧和注意事项\n- 常见问题解答',
                'start_time': 120,
                'end_time': 300
            },
            {
                'title': '总结与联系方式',
                'content': f'这是视频文件 "{base_name}" 的第三段内容演示。本段总结了产品的主要优势和适用场景，提供了联系方式。\n\n主要内容：\n- 产品优势总结\n- 价格和套餐信息\n- 适用场景和行业\n- 联系方式和售后服务',
                'start_time': 300,
                'end_time': 420
            }
        ]
    
    def _get_media_default_segments(self, file_name: str) -> List[dict]:
        """生成其他媒体文件的默认段落"""
        base_name = file_name.split('.')[0]
        
        return [
            {
                'title': '文件概述',
                'content': f'这是媒体文件 "{base_name}" 的第一段内容演示。本段包含了文件的基本信息和主要内容概述。\n\n主要内容：\n- 文件基本信息\n- 内容类型说明\n- 主要用途和价值\n- 处理建议和注意事项',
                'start_time': 0,
                'end_time': 120
            },
            {
                'title': '详细内容',
                'content': f'这是媒体文件 "{base_name}" 的第二段内容演示。本段详细介绍了文件的核心内容和关键信息。\n\n主要内容：\n- 核心内容分析\n- 关键信息提取\n- 重要要点总结\n- 后续处理建议',
                'start_time': 120,
                'end_time': 240
            }
        ]
    
    def get_content(self, file, save_image):
        """获取文件内容（用于预览）"""
        try:
            file_name = file.name
            # 判断媒体类型
            file_ext = file_name.lower().split('.')[-1]
            video_exts = {'mp4', 'avi', 'mov', 'mkv', 'webm', 'flv', 'wmv'}
            
            if file_ext in video_exts:
                return f"[视频文件: {file_name}]\n\n该文件需要进行音频提取和语音识别处理。"
            else:
                return f"[音频文件: {file_name}]\n\n该文件需要进行语音识别处理。"
        except Exception as e:
            return f"读取文件失败: {str(e)}"
    
    def _format_time(self, seconds: float) -> str:
        """格式化时间"""
        if seconds is None:
            return "00:00"
        
        hours = int(seconds // 3600)
        minutes = int((seconds % 3600) // 60)
        secs = int(seconds % 60)
        
        if hours > 0:
            return f"{hours:02d}:{minutes:02d}:{secs:02d}"
        else:
            return f"{minutes:02d}:{secs:02d}"