Merge pull request #2 from JefferyHcool/dev

feat(events): 实现转写完成后的文件清理功能

backend/app/transcriber/base.py

@@ -12,3 +12,12 @@ class Transcriber(ABC):
         :return: 返回一个 TranscriptResult 类
         '''
         pass
+
+    def on_finish(self,video_path:str,result: TranscriptResult)->None:
+        '''
+        当音频转录完成时调用
+        :param video_path: 视频路径
+        :param result: 识别结果
+        :return:
+        '''
+        pass

backend/app/transcriber/whisper.py

@@ -6,6 +6,8 @@ from app.transcriber.base import Transcriber
 from app.utils.env_checker import is_cuda_available, is_torch_installed
 from app.utils.path_helper import get_model_dir
 
+from events import transcription_finished
+
 '''
  Size of the model to use (tiny, tiny.en, base, base.en, small, small.en, distil-small.en, medium, medium.en, distil-medium.en, large-v1, large-v2, large-v3, large, distil-large-v2, distil-large-v3, large-v3-turbo, or turbo
 '''
@@ -64,29 +66,37 @@ class WhisperTranscriber(Transcriber):
 
     @timeit
     def transcript(self, file_path: str) -> TranscriptResult:
+        try:
 
-        segments_raw, info = self.model.transcribe(file_path)
+            segments_raw, info = self.model.transcribe(file_path)
 
-        segments = []
+            segments = []
-        full_text = ""
+            full_text = ""
 
-        for seg in segments_raw:
+            for seg in segments_raw:
-            text = seg.text.strip()
+                text = seg.text.strip()
-            full_text += text + " "
+                full_text += text + " "
-            segments.append(TranscriptSegment(
+                segments.append(TranscriptSegment(
-                start=seg.start,
+                    start=seg.start,
-                end=seg.end,
+                    end=seg.end,
-                text=text
+                    text=text
-            ))
+                ))
 
-        return TranscriptResult(
+            result= TranscriptResult(
-            language=info.language,
+                language=info.language,
-            full_text=full_text.strip(),
+                full_text=full_text.strip(),
-            segments=segments,
+                segments=segments,
-            raw=info
+                raw=info
-        )
+            )
+            self.on_finish(file_path, result)
+            return result
+        except Exception as e:
+            print(f"转写失败：{e}")
 
 
-if __name__ == '__main__':
+    def on_finish(self,video_path:str,result: TranscriptResult)->None:
-    print(WhisperTranscriber(cpu_threads=8).transcript(
+        print("转写完成")
-        '''D:\\data_backup_from_ssd\\02_个人项目\\11_BiliNote\\backend\\data\\BV1vcZ5YQE9X.mp3'''))
+        transcription_finished.send({
+            "file_path": video_path,
+        })
+

backend/events/__init__.py

@@ -0,0 +1,8 @@
+# 注册监听器
+from events.handlers import cleanup_temp_files
+from events.signals import transcription_finished
+
+
+def  register_handler():
+    transcription_finished.connect(cleanup_temp_files)
+

backend/events/handlers.py

@@ -0,0 +1,8 @@
+import os
+
+
+def cleanup_temp_files(data):
+    print(f"🧹 清理转写文件：{data['file_path']}")
+    os.remove(data['file_path'])
+
+

backend/events/signals.py

@@ -0,0 +1,2 @@
+from blinker import signal
+transcription_finished = signal("transcription_finished")

backend/main.py

@@ -7,6 +7,7 @@ from dotenv import load_dotenv
 from app import create_app
 from app.db.video_task_dao import init_video_task_table
 from app.transcriber.transcriber_provider import get_transcriber
+from events import register_handler
 from ffmpeg_helper import ensure_ffmpeg_or_raise
 
 load_dotenv()
@@ -26,15 +27,13 @@ if not os.path.exists(out_dir):
 app = create_app()
 app.mount(static_path, StaticFiles(directory=static_dir), name="static")
 
+async def startup_event():
+    register_handler()
 @app.on_event("startup")
-def check_env():
+async def startup_event():
+    register_handler()
     ensure_ffmpeg_or_raise()
-@app.on_event("startup")
-async def load_model_on_startup():
     get_transcriber()
-
-@app.on_event("startup")
-def startup():
     init_video_task_table()
 
 if __name__ == "__main__":