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