12.26

services/a2f_api/a2f_service.py

@@ -3,38 +3,51 @@ import sys
 import os
 from pathlib import Path
 import glob
+import tempfile
+import shutil
+from datetime import datetime
 
 class A2FService:
     def __init__(self, a2f_url="192.168.1.39:52000"):
         self.base_dir = Path(__file__).parent.parent.parent
-        self.output_dir = self.base_dir / "data" / "output"
         self.a2f_script = self.base_dir / "external" / "Audio2Face-3D-Samples" / "scripts" / "audio2face_3d_microservices_interaction_app" / "a2f_3d.py"
         self.config_file = self.base_dir / "external" / "Audio2Face-3D-Samples" / "scripts" / "audio2face_3d_microservices_interaction_app" / "config" / "config_james.yml"
         self.a2f_url = a2f_url
-        os.makedirs(self.output_dir, exist_ok=True)
 
-    def audio_to_csv(self, audio_path: str) -> str:
-        cmd = [
-            sys.executable,
-            str(self.a2f_script),
-            "run_inference",
-            audio_path,
-            str(self.config_file),
-            "--url",
-            self.a2f_url
-        ]
+    def audio_to_csv(self, audio_path: str) -> tuple[str, str]:
+        # 使用时间戳创建独立的临时工作目录
+        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S_%f')
+        temp_work_dir = tempfile.mkdtemp(prefix=f"a2f_work_{timestamp}_")
 
-        result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, cwd=str(self.output_dir))
+        try:
+            cmd = [
+                sys.executable,
+                str(self.a2f_script),
+                "run_inference",
+                audio_path,
+                str(self.config_file),
+                "--url",
+                self.a2f_url
+            ]
 
-        if result.returncode != 0:
-            raise RuntimeError(f"A2F inference failed: {result.stdout}")
+            # 在独立的工作目录中运行
+            result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, cwd=temp_work_dir)
 
-        output_dirs = sorted(glob.glob(str(self.output_dir / "output_*")))
-        if not output_dirs:
-            raise RuntimeError("No output directory found")
+            if result.returncode != 0:
+                raise RuntimeError(f"A2F inference failed: {result.stdout}")
 
-        csv_path = os.path.join(output_dirs[-1], "animation_frames.csv")
-        if not os.path.exists(csv_path):
-            raise RuntimeError(f"CSV file not found: {csv_path}")
+            # 在工作目录中查找输出
+            output_dirs = sorted(glob.glob(os.path.join(temp_work_dir, "output_*")))
+            if not output_dirs:
+                raise RuntimeError(f"No output directory found in {temp_work_dir}")
 
-        return csv_path
+            csv_path = os.path.join(output_dirs[-1], "animation_frames.csv")
+            if not os.path.exists(csv_path):
+                raise RuntimeError(f"CSV file not found: {csv_path}")
+
+            # 返回CSV路径和临时目录路径（用于后续清理）
+            return csv_path, temp_work_dir
+        except Exception as e:
+            # 出错时清理临时目录
+            shutil.rmtree(temp_work_dir, ignore_errors=True)
+            raise e