出力ディレクトリの作成とUUIDを用いたファイル名生成を追加しました。また、クリーンアップ機能の初期化と状態確認ボタンをGradioインターフェースに追加しました。これにより、ファイル管理が改善され、ユーザーがクリーンアップ状況を確認できるようになりました。

app.py

@@ -7,6 +7,10 @@ from model_manager import ModelManager
 
 # サンプルファイルのディレクトリを定義（絶対パスに解決）
 EXAMPLES_DIR = (Path(__file__).parent / "example").resolve()
+OUTPUT_DIR = (Path(__file__).parent / "output").resolve()
+
+# 出力ディレクトリの作成
+OUTPUT_DIR.mkdir(exist_ok=True)
 
 # インポートエラーのデバッグ情報を表示
 try:
@@ -60,9 +64,10 @@ def process_talking_head(audio_file, source_image):
         return None, "ソース画像をアップロードしてください。"
     
     try:
-        # 一時ファイルの作成
-        with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as tmp_output:
-            output_path = tmp_output.name
+        # 出力ファイルの作成（出力ディレクトリ内）
+        import uuid
+        output_filename = f"{uuid.uuid4()}.mp4"
+        output_path = str(OUTPUT_DIR / output_filename)
         
         # 処理実行
         print(f"処理開始: audio={audio_file}, image={source_image}")
@@ -149,5 +154,5 @@ if __name__ == "__main__":
         server_name="0.0.0.0",
         server_port=7860,
         share=False,
-        allowed_paths=[str(EXAMPLES_DIR)]
+        allowed_paths=[str(EXAMPLES_DIR), str(OUTPUT_DIR)]
     )

app_optimized.py

@@ -25,9 +25,17 @@ from core.optimization import (
     ParallelInference,
     OptimizedInferenceWrapper
 )
+from cleanup_old_files import initialize_cleanup, get_cleanup_status
 
 # サンプルファイルのディレクトリを定義
 EXAMPLES_DIR = (Path(__file__).parent / "example").resolve()
+OUTPUT_DIR = (Path(__file__).parent / "output").resolve()
+
+# 出力ディレクトリの作成
+OUTPUT_DIR.mkdir(exist_ok=True)
+
+# ファイルクリーンアップの初期化（24時間後に自動削除）
+initialize_cleanup(OUTPUT_DIR, max_age_hours=24)
 
 # 初期化フラグ
 print("=== Phase 3 最適化版 - 初期化開始 ===")
@@ -202,9 +210,10 @@ def process_talking_head_optimized(
     try:
         start_time = time.time()
         
-        # 一時ファイルの作成
-        with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as tmp_output:
-            output_path = tmp_output.name
+        # 出力ファイルの作成（出力ディレクトリ内）
+        import uuid
+        output_filename = f"{uuid.uuid4()}.mp4"
+        output_path = str(OUTPUT_DIR / output_filename)
         
         # アバタートークンから埋め込みを取得
         if avatar_token:
@@ -425,6 +434,7 @@ with gr.Blocks(title="DittoTalkingHead - Phase 3 最適化版") as demo:
             with gr.Row():
                 clear_inference_cache_btn = gr.Button("🗑️ 推論キャッシュをクリア", variant="secondary")
                 clear_avatar_cache_btn = gr.Button("🗑️ アバターキャッシュをクリア", variant="secondary")
+                cleanup_status_btn = gr.Button("📊 クリーンアップ状態", variant="secondary")
             
             cache_status = gr.Textbox(label="キャッシュ操作ステータス", lines=2)
     
@@ -497,9 +507,17 @@ with gr.Blocks(title="DittoTalkingHead - Phase 3 最適化版") as demo:
         fn=clear_avatar_cache,
         outputs=[cache_status]
     )
+    
+    cleanup_status_btn.click(
+        fn=lambda: get_cleanup_status(),
+        outputs=[cache_status]
+    )
 
 if __name__ == "__main__":
     # Cold Start最適化設定でGradioを起動
     launch_settings = cold_start_optimizer.optimize_gradio_settings()
     
+    # allowed_pathsを追加
+    launch_settings['allowed_paths'] = [str(EXAMPLES_DIR), str(OUTPUT_DIR)]
+    
     demo.launch(**launch_settings)

app_streaming.py

@@ -12,6 +12,10 @@ DATA_ROOT = "checkpoints/ditto_pytorch"
 
 # サンプルファイルのディレクトリ
 EXAMPLES_DIR = (Path(__file__).parent / "example").resolve()
+OUTPUT_DIR = (Path(__file__).parent / "output").resolve()
+
+# 出力ディレクトリの作成
+OUTPUT_DIR.mkdir(exist_ok=True)
 
 # グローバルで一度だけロード（concurrency_count=1 前提）
 sdk: StreamSDK | None = None
@@ -44,7 +48,9 @@ def generator(mic, src_img):
         sdk = init_sdk()
         
         # setup: online_mode=True でストリーミング
-        tmp_out = tempfile.mktemp(suffix=".mp4")
+        import uuid
+        output_filename = f"{uuid.uuid4()}.mp4"
+        tmp_out = str(OUTPUT_DIR / output_filename)
         sdk.setup(src_img, tmp_out, online_mode=True, max_size=1024)
         N_total = int(np.ceil(len(wav_full) / sr * 20))  # 概算フレーム数
         sdk.setup_Nd(N_total)
@@ -191,5 +197,6 @@ if __name__ == "__main__":
     demo.queue(concurrency_count=1, max_size=8).launch(
         server_name="0.0.0.0",
         server_port=7860,
-        share=False
+        share=False,
+        allowed_paths=[str(EXAMPLES_DIR), str(OUTPUT_DIR)]
     )

cleanup_old_files.py

@@ -0,0 +1,167 @@
+"""
+出力ディレクトリの古いファイルをクリーンアップする機能
+"""
+
+import os
+import time
+from pathlib import Path
+from datetime import datetime, timedelta
+import threading
+import logging
+
+# ロギング設定
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class FileCleanupManager:
+    """定期的に古いファイルを削除するマネージャー"""
+    
+    def __init__(self, output_dir: Path, max_age_hours: float = 24, check_interval_minutes: int = 60):
+        """
+        Args:
+            output_dir: 監視対象のディレクトリ
+            max_age_hours: ファイルを保持する最大時間（時間）
+            check_interval_minutes: チェック間隔（分）
+        """
+        self.output_dir = output_dir
+        self.max_age_seconds = max_age_hours * 3600
+        self.check_interval = check_interval_minutes * 60
+        self._running = False
+        self._thread = None
+        
+    def start(self):
+        """クリーンアップ処理を開始"""
+        if self._running:
+            logger.warning("Cleanup manager is already running")
+            return
+            
+        self._running = True
+        self._thread = threading.Thread(target=self._cleanup_loop, daemon=True)
+        self._thread.start()
+        logger.info(f"Started cleanup manager for {self.output_dir}")
+        
+    def stop(self):
+        """クリーンアップ処理を停止"""
+        self._running = False
+        if self._thread:
+            self._thread.join()
+        logger.info("Stopped cleanup manager")
+        
+    def _cleanup_loop(self):
+        """定期的にクリーンアップを実行"""
+        while self._running:
+            try:
+                self.cleanup_now()
+            except Exception as e:
+                logger.error(f"Error during cleanup: {e}")
+            
+            # 次のチェックまで待機
+            time.sleep(self.check_interval)
+            
+    def cleanup_now(self):
+        """今すぐクリーンアップを実行"""
+        if not self.output_dir.exists():
+            logger.warning(f"Output directory does not exist: {self.output_dir}")
+            return
+            
+        current_time = time.time()
+        deleted_count = 0
+        deleted_size = 0
+        
+        # MP4ファイルをチェック
+        for file_path in self.output_dir.glob("*.mp4"):
+            try:
+                # ファイルの最終変更時刻を取得
+                file_age = current_time - file_path.stat().st_mtime
+                
+                if file_age > self.max_age_seconds:
+                    file_size = file_path.stat().st_size
+                    file_path.unlink()
+                    deleted_count += 1
+                    deleted_size += file_size
+                    logger.debug(f"Deleted old file: {file_path.name} (age: {file_age/3600:.1f} hours)")
+                    
+            except Exception as e:
+                logger.error(f"Failed to delete {file_path}: {e}")
+                
+        if deleted_count > 0:
+            logger.info(f"Cleanup completed: deleted {deleted_count} files, freed {deleted_size/1024/1024:.1f} MB")
+            
+    def get_directory_stats(self):
+        """ディレクトリの統計情報を取得"""
+        if not self.output_dir.exists():
+            return {
+                "total_files": 0,
+                "total_size_mb": 0,
+                "oldest_file_hours": None
+            }
+            
+        files = list(self.output_dir.glob("*.mp4"))
+        if not files:
+            return {
+                "total_files": 0,
+                "total_size_mb": 0,
+                "oldest_file_hours": None
+            }
+            
+        total_size = sum(f.stat().st_size for f in files)
+        current_time = time.time()
+        oldest_age = max((current_time - f.stat().st_mtime) for f in files)
+        
+        return {
+            "total_files": len(files),
+            "total_size_mb": total_size / 1024 / 1024,
+            "oldest_file_hours": oldest_age / 3600
+        }
+
+
+# グローバルインスタンス（アプリケーションで使用）
+cleanup_manager = None
+
+def initialize_cleanup(output_dir: Path, max_age_hours: float = 24):
+    """クリーンアップマネージャーを初期化して開始"""
+    global cleanup_manager
+    
+    if cleanup_manager is not None:
+        cleanup_manager.stop()
+        
+    cleanup_manager = FileCleanupManager(output_dir, max_age_hours)
+    cleanup_manager.start()
+    
+    return cleanup_manager
+
+
+def get_cleanup_status():
+    """クリーンアップの状態を取得"""
+    if cleanup_manager is None:
+        return "Cleanup manager not initialized"
+        
+    stats = cleanup_manager.get_directory_stats()
+    return f"""
+クリーンアップ状態:
+- ファイル数: {stats['total_files']}
+- 合計サイズ: {stats['total_size_mb']:.1f} MB
+- 最古のファイル: {stats['oldest_file_hours']:.1f} 時間前 (最古のファイルがある場合)
+"""
+
+
+if __name__ == "__main__":
+    # テスト実行
+    output_dir = Path("./output")
+    output_dir.mkdir(exist_ok=True)
+    
+    # テスト用のダミーファイルを作成
+    test_file = output_dir / "test.mp4"
+    test_file.write_text("dummy content")
+    
+    # 古いファイルをシミュレート（2日前）
+    old_time = time.time() - (48 * 3600)
+    os.utime(test_file, (old_time, old_time))
+    
+    # クリーンアップマネージャーを初期化
+    manager = FileCleanupManager(output_dir, max_age_hours=24)
+    
+    # 即座にクリーンアップを実行
+    print("Before cleanup:", manager.get_directory_stats())
+    manager.cleanup_now()
+    print("After cleanup:", manager.get_directory_stats())