386 lines
15 KiB
Python
386 lines
15 KiB
Python
"""
|
|
Drop Folder Monitor - Watches a folder for new music files and imports them
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
import asyncio
|
|
import threading
|
|
import shutil
|
|
from typing import Optional, Dict
|
|
from mutagen import File as MutagenFile
|
|
from watchdog.observers import Observer
|
|
from watchdog.events import FileSystemEventHandler
|
|
|
|
from utils.logging_config import get_logger
|
|
from config.settings import config_manager
|
|
|
|
logger = get_logger("drop_folder_monitor")
|
|
|
|
|
|
class DropFolderHandler(FileSystemEventHandler):
|
|
"""Handles file system events for the drop folder"""
|
|
|
|
def __init__(self, processor):
|
|
self.processor = processor
|
|
self.supported_formats = config_manager.get('drop_folder.supported_formats',
|
|
['.mp3', '.flac', '.ogg', '.aac', '.wma', '.wav', '.m4a'])
|
|
|
|
def on_created(self, event):
|
|
"""Handle new file creation"""
|
|
if event.is_directory:
|
|
return
|
|
|
|
file_path = Path(event.src_path)
|
|
|
|
# Check if it's a supported audio file
|
|
if file_path.suffix.lower() in self.supported_formats:
|
|
logger.info(f"New file detected: {file_path.name}")
|
|
self.processor.queue_file(file_path)
|
|
|
|
|
|
class DropFolderProcessor:
|
|
"""Processes audio files dropped into the watch folder"""
|
|
|
|
def __init__(self):
|
|
self.processing_queue = []
|
|
self._queue_lock = threading.Lock()
|
|
self.watch_path = None
|
|
self._observer = None
|
|
self._running = False
|
|
self._spotify_client = None
|
|
self._processor_thread = None
|
|
self._seen_files = set()
|
|
|
|
def _get_spotify_client(self):
|
|
"""Lazy load Spotify client only when needed"""
|
|
if self._spotify_client is None:
|
|
try:
|
|
from core.spotify_client import SpotifyClient
|
|
self._spotify_client = SpotifyClient()
|
|
if not self._spotify_client.is_authenticated():
|
|
logger.warning("Spotify not authenticated - metadata lookup disabled")
|
|
self._spotify_client = False # Mark as unavailable
|
|
except Exception as e:
|
|
logger.warning(f"Could not initialize Spotify client: {e}")
|
|
self._spotify_client = False
|
|
return self._spotify_client if self._spotify_client else None
|
|
|
|
def start(self):
|
|
"""Start monitoring the drop folder"""
|
|
watch_path = config_manager.get('drop_folder.watch_path')
|
|
enabled = config_manager.get('drop_folder.enabled', False)
|
|
|
|
if not enabled:
|
|
logger.info("Drop folder monitoring is disabled")
|
|
return False
|
|
|
|
if not watch_path or not os.path.exists(watch_path):
|
|
logger.warning(f"Drop folder path not configured or doesn't exist: {watch_path}")
|
|
return False
|
|
|
|
self.watch_path = Path(watch_path)
|
|
self._running = True
|
|
|
|
# Set up file system watcher
|
|
event_handler = DropFolderHandler(self)
|
|
self._observer = Observer()
|
|
self._observer.schedule(event_handler, str(self.watch_path), recursive=False)
|
|
self._observer.start()
|
|
|
|
logger.info(f"Drop folder monitoring started: {self.watch_path}")
|
|
|
|
self._scan_existing_files()
|
|
|
|
return True
|
|
|
|
def stop(self):
|
|
"""Stop monitoring"""
|
|
self._running = False
|
|
if self._observer:
|
|
self._observer.stop()
|
|
self._observer.join()
|
|
self._observer = None
|
|
if hasattr(self, '_processor_thread') and self._processor_thread:
|
|
self._processor_thread.join(timeout=2)
|
|
self._processor_thread = None
|
|
logger.info("Drop folder monitoring stopped")
|
|
|
|
def reload_config(self):
|
|
"""Reload configuration and restart if settings changed"""
|
|
enabled = config_manager.get('drop_folder.enabled', False)
|
|
watch_path = config_manager.get('drop_folder.watch_path')
|
|
|
|
# Check if we need to stop
|
|
if not enabled:
|
|
if self._observer:
|
|
logger.info("Drop folder disabled - stopping monitor")
|
|
self.stop()
|
|
return
|
|
|
|
# Check if we need to start or restart
|
|
new_path = Path(watch_path) if watch_path else None
|
|
needs_restart = self._observer is None or new_path != self.watch_path
|
|
|
|
if needs_restart:
|
|
logger.info("Drop folder config changed - (re)starting monitor")
|
|
self.stop()
|
|
if self.start():
|
|
self.start_background_processing()
|
|
|
|
def start_background_processing(self):
|
|
"""Start the async queue processor in a background thread.
|
|
Call this after start() returns True."""
|
|
def run_async_loop():
|
|
asyncio.run(self._run_async_tasks())
|
|
|
|
self._processor_thread = threading.Thread(target=run_async_loop, daemon=True)
|
|
self._processor_thread.start()
|
|
logger.info("Background queue processor started")
|
|
|
|
async def _run_async_tasks(self):
|
|
await asyncio.gather(
|
|
self.process_queue(),
|
|
self.poll_folder(interval=15)
|
|
)
|
|
|
|
def _scan_existing_files(self):
|
|
"""Scan for existing files in the drop folder"""
|
|
supported_formats = config_manager.get('drop_folder.supported_formats',
|
|
['.mp3', '.flac', '.ogg', '.aac', '.wma', '.wav', '.m4a'])
|
|
|
|
for file_path in self.watch_path.glob('*'):
|
|
if file_path.is_file() and file_path.suffix.lower() in supported_formats:
|
|
self.queue_file(file_path)
|
|
|
|
logger.info(f"Found {len(self.processing_queue)} existing files to process")
|
|
|
|
def queue_file(self, file_path: Path):
|
|
"""Add file to processing queue"""
|
|
with self._queue_lock:
|
|
if file_path not in self.processing_queue:
|
|
self.processing_queue.append(file_path)
|
|
self._seen_files.add(file_path)
|
|
logger.debug(f"Queued for processing: {file_path.name}")
|
|
|
|
async def poll_folder(self, interval: int = 15):
|
|
""""Fallback polling for docker mounts (watchdog not deterministic)"""
|
|
supported_formats = config_manager.get(
|
|
'drop_folder.supported_formats',
|
|
['.mp3', '.flac', '.ogg', '.aac', '.wma', '.wav', '.m4a']
|
|
)
|
|
|
|
while self._running:
|
|
try:
|
|
for file_path in self.watch_path.glob('*'):
|
|
if (
|
|
file_path.is_file()
|
|
and file_path.suffix.lower() in supported_formats
|
|
and file_path not in self._seen_files
|
|
):
|
|
logger.info(f"[Polling] Detected new file: {file_path.name}")
|
|
self.queue_file(file_path)
|
|
except Exception as e:
|
|
logger.warning(f"Drop folder polling error: {e}")
|
|
|
|
await asyncio.sleep(interval)
|
|
|
|
async def process_queue(self):
|
|
"""Process queued files"""
|
|
while self._running:
|
|
file_path = None
|
|
with self._queue_lock:
|
|
if self.processing_queue:
|
|
file_path = self.processing_queue.pop(0)
|
|
if file_path:
|
|
await self._process_file(file_path)
|
|
else:
|
|
await asyncio.sleep(5)
|
|
|
|
def _read_file_metadata(self, file_path: Path) -> dict:
|
|
"""Read metadata from audio file using mutagen"""
|
|
try:
|
|
audio = MutagenFile(file_path, easy=True)
|
|
if audio is None:
|
|
return self._parse_filename_metadata(file_path)
|
|
|
|
return {
|
|
'title': audio.get('title', [''])[0] or file_path.stem,
|
|
'artist': audio.get('artist', [''])[0] or 'Unknown Artist',
|
|
'album': audio.get('album', [''])[0] if audio.get('album') else None,
|
|
'genre': audio.get('genre', [''])[0] if audio.get('genre') else None,
|
|
'date': audio.get('date', [''])[0] if audio.get('date') else None,
|
|
}
|
|
except Exception as e:
|
|
logger.warning(f"Could not read metadata from {file_path.name}: {e}")
|
|
return self._parse_filename_metadata(file_path)
|
|
|
|
def _parse_filename_metadata(self, file_path: Path) -> dict:
|
|
"""Fallback: parse metadata from filename (Artist - Title.ext)"""
|
|
stem = file_path.stem
|
|
if ' - ' in stem:
|
|
parts = stem.split(' - ', 1)
|
|
return {'artist': parts[0].strip(), 'title': parts[1].strip(), 'album': None}
|
|
return {'title': stem, 'artist': 'Unknown Artist', 'album': None}
|
|
|
|
def _lookup_spotify_metadata(self, artist: str, title: str) -> Optional[Dict]:
|
|
"""Try to find better metadata from Spotify"""
|
|
spotify = self._get_spotify_client()
|
|
if not spotify:
|
|
return None
|
|
|
|
try:
|
|
# Search Spotify with artist and title
|
|
query = f"artist:{artist} track:{title}"
|
|
tracks = spotify.search_tracks(query, limit=5)
|
|
|
|
if not tracks:
|
|
# Try simpler search
|
|
query = f"{artist} {title}"
|
|
tracks = spotify.search_tracks(query, limit=5)
|
|
|
|
if tracks:
|
|
# Find best match (first result is usually best)
|
|
best_track = tracks[0]
|
|
|
|
# Basic validation - check if artist name is somewhat similar
|
|
spotify_artist = best_track.artists[0] if best_track.artists else ''
|
|
if self._similarity(artist.lower(), spotify_artist.lower()) > 0.5:
|
|
logger.info(f" Spotify match: {best_track.name} by {spotify_artist}")
|
|
return {
|
|
'title': best_track.name,
|
|
'artist': spotify_artist,
|
|
'album': best_track.album_name,
|
|
'spotify_id': best_track.id,
|
|
}
|
|
else:
|
|
logger.debug(f" Spotify result '{spotify_artist}' didn't match '{artist}'")
|
|
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Spotify lookup failed: {e}")
|
|
return None
|
|
|
|
def _similarity(self, a: str, b: str) -> float:
|
|
"""Simple similarity check between two strings"""
|
|
if not a or not b:
|
|
return 0.0
|
|
if a == b:
|
|
return 1.0
|
|
# Check if one contains the other
|
|
if a in b or b in a:
|
|
return 0.8
|
|
# Count common words
|
|
words_a = set(a.split())
|
|
words_b = set(b.split())
|
|
if not words_a or not words_b:
|
|
return 0.0
|
|
common = words_a & words_b
|
|
return len(common) / max(len(words_a), len(words_b))
|
|
|
|
def _sanitize_filename(self, filename: str) -> str:
|
|
"""Sanitize filename for file system compatibility"""
|
|
# Replace invalid characters with underscores
|
|
sanitized = re.sub(r'[<>:"/\\|?*]', '_', filename)
|
|
# Remove multiple spaces and trim
|
|
sanitized = re.sub(r'\s+', ' ', sanitized).strip()
|
|
# Limit length to avoid filesystem issues
|
|
return sanitized[:200] if len(sanitized) > 200 else sanitized
|
|
|
|
async def _wait_for_file_stable(self, file_path: Path, timeout: int = 60) -> bool:
|
|
"""Wait for file to finish being written (size stops changing)"""
|
|
if not file_path.exists():
|
|
return False
|
|
|
|
last_size = -1
|
|
stable_count = 0
|
|
elapsed = 0
|
|
|
|
while elapsed < timeout:
|
|
try:
|
|
current_size = file_path.stat().st_size
|
|
if current_size == last_size and current_size > 0:
|
|
stable_count += 1
|
|
if stable_count >= 2: # Size stable for 2 checks
|
|
return True
|
|
else:
|
|
stable_count = 0
|
|
last_size = current_size
|
|
except OSError:
|
|
return False
|
|
|
|
await asyncio.sleep(0.5)
|
|
elapsed += 0.5
|
|
|
|
logger.warning(f"Timeout waiting for file to stabilize: {file_path.name}")
|
|
return False
|
|
|
|
async def _process_file(self, file_path: Path):
|
|
"""Process a single audio file - organize into transfer folder structure"""
|
|
try:
|
|
if not file_path.exists():
|
|
logger.warning(f"File no longer exists {file_path}")
|
|
return
|
|
|
|
# Wait for file to finish being copied/written
|
|
if not await self._wait_for_file_stable(file_path):
|
|
logger.warning(f"Skipping unstable file: {file_path.name}")
|
|
return
|
|
|
|
logger.info(f"Processing: {file_path.name}")
|
|
|
|
# Read metadata from file
|
|
metadata = self._read_file_metadata(file_path)
|
|
title = metadata.get('title', 'Unknown')
|
|
artist = metadata.get('artist', 'Unknown Artist')
|
|
album = metadata.get('album')
|
|
|
|
logger.info(f" File metadata - Title: {title}, Artist: {artist}, Album: {album or 'None'}")
|
|
|
|
# Try to enhance metadata with Spotify lookup (automatically skipped if not authenticated)
|
|
spotify_meta = self._lookup_spotify_metadata(artist, title)
|
|
if spotify_meta:
|
|
title = spotify_meta.get('title', title)
|
|
artist = spotify_meta.get('artist', artist)
|
|
album = spotify_meta.get('album', album)
|
|
logger.info(f" Enhanced metadata - Title: {title}, Artist: {artist}, Album: {album or 'None'}")
|
|
|
|
# Get transfer path from config
|
|
transfer_path = config_manager.get('soulseek.transfer_path', './Transfer')
|
|
transfer_dir = Path(transfer_path)
|
|
transfer_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Create artist directory
|
|
artist_dir = transfer_dir / self._sanitize_filename(artist)
|
|
artist_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Determine folder structure based on album info
|
|
file_ext = file_path.suffix
|
|
|
|
if album and album not in ['Unknown Album', 'None', '']:
|
|
# Album track: Transfer/Artist/Artist - Album/Title.ext
|
|
album_folder = f"{self._sanitize_filename(artist)} - {self._sanitize_filename(album)}"
|
|
album_dir = artist_dir / album_folder
|
|
album_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
dest_filename = f"{self._sanitize_filename(title)}{file_ext}"
|
|
dest_path = album_dir / dest_filename
|
|
logger.info(f" -> Album track: {artist}/{album_folder}/{dest_filename}")
|
|
else:
|
|
# Single track: Transfer/Artist/Title.ext (no subfolder)
|
|
dest_filename = f"{self._sanitize_filename(title)}{file_ext}"
|
|
dest_path = artist_dir / dest_filename
|
|
logger.info(f" -> Single: {artist}/{dest_filename}")
|
|
|
|
# Handle existing file
|
|
if dest_path.exists():
|
|
dest_path.unlink()
|
|
|
|
# Move file to organized location
|
|
shutil.move(str(file_path), str(dest_path))
|
|
logger.info(f" Moved to: {dest_path}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error processing {file_path.name}: {e}")
|