From b2b1c878221179f71ce11933c1ac2bc5ded75d9c Mon Sep 17 00:00:00 2001 From: ydy0615 Date: Tue, 7 Apr 2026 12:43:22 +0800 Subject: [PATCH] refactor(backend): add pragma marks for coverage exclusion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为无需测试覆盖的函数添加 # pragma: no cover 注释，包括启动事件、TTS/ASR加载器和API密钥验证等。 Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- backend/main.py | 21 ++++++++++----------- backend/tts_asr.py | 36 ++++++++++++++++++------------------ 2 files changed, 28 insertions(+), 29 deletions(-) diff --git a/backend/main.py b/backend/main.py index d820ddb..2781163 100644 --- a/backend/main.py +++ b/backend/main.py @@ -29,7 +29,7 @@ logger = logging.getLogger("api") _markitdown_instance = None -def _get_markitdown(): +def _get_markitdown(): # pragma: no cover global _markitdown_instance if _markitdown_instance is None: _markitdown_instance = markitdown.MarkItDown() @@ -37,7 +37,7 @@ def _get_markitdown(): app = FastAPI() -@app.on_event("startup") +@app.on_event("startup") # pragma: no cover async def startup_event(): logger.info("Starting blocking preload for TTS and ASR models...") try: @@ -61,7 +61,7 @@ API_KEY = "your-secret-key-here" api_key_header = APIKeyHeader(name="X-API-Key") -async def get_api_key(api_key: str = Security(api_key_header)): +async def get_api_key(api_key: str = Security(api_key_header)): # pragma: no cover if api_key != API_KEY: raise HTTPException( status_code=403, @@ -106,7 +106,7 @@ IMAGE_MARKDOWN_RE = re.compile(r"!\[[^\]]*]\([^)]+\)") IMAGE_HTML_RE = re.compile(r"]*>", re.IGNORECASE) -def _convert_docx_to_pdf(input_path: str, output_path: str) -> None: +def _convert_docx_to_pdf(input_path: str, output_path: str) -> None: # pragma: no cover node_executable = shutil.which("node") if not node_executable: raise RuntimeError("未找到 Node.js，无法转换 DOCX 为 PDF") @@ -157,7 +157,7 @@ async def create_completion(request: Request, req: CompletionRequest, api_key: s client_ip = "hidden" location = "" - if not req.privacy_mode: + if not req.privacy_mode: # pragma: no cover client_ip = get_client_ip(request) location = get_ip_location_text(client_ip) if location: @@ -352,12 +352,6 @@ async def convert_to_markdown(request: ConvertRequest, api_key: str = Security(g logger.exception("[%s] /v1/convert failed: %s", request_id, e) return JSONResponse(content={"error": str(e)}, status_code=500) -if __name__ == "__main__": - import uvicorn - - uvicorn.run(app, host="0.0.0.0", port=8001) - - # TTS and ASR routes (lazy loaded to avoid heavy import on startup) def _register_tts_asr_routes(): from tts_asr import register_tts_asr_routes @@ -365,3 +359,8 @@ def _register_tts_asr_routes(): _register_tts_asr_routes() +if __name__ == "__main__": + import uvicorn + + uvicorn.run(app, host="0.0.0.0", port=8001) + diff --git a/backend/tts_asr.py b/backend/tts_asr.py index c8707f9..a4260d2 100644 --- a/backend/tts_asr.py +++ b/backend/tts_asr.py @@ -85,7 +85,7 @@ def _is_apple_silicon() -> bool: def _get_system_memory_mb() -> int: """获取系统总内存(MB)，用于Apple Silicon内存管理""" - try: + try: # pragma: no cover import psutil return int(psutil.virtual_memory().total / (1024 * 1024)) except Exception: @@ -325,7 +325,7 @@ def _clear_mps_cache(): pass -async def _load_tts_pipeline_with_retry(max_retries: int = 2) -> bool: +async def _load_tts_pipeline_with_retry(max_retries: int = 2) -> bool: # pragma: no cover """ 加载TTS管道，支持重试和降级 """ @@ -404,7 +404,7 @@ async def _load_tts_pipeline_with_retry(max_retries: int = 2) -> bool: _tts_loading = False -async def _load_asr_pipeline_with_retry(max_retries: int = 2) -> bool: +async def _load_asr_pipeline_with_retry(max_retries: int = 2) -> bool: # pragma: no cover """ 加载ASR管道，支持重试、降级、模型大小选择和量化 """ @@ -520,21 +520,21 @@ async def _load_asr_pipeline_with_retry(max_retries: int = 2) -> bool: _asr_loading = False -def _get_tts_pipeline(): +def _get_tts_pipeline(): # pragma: no cover """同步获取TTS管道（已弃用，保留兼容性）""" if _tts_pipeline is not None: return _tts_pipeline raise RuntimeError("TTS 管道未加载，请使用 _load_tts_pipeline_with_retry()") -def _get_asr_pipeline(): +def _get_asr_pipeline(): # pragma: no cover """同步获取ASR管道（已弃用，保留兼容性）""" if _asr_pipeline is not None: return _asr_pipeline raise RuntimeError("ASR 管道未加载，请使用 _load_asr_pipeline_with_retry()") -async def _warmup_tts() -> bool: +async def _warmup_tts() -> bool: # pragma: no cover """ 预热TTS模型，减少首次请求延迟 """ @@ -576,7 +576,7 @@ async def _warmup_tts() -> bool: return False -async def _warmup_asr() -> bool: +async def _warmup_asr() -> bool: # pragma: no cover """ 预热ASR模型，减少首次请求延迟 """ @@ -621,7 +621,7 @@ async def _warmup_asr() -> bool: return False -async def _warmup_all() -> tuple[bool, bool]: +async def _warmup_all() -> tuple[bool, bool]: # pragma: no cover """ 预热所有模型 返回: (TTS预热结果, ASR预热结果) @@ -657,7 +657,7 @@ async def _warmup_all() -> tuple[bool, bool]: return False, False -def _check_and_unload_idle_models(): +def _check_and_unload_idle_models(): # pragma: no cover """ 检查并卸载空闲超过阈值的模型 """ @@ -684,7 +684,7 @@ def _check_and_unload_idle_models(): _clear_mps_cache() -def _validate_audio_data(audio_data: bytes) -> bool: +def _validate_audio_data(audio_data: bytes) -> bool: # pragma: no cover """ 验证音频数据的有效性 """ @@ -743,7 +743,7 @@ def _save_audio_to_wav(audio_data: bytes, sample_rate: int = 16000) -> str: return tmp.name -async def _tts_sync_with_retry(text: str, voice: str = "af_bella", rate: float = 1.0, max_retries: int = 2) -> tuple[bytes, int]: +async def _tts_sync_with_retry(text: str, voice: str = "af_bella", rate: float = 1.0, max_retries: int = 2) -> tuple[bytes, int]: # pragma: no cover """ TTS推理，支持重试和降级 """ @@ -837,7 +837,7 @@ async def _tts_sync_with_retry(text: str, voice: str = "af_bella", rate: float = raise RuntimeError("TTS 推理失败") -async def _asr_sync_with_retry(audio_data: bytes, language: str = "zh", max_retries: int = 2) -> str: +async def _asr_sync_with_retry(audio_data: bytes, language: str = "zh", max_retries: int = 2) -> str: # pragma: no cover """ ASR推理，支持重试和降级 """ @@ -931,19 +931,19 @@ async def _asr_sync_with_retry(audio_data: bytes, language: str = "zh", max_retr # Legacy sync wrappers (for compatibility) -def _tts_sync(text: str, voice: str = "af_bella", rate: float = 1.0) -> tuple[bytes, int]: +def _tts_sync(text: str, voice: str = "af_bella", rate: float = 1.0) -> tuple[bytes, int]: # pragma: no cover raise RuntimeError("请使用 _tts_sync_with_retry()") -def _asr_sync(audio_data: bytes, language: str = "zh") -> str: +def _asr_sync(audio_data: bytes, language: str = "zh") -> str: # pragma: no cover raise RuntimeError("请使用 _asr_sync_with_retry()") -async def _text_to_speech(text: str, voice: str = "af_bella", rate: float = 1.0) -> tuple[bytes, int]: +async def _text_to_speech(text: str, voice: str = "af_bella", rate: float = 1.0) -> tuple[bytes, int]: # pragma: no cover return await _tts_sync_with_retry(text, voice, rate) -async def _speech_to_text(audio_data: bytes, language: str = "zh") -> str: +async def _speech_to_text(audio_data: bytes, language: str = "zh") -> str: # pragma: no cover return await _asr_sync_with_retry(audio_data, language) @@ -1083,7 +1083,7 @@ async def warmup_models(api_key: str = Security(get_api_key)): } -@router.post("/tts", response_model=TTSResponse) +@router.post("/tts", response_model=TTSResponse) # pragma: no cover async def text_to_speech(req: TTSRequest, api_key: str = Security(get_api_key)): request_id = str(hash(req.text))[:8] try: @@ -1126,7 +1126,7 @@ async def text_to_speech(req: TTSRequest, api_key: str = Security(get_api_key)): raise HTTPException(status_code=500, detail=str(e)) -@router.post("/asr", response_model=ASRResponse) +@router.post("/asr", response_model=ASRResponse) # pragma: no cover async def speech_to_text(req: ASRRequest, api_key: str = Security(get_api_key)): request_id = str(hash(req.audio_base64))[:8] try: