Maas-Model-platform/backend/app/services/audio_v2/base_service.py

"""
V2服务基类

提供所有V2服务的通用功能和工具方法
"""
import logging
import os
from typing import Optional
from sqlalchemy.orm import Session

from app.services.oss_service import get_oss_service

logger = logging.getLogger(__name__)


class BaseV2Service:
    """V2服务基类"""
    
    def __init__(self, db: Session, user_id: str, api_key: str = None):
        """
        初始化基础服务
        
        Args:
            db: 数据库会话
            user_id: 用户ID
            api_key: DashScope API密钥
        """
        self.db = db
        self.user_id = user_id
        self.api_key = api_key or os.getenv("DASHSCOPE_API_KEY")
        self.oss_service = get_oss_service()
        
        # 设置DashScope API Key
        import dashscope
        dashscope.api_key = self.api_key
    
    def _generate_task_id(self) -> str:
        """
        生成本地任务ID（UUID）
        
        Returns:
            UUID字符串
        """
        import uuid
        return str(uuid.uuid4())
    
    def _calculate_char_count(self, text: str) -> int:
        """
        计算字符数（汉字算两个字符）
        
        Args:
            text: 待计算的文本
            
        Returns:
            字符数（汉字算两个字符）
        """
        count = 0
        for char in text:
            # 判断是否为汉字（CJK统一汉字范围）
            if '\u4e00' <= char <= '\u9fff':
                count += 2
            else:
                count += 1
        return count
    
    def _estimate_duration_from_bytes(
        self, 
        audio_data: bytes, 
        format_str: str, 
        sample_rate: int
    ) -> float:
        """
        从音频字节数据估算时长
        
        Args:
            audio_data: 音频数据
            format_str: 音频格式
            sample_rate: 采样率
            
        Returns:
            估算的时长（秒）
        """
        if format_str == "pcm":
            # PCM: 16bit mono = 2 bytes per sample
            return len(audio_data) / (sample_rate * 2)
        elif format_str == "wav":
            # WAV: 跳过44字节头，16bit mono
            return (len(audio_data) - 44) / (sample_rate * 2)
        elif format_str == "mp3":
            # MP3: 粗略估算，128kbps = 16KB/s
            return len(audio_data) / 16000
        else:
            # 默认估算
            return len(audio_data) / 16000