Refactoring

Author: MrDesjardins

CLAUDE.md

@@ -45,23 +45,33 @@ Audio Stream Server is a FastAPI application that streams audio from YouTube vid
 
 ### Path Handling
 
-**CRITICAL: Always use .expanduser().resolve() when creating Path objects from user-provided or config paths.**
+**CRITICAL: Always use path_utils helpers for handling file paths.**
 
-- ALWAYS call `.expanduser().resolve()` on Path objects to handle `~` (home directory) expansion and resolve symbolic links
-- This is required for paths that come from:
+- Use `expand_path()` and `expand_path_str()` from `services.path_utils` for all user/config paths
+- `expand_path(path)` - Returns Path object with ~ expansion and symbolic link resolution
+- `expand_path_str(path)` - Returns string (for external commands like ffmpeg, yt-dlp)
+- This is required for paths from:
   - Configuration files (environment variables, .env)
   - User input
   - Function parameters that represent file paths
-- Temporary file paths (from `tempfile.mkstemp()`, etc.) do NOT need `.expanduser().resolve()`
+- Temporary file paths (from `tempfile.mkstemp()`, etc.) do NOT need expansion
 - Example:
   ```python
+  from services.path_utils import expand_path, expand_path_str
+
   # CORRECT - user/config paths
-  audio_path = Path(config.temp_audio_dir).expanduser().resolve()
-  file_size = Path(audio_path_param).expanduser().resolve().stat().st_size
+  audio_path = expand_path(config.temp_audio_dir)
+  file_size = expand_path(audio_path_param).stat().st_size
+
+  # CORRECT - external commands
+  subprocess.run(["ffmpeg", "-i", expand_path_str(audio_path), ...])
 
   # CORRECT - temporary file paths (no expansion needed)
   temp_fd, temp_path = tempfile.mkstemp()
-  temp_file = Path(temp_path)  # No .expanduser().resolve() needed
+  temp_file = Path(temp_path)  # No expansion needed
+
+  # WRONG - manual expansion (use helper instead)
+  audio_path = Path(config.temp_audio_dir).expanduser().resolve()  # ❌ Use expand_path()
 
   # WRONG - missing expansion for user path
   audio_path = Path(config.temp_audio_dir)  # ❌ May fail if path contains ~

migrate_add_queue_columns.py

@@ -9,11 +9,17 @@
 import sys
 from pathlib import Path
 
+# Add parent directory to path for imports
+
+sys.path.insert(0, str(Path(__file__).parent))
+
+from services.path_utils import expand_path
+
 
 def migrate_queue_table() -> None:
     """Add type and week_year columns to queue table if they don't exist."""
     db_path_str = os.getenv("DATABASE_PATH", "./audio_history.db")
-    db_path = Path(db_path_str).expanduser().resolve()
+    db_path = expand_path(db_path_str)
 
     print(f"Migrating database: {db_path}")
 

routes/stream.py

@@ -3,14 +3,14 @@
 """
 
 import logging
-from pathlib import Path
 import threading
 from fastapi import APIRouter, HTTPException
 from fastapi.responses import FileResponse, JSONResponse
 from pydantic import BaseModel
 from config import get_config
 from services.background_tasks import get_transcription_queue, TranscriptionJob
 from services.database import add_to_history, get_history, clear_history
+from services.path_utils import expand_path
 from services.youtube import get_video_metadata, extract_video_id
 
 logger = logging.getLogger(__name__)
@@ -162,14 +162,14 @@ def _audio_is_ready(video_id: str) -> bool:
     """Check if the audio file exists and is not still being downloaded."""
     from services.streaming import is_download_in_progress
 
-    audio_path = Path(config.get_audio_path(video_id)).expanduser().resolve()
+    audio_path = expand_path(config.get_audio_path(video_id))
     return audio_path.exists() and not is_download_in_progress(video_id)
 
 
 @router.get("/audio/{video_id}")
 def get_audio_file(video_id: str):
     """Serve the actual MP3 file for the player with mobile-optimized headers."""
-    audio_path = Path(config.get_audio_path(video_id)).expanduser().resolve()
+    audio_path = expand_path(config.get_audio_path(video_id))
 
     if _audio_is_ready(video_id):
         file_size = audio_path.stat().st_size
@@ -198,7 +198,7 @@ def get_audio_file(video_id: str):
 @router.head("/audio/{video_id}")
 def check_audio_file(video_id: str):
     """Check if audio file exists and is ready (for polling). HEAD request."""
-    audio_path = Path(config.get_audio_path(video_id)).expanduser().resolve()
+    audio_path = expand_path(config.get_audio_path(video_id))
 
     if _audio_is_ready(video_id):
         file_size = audio_path.stat().st_size

routes/weekly_summaries.py

@@ -1,7 +1,6 @@
 """API routes for weekly summaries."""
 
 import logging
-from pathlib import Path
 from fastapi import APIRouter, HTTPException
 from fastapi.responses import FileResponse
 from typing import List, Dict
@@ -11,6 +10,7 @@
     get_summary_by_week_year,
     add_summary_to_queue,
 )
+from services.path_utils import expand_path
 
 logger = logging.getLogger(__name__)
 
@@ -63,7 +63,7 @@ async def stream_summary_audio(week_year: str):
             )
 
         # Check if file exists
-        if not Path(audio_path).expanduser().resolve().exists():
+        if not expand_path(audio_path).exists():
             logger.error(f"Audio file not found: {audio_path}")
             raise HTTPException(
                 status_code=404, detail=f"Audio file not found: {audio_path}"

services/background_tasks.py

@@ -1,7 +1,6 @@
 """Background task processing for audio transcription."""
 
 import logging
-from pathlib import Path
 import threading
 import time
 import random
@@ -12,6 +11,7 @@
 from typing import Optional, Dict
 
 from services.cache import get_transcript_cache, get_audio_cache
+from services.path_utils import expand_path
 
 logger = logging.getLogger(__name__)
 
@@ -413,7 +413,7 @@ def _wait_for_file(
         start_time = time.time()
 
         while time.time() - start_time < timeout:
-            path = Path(audio_path).expanduser().resolve()
+            path = expand_path(audio_path)
             file_exists = path.exists()
             still_downloading = is_download_in_progress(video_id)
 

services/cache.py

@@ -9,6 +9,7 @@
 from datetime import datetime
 
 from config import get_config
+from services.path_utils import expand_path
 
 logger = logging.getLogger(__name__)
 
@@ -22,7 +23,7 @@ class TranscriptionCache:
     """Manages caching of transcripts and summaries."""
 
     def __init__(self, cache_dir: str = "/tmp/transcription-cache"):
-        self.cache_dir = Path(cache_dir).expanduser().resolve()
+        self.cache_dir = expand_path(cache_dir)
         self.cache_dir.mkdir(parents=True, exist_ok=True)
         self._lock = threading.Lock()
         logger.info(f"Transcription cache initialized at {self.cache_dir}")
@@ -106,7 +107,7 @@ class AudioCache:
     def __init__(self, max_files: int = 10):
         self.max_files = max_files
         config = get_config()
-        self.audio_dir = Path(config.temp_audio_dir).expanduser().resolve()
+        self.audio_dir = expand_path(config.temp_audio_dir)
         self.audio_dir.mkdir(parents=True, exist_ok=True)
         logger.info(
             f"Audio cache initialized: max {max_files} files in {self.audio_dir}"

services/path_utils.py

@@ -0,0 +1,51 @@
+"""
+Path utilities for handling file paths with proper expansion.
+
+Provides helper functions for expanding user paths (~) and resolving
+symbolic links consistently across the codebase.
+"""
+
+from pathlib import Path
+from typing import Union
+
+
+def expand_path(path: Union[str, Path]) -> Path:
+    """
+    Expand user path (~) and resolve to absolute path.
+
+    This function handles:
+    - Tilde (~) expansion to home directory
+    - Resolving symbolic links
+    - Converting to absolute path
+
+    Args:
+        path: Path as string or Path object
+
+    Returns:
+        Fully expanded and resolved Path object
+
+    Example:
+        >>> expand_path("~/documents/file.txt")
+        PosixPath('/home/user/documents/file.txt')
+    """
+    return Path(path).expanduser().resolve()
+
+
+def expand_path_str(path: Union[str, Path]) -> str:
+    """
+    Expand user path (~) and resolve to absolute path string.
+
+    Same as expand_path() but returns a string instead of Path object.
+    Useful for external commands (ffmpeg, yt-dlp) that don't handle ~ paths.
+
+    Args:
+        path: Path as string or Path object
+
+    Returns:
+        Fully expanded and resolved path as string
+
+    Example:
+        >>> expand_path_str("~/documents/file.txt")
+        '/home/user/documents/file.txt'
+    """
+    return str(expand_path(path))

services/streaming.py

@@ -8,8 +8,8 @@
 import os
 import subprocess
 
-from pathlib import Path
 from services.cache import get_audio_cache
+from services.path_utils import expand_path
 from config import get_config
 
 logger = logging.getLogger(__name__)
@@ -23,7 +23,7 @@ def _get_download_marker(youtube_video_id: str) -> str:
 
 def is_download_in_progress(youtube_video_id: str) -> bool:
     """Check if a download is currently in progress for this video."""
-    return Path(_get_download_marker(youtube_video_id)).expanduser().resolve().exists()
+    return expand_path(_get_download_marker(youtube_video_id)).exists()
 
 
 def start_youtube_download(youtube_video_id: str):
@@ -41,7 +41,7 @@ def start_youtube_download(youtube_video_id: str):
         proc or None if already cached
     """
     audio_cache = get_audio_cache()
-    audio_path = Path(config.get_audio_path(youtube_video_id)).expanduser().resolve()
+    audio_path = expand_path(config.get_audio_path(youtube_video_id))
 
     if audio_cache.check_file_exists(youtube_video_id):
         logger.info(f"Audio file for video {youtube_video_id} already exists in cache")
@@ -51,7 +51,7 @@ def start_youtube_download(youtube_video_id: str):
     url = f"https://www.youtube.com/watch?v={youtube_video_id}"
 
     # Create marker file so the /audio endpoint won't serve a partial file
-    marker_path = Path(_get_download_marker(youtube_video_id)).expanduser().resolve()
+    marker_path = expand_path(_get_download_marker(youtube_video_id))
     try:
         marker_path.touch(exist_ok=True)
     except Exception as e:
@@ -115,11 +115,9 @@ def finish_youtube_download(youtube_video_id: str, returncode: int):
         youtube_video_id: YouTube video ID
         returncode: Process exit code (0 = success)
     """
-    audio_path = Path(config.get_audio_path(youtube_video_id)).expanduser().resolve()
-    marker_path = Path(_get_download_marker(youtube_video_id)).expanduser().resolve()
-    stderr_path = (
-        Path(config.get_audio_path(youtube_video_id) + ".err").expanduser().resolve()
-    )
+    audio_path = expand_path(config.get_audio_path(youtube_video_id))
+    marker_path = expand_path(_get_download_marker(youtube_video_id))
+    stderr_path = expand_path(config.get_audio_path(youtube_video_id) + ".err")
 
     # Read stderr for error reporting
     error_output = ""

services/transcription.py

@@ -10,6 +10,7 @@
 
 from config import get_config
 from services.api_clients import get_openai_client
+from services.path_utils import expand_path, expand_path_str
 
 logger = logging.getLogger(__name__)
 
@@ -37,7 +38,7 @@ def compress_audio_for_whisper(audio_path: str) -> str:
         Exception: If compression fails
     """
     # Expand ~ in path for ffmpeg (ffmpeg doesn't handle ~ paths)
-    expanded_audio_path = str(Path(audio_path).expanduser().resolve())
+    expanded_audio_path = expand_path_str(audio_path)
 
     logger.info(
         f"Compressing audio file {expanded_audio_path} for Whisper (1.5x speed, saves API costs)"
@@ -135,7 +136,7 @@ def transcribe_audio_openai(audio_path: str, retries: int = 3) -> str:
     client = get_openai_client()
 
     # Always compress audio to save costs and reduce upload time
-    file_size = Path(audio_path).expanduser().resolve().stat().st_size
+    file_size = expand_path(audio_path).stat().st_size
     logger.info(
         f"Audio file size: {file_size / 1024 / 1024:.2f}MB - "
         f"compressing for Whisper (saves 33% API costs)"
@@ -222,7 +223,7 @@ def transcribe_audio_gemini(audio_path: str, retries: int = 3) -> str:
         raise ValueError("Gemini API key not configured")
 
     # Expand ~ in path (Python's open() doesn't handle ~ paths)
-    expanded_audio_path = str(Path(audio_path).expanduser().resolve())
+    expanded_audio_path = expand_path_str(audio_path)
 
     last_error = None
 

services/tts.py

@@ -5,11 +5,12 @@
 """
 
 import re
-from pathlib import Path
 from typing import Optional
 from elevenlabs.client import ElevenLabs
 from bs4 import BeautifulSoup
 
+from services.path_utils import expand_path
+
 
 class TTSError(Exception):
     """Base exception for TTS-related errors."""
@@ -178,7 +179,7 @@ def save_audio_file(audio_data: bytes, file_path: str) -> int:
     Raises:
         IOError: If file cannot be written
     """
-    path = Path(file_path).expanduser().resolve()
+    path = expand_path(file_path)
 
     # Create parent directory if needed
     path.parent.mkdir(parents=True, exist_ok=True)
@@ -205,7 +206,7 @@ def get_audio_duration(file_path: str) -> Optional[int]:
     Returns:
         Duration in seconds, or None if file doesn't exist
     """
-    path = Path(file_path).expanduser().resolve()
+    path = expand_path(file_path)
     if not path.exists():
         return None