105 lines
3.7 KiB
Python
105 lines
3.7 KiB
Python
"""
|
|
Server management utilities.
|
|
Handles server songs loading and server duplicates tracking.
|
|
"""
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
|
|
def load_server_songs(songs_path="data/songs.json"):
|
|
"""Load the list of songs already available on the server."""
|
|
songs_file = Path(songs_path)
|
|
if not songs_file.exists():
|
|
print(f"⚠️ Server songs file not found: {songs_path}")
|
|
return set()
|
|
try:
|
|
with open(songs_file, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
server_songs = set()
|
|
for song in data:
|
|
if "artist" in song and "title" in song:
|
|
artist = song["artist"].strip()
|
|
title = song["title"].strip()
|
|
key = f"{artist.lower()}_{normalize_title(title)}"
|
|
server_songs.add(key)
|
|
print(f"📋 Loaded {len(server_songs)} songs from server (songs.json)")
|
|
return server_songs
|
|
except (json.JSONDecodeError, FileNotFoundError) as e:
|
|
print(f"⚠️ Could not load server songs: {e}")
|
|
return set()
|
|
|
|
|
|
def is_song_on_server(server_songs, artist, title):
|
|
"""Check if a song is already available on the server."""
|
|
key = f"{artist.lower()}_{normalize_title(title)}"
|
|
return key in server_songs
|
|
|
|
|
|
def load_server_duplicates_tracking(
|
|
tracking_path="data/server_duplicates_tracking.json",
|
|
):
|
|
"""Load the tracking of songs found to be duplicates on the server."""
|
|
tracking_file = Path(tracking_path)
|
|
if not tracking_file.exists():
|
|
return {}
|
|
try:
|
|
with open(tracking_file, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except (json.JSONDecodeError, FileNotFoundError) as e:
|
|
print(f"⚠️ Could not load server duplicates tracking: {e}")
|
|
return {}
|
|
|
|
|
|
def save_server_duplicates_tracking(
|
|
tracking, tracking_path="data/server_duplicates_tracking.json"
|
|
):
|
|
"""Save the tracking of songs found to be duplicates on the server."""
|
|
try:
|
|
with open(tracking_path, "w", encoding="utf-8") as f:
|
|
json.dump(tracking, f, indent=2, ensure_ascii=False)
|
|
except Exception as e:
|
|
print(f"⚠️ Could not save server duplicates tracking: {e}")
|
|
|
|
|
|
def is_song_marked_as_server_duplicate(tracking, artist, title):
|
|
"""Check if a song has been marked as a server duplicate."""
|
|
key = f"{artist.lower()}_{normalize_title(title)}"
|
|
return key in tracking
|
|
|
|
|
|
def mark_song_as_server_duplicate(tracking, artist, title, video_title, channel_name):
|
|
"""Mark a song as a server duplicate for future skipping."""
|
|
key = f"{artist.lower()}_{normalize_title(title)}"
|
|
tracking[key] = {
|
|
"artist": artist,
|
|
"title": title,
|
|
"video_title": video_title,
|
|
"channel": channel_name,
|
|
"marked_at": datetime.now().isoformat(),
|
|
"reason": "already_on_server",
|
|
}
|
|
save_server_duplicates_tracking(tracking)
|
|
|
|
|
|
def check_and_mark_server_duplicate(
|
|
server_songs, server_duplicates_tracking, artist, title, video_title, channel_name
|
|
):
|
|
"""Check if a song is on server and mark it as duplicate if so. Returns True if it's a duplicate."""
|
|
if is_song_on_server(server_songs, artist, title):
|
|
if not is_song_marked_as_server_duplicate(
|
|
server_duplicates_tracking, artist, title
|
|
):
|
|
mark_song_as_server_duplicate(
|
|
server_duplicates_tracking, artist, title, video_title, channel_name
|
|
)
|
|
return True
|
|
return False
|
|
|
|
|
|
def normalize_title(title):
|
|
"""Normalize a title for consistent key generation."""
|
|
normalized = title.replace("(Karaoke Version)", "").replace("(Karaoke)", "").strip()
|
|
return " ".join(normalized.split()).lower()
|