109 lines
3.9 KiB
Python
109 lines
3.9 KiB
Python
import os
|
|
import json
|
|
import time
|
|
import sqlite3
|
|
import requests
|
|
from typing import Dict, Any
|
|
from rapidfuzz import fuzz
|
|
|
|
class MusicBrainzVerifier:
|
|
"""
|
|
Cross-references artist names against MusicBrainz with local SQLite caching.
|
|
Free, no API key required (Rate limit: 1 request/second).
|
|
"""
|
|
BASE_URL = "https://musicbrainz.org/ws/2/artist"
|
|
|
|
def __init__(self, db_path: str = ".cache/verifiers_cache.sqlite", app_name: str = "HiffiArtistIntelligence", version: str = "1.0"):
|
|
self.headers = {
|
|
"User-Agent": f"{app_name}/{version} ( contact@superlabs.co )"
|
|
}
|
|
self.last_call_time = 0.0
|
|
self.db_path = db_path
|
|
os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
|
|
self._init_db()
|
|
|
|
def _init_db(self):
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
conn.execute("""
|
|
CREATE TABLE IF NOT EXISTS musicbrainz_cache (
|
|
artist_name TEXT PRIMARY KEY,
|
|
data_json TEXT,
|
|
cached_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
|
|
def _get_cache(self, name: str) -> Dict[str, Any]:
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT data_json FROM musicbrainz_cache WHERE artist_name = ?", (name.lower(),))
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return json.loads(row[0])
|
|
return None
|
|
|
|
def _set_cache(self, name: str, data: Dict[str, Any]):
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
conn.execute(
|
|
"INSERT OR REPLACE INTO musicbrainz_cache (artist_name, data_json) VALUES (?, ?)",
|
|
(name.lower(), json.dumps(data))
|
|
)
|
|
|
|
def _rate_limit(self):
|
|
elapsed = time.time() - self.last_call_time
|
|
if elapsed < 1.0:
|
|
time.sleep(1.0 - elapsed)
|
|
self.last_call_time = time.time()
|
|
|
|
def verify_artist(self, name: str) -> Dict[str, Any]:
|
|
if not name or not name.strip():
|
|
return {"verified": False, "confidence": 0.0, "reason": "empty_name"}
|
|
|
|
cleaned_name = name.strip()
|
|
|
|
# Check local cache first
|
|
cached = self._get_cache(cleaned_name)
|
|
if cached:
|
|
return cached
|
|
|
|
self._rate_limit()
|
|
query = f'artist:"{cleaned_name}"'
|
|
params = {"query": query, "fmt": "json", "limit": 3}
|
|
|
|
try:
|
|
resp = requests.get(self.BASE_URL, headers=self.headers, params=params, timeout=10)
|
|
if resp.status_code != 200:
|
|
result = {"verified": False, "confidence": 0.0, "status_code": resp.status_code}
|
|
return result
|
|
|
|
data = resp.json()
|
|
artists = data.get("artists", [])
|
|
if not artists:
|
|
result = {"verified": False, "confidence": 0.0, "reason": "not_found"}
|
|
self._set_cache(cleaned_name, result)
|
|
return result
|
|
|
|
best_match = artists[0]
|
|
mb_name = best_match.get("name", "")
|
|
match_score = fuzz.token_sort_ratio(cleaned_name.lower(), mb_name.lower())
|
|
|
|
confidence = round(match_score / 100.0, 2)
|
|
is_verified = confidence >= 0.80
|
|
|
|
result = {
|
|
"verified": is_verified,
|
|
"confidence": confidence,
|
|
"mbid": best_match.get("id"),
|
|
"matched_name": mb_name,
|
|
"disambiguation": best_match.get("disambiguation", ""),
|
|
"country": best_match.get("country", ""),
|
|
"type": best_match.get("type", "")
|
|
}
|
|
self._set_cache(cleaned_name, result)
|
|
return result
|
|
except Exception as e:
|
|
return {"verified": False, "confidence": 0.0, "error": str(e)}
|
|
|
|
if __name__ == "__main__":
|
|
verifier = MusicBrainzVerifier()
|
|
print("Testing MusicBrainz Caching...")
|
|
print(verifier.verify_artist("Kendrick Lamar")) |