Implement SessionManager for managing session data and directories

tests/unit/test_session_manager.py

@@ -0,0 +1,45 @@
+"""Tests for SessionManager module."""
+
+from yomitalk.utils.session_manager import SessionManager
+
+
+def test_session_manager_initialization():
+    """Test that SessionManager is initialized properly."""
+    session_manager = SessionManager()
+    assert session_manager.session_id is not None
+    assert isinstance(session_manager.session_id, str)
+    assert len(session_manager.session_id) > 0
+
+    # セッションIDのフォーマットを確認
+    assert session_manager.session_id.startswith("session_")
+
+
+def test_session_dirs_creation():
+    """Test that SessionManager creates session-specific directories."""
+    session_manager = SessionManager()
+
+    # テンポラリディレクトリのテスト
+    temp_dir = session_manager.get_temp_dir()
+    assert temp_dir.exists()
+    assert temp_dir.is_dir()
+    assert session_manager.session_id in str(temp_dir)
+
+    # 出力ディレクトリのテスト
+    output_dir = session_manager.get_output_dir()
+    assert output_dir.exists()
+    assert output_dir.is_dir()
+    assert session_manager.session_id in str(output_dir)
+
+    # トーク一時ディレクトリのテスト
+    talk_temp_dir = session_manager.get_talk_temp_dir()
+    assert talk_temp_dir.exists()
+    assert talk_temp_dir.is_dir()
+    assert session_manager.session_id in str(talk_temp_dir)
+    assert "talks" in str(talk_temp_dir)
+
+
+def test_unique_session_ids():
+    """Test that consecutive session managers get different session IDs."""
+    session1 = SessionManager()
+    session2 = SessionManager()
+    assert session1.session_id != session2.session_id

yomitalk/app.py

@@ -7,7 +7,6 @@ Builds the Paper Podcast Generator application using Gradio.
 
 import math
 import os
-import uuid
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
@@ -18,8 +17,9 @@ from yomitalk.components.file_uploader import FileUploader
 from yomitalk.components.text_processor import TextProcessor
 from yomitalk.prompt_manager import DocumentType, PodcastMode
 from yomitalk.utils.logger import logger
+from yomitalk.utils.session_manager import SessionManager
 
-# Check for temporary file directories
+# Check for base directories
 os.makedirs("data/temp", exist_ok=True)
 os.makedirs("data/output", exist_ok=True)
 
@@ -39,9 +39,19 @@ class PaperPodcastApp:
 
         Creates instances of FileUploader, TextProcessor, and AudioGenerator.
         """
-        self.file_uploader = FileUploader()
+        # セッション管理の初期化
+        self.session_manager = SessionManager()
+        logger.info(
+            f"Initializing app with session ID: {self.session_manager.get_session_id()}"
+        )
+
+        # 各コンポーネントにセッション固有のディレクトリを渡す
+        self.file_uploader = FileUploader(temp_dir=self.session_manager.get_temp_dir())
         self.text_processor = TextProcessor()
-        self.audio_generator = AudioGenerator()
+        self.audio_generator = AudioGenerator(
+            session_output_dir=self.session_manager.get_output_dir(),
+            session_temp_dir=self.session_manager.get_talk_temp_dir(),
+        )
 
         # Check if VOICEVOX Core is available
         self.voicevox_core_available = (
@@ -136,6 +146,7 @@ class PaperPodcastApp:
         Process file uploads.
 
         Properly handles file objects from Gradio's file upload component.
+        This is now a wrapper around FileUploader's handle_file_upload method.
 
         Args:
             file_obj: Gradio's file object
@@ -146,45 +157,8 @@ class PaperPodcastApp:
         if file_obj is None:
             return None
 
-        try:
-            # Temporary directory path
-            temp_dir = Path("data/temp")
-            temp_dir.mkdir(parents=True, exist_ok=True)
-
-            # Get filename
-            if isinstance(file_obj, list) and len(file_obj) > 0:
-                file_obj = file_obj[0]  # Get first element if it's a list
-
-            # セキュリティのため、オリジナルファイル名は使用せず、一意のIDを生成
-            # ただし、元のファイル拡張子は保持する
-            original_extension = ".txt"  # デフォルト拡張子
-            if hasattr(file_obj, "name"):
-                # 元のファイルの拡張子を取得
-                original_extension = os.path.splitext(Path(file_obj.name).name)[1]
-                # 拡張子がない場合はデフォルト値を使用
-                if not original_extension:
-                    original_extension = ".txt"
-
-            # 安全なファイル名を生成（UUIDと元の拡張子を組み合わせる）
-            filename = f"uploaded_{uuid.uuid4().hex}{original_extension}"
-
-            # Create temporary file path
-            temp_path = temp_dir / filename
-
-            # Get and save file data
-            if hasattr(file_obj, "read") and callable(file_obj.read):
-                with open(temp_path, "wb") as f:
-                    f.write(file_obj.read())
-            elif hasattr(file_obj, "name"):
-                with open(temp_path, "wb") as f:
-                    with open(file_obj.name, "rb") as source:
-                        f.write(source.read())
-
-            return str(temp_path)
-
-        except Exception as e:
-            logger.error(f"File processing error: {e}")
-            return None
+        # FileUploaderのhandle_file_uploadメソッドを使用
+        return self.file_uploader.handle_file_upload(file_obj)
 
     def extract_file_text(self, file_obj) -> str:
         """

