add cache and watcher and librosa

Author: formeo

pyproject.toml

@@ -1,91 +1,62 @@
 [build-system]
-requires = ["setuptools>=61.0", "wheel"]
-build-backend = "setuptools.build_meta"
+requires = ["hatchling"]
+build-backend = "hatchling.build"
 
 [project]
-name = "music-recognition"
+name = "music-recognition-tool"
 version = "1.0.0"
-description = "Identify unknown music files using Shazam, write ID3 tags, rename and organize"
+description = "Identify music files using Shazam, write ID3 tags, rename and organize"
 readme = "README.md"
-license = {text = "MIT"}
+license = "MIT"
 authors = [
-    {name = "formeo", email = "formeo@example.com"}
-]
-keywords = [
-    "music",
-    "recognition", 
-    "shazam",
-    "id3",
-    "tags",
-    "mp3",
-    "audio",
-    "fingerprinting",
-    "organize",
-    "rename",
+    { name = "Roman Gordienko", email = "grlformeo@gmail.com" }
 ]
+keywords = ["music", "recognition", "shazam", "id3", "mp3", "tags"]
 classifiers = [
     "Development Status :: 4 - Beta",
     "Environment :: Console",
     "Intended Audience :: End Users/Desktop",
-    "Intended Audience :: Developers",
     "License :: OSI Approved :: MIT License",
     "Operating System :: OS Independent",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
     "Topic :: Multimedia :: Sound/Audio",
     "Topic :: Multimedia :: Sound/Audio :: Analysis",
 ]
-requires-python = ">=3.9"
+requires-python = ">=3.10"
 dependencies = [
     "shazamio>=0.6.0",
     "mutagen>=1.47.0",
     "pydub>=0.25.1",
-    "aiohttp>=3.9.0",
-    "aiofiles>=23.0.0",
 ]
 
 [project.optional-dependencies]
+analyze = ["librosa>=0.10.0", "numpy>=1.24.0"]
 dev = [
     "pytest>=7.0.0",
     "pytest-asyncio>=0.21.0",
-    "pytest-cov>=4.0.0",
-    "black>=23.0.0",
-    "isort>=5.12.0",
+    "ruff>=0.1.0",
     "mypy>=1.0.0",
-    "flake8>=6.0.0",
 ]
 
 [project.scripts]
 music-recognize = "music_recognition.cli:main"
 
 [project.urls]
 Homepage = "https://github.com/formeo/music_recognition"
-Documentation = "https://github.com/formeo/music_recognition#readme"
 Repository = "https://github.com/formeo/music_recognition"
 Issues = "https://github.com/formeo/music_recognition/issues"
-
-[tool.setuptools.packages.find]
-where = ["src"]
-
-[tool.pytest.ini_options]
-asyncio_mode = "auto"
-testpaths = ["tests"]
-python_files = ["test_*.py"]
-
-[tool.black]
-line-length = 100
-target-version = ['py39', 'py310', 'py311', 'py312']
-
-[tool.isort]
-profile = "black"
-line_length = 100
-src_paths = ["src"]
-
-[tool.mypy]
-python_version = "3.9"
-warn_return_any = true
-warn_unused_configs = true
-ignore_missing_imports = true
+Changelog = "https://github.com/formeo/music_recognition/blob/main/CHANGELOG.md"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/music_recognition"]
+
+[tool.hatch.build.targets.sdist]
+include = [
+    "/src",
+    "/tests",
+    "/README.md",
+    "/LICENSE",
+]

requirements.txt

@@ -4,3 +4,5 @@ mutagen>=1.47.0
 pydub>=0.25.1
 aiohttp>=3.9.0
 aiofiles>=23.0.0
+librosa==0.11.0
+watchdog==6.0.0

src/music_recognition/__init__.py

@@ -1,12 +1,8 @@
-"""
-Music Recognition - Identify, tag, rename and organize your music collection.
+# src/music_recognition/__init__.py
+"""Music Recognition - Identify, tag, rename and organize music files."""
 
-Usage:
-    from music_recognition import MusicRecognizer
-    
-    recognizer = MusicRecognizer()
-    stats = await recognizer.process_directory("/music", rename=True)
-"""
+__version__ = "1.1.0"
+__author__ = "formeo"
 
 from .core import (
     MusicRecognizer,
@@ -15,17 +11,14 @@
     ProcessingStats,
     recognize_and_tag,
     setup_logging,
-    __version__,
-    __author__,
 )
 
 __all__ = [
-    'MusicRecognizer',
-    'TrackInfo',
-    'ProcessingResult',
-    'ProcessingStats',
-    'recognize_and_tag',
-    'setup_logging',
-    '__version__',
-    '__author__',
-]
+    "MusicRecognizer",
+    "TrackInfo", 
+    "ProcessingResult",
+    "ProcessingStats",
+    "recognize_and_tag",
+    "setup_logging",
+    "__version__",
+]

src/music_recognition/analyzer.py

@@ -0,0 +1,71 @@
+from dataclasses import dataclass
+from typing import Optional
+import numpy as np
+
+
+@dataclass
+class AudioAnalysis:
+    """Audio analysis results."""
+    bpm: float = 0.0
+    key: str = ""
+    energy: float = 0.0
+    duration: float = 0.0
+    loudness_db: float = 0.0
+
+    def to_comment(self) -> str:
+        """Format as ID3 comment."""
+        return f"BPM: {self.bpm:.0f} | Key: {self.key} | Energy: {self.energy:.2f}"
+
+
+class AudioAnalyzer:
+    """Analyze audio characteristics using librosa."""
+
+    KEY_NAMES = ['C', 'C#', 'D', 'D#', 'E', 'F', 'F#', 'G', 'G#', 'A', 'A#', 'B']
+    MODE_NAMES = ['minor', 'major']
+
+    def __init__(self):
+        try:
+            import librosa
+            self.librosa = librosa
+        except ImportError:
+            raise ImportError("librosa is required: pip install librosa")
+
+    def analyze(self, file_path: str) -> AudioAnalysis:
+        """Perform full audio analysis."""
+        y, sr = self.librosa.load(file_path, sr=22050, duration=60)  # First 60 sec
+
+        return AudioAnalysis(
+            bpm=self._detect_bpm(y, sr),
+            key=self._detect_key(y, sr),
+            energy=self._calculate_energy(y),
+            duration=self.librosa.get_duration(y=y, sr=sr),
+            loudness_db=self._calculate_loudness(y),
+        )
+
+    def _detect_bpm(self, y, sr) -> float:
+        """Detect tempo in BPM."""
+        tempo, _ = self.librosa.beat.beat_track(y=y, sr=sr)
+        return float(tempo[0]) if hasattr(tempo, '__iter__') else float(tempo)
+
+    def _detect_key(self, y, sr) -> str:
+        """Detect musical key."""
+        # Compute chromagram
+        chroma = self.librosa.feature.chroma_cqt(y=y, sr=sr)
+        chroma_mean = np.mean(chroma, axis=1)
+
+        # Find dominant pitch class
+        key_idx = int(np.argmax(chroma_mean))
+
+        # Simple major/minor detection based on intervals
+        # (more sophisticated methods exist)
+        return f"{self.KEY_NAMES[key_idx]}"
+
+    def _calculate_energy(self, y) -> float:
+        """Calculate average RMS energy."""
+        rms = self.librosa.feature.rms(y=y)
+        return float(np.mean(rms))
+
+    def _calculate_loudness(self, y) -> float:
+        """Calculate loudness in dB."""
+        rms = np.sqrt(np.mean(y ** 2))
+        return float(20 * np.log10(rms + 1e-10))

src/music_recognition/cache.py

@@ -0,0 +1,80 @@
+# В начало файла
+import hashlib
+import sqlite3
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Optional
+
+from music_recognition import TrackInfo
+
+
+class RecognitionCache:
+    """Cache for recognition results to avoid duplicate API calls."""
+
+    def __init__(self, db_path: str = "~/.music_recognition/cache.db"):
+        self.db_path = Path(db_path).expanduser()
+        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._init_db()
+
+    def _init_db(self):
+        with self._connect() as conn:
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS cache (
+                    audio_hash TEXT PRIMARY KEY,
+                    track_info TEXT NOT NULL,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                )
+            """)
+            conn.execute("""
+                CREATE INDEX IF NOT EXISTS idx_created_at ON cache(created_at)
+            """)
+
+    @contextmanager
+    def _connect(self):
+        conn = sqlite3.connect(self.db_path)
+        try:
+            yield conn
+            conn.commit()
+        finally:
+            conn.close()
+
+    def get(self, file_path: str) -> Optional[TrackInfo]:
+        """Get cached result by file hash."""
+        audio_hash = self._hash_audio(file_path)
+        with self._connect() as conn:
+            row = conn.execute(
+                "SELECT track_info FROM cache WHERE audio_hash = ?",
+                (audio_hash,)
+            ).fetchone()
+
+            if row:
+                import json
+                data = json.loads(row[0])
+                return TrackInfo(**data)
+        return None
+
+    def set(self, file_path: str, info: TrackInfo):
+        """Cache recognition result."""
+        import json
+        audio_hash = self._hash_audio(file_path)
+        with self._connect() as conn:
+            conn.execute(
+                "INSERT OR REPLACE INTO cache (audio_hash, track_info) VALUES (?, ?)",
+                (audio_hash, json.dumps(info.to_dict()))
+            )
+
+    def clear(self, older_than_days: int = 30):
+        """Remove old cache entries."""
+        with self._connect() as conn:
+            conn.execute(
+                "DELETE FROM cache WHERE created_at < datetime('now', ?)",
+                (f'-{older_than_days} days',)
+            )
+
+    @staticmethod
+    def _hash_audio(file_path: str, chunk_size: int = 1024 * 1024) -> str:
+        """Hash first ~1MB of audio file (enough for fingerprint)."""
+        hasher = hashlib.sha256()
+        with open(file_path, 'rb') as f:
+            hasher.update(f.read(chunk_size))
+        return hasher.hexdigest()

src/music_recognition/core.py

@@ -25,15 +25,12 @@
 from pydub import AudioSegment
 from shazamio import Shazam
 
-__version__ = "1.0.0"
+__version__ = "1.1.0"
 __author__ = "formeo"
 
-logger = logging.getLogger(__name__)
-
+from music_recognition.cache import RecognitionCache
 
-# ============================================================
-# Data Classes
-# ============================================================
+logger = logging.getLogger(__name__)
 
 @dataclass
 class TrackInfo:
@@ -149,6 +146,8 @@ def __init__(
         max_concurrent: int = 5,
         delay_between_requests: float = 0.5,
         retry_attempts: int = 2,
+        use_cache: bool = True,
+        cache_path: Optional[str] = None,
     ):
         """
         Initialize the recognizer.
@@ -162,10 +161,26 @@ def __init__(
         self.semaphore = asyncio.Semaphore(max_concurrent)
         self.delay = delay_between_requests
         self.retry_attempts = retry_attempts
+        self.cache = RecognitionCache(cache_path) if use_cache else None
 
     # ==================== Recognition ====================
-    
-    async def recognize_file(self, file_path: str) -> TrackInfo:
+
+    async def recognize_file(self, file_path: str, skip_cache: bool = False) -> TrackInfo:
+        """Recognize with cache support."""
+        if self.cache and not skip_cache:
+            cached = self.cache.get(file_path)
+            if cached:
+                logger.debug(f"Cache hit: {file_path}")
+                return cached
+
+        info = await self._recognize_shazam(file_path)
+
+        if self.cache and info.is_recognized:
+            self.cache.set(file_path, info)
+
+        return info
+
+    async def _recognize_shazam(self, file_path: str) -> TrackInfo:
         """
         Recognize a single audio file using Shazam.