80 lines
3.0 KiB
Python
80 lines
3.0 KiB
Python
|
"""
|
||
|
|
collectors/sync_spotify_complete.py
|
||
|
|
Spotify sync engine using credentials from .env.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import re
|
||
|
|
import time
|
||
|
|
import base64
|
||
|
|
import sqlite3
|
||
|
|
import logging
|
||
|
|
from pathlib import Path
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from typing import Dict, List, Optional
|
||
|
|
import requests
|
||
|
|
from dotenv import load_dotenv
|
||
|
|
|
||
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
|
|
load_dotenv(PROJECT_ROOT / ".env")
|
||
|
|
|
||
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
|
||
|
|
logger = logging.getLogger("SpotifySync")
|
||
|
|
|
||
|
|
DB_PATH = PROJECT_ROOT / "data" / "hiffi_catalog.db"
|
||
|
|
|
||
|
|
class SpotifySyncEngine:
|
||
|
|
def __init__(self, db_path: Path = DB_PATH):
|
||
|
|
self.db_path = db_path
|
||
|
|
self.client_id = os.getenv("SPOTIFY_CLIENT_ID", "").strip()
|
||
|
|
self.client_secret = os.getenv("SPOTIFY_CLIENT_SECRET", "").strip()
|
||
|
|
self.access_token = None
|
||
|
|
|
||
|
|
def authenticate(self) -> bool:
|
||
|
|
if not self.client_id or not self.client_secret:
|
||
|
|
logger.warning("Spotify credentials missing in .env")
|
||
|
|
return False
|
||
|
|
url = "https://accounts.spotify.com/api/token"
|
||
|
|
auth_bytes = f"{self.client_id}:{self.client_secret}".encode("utf-8")
|
||
|
|
headers = {
|
||
|
|
"Authorization": f"Basic {base64.b64encode(auth_bytes).decode('utf-8')}",
|
||
|
|
"Content-Type": "application/x-www-form-urlencoded"
|
||
|
|
}
|
||
|
|
data = {"grant_type": "client_credentials"}
|
||
|
|
try:
|
||
|
|
resp = requests.post(url, headers=headers, data=data, timeout=10)
|
||
|
|
if resp.status_code == 200:
|
||
|
|
self.access_token = resp.json().get("access_token")
|
||
|
|
return True
|
||
|
|
except Exception as e:
|
||
|
|
logger.error(f"Spotify auth error: {e}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
def fetch_batch(self, sp_ids: List[str]) -> Dict[str, Dict]:
|
||
|
|
results = {}
|
||
|
|
if not self.access_token or not sp_ids:
|
||
|
|
return results
|
||
|
|
headers = {"Authorization": f"Bearer {self.access_token}"}
|
||
|
|
for i in range(0, len(sp_ids), 50):
|
||
|
|
chunk = sp_ids[i:i + 50]
|
||
|
|
url = f"https://api.spotify.com/v1/artists?ids={','.join(chunk)}"
|
||
|
|
try:
|
||
|
|
resp = requests.get(url, headers=headers, timeout=10)
|
||
|
|
if resp.status_code == 200:
|
||
|
|
for art in resp.json().get("artists", []):
|
||
|
|
if art:
|
||
|
|
results[art["id"]] = {
|
||
|
|
"followers": art.get("followers", {}).get("total", 0),
|
||
|
|
"popularity": art.get("popularity", 0)
|
||
|
|
}
|
||
|
|
except Exception as e:
|
||
|
|
logger.error(f"Batch fetch error: {e}")
|
||
|
|
time.sleep(0.05)
|
||
|
|
return results
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
engine = SpotifySyncEngine()
|
||
|
|
if engine.authenticate():
|
||
|
|
logger.info("Spotify API credentials authenticated successfully.")
|
||
|
|
else:
|
||
|
|
logger.error("Authentication failed. Check SPOTIFY_CLIENT_ID and SPOTIFY_CLIENT_SECRET in .env.")
|