yomitalk/components/audio_generator.py

@@ -123,10 +123,31 @@ class AudioGenerator:
 
     CONVERSION_OVERRIDE = {"this": "ディス", "to": "トゥ", "a": "ア"}
 
-    def __init__(self) -> None:
-        """Initialize AudioGenerator."""
-        self.output_dir = Path("data/output")
-        self.temp_dir = Path("data/temp/talks")
+    def __init__(
+        self,
+        session_output_dir: Optional[Path] = None,
+        session_temp_dir: Optional[Path] = None,
+    ) -> None:
+        """
+        Initialize AudioGenerator.
+
+        Args:
+            session_output_dir (Optional[Path]): Session-specific output directory.
+                If not provided, defaults to "data/output"
+            session_temp_dir (Optional[Path]): Session-specific temporary directory.
+                If not provided, defaults to "data/temp/talks"
+        """
+        # Use session-specific directories if provided
+        self.output_dir = (
+            session_output_dir if session_output_dir else Path("data/output")
+        )
+        self.temp_dir = (
+            session_temp_dir if session_temp_dir else Path("data/temp/talks")
+        )
+
+        # Make sure directories exist
+        self.output_dir.mkdir(parents=True, exist_ok=True)
+        self.temp_dir.mkdir(parents=True, exist_ok=True)
 
         # VOICEVOX Core
         self.core_initialized = False

yomitalk/components/file_uploader.py

@@ -4,7 +4,6 @@ Provides text extraction functionality for the Paper Podcast Generator applicati
 """
 
 import os
-from pathlib import Path
 from typing import List
 
 from yomitalk.utils.logger import logger
@@ -14,10 +13,14 @@ from yomitalk.utils.pdf_extractor import PDFExtractor
 class FileUploader:
     """Class for uploading files and extracting text."""
 
-    def __init__(self) -> None:
-        """Initialize FileUploader."""
-        self.temp_dir = Path("data/temp")
-        self.temp_dir.mkdir(parents=True, exist_ok=True)
+    def __init__(self, temp_dir=None) -> None:
+        """
+        Initialize FileUploader.
+
+        Args:
+            temp_dir (Optional[Path]): Session-specific temporary directory path.
+                If not provided, defaults to "data/temp"
+        """
         self.supported_text_extensions = [".txt", ".md", ".text", ".tmp"]
         self.supported_pdf_extensions = [".pdf"]
         self.supported_extensions = (
@@ -25,6 +28,12 @@ class FileUploader:
         )
         self.pdf_extractor = PDFExtractor()
 
+        # Set temporary directory
+        from pathlib import Path
+
+        self.temp_dir = Path(temp_dir) if temp_dir else Path("data/temp")
+        self.temp_dir.mkdir(parents=True, exist_ok=True)
+
     def extract_text_from_path(self, file_path: str) -> str:
         """
         Extract text from a file based on its extension.
@@ -76,6 +85,61 @@ class FileUploader:
             logger.error(f"Text file reading error: {e}")
             return f"Text file reading failed: {str(e)}"
 
+    def handle_file_upload(self, file_obj):
+        """
+        Process file uploads.
+
+        Properly handles file objects from Gradio's file upload component.
+
+        Args:
+            file_obj: Gradio's file object
+
+        Returns:
+            str: Path to the temporary file
+        """
+        if file_obj is None:
+            return None
+
+        try:
+            # Get filename
+            if isinstance(file_obj, list) and len(file_obj) > 0:
+                file_obj = file_obj[0]  # Get first element if it's a list
+
+            # セキュリティのため、オリジナルファイル名は使用せず、一意のIDを生成
+            # ただし、元のファイル拡張子は保持する
+            import os
+            import uuid
+            from pathlib import Path
+
+            original_extension = ".txt"  # デフォルト拡張子
+            if hasattr(file_obj, "name"):
+                # 元のファイルの拡張子を取得
+                original_extension = os.path.splitext(Path(file_obj.name).name)[1]
+                # 拡張子がない場合はデフォルト値を使用
+                if not original_extension:
+                    original_extension = ".txt"
+
+            # 安全なファイル名を生成（UUIDと元の拡張子を組み合わせる）
+            filename = f"uploaded_{uuid.uuid4().hex}{original_extension}"
+
+            # セッション固有のtemp_dirを使用
+            temp_path = self.temp_dir / filename
+
+            # Get and save file data
+            if hasattr(file_obj, "read") and callable(file_obj.read):
+                with open(temp_path, "wb") as f:
+                    f.write(file_obj.read())
+            elif hasattr(file_obj, "name"):
+                with open(temp_path, "wb") as f:
+                    with open(file_obj.name, "rb") as source:
+                        f.write(source.read())
+
+            return str(temp_path)
+
+        except Exception as e:
+            logger.error(f"File processing error: {e}")
+            return None
+
     def get_supported_extensions(self) -> List[str]:
         """
         Get list of supported file extensions.

yomitalk/utils/session_manager.py

@@ -0,0 +1,78 @@
+"""Session Manager Module.
+
+This module provides functionality for managing Hugging Face Space sessions.
+"""
+
+import time
+import uuid
+from pathlib import Path
+
+from yomitalk.utils.logger import logger
+
+
+class SessionManager:
+    """Class for managing session data across Hugging Face Space sessions."""
+
+    def __init__(self):
+        """Initialize SessionManager."""
+        self.session_id = self._generate_session_id()
+        self.base_temp_dir = Path("data/temp")
+        self.base_output_dir = Path("data/output")
+        logger.info(f"Session initialized with ID: {self.session_id}")
+
+    def _generate_session_id(self) -> str:
+        """
+        Generate a unique session ID.
+
+        Creates a unique ID based on timestamp and UUID to ensure uniqueness
+        across all environments.
+
+        Returns:
+            str: A unique session ID
+        """
+        # Always use UUID-based session ID
+        timestamp = int(time.time())
+        random_id = uuid.uuid4().hex[:8]
+        return f"session_{timestamp}_{random_id}"
+
+    def get_session_id(self) -> str:
+        """
+        Get the current session ID.
+
+        Returns:
+            str: The current session ID
+        """
+        return self.session_id
+
+    def get_temp_dir(self) -> Path:
+        """
+        Get the temporary directory for the current session.
+
+        Returns:
+            Path: Path to the session's temporary directory
+        """
+        session_temp_dir = self.base_temp_dir / self.session_id
+        session_temp_dir.mkdir(parents=True, exist_ok=True)
+        return session_temp_dir
+
+    def get_output_dir(self) -> Path:
+        """
+        Get the output directory for the current session.
+
+        Returns:
+            Path: Path to the session's output directory
+        """
+        session_output_dir = self.base_output_dir / self.session_id
+        session_output_dir.mkdir(parents=True, exist_ok=True)
+        return session_output_dir
+
+    def get_talk_temp_dir(self) -> Path:
+        """
+        Get the talks temporary directory for the current session.
+
+        Returns:
+            Path: Path to the session's talks temporary directory
+        """
+        talk_temp_dir = self.get_temp_dir() / "talks"
+        talk_temp_dir.mkdir(parents=True, exist_ok=True)
+        return talk_temp_dir