Signed-off-by: mbrucedogs <mbrucedogs@gmail.com>
This commit is contained in:
parent
53b90a3027
commit
f3723137fa
@ -1,5 +1,5 @@
|
||||
|
||||
# 🎤 Karaoke Video Downloader – PRD (v2.0)
|
||||
# 🎤 Karaoke Video Downloader – PRD (v2.1)
|
||||
|
||||
## ✅ Overview
|
||||
A Python-based Windows CLI tool to download karaoke videos from YouTube channels/playlists using `yt-dlp.exe`, with advanced tracking, songlist prioritization, and flexible configuration.
|
||||
@ -30,13 +30,13 @@ A Python-based Windows CLI tool to download karaoke videos from YouTube channels
|
||||
|
||||
## 📥 Input
|
||||
- YouTube channel or playlist URLs (e.g. `https://www.youtube.com/@SingKingKaraoke/videos`)
|
||||
- Optional: `channels.txt` file with multiple channel URLs (one per line)
|
||||
- Optional: `docs/songList.json` for prioritized song downloads
|
||||
- Optional: `data/channels.txt` file with multiple channel URLs (one per line)
|
||||
- Optional: `data/songList.json` for prioritized song downloads
|
||||
|
||||
### Example Usage
|
||||
```bash
|
||||
python download_karaoke.py https://www.youtube.com/@SingKingKaraoke/videos
|
||||
python download_karaoke.py --file channels.txt
|
||||
python download_karaoke.py --file data/channels.txt
|
||||
python download_karaoke.py --songlist-only
|
||||
```
|
||||
|
||||
@ -44,8 +44,8 @@ python download_karaoke.py --songlist-only
|
||||
|
||||
## 📤 Output
|
||||
- MP4 files in `downloads/<ChannelName>/` subfolders
|
||||
- All videos tracked in `karaoke_tracking.json`
|
||||
- Songlist progress tracked in `songlist_tracking.json`
|
||||
- All videos tracked in `data/karaoke_tracking.json`
|
||||
- Songlist progress tracked in `data/songlist_tracking.json`
|
||||
- Logs in `logs/`
|
||||
|
||||
---
|
||||
@ -54,7 +54,7 @@ python download_karaoke.py --songlist-only
|
||||
- ✅ Channel-based downloads (with per-channel folders)
|
||||
- ✅ Robust JSON tracking (downloaded, partial, failed, etc.)
|
||||
- ✅ Batch saving and channel video caching for performance
|
||||
- ✅ Configurable download resolution and yt-dlp options (`config.json`)
|
||||
- ✅ Configurable download resolution and yt-dlp options (`data/config.json`)
|
||||
- ✅ Songlist integration: prioritize and track custom songlists
|
||||
- ✅ Songlist-only mode: download only songs from the songlist
|
||||
- ✅ Global songlist tracking to avoid duplicates across channels
|
||||
@ -67,24 +67,40 @@ python download_karaoke.py --songlist-only
|
||||
## 📂 Folder Structure
|
||||
```
|
||||
KaroakeVideoDownloader/
|
||||
├── download_karaoke.py # Main script
|
||||
├── tracking_manager.py # Tracking logic
|
||||
├── manage_tracking.py # Tracking management utility
|
||||
├── update_resolution.py # Resolution config utility
|
||||
├── config.json # Main config
|
||||
├── downloader/yt-dlp.exe # yt-dlp binary
|
||||
├── downloads/ # All video output
|
||||
│ └── [ChannelName]/ # Per-channel folders
|
||||
├── logs/ # Download logs
|
||||
├── karaoke_tracking.json # Main tracking DB
|
||||
├── songlist_tracking.json # Songlist tracking DB
|
||||
├── docs/songList.json # Songlist for prioritization
|
||||
├── karaoke_downloader/ # All core Python code and utilities
|
||||
│ ├── downloader.py # Main downloader class
|
||||
│ ├── cli.py # CLI entry point
|
||||
│ ├── id3_utils.py # ID3 tagging helpers
|
||||
│ ├── songlist_manager.py # Songlist logic
|
||||
│ ├── youtube_utils.py # YouTube helpers
|
||||
│ ├── tracking_manager.py # Tracking logic
|
||||
│ ├── check_resolution.py # Resolution checker utility
|
||||
│ ├── resolution_cli.py # Resolution config CLI
|
||||
│ └── tracking_cli.py # Tracking management CLI
|
||||
├── data/ # All config, tracking, cache, and songlist files
|
||||
│ ├── config.json
|
||||
│ ├── karaoke_tracking.json
|
||||
│ ├── songlist_tracking.json
|
||||
│ ├── channel_cache.json
|
||||
│ ├── channels.txt
|
||||
│ └── songList.json
|
||||
├── downloads/ # All video output
|
||||
│ └── [ChannelName]/ # Per-channel folders
|
||||
├── logs/ # Download logs
|
||||
├── downloader/yt-dlp.exe # yt-dlp binary
|
||||
├── tests/ # Diagnostic and test scripts
|
||||
│ └── test_installation.py
|
||||
├── download_karaoke.py # Main entry point (thin wrapper)
|
||||
├── README.md
|
||||
├── PRD.md
|
||||
├── requirements.txt
|
||||
└── download_karaoke.bat # (optional Windows launcher)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 🚦 CLI Options (Summary)
|
||||
- `--file <channels.txt>`: Download from a list of channels
|
||||
- `--file <data/channels.txt>`: Download from a list of channels
|
||||
- `--songlist-priority`: Prioritize songlist songs in download queue
|
||||
- `--songlist-only`: Download only songs from the songlist
|
||||
- `--songlist-status`: Show songlist download progress
|
||||
@ -96,7 +112,7 @@ KaroakeVideoDownloader/
|
||||
|
||||
## 🧠 Logic Highlights
|
||||
- **Tracking:** All downloads, statuses, and formats are tracked in JSON files for reliability and deduplication.
|
||||
- **Songlist:** Loads and normalizes `docs/songList.json`, matches against available videos, and prioritizes or restricts downloads accordingly.
|
||||
- **Songlist:** Loads and normalizes `data/songList.json`, matches against available videos, and prioritizes or restricts downloads accordingly.
|
||||
- **Batch/Caching:** Channel video lists are cached to minimize API calls; tracking is batch-saved for performance.
|
||||
- **ID3 Tagging:** Artist/title extracted from video title and embedded in MP4 files.
|
||||
- **Cleanup:** Extra files from yt-dlp (e.g., `.info.json`) are automatically removed after download.
|
||||
64
README.md
64
README.md
@ -27,7 +27,7 @@ python download_karaoke.py https://www.youtube.com/@SingKingKaraoke/videos
|
||||
|
||||
### Download from a List of Channels
|
||||
```bash
|
||||
python download_karaoke.py --file channels.txt
|
||||
python download_karaoke.py --file data/channels.txt
|
||||
```
|
||||
|
||||
### Download Only Songlist Songs
|
||||
@ -56,12 +56,12 @@ python download_karaoke.py --resolution 1080p
|
||||
```
|
||||
|
||||
## 🧠 Songlist Integration
|
||||
- Place your prioritized song list in `docs/songList.json` (see example format below).
|
||||
- Place your prioritized song list in `data/songList.json` (see example format below).
|
||||
- The tool will match and prioritize these songs across all available channel videos.
|
||||
- Use `--songlist-only` to download only these songs, or `--songlist-priority` to prioritize them in the queue.
|
||||
- Download progress for the songlist is tracked globally in `songlist_tracking.json`.
|
||||
- Download progress for the songlist is tracked globally in `data/songlist_tracking.json`.
|
||||
|
||||
#### Example `docs/songList.json`
|
||||
#### Example `data/songList.json`
|
||||
```json
|
||||
[
|
||||
{ "artist": "Taylor Swift", "title": "Cruel Summer" },
|
||||
@ -70,29 +70,45 @@ python download_karaoke.py --resolution 1080p
|
||||
```
|
||||
|
||||
## 🛠️ Tracking & Caching
|
||||
- **karaoke_tracking.json**: Tracks all downloads, statuses, and formats
|
||||
- **songlist_tracking.json**: Tracks global songlist download progress
|
||||
- **channel_cache.json**: Caches channel video lists for performance
|
||||
- **data/karaoke_tracking.json**: Tracks all downloads, statuses, and formats
|
||||
- **data/songlist_tracking.json**: Tracks global songlist download progress
|
||||
- **data/channel_cache.json**: Caches channel video lists for performance
|
||||
|
||||
## 📂 Folder Structure
|
||||
```
|
||||
KaroakeVideoDownloader/
|
||||
├── download_karaoke.py # Main script
|
||||
├── tracking_manager.py # Tracking logic
|
||||
├── manage_tracking.py # Tracking management utility
|
||||
├── update_resolution.py # Resolution config utility
|
||||
├── config.json # Main config
|
||||
├── downloader/yt-dlp.exe # yt-dlp binary
|
||||
├── downloads/ # All video output
|
||||
│ └── [ChannelName]/ # Per-channel folders
|
||||
├── logs/ # Download logs
|
||||
├── karaoke_tracking.json # Main tracking DB
|
||||
├── songlist_tracking.json # Songlist tracking DB
|
||||
├── docs/songList.json # Songlist for prioritization
|
||||
├── karaoke_downloader/ # All core Python code and utilities
|
||||
│ ├── downloader.py # Main downloader class
|
||||
│ ├── cli.py # CLI entry point
|
||||
│ ├── id3_utils.py # ID3 tagging helpers
|
||||
│ ├── songlist_manager.py # Songlist logic
|
||||
│ ├── youtube_utils.py # YouTube helpers
|
||||
│ ├── tracking_manager.py # Tracking logic
|
||||
│ ├── check_resolution.py # Resolution checker utility
|
||||
│ ├── resolution_cli.py # Resolution config CLI
|
||||
│ └── tracking_cli.py # Tracking management CLI
|
||||
├── data/ # All config, tracking, cache, and songlist files
|
||||
│ ├── config.json
|
||||
│ ├── karaoke_tracking.json
|
||||
│ ├── songlist_tracking.json
|
||||
│ ├── channel_cache.json
|
||||
│ ├── channels.txt
|
||||
│ └── songList.json
|
||||
├── downloads/ # All video output
|
||||
│ └── [ChannelName]/ # Per-channel folders
|
||||
├── logs/ # Download logs
|
||||
├── downloader/yt-dlp.exe # yt-dlp binary
|
||||
├── tests/ # Diagnostic and test scripts
|
||||
│ └── test_installation.py
|
||||
├── download_karaoke.py # Main entry point (thin wrapper)
|
||||
├── README.md
|
||||
├── PRD.md
|
||||
├── requirements.txt
|
||||
└── download_karaoke.bat # (optional Windows launcher)
|
||||
```
|
||||
|
||||
## 🚦 CLI Options
|
||||
- `--file <channels.txt>`: Download from a list of channels
|
||||
- `--file <data/channels.txt>`: Download from a list of channels
|
||||
- `--songlist-priority`: Prioritize songlist songs in download queue
|
||||
- `--songlist-only`: Download only songs from the songlist
|
||||
- `--songlist-status`: Show songlist download progress
|
||||
@ -103,24 +119,24 @@ KaroakeVideoDownloader/
|
||||
## 📝 Example Usage
|
||||
```bash
|
||||
python download_karaoke.py https://www.youtube.com/@SingKingKaraoke/videos --songlist-priority --limit 10
|
||||
python download_karaoke.py --file channels.txt --songlist-only
|
||||
python download_karaoke.py --file data/channels.txt --songlist-only
|
||||
python download_karaoke.py --songlist-status
|
||||
```
|
||||
|
||||
## 🏷️ ID3 Tagging
|
||||
- Adds artist/title/album/year/genre to MP4 files using mutagen (if installed)
|
||||
- Adds artist/title/album/genre to MP4 files using mutagen (if installed)
|
||||
|
||||
## 🧹 Cleanup
|
||||
- Removes `.info.json` and `.meta` files after download
|
||||
|
||||
## 🧩 Configuration
|
||||
- All options are in `config.json` (format, resolution, metadata, etc.)
|
||||
- All options are in `data/config.json` (format, resolution, metadata, etc.)
|
||||
- You can edit this file or use CLI flags to override
|
||||
|
||||
## 🐞 Troubleshooting
|
||||
- Ensure `yt-dlp.exe` is in the `downloader/` folder
|
||||
- Check `logs/` for error details
|
||||
- Use `python check_resolution.py` to verify video quality
|
||||
- Use `python -m karaoke_downloader.check_resolution` to verify video quality
|
||||
|
||||
---
|
||||
|
||||
|
||||
@ -4,983 +4,7 @@ Karaoke Playlist Downloader
|
||||
A Python-based Windows command-line tool that wraps yt-dlp.exe to batch-download karaoke videos from YouTube playlists.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import argparse
|
||||
import subprocess
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse, parse_qs
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from tracking_manager import TrackingManager, SongStatus, FormatType
|
||||
|
||||
# Try to import mutagen for ID3 tagging
|
||||
try:
|
||||
from mutagen.mp4 import MP4, MP4Tags
|
||||
MUTAGEN_AVAILABLE = True
|
||||
except ImportError:
|
||||
MUTAGEN_AVAILABLE = False
|
||||
print("⚠️ mutagen not available - ID3 tagging will be disabled")
|
||||
print(" Install with: pip install mutagen")
|
||||
|
||||
|
||||
class KaraokeDownloader:
|
||||
def __init__(self):
|
||||
self.yt_dlp_path = Path("downloader/yt-dlp.exe")
|
||||
self.downloads_dir = Path("downloads")
|
||||
self.logs_dir = Path("logs")
|
||||
|
||||
# Create necessary directories
|
||||
self.downloads_dir.mkdir(exist_ok=True)
|
||||
self.logs_dir.mkdir(exist_ok=True)
|
||||
|
||||
# Initialize tracking manager
|
||||
self.tracker = TrackingManager()
|
||||
|
||||
# Load configuration
|
||||
self.config = self._load_config()
|
||||
|
||||
# Songlist tracking file
|
||||
self.songlist_tracking_file = Path("songlist_tracking.json")
|
||||
self.songlist_tracking = self._load_songlist_tracking()
|
||||
|
||||
def _get_channel_info(self, channel_url):
|
||||
"""Extract channel information from URL."""
|
||||
# Fallback: extract channel info from URL (faster and more reliable)
|
||||
if '@' in channel_url:
|
||||
# Handle @username format
|
||||
channel_name = channel_url.split('@')[1].split('/')[0]
|
||||
channel_id = f"@{channel_name}"
|
||||
else:
|
||||
# Handle other formats
|
||||
channel_name = "unknown_channel"
|
||||
channel_id = "unknown_channel"
|
||||
|
||||
# Clean the channel name for use as directory name
|
||||
channel_name = re.sub(r'[<>:"/\\|?*]', '_', channel_name)
|
||||
return channel_name, channel_id
|
||||
|
||||
def _get_playlist_info(self, playlist_url):
|
||||
"""Extract playlist information from URL (backward compatibility)."""
|
||||
return self._get_channel_info(playlist_url)
|
||||
|
||||
def _load_config(self):
|
||||
"""Load configuration from config.json file."""
|
||||
config_file = Path("config.json")
|
||||
if config_file.exists():
|
||||
try:
|
||||
with open(config_file, 'r', encoding='utf-8') as f:
|
||||
return json.load(f)
|
||||
except (json.JSONDecodeError, FileNotFoundError) as e:
|
||||
print(f"Warning: Could not load config.json: {e}")
|
||||
|
||||
# Return default configuration
|
||||
return {
|
||||
"download_settings": {
|
||||
"format": "best[height<=720][ext=mp4]/best[height<=720]/best[ext=mp4]/best",
|
||||
"preferred_resolution": "720p",
|
||||
"audio_format": "mp3",
|
||||
"audio_quality": "0",
|
||||
"subtitle_language": "en",
|
||||
"subtitle_format": "srt",
|
||||
"write_metadata": False,
|
||||
"write_thumbnail": False,
|
||||
"write_description": False,
|
||||
"write_annotations": False,
|
||||
"write_comments": False,
|
||||
"write_subtitles": False,
|
||||
"embed_metadata": False,
|
||||
"add_metadata": False,
|
||||
"continue_downloads": True,
|
||||
"no_overwrites": True,
|
||||
"ignore_errors": True,
|
||||
"no_warnings": False
|
||||
},
|
||||
"folder_structure": {
|
||||
"downloads_dir": "downloads",
|
||||
"logs_dir": "logs",
|
||||
"tracking_file": "karaoke_tracking.json"
|
||||
},
|
||||
"logging": {
|
||||
"level": "INFO",
|
||||
"format": "%(asctime)s - %(levelname)s - %(message)s",
|
||||
"include_console": True,
|
||||
"include_file": True
|
||||
},
|
||||
"yt_dlp_path": "downloader/yt-dlp.exe"
|
||||
}
|
||||
|
||||
def _load_songlist(self):
|
||||
"""Load songlist from docs/songList.json file."""
|
||||
songlist_file = Path("docs/songList.json")
|
||||
if not songlist_file.exists():
|
||||
print("⚠️ Songlist file not found: docs/songList.json")
|
||||
return []
|
||||
|
||||
try:
|
||||
with open(songlist_file, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
|
||||
# Extract all songs from all categories
|
||||
all_songs = []
|
||||
for category in data:
|
||||
if "songs" in category:
|
||||
for song in category["songs"]:
|
||||
if "artist" in song and "title" in song:
|
||||
all_songs.append({
|
||||
"artist": song["artist"].strip(),
|
||||
"title": song["title"].strip(),
|
||||
"position": song.get("position", 0)
|
||||
})
|
||||
|
||||
print(f"📋 Loaded {len(all_songs)} songs from songlist")
|
||||
return all_songs
|
||||
|
||||
except (json.JSONDecodeError, FileNotFoundError) as e:
|
||||
print(f"⚠️ Could not load songlist: {e}")
|
||||
return []
|
||||
|
||||
def _load_songlist_tracking(self):
|
||||
"""Load songlist tracking data."""
|
||||
if not self.songlist_tracking_file.exists():
|
||||
return {}
|
||||
|
||||
try:
|
||||
with open(self.songlist_tracking_file, 'r', encoding='utf-8') as f:
|
||||
return json.load(f)
|
||||
except (json.JSONDecodeError, FileNotFoundError) as e:
|
||||
print(f"⚠️ Could not load songlist tracking: {e}")
|
||||
return {}
|
||||
|
||||
def _save_songlist_tracking(self):
|
||||
"""Save songlist tracking data."""
|
||||
try:
|
||||
with open(self.songlist_tracking_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(self.songlist_tracking, f, indent=2, ensure_ascii=False)
|
||||
except Exception as e:
|
||||
print(f"⚠️ Could not save songlist tracking: {e}")
|
||||
|
||||
def _is_songlist_song_downloaded(self, artist, title):
|
||||
"""Check if a songlist song has been downloaded from any channel."""
|
||||
key = f"{artist.lower()}_{self._normalize_title(title)}"
|
||||
return key in self.songlist_tracking
|
||||
|
||||
def _mark_songlist_song_downloaded(self, artist, title, channel_name, file_path):
|
||||
"""Mark a songlist song as downloaded."""
|
||||
key = f"{artist.lower()}_{self._normalize_title(title)}"
|
||||
self.songlist_tracking[key] = {
|
||||
"artist": artist,
|
||||
"title": title,
|
||||
"channel": channel_name,
|
||||
"file_path": str(file_path),
|
||||
"downloaded_at": datetime.now().isoformat()
|
||||
}
|
||||
self._save_songlist_tracking()
|
||||
|
||||
def _normalize_title(self, title):
|
||||
"""Normalize title for comparison (remove karaoke suffixes, etc.)."""
|
||||
# Remove common karaoke suffixes
|
||||
normalized = title.replace("(Karaoke Version)", "").replace("(Karaoke)", "").strip()
|
||||
# Remove extra spaces and convert to lowercase for comparison
|
||||
return " ".join(normalized.split()).lower()
|
||||
|
||||
def _find_songlist_matches(self, cached_videos, songlist, songlist_only=False):
|
||||
"""Find videos that match songs in the songlist."""
|
||||
matches = []
|
||||
songlist_lookup = {}
|
||||
unique_matches = set()
|
||||
already_downloaded = 0
|
||||
new_downloads = 0
|
||||
|
||||
# Create lookup for songlist songs (normalized)
|
||||
for song in songlist:
|
||||
normalized_title = self._normalize_title(song["title"])
|
||||
key = f"{song['artist'].lower()}_{normalized_title}"
|
||||
songlist_lookup[key] = song
|
||||
|
||||
print(f"🔍 Searching for {len(songlist)} songlist songs in {len(cached_videos)} cached videos...")
|
||||
|
||||
for video in cached_videos:
|
||||
# Extract artist and title from video title
|
||||
artist, title = self._extract_artist_title(video['title'])
|
||||
normalized_title = self._normalize_title(title)
|
||||
key = f"{artist.lower()}_{normalized_title}"
|
||||
|
||||
if key in songlist_lookup:
|
||||
songlist_song = songlist_lookup[key]
|
||||
|
||||
# Check if already downloaded from any channel
|
||||
if self._is_songlist_song_downloaded(artist, title):
|
||||
already_downloaded += 1
|
||||
print(f"⏭️ Already downloaded: {artist} - {title}")
|
||||
continue
|
||||
|
||||
matches.append({
|
||||
'video': video,
|
||||
'songlist_song': songlist_song,
|
||||
'priority_score': songlist_song.get('position', 9999) # Lower position = higher priority
|
||||
})
|
||||
unique_matches.add(key)
|
||||
new_downloads += 1
|
||||
print(f"🎯 Found match: {artist} - {title} (position {songlist_song.get('position', 'N/A')})")
|
||||
|
||||
print(f"🎯 Matching Results:")
|
||||
print(f" • Total matches found: {len(matches) + already_downloaded}")
|
||||
print(f" • Unique songs matched: {len(unique_matches) + already_downloaded}")
|
||||
print(f" • Already downloaded: {already_downloaded}")
|
||||
print(f" • New downloads available: {new_downloads}")
|
||||
print(f" • Duplicate matches: {len(matches) - len(unique_matches)}")
|
||||
|
||||
return matches
|
||||
|
||||
def _get_playlist_name(self, playlist_url):
|
||||
"""Extract playlist name from URL (backward compatibility)."""
|
||||
return self._get_playlist_info(playlist_url)[0]
|
||||
|
||||
def _setup_logging(self, playlist_name):
|
||||
"""Setup logging for the current playlist."""
|
||||
log_file = self.logs_dir / f"{playlist_name}.log"
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(levelname)s - %(message)s',
|
||||
handlers=[
|
||||
logging.FileHandler(log_file),
|
||||
logging.StreamHandler(sys.stdout)
|
||||
]
|
||||
)
|
||||
|
||||
return logging.getLogger(__name__)
|
||||
|
||||
def _get_channel_videos(self, channel_url):
|
||||
"""Get list of videos from a channel."""
|
||||
try:
|
||||
cmd = [
|
||||
str(self.yt_dlp_path),
|
||||
"--flat-playlist",
|
||||
"--print", "id,title,duration,upload_date",
|
||||
channel_url
|
||||
]
|
||||
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
|
||||
if result.returncode == 0 and result.stdout.strip():
|
||||
videos = []
|
||||
lines = result.stdout.strip().split('\n')
|
||||
for i in range(0, len(lines), 4):
|
||||
if i + 3 < len(lines):
|
||||
video_id = lines[i].strip()
|
||||
title = lines[i + 1].strip()
|
||||
duration = lines[i + 2].strip()
|
||||
upload_date = lines[i + 3].strip()
|
||||
try:
|
||||
duration_int = int(duration) if duration != 'NA' else None
|
||||
except ValueError:
|
||||
duration_int = None
|
||||
|
||||
videos.append({
|
||||
'video_id': video_id,
|
||||
'title': title,
|
||||
'duration': duration_int,
|
||||
'upload_date': upload_date
|
||||
})
|
||||
return videos
|
||||
except (subprocess.TimeoutExpired, subprocess.CalledProcessError) as e:
|
||||
print(f"Error getting channel videos: {e}")
|
||||
|
||||
return []
|
||||
|
||||
def _get_playlist_songs(self, playlist_url):
|
||||
"""Get list of songs in a playlist (backward compatibility)."""
|
||||
return self._get_channel_videos(playlist_url)
|
||||
|
||||
def _extract_artist_title(self, video_title):
|
||||
"""Extract artist and title from karaoke video title."""
|
||||
# Remove common karaoke suffixes
|
||||
title = video_title.replace("(Karaoke Version)", "").replace("(Karaoke)", "").strip()
|
||||
|
||||
# Look for common patterns
|
||||
if " - " in title:
|
||||
# Format: "Artist - Song Title"
|
||||
parts = title.split(" - ", 1)
|
||||
if len(parts) == 2:
|
||||
artist = parts[0].strip()
|
||||
song_title = parts[1].strip()
|
||||
|
||||
# Handle special cases
|
||||
if artist == "d4vd, Hyunjin":
|
||||
return "d4vd, Hyunjin", song_title
|
||||
elif artist == "Adrianne Lenker":
|
||||
return "Adrianne Lenker", song_title
|
||||
else:
|
||||
return artist, song_title
|
||||
|
||||
# Handle special cases without clear artist - title pattern
|
||||
if "Nothing Beats A Jet2 Holiday" in title:
|
||||
return "Jet2", "Nothing Beats A Jet2 Holiday"
|
||||
elif "Spin The Wheel" in title:
|
||||
return "SingKingKaraoke", "Spin The Wheel - Your Song Requests"
|
||||
|
||||
# If no clear pattern, assume the whole title is the song name
|
||||
return "Unknown Artist", title
|
||||
|
||||
def _add_id3_tags(self, file_path, video_title, channel_name):
|
||||
"""Add ID3 tags to the downloaded MP4 file."""
|
||||
if not MUTAGEN_AVAILABLE:
|
||||
print("⚠️ mutagen not available - skipping ID3 tagging")
|
||||
return
|
||||
|
||||
try:
|
||||
artist, title = self._extract_artist_title(video_title)
|
||||
|
||||
# Load the MP4 file
|
||||
mp4 = MP4(str(file_path))
|
||||
|
||||
# Add metadata tags
|
||||
mp4['\xa9nam'] = title # Title
|
||||
mp4['\xa9ART'] = artist # Artist
|
||||
mp4['\xa9alb'] = f"{channel_name} Karaoke" # Album
|
||||
mp4['\xa9gen'] = "Karaoke" # Genre
|
||||
mp4['\xa9day'] = str(datetime.now().year) # Year
|
||||
|
||||
# Save the metadata
|
||||
mp4.save()
|
||||
|
||||
print(f"📝 Added ID3 tags: Artist='{artist}', Title='{title}'")
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ Could not add ID3 tags: {e}")
|
||||
|
||||
def _cleanup_extra_files(self, mp4_file_path):
|
||||
"""Remove extra files (info.json, meta) that yt-dlp creates alongside MP4 files."""
|
||||
try:
|
||||
# Get the base path without extension
|
||||
base_path = mp4_file_path.with_suffix('')
|
||||
|
||||
# Files to remove
|
||||
extra_files = [
|
||||
base_path.with_suffix('.info.json'),
|
||||
base_path.with_suffix('.meta')
|
||||
]
|
||||
|
||||
removed_count = 0
|
||||
for extra_file in extra_files:
|
||||
if extra_file.exists():
|
||||
extra_file.unlink()
|
||||
removed_count += 1
|
||||
print(f"🧹 Removed extra file: {extra_file.name}")
|
||||
|
||||
if removed_count > 0:
|
||||
print(f"🧹 Cleaned up {removed_count} extra file(s)")
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ Could not clean up extra files: {e}")
|
||||
|
||||
def download_channel_videos(self, channel_url, force_refresh=False):
|
||||
"""Download all videos from a YouTube channel."""
|
||||
print(f"🎤 Starting download for channel: {channel_url}")
|
||||
|
||||
# Get channel info
|
||||
print("🔍 Extracting channel information...")
|
||||
channel_name, channel_id = self._get_channel_info(channel_url)
|
||||
print(f"📺 Channel name: {channel_name}")
|
||||
print(f"🆔 Channel ID: {channel_id}")
|
||||
|
||||
# Setup logging
|
||||
print("📝 Setting up logging...")
|
||||
logger = self._setup_logging(channel_name)
|
||||
logger.info(f"Starting download for channel: {channel_url}")
|
||||
|
||||
# Add channel to tracking
|
||||
print("📋 Adding channel to tracking system...")
|
||||
self.tracker.add_playlist(channel_id, channel_name, channel_url)
|
||||
|
||||
# Check cache first (unless force refresh is requested)
|
||||
print("💾 Checking channel cache...")
|
||||
current_videos = None
|
||||
if not force_refresh:
|
||||
current_videos = self.tracker.get_cached_channel_videos(channel_id)
|
||||
if current_videos:
|
||||
print(f"📋 Using cached channel data: {len(current_videos)} videos")
|
||||
else:
|
||||
print("📋 Cache miss or expired, fetching fresh channel data...")
|
||||
|
||||
# Fetch fresh data if needed
|
||||
if not current_videos:
|
||||
print("📋 Fetching channel videos from YouTube...")
|
||||
current_videos = self._get_channel_videos(channel_url)
|
||||
print(f"🎵 Found {len(current_videos)} videos in channel")
|
||||
|
||||
# Cache the results
|
||||
if current_videos:
|
||||
print("💾 Caching channel data for future use...")
|
||||
self.tracker.cache_channel_videos(channel_id, current_videos)
|
||||
print(f"💾 Cached channel data for future use")
|
||||
else:
|
||||
print(f"🎵 Using {len(current_videos)} cached videos")
|
||||
|
||||
# Sync channel with tracking
|
||||
print("🔄 Syncing channel with tracking system...")
|
||||
sync_result = self.tracker.sync_playlist(channel_id, current_videos)
|
||||
print(f"🔄 Sync result: {sync_result['added']} new, {sync_result['removed']} removed, {sync_result['total']} total")
|
||||
|
||||
# Create channel-specific downloads directory
|
||||
print("📁 Creating downloads directory...")
|
||||
downloads_dir = self.downloads_dir / channel_name
|
||||
downloads_dir.mkdir(parents=True, exist_ok=True)
|
||||
print(f"📁 Downloads will be saved to: {downloads_dir}")
|
||||
|
||||
# Load songlist for prioritization (if enabled)
|
||||
songlist = []
|
||||
if hasattr(self, 'use_songlist_priority') and self.use_songlist_priority:
|
||||
print("📋 Loading songlist for prioritization...")
|
||||
songlist = self._load_songlist()
|
||||
|
||||
# Show songlist statistics
|
||||
unique_songs = len(set(f"{song['artist']}_{song['title']}" for song in songlist))
|
||||
print(f"📊 Songlist Statistics:")
|
||||
print(f" • Total songs in list: {len(songlist)}")
|
||||
print(f" • Unique songs: {unique_songs}")
|
||||
print(f" • Duplicates in list: {len(songlist) - unique_songs}")
|
||||
else:
|
||||
print("📋 Songlist prioritization disabled")
|
||||
|
||||
# Get videos that need downloading
|
||||
print("🔍 Checking which videos need to be downloaded...")
|
||||
videos_to_download = []
|
||||
found_existing = 0
|
||||
|
||||
# Get list of existing MP4 files once (more efficient)
|
||||
existing_files = set()
|
||||
for file_path in downloads_dir.glob(f"*.mp4"):
|
||||
if file_path.is_file():
|
||||
existing_files.add(file_path.stem)
|
||||
|
||||
# Also check root downloads directory
|
||||
for file_path in self.downloads_dir.glob(f"*.mp4"):
|
||||
if file_path.is_file():
|
||||
existing_files.add(file_path.stem)
|
||||
|
||||
print(f"📁 Found {len(existing_files)} existing MP4 files in directories")
|
||||
|
||||
# Separate videos into songlist matches and others
|
||||
songlist_matches = []
|
||||
other_videos = []
|
||||
|
||||
for video in current_videos:
|
||||
if not self.tracker.is_song_downloaded(video['video_id'], channel_id):
|
||||
if video['title'] in existing_files:
|
||||
# File exists but not tracked, mark as downloaded
|
||||
self.tracker.update_song_status(
|
||||
video['video_id'],
|
||||
channel_id,
|
||||
SongStatus.DOWNLOADED,
|
||||
FormatType.MP4,
|
||||
downloads_dir / f"{video['title']}.mp4" # Assume it's in channel dir
|
||||
)
|
||||
found_existing += 1
|
||||
else:
|
||||
other_videos.append(video)
|
||||
|
||||
# Find songlist matches and prioritize them
|
||||
if songlist:
|
||||
songlist_matches = self._find_songlist_matches(other_videos, songlist, songlist_only=getattr(self, 'songlist_only', False))
|
||||
|
||||
# Remove matched videos from other_videos to avoid duplicates
|
||||
matched_video_ids = {match['video']['video_id'] for match in songlist_matches}
|
||||
other_videos = [v for v in other_videos if v['video_id'] not in matched_video_ids]
|
||||
|
||||
# Sort songlist matches by priority (lower position = higher priority)
|
||||
songlist_matches.sort(key=lambda x: x['priority_score'])
|
||||
|
||||
# Add songlist matches first, then other videos (unless songlist_only is enabled)
|
||||
if getattr(self, 'songlist_only', False):
|
||||
videos_to_download = [match['video'] for match in songlist_matches]
|
||||
print(f"🎯 Songlist-only mode: {len(songlist_matches)} songlist matches")
|
||||
else:
|
||||
videos_to_download = [match['video'] for match in songlist_matches] + other_videos
|
||||
print(f"🎯 Prioritized {len(songlist_matches)} songlist matches")
|
||||
print(f"📝 Added {len(other_videos)} other videos")
|
||||
else:
|
||||
videos_to_download = other_videos
|
||||
|
||||
if found_existing > 0:
|
||||
print(f"📁 Found {found_existing} existing files that weren't tracked")
|
||||
|
||||
# Apply limit if specified
|
||||
if hasattr(self, 'download_limit') and self.download_limit:
|
||||
videos_to_download = videos_to_download[:self.download_limit]
|
||||
print(f"📥 Videos to download: {len(videos_to_download)} (limited to {self.download_limit})")
|
||||
else:
|
||||
print(f"📥 Videos to download: {len(videos_to_download)}")
|
||||
|
||||
if not videos_to_download:
|
||||
print("✅ All videos already downloaded!")
|
||||
return True
|
||||
|
||||
# Build yt-dlp command for each video
|
||||
for i, video in enumerate(videos_to_download, 1):
|
||||
print(f"\n🎵 Downloading {i} of {len(videos_to_download)}: {video['title']}")
|
||||
|
||||
# Update status to downloading
|
||||
self.tracker.update_song_status(
|
||||
video['video_id'],
|
||||
channel_id,
|
||||
SongStatus.DOWNLOADING
|
||||
)
|
||||
|
||||
# Build yt-dlp command for this specific video using config
|
||||
download_settings = self.config["download_settings"]
|
||||
|
||||
cmd = [
|
||||
str(self.yt_dlp_path),
|
||||
"--format", download_settings["format"],
|
||||
"--output", str(downloads_dir / "%(title)s.%(ext)s"),
|
||||
]
|
||||
|
||||
# Only add metadata options if enabled in config
|
||||
if download_settings.get("add_metadata", False):
|
||||
cmd.append("--add-metadata")
|
||||
if download_settings.get("embed_metadata", False):
|
||||
cmd.append("--embed-metadata")
|
||||
if download_settings.get("write_metadata", False):
|
||||
cmd.append("--write-info-json")
|
||||
|
||||
# Add optional settings based on config (only essential ones)
|
||||
if download_settings.get("no_overwrites", True):
|
||||
cmd.append("--no-overwrites")
|
||||
if download_settings.get("continue_downloads", True):
|
||||
cmd.append("--continue")
|
||||
if download_settings.get("ignore_errors", True):
|
||||
cmd.append("--ignore-errors")
|
||||
if download_settings.get("no_warnings", False):
|
||||
cmd.append("--no-warnings")
|
||||
|
||||
# Add progress and display options
|
||||
cmd.extend([
|
||||
"--progress",
|
||||
"--newline",
|
||||
"--console-title",
|
||||
f"https://www.youtube.com/watch?v={video['video_id']}"
|
||||
])
|
||||
|
||||
try:
|
||||
print(f"🚀 Starting download for: {video['title']}")
|
||||
logger.info(f"Starting download for video ID: {video['video_id']}")
|
||||
|
||||
# Run yt-dlp
|
||||
process = subprocess.Popen(
|
||||
cmd,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
text=True,
|
||||
bufsize=1,
|
||||
universal_newlines=True
|
||||
)
|
||||
|
||||
# Process output in real-time
|
||||
for line in process.stdout:
|
||||
line = line.strip()
|
||||
if line:
|
||||
print(line)
|
||||
logger.info(line)
|
||||
|
||||
process.wait()
|
||||
|
||||
# Check if download was successful (more lenient - if MP4 exists, consider success)
|
||||
downloaded_file = None
|
||||
|
||||
# Look for the file in the channel directory by title (since yt-dlp uses title as filename)
|
||||
for file_path in downloads_dir.glob(f"*.mp4"):
|
||||
if file_path.is_file():
|
||||
# Check if this file matches the video title
|
||||
file_name = file_path.stem # filename without extension
|
||||
if file_name == video['title']:
|
||||
downloaded_file = file_path
|
||||
break
|
||||
|
||||
# If not found, also check the root downloads directory
|
||||
if not downloaded_file:
|
||||
for file_path in self.downloads_dir.glob(f"*.mp4"):
|
||||
if file_path.is_file():
|
||||
# Check if this file matches the video title
|
||||
file_name = file_path.stem # filename without extension
|
||||
if file_name == video['title']:
|
||||
downloaded_file = file_path
|
||||
break
|
||||
|
||||
if downloaded_file:
|
||||
# MP4 file exists, consider download successful
|
||||
format_type = FormatType.MP4
|
||||
|
||||
# Update tracking with successful download
|
||||
self.tracker.update_song_status(
|
||||
video['video_id'],
|
||||
channel_id,
|
||||
SongStatus.DOWNLOADED,
|
||||
format_type,
|
||||
downloaded_file
|
||||
)
|
||||
|
||||
print(f"✅ Successfully downloaded: {video['title']} ({format_type})")
|
||||
logger.info(f"Successfully downloaded: {video['title']} ({format_type})")
|
||||
|
||||
# Add ID3 tags to the downloaded file
|
||||
self._add_id3_tags(downloaded_file, video['title'], channel_name)
|
||||
|
||||
# Clean up extra files (info.json, meta files)
|
||||
self._cleanup_extra_files(downloaded_file)
|
||||
|
||||
# Mark as downloaded in songlist tracking if it's a songlist song
|
||||
if hasattr(self, 'use_songlist_priority') and self.use_songlist_priority:
|
||||
artist, title = self._extract_artist_title(video['title'])
|
||||
self._mark_songlist_song_downloaded(artist, title, channel_name, downloaded_file)
|
||||
print(f"📋 Marked songlist song as downloaded: {artist} - {title}")
|
||||
else:
|
||||
# No MP4 file found, mark as failed
|
||||
self.tracker.update_song_status(
|
||||
video['video_id'],
|
||||
channel_id,
|
||||
SongStatus.FAILED,
|
||||
error_message=f"Download failed - no MP4 file found (return code: {process.returncode})"
|
||||
)
|
||||
print(f"❌ Download failed for: {video['title']} (return code: {process.returncode})")
|
||||
logger.error(f"Download failed for: {video['title']} (return code: {process.returncode})")
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
error_msg = f"Error running yt-dlp for {video['title']}: {e}"
|
||||
print(f"❌ {error_msg}")
|
||||
logger.error(error_msg)
|
||||
|
||||
self.tracker.update_song_status(
|
||||
video['video_id'],
|
||||
channel_id,
|
||||
SongStatus.FAILED,
|
||||
error_message=str(e)
|
||||
)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n⏹️ Download interrupted by user")
|
||||
logger.info("Download interrupted by user")
|
||||
|
||||
# Mark current video as partial
|
||||
self.tracker.update_song_status(
|
||||
video['video_id'],
|
||||
channel_id,
|
||||
SongStatus.PARTIAL
|
||||
)
|
||||
return False
|
||||
|
||||
# Force save tracking data
|
||||
self.tracker.force_save()
|
||||
|
||||
# Show final statistics
|
||||
stats = self.tracker.get_statistics()
|
||||
print(f"\n📊 Download Statistics:")
|
||||
print(f" Total songs: {stats['total_songs']}")
|
||||
print(f" Downloaded: {stats['downloaded_songs']}")
|
||||
print(f" Failed: {stats['failed_songs']}")
|
||||
print(f" Partial: {stats['partial_songs']}")
|
||||
print(f" Total size: {stats['total_size_mb']} MB")
|
||||
|
||||
return True
|
||||
|
||||
def download_playlist(self, playlist_url):
|
||||
"""Download all videos from a YouTube playlist (backward compatibility)."""
|
||||
return self.download_channel_videos(playlist_url)
|
||||
|
||||
def download_from_file(self, file_path, force_refresh=False):
|
||||
"""Download multiple playlists from a text file."""
|
||||
file_path = Path(file_path)
|
||||
|
||||
if not file_path.exists():
|
||||
print(f"❌ File not found: {file_path}")
|
||||
return False
|
||||
|
||||
print(f"📄 Reading playlists from: {file_path}")
|
||||
|
||||
with open(file_path, 'r') as f:
|
||||
playlist_urls = [line.strip() for line in f if line.strip()]
|
||||
|
||||
print(f"📋 Found {len(playlist_urls)} URL(s) to download")
|
||||
|
||||
success_count = 0
|
||||
for i, url in enumerate(playlist_urls, 1):
|
||||
print(f"\n{'='*50}")
|
||||
print(f"📥 Processing URL {i}/{len(playlist_urls)}: {url}")
|
||||
print(f"{'='*50}")
|
||||
|
||||
if self.download_channel_videos(url, force_refresh=force_refresh):
|
||||
success_count += 1
|
||||
|
||||
print(f"\n🎉 Download summary: {success_count}/{len(playlist_urls)} URLs completed successfully")
|
||||
|
||||
# Force save any pending tracking data
|
||||
self.tracker.force_save()
|
||||
|
||||
return success_count == len(playlist_urls)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Karaoke Video Downloader - Download YouTube playlists and channel videos for karaoke",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="""
|
||||
Examples:
|
||||
python download_karaoke.py https://www.youtube.com/playlist?list=XYZ
|
||||
python download_karaoke.py https://www.youtube.com/@SingKingKaraoke/videos
|
||||
python download_karaoke.py --file playlists.txt
|
||||
python download_karaoke.py --limit 5 https://www.youtube.com/@SingKingKaraoke/videos
|
||||
python download_karaoke.py --refresh https://www.youtube.com/@SingKingKaraoke/videos
|
||||
python download_karaoke.py --cache-info
|
||||
python download_karaoke.py --clear-cache all
|
||||
"""
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'url',
|
||||
nargs='?',
|
||||
help='YouTube playlist or channel URL to download'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--file', '-f',
|
||||
help='Text file containing playlist or channel URLs (one per line)'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--status', '-s',
|
||||
action='store_true',
|
||||
help='Show download status and statistics'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--report', '-r',
|
||||
metavar='PLAYLIST_ID',
|
||||
help='Generate a detailed report for a specific playlist'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--sync', '--sync-only',
|
||||
action='store_true',
|
||||
help='Only sync playlist without downloading (update tracking)'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--cleanup',
|
||||
action='store_true',
|
||||
help='Clean up orphaned tracking entries'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--resolution', '--res',
|
||||
choices=['480p', '720p', '1080p', '1440p', '2160p'],
|
||||
default='720p',
|
||||
help='Preferred video resolution (default: 720p)'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--limit', '-l',
|
||||
type=int,
|
||||
help='Limit the number of videos to download (e.g., --limit 10)'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--refresh', '--force-refresh',
|
||||
action='store_true',
|
||||
help='Force refresh channel cache (ignore cached data)'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--cache-info',
|
||||
action='store_true',
|
||||
help='Show channel cache information'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--clear-cache',
|
||||
metavar='CHANNEL_ID',
|
||||
nargs='?',
|
||||
const='all',
|
||||
help='Clear cache for specific channel or all channels (use --clear-cache all)'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--cache-duration',
|
||||
type=int,
|
||||
metavar='HOURS',
|
||||
help='Set cache duration in hours (default: 24)'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--songlist-priority',
|
||||
action='store_true',
|
||||
help='Prioritize downloads based on docs/songList.json (default: enabled)'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--no-songlist-priority',
|
||||
action='store_true',
|
||||
help='Disable songlist prioritization'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--songlist-only',
|
||||
action='store_true',
|
||||
help='Only download songs that are in the songlist (skip all others)'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--songlist-status',
|
||||
action='store_true',
|
||||
help='Show songlist download status and statistics'
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--version', '-v',
|
||||
action='version',
|
||||
version='Karaoke Playlist Downloader v1.0'
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# Check if yt-dlp.exe exists
|
||||
yt_dlp_path = Path("downloader/yt-dlp.exe")
|
||||
if not yt_dlp_path.exists():
|
||||
print("❌ Error: yt-dlp.exe not found in downloader/ directory")
|
||||
print("Please ensure yt-dlp.exe is present in the downloader/ folder")
|
||||
sys.exit(1)
|
||||
|
||||
# Initialize downloader
|
||||
downloader = KaraokeDownloader()
|
||||
|
||||
# Set download limit if specified
|
||||
if args.limit:
|
||||
downloader.download_limit = args.limit
|
||||
print(f"🎯 Download limit set to: {args.limit} videos")
|
||||
|
||||
# Set songlist priority setting
|
||||
if args.no_songlist_priority:
|
||||
downloader.use_songlist_priority = False
|
||||
print("🎯 Songlist prioritization disabled")
|
||||
else:
|
||||
downloader.use_songlist_priority = True
|
||||
print("🎯 Songlist prioritization enabled")
|
||||
|
||||
# Set songlist-only mode
|
||||
if args.songlist_only:
|
||||
downloader.songlist_only = True
|
||||
print("🎯 Songlist-only mode enabled (will only download songlist songs)")
|
||||
|
||||
# Update resolution if specified
|
||||
if args.resolution != '720p':
|
||||
resolution_map = {
|
||||
'480p': '480',
|
||||
'720p': '720',
|
||||
'1080p': '1080',
|
||||
'1440p': '1440',
|
||||
'2160p': '2160'
|
||||
}
|
||||
height = resolution_map[args.resolution]
|
||||
downloader.config["download_settings"]["format"] = f"best[height<={height}][ext=mp4]/best[height<={height}]/best[ext=mp4]/best"
|
||||
downloader.config["download_settings"]["preferred_resolution"] = args.resolution
|
||||
print(f"🎬 Using resolution: {args.resolution}")
|
||||
|
||||
# Process arguments
|
||||
if args.status:
|
||||
# Show status and statistics
|
||||
stats = downloader.tracker.get_statistics()
|
||||
print("🎤 Karaoke Downloader Status")
|
||||
print("=" * 40)
|
||||
print(f"Total Songs: {stats['total_songs']}")
|
||||
print(f"Total Playlists: {stats['total_playlists']}")
|
||||
print(f"Downloaded Songs: {stats['downloaded_songs']}")
|
||||
print(f"Failed Songs: {stats['failed_songs']}")
|
||||
print(f"Partial Downloads: {stats['partial_songs']}")
|
||||
print(f"Total Size: {stats['total_size_mb']} MB")
|
||||
print(f"Last Updated: {stats['last_updated']}")
|
||||
sys.exit(0)
|
||||
|
||||
elif args.cache_info:
|
||||
# Show cache information
|
||||
cache_info = downloader.tracker.get_cache_info()
|
||||
print("💾 Channel Cache Information")
|
||||
print("=" * 40)
|
||||
print(f"Total Channels: {cache_info['total_channels']}")
|
||||
print(f"Total Cached Videos: {cache_info['total_cached_videos']}")
|
||||
print(f"Cache Duration: {cache_info['cache_duration_hours']} hours")
|
||||
print(f"Last Updated: {cache_info['last_updated']}")
|
||||
sys.exit(0)
|
||||
|
||||
elif args.clear_cache:
|
||||
# Clear cache
|
||||
if args.clear_cache == 'all':
|
||||
downloader.tracker.clear_channel_cache()
|
||||
print("🧹 Cleared all channel caches")
|
||||
else:
|
||||
downloader.tracker.clear_channel_cache(args.clear_cache)
|
||||
print(f"🧹 Cleared cache for channel: {args.clear_cache}")
|
||||
sys.exit(0)
|
||||
|
||||
elif args.cache_duration:
|
||||
# Set cache duration
|
||||
downloader.tracker.set_cache_duration(args.cache_duration)
|
||||
print(f"⏰ Cache duration set to {args.cache_duration} hours")
|
||||
sys.exit(0)
|
||||
|
||||
elif args.report:
|
||||
# Generate playlist report
|
||||
report = downloader.tracker.export_playlist_report(args.report)
|
||||
print(report)
|
||||
sys.exit(0)
|
||||
|
||||
elif args.cleanup:
|
||||
# Clean up orphaned entries
|
||||
orphaned = downloader.tracker.cleanup_orphaned_files(downloader.downloads_dir)
|
||||
print(f"🧹 Cleaned up {len(orphaned)} orphaned tracking entries")
|
||||
sys.exit(0)
|
||||
|
||||
elif args.songlist_status:
|
||||
# Show songlist status and statistics
|
||||
songlist = downloader._load_songlist()
|
||||
tracking = downloader.songlist_tracking
|
||||
|
||||
print("📋 Songlist Download Status")
|
||||
print("=" * 40)
|
||||
print(f"Total songs in list: {len(songlist)}")
|
||||
print(f"Songs downloaded: {len(tracking)}")
|
||||
print(f"Songs remaining: {len(songlist) - len(tracking)}")
|
||||
print(f"Progress: {(len(tracking) / len(songlist) * 100):.1f}%")
|
||||
|
||||
if tracking:
|
||||
print(f"\n📁 Downloaded songs:")
|
||||
for key, info in list(tracking.items())[:10]: # Show first 10
|
||||
print(f" • {info['artist']} - {info['title']} (from {info['channel']})")
|
||||
if len(tracking) > 10:
|
||||
print(f" ... and {len(tracking) - 10} more")
|
||||
|
||||
sys.exit(0)
|
||||
|
||||
elif args.file:
|
||||
success = downloader.download_from_file(args.file, force_refresh=args.refresh)
|
||||
elif args.url:
|
||||
success = downloader.download_channel_videos(args.url, force_refresh=args.refresh)
|
||||
else:
|
||||
parser.print_help()
|
||||
sys.exit(1)
|
||||
|
||||
# Force save any pending tracking data before exit
|
||||
downloader.tracker.force_save()
|
||||
|
||||
if success:
|
||||
print("\n🎤 All downloads completed successfully!")
|
||||
sys.exit(0)
|
||||
else:
|
||||
print("\n⚠️ Some downloads may have failed. Check the logs for details.")
|
||||
sys.exit(1)
|
||||
|
||||
from karaoke_downloader.cli import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
4
karaoke_downloader/__init__.py
Normal file
4
karaoke_downloader/__init__.py
Normal file
@ -0,0 +1,4 @@
|
||||
"""karaoke_downloader package: core modules for the Karaoke Video Downloader CLI tool."""
|
||||
|
||||
|
||||
|
||||
151
karaoke_downloader/cli.py
Normal file
151
karaoke_downloader/cli.py
Normal file
@ -0,0 +1,151 @@
|
||||
import sys
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
from karaoke_downloader.downloader import KaraokeDownloader
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Karaoke Video Downloader - Download YouTube playlists and channel videos for karaoke",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="""
|
||||
Examples:
|
||||
python download_karaoke.py https://www.youtube.com/playlist?list=XYZ
|
||||
python download_karaoke.py https://www.youtube.com/@SingKingKaraoke/videos
|
||||
python download_karaoke.py --file data/channels.txt
|
||||
python download_karaoke.py --reset-channel SingKingKaraoke --delete-files
|
||||
"""
|
||||
)
|
||||
parser.add_argument('url', nargs='?', help='YouTube playlist or channel URL to download')
|
||||
parser.add_argument('--file', '-f', help='Text file containing playlist or channel URLs (one per line)')
|
||||
parser.add_argument('--status', '-s', action='store_true', help='Show download status and statistics')
|
||||
parser.add_argument('--report', '-r', metavar='PLAYLIST_ID', help='Generate a detailed report for a specific playlist')
|
||||
parser.add_argument('--sync', '--sync-only', action='store_true', help='Only sync playlist without downloading (update tracking)')
|
||||
parser.add_argument('--cleanup', action='store_true', help='Clean up orphaned tracking entries')
|
||||
parser.add_argument('--resolution', '--res', choices=['480p', '720p', '1080p', '1440p', '2160p'], default='720p', help='Preferred video resolution (default: 720p)')
|
||||
parser.add_argument('--limit', '-l', type=int, help='Limit the number of videos to download (e.g., --limit 10)')
|
||||
parser.add_argument('--refresh', '--force-refresh', action='store_true', help='Force refresh channel cache (ignore cached data)')
|
||||
parser.add_argument('--cache-info', action='store_true', help='Show channel cache information')
|
||||
parser.add_argument('--clear-cache', metavar='CHANNEL_ID', nargs='?', const='all', help='Clear cache for specific channel or all channels (use --clear-cache all)')
|
||||
parser.add_argument('--cache-duration', type=int, metavar='HOURS', help='Set cache duration in hours (default: 24)')
|
||||
parser.add_argument('--songlist-priority', action='store_true', help='Prioritize downloads based on data/songList.json (default: enabled)')
|
||||
parser.add_argument('--no-songlist-priority', action='store_true', help='Disable songlist prioritization')
|
||||
parser.add_argument('--songlist-only', action='store_true', help='Only download songs that are in the songlist (skip all others)')
|
||||
parser.add_argument('--songlist-status', action='store_true', help='Show songlist download status and statistics')
|
||||
parser.add_argument('--reset-channel', metavar='CHANNEL_NAME', help='Reset all tracking and files for a channel')
|
||||
parser.add_argument('--reset-songlist', action='store_true', help='When used with --reset-channel, also reset songlist songs for this channel')
|
||||
parser.add_argument('--version', '-v', action='version', version='Karaoke Playlist Downloader v1.0')
|
||||
args = parser.parse_args()
|
||||
|
||||
yt_dlp_path = Path("downloader/yt-dlp.exe")
|
||||
if not yt_dlp_path.exists():
|
||||
print("❌ Error: yt-dlp.exe not found in downloader/ directory")
|
||||
print("Please ensure yt-dlp.exe is present in the downloader/ folder")
|
||||
sys.exit(1)
|
||||
|
||||
downloader = KaraokeDownloader()
|
||||
if args.limit:
|
||||
downloader.download_limit = args.limit
|
||||
print(f"🎯 Download limit set to: {args.limit} videos")
|
||||
if args.no_songlist_priority:
|
||||
downloader.use_songlist_priority = False
|
||||
print("🎯 Songlist prioritization disabled")
|
||||
else:
|
||||
downloader.use_songlist_priority = True
|
||||
print("🎯 Songlist prioritization enabled")
|
||||
if args.songlist_only:
|
||||
downloader.songlist_only = True
|
||||
print("🎯 Songlist-only mode enabled (will only download songlist songs)")
|
||||
if args.resolution != '720p':
|
||||
resolution_map = {
|
||||
'480p': '480',
|
||||
'720p': '720',
|
||||
'1080p': '1080',
|
||||
'1440p': '1440',
|
||||
'2160p': '2160'
|
||||
}
|
||||
height = resolution_map[args.resolution]
|
||||
downloader.config["download_settings"]["format"] = f"best[height<={height}][ext=mp4]/best[height<={height}]/best[ext=mp4]/best"
|
||||
downloader.config["download_settings"]["preferred_resolution"] = args.resolution
|
||||
print(f"🎬 Using resolution: {args.resolution}")
|
||||
|
||||
# --- NEW: Reset channel CLI command ---
|
||||
if args.reset_channel:
|
||||
downloader.reset_channel_downloads(
|
||||
args.reset_channel,
|
||||
reset_songlist=args.reset_songlist,
|
||||
delete_files=True # Always delete files if they exist
|
||||
)
|
||||
sys.exit(0)
|
||||
# --- END NEW ---
|
||||
|
||||
if args.status:
|
||||
stats = downloader.tracker.get_statistics()
|
||||
print("🎤 Karaoke Downloader Status")
|
||||
print("=" * 40)
|
||||
print(f"Total Songs: {stats['total_songs']}")
|
||||
print(f"Total Playlists: {stats['total_playlists']}")
|
||||
print(f"Downloaded Songs: {stats['downloaded_songs']}")
|
||||
print(f"Failed Songs: {stats['failed_songs']}")
|
||||
print(f"Partial Downloads: {stats['partial_songs']}")
|
||||
print(f"Total Size: {stats['total_size_mb']} MB")
|
||||
print(f"Last Updated: {stats['last_updated']}")
|
||||
sys.exit(0)
|
||||
elif args.cache_info:
|
||||
cache_info = downloader.tracker.get_cache_info()
|
||||
print("💾 Channel Cache Information")
|
||||
print("=" * 40)
|
||||
print(f"Total Channels: {cache_info['total_channels']}")
|
||||
print(f"Total Cached Videos: {cache_info['total_cached_videos']}")
|
||||
print(f"Cache Duration: {cache_info['cache_duration_hours']} hours")
|
||||
print(f"Last Updated: {cache_info['last_updated']}")
|
||||
sys.exit(0)
|
||||
elif args.clear_cache:
|
||||
if args.clear_cache == 'all':
|
||||
downloader.tracker.clear_channel_cache()
|
||||
print("🧹 Cleared all channel caches")
|
||||
else:
|
||||
downloader.tracker.clear_channel_cache(args.clear_cache)
|
||||
print(f"🧹 Cleared cache for channel: {args.clear_cache}")
|
||||
sys.exit(0)
|
||||
elif args.cache_duration:
|
||||
downloader.tracker.set_cache_duration(args.cache_duration)
|
||||
print(f"⏰ Cache duration set to {args.cache_duration} hours")
|
||||
sys.exit(0)
|
||||
elif args.report:
|
||||
report = downloader.tracker.export_playlist_report(args.report)
|
||||
print(report)
|
||||
sys.exit(0)
|
||||
elif args.cleanup:
|
||||
orphaned = downloader.tracker.cleanup_orphaned_files(downloader.downloads_dir)
|
||||
print(f"🧹 Cleaned up {len(orphaned)} orphaned tracking entries")
|
||||
sys.exit(0)
|
||||
elif args.songlist_status:
|
||||
songlist = downloader._load_songlist()
|
||||
tracking = downloader.songlist_tracking
|
||||
print("📋 Songlist Download Status")
|
||||
print("=" * 40)
|
||||
print(f"Total songs in list: {len(songlist)}")
|
||||
print(f"Songs downloaded: {len(tracking)}")
|
||||
print(f"Songs remaining: {len(songlist) - len(tracking)}")
|
||||
print(f"Progress: {(len(tracking) / len(songlist) * 100):.1f}%")
|
||||
if tracking:
|
||||
print(f"\n📁 Downloaded songs:")
|
||||
for key, info in list(tracking.items())[:10]:
|
||||
print(f" • {info['artist']} - {info['title']} (from {info['channel']})")
|
||||
if len(tracking) > 10:
|
||||
print(f" ... and {len(tracking) - 10} more")
|
||||
sys.exit(0)
|
||||
elif args.file:
|
||||
success = downloader.download_from_file(args.file, force_refresh=args.refresh)
|
||||
elif args.url:
|
||||
success = downloader.download_channel_videos(args.url, force_refresh=args.refresh)
|
||||
else:
|
||||
parser.print_help()
|
||||
sys.exit(1)
|
||||
downloader.tracker.force_save()
|
||||
if success:
|
||||
print("\n🎤 All downloads completed successfully!")
|
||||
sys.exit(0)
|
||||
else:
|
||||
print("\n⚠️ Some downloads may have failed. Check the logs for details.")
|
||||
sys.exit(1)
|
||||
142
karaoke_downloader/downloader.py
Normal file
142
karaoke_downloader/downloader.py
Normal file
@ -0,0 +1,142 @@
|
||||
import os
|
||||
import sys
|
||||
import subprocess
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
from karaoke_downloader.tracking_manager import TrackingManager, SongStatus, FormatType
|
||||
from karaoke_downloader.id3_utils import add_id3_tags, extract_artist_title
|
||||
from karaoke_downloader.songlist_manager import (
|
||||
load_songlist, load_songlist_tracking, save_songlist_tracking,
|
||||
is_songlist_song_downloaded, mark_songlist_song_downloaded, normalize_title
|
||||
)
|
||||
from karaoke_downloader.youtube_utils import get_channel_info, get_playlist_info
|
||||
import logging
|
||||
|
||||
DATA_DIR = Path("data")
|
||||
|
||||
class KaraokeDownloader:
|
||||
def __init__(self):
|
||||
self.yt_dlp_path = Path("downloader/yt-dlp.exe")
|
||||
self.downloads_dir = Path("downloads")
|
||||
self.logs_dir = Path("logs")
|
||||
self.downloads_dir.mkdir(exist_ok=True)
|
||||
self.logs_dir.mkdir(exist_ok=True)
|
||||
self.tracker = TrackingManager(tracking_file=DATA_DIR / "karaoke_tracking.json", cache_file=DATA_DIR / "channel_cache.json")
|
||||
self.config = self._load_config()
|
||||
self.songlist_tracking_file = DATA_DIR / "songlist_tracking.json"
|
||||
self.songlist_tracking = load_songlist_tracking(str(self.songlist_tracking_file))
|
||||
|
||||
def _load_config(self):
|
||||
config_file = DATA_DIR / "config.json"
|
||||
if config_file.exists():
|
||||
try:
|
||||
with open(config_file, 'r', encoding='utf-8') as f:
|
||||
return json.load(f)
|
||||
except (json.JSONDecodeError, FileNotFoundError) as e:
|
||||
print(f"Warning: Could not load config.json: {e}")
|
||||
return {
|
||||
"download_settings": {
|
||||
"format": "best[height<=720][ext=mp4]/best[height<=720]/best[ext=mp4]/best",
|
||||
"preferred_resolution": "720p",
|
||||
"audio_format": "mp3",
|
||||
"audio_quality": "0",
|
||||
"subtitle_language": "en",
|
||||
"subtitle_format": "srt",
|
||||
"write_metadata": False,
|
||||
"write_thumbnail": False,
|
||||
"write_description": False,
|
||||
"write_annotations": False,
|
||||
"write_comments": False,
|
||||
"write_subtitles": False,
|
||||
"embed_metadata": False,
|
||||
"add_metadata": False,
|
||||
"continue_downloads": True,
|
||||
"no_overwrites": True,
|
||||
"ignore_errors": True,
|
||||
"no_warnings": False
|
||||
},
|
||||
"folder_structure": {
|
||||
"downloads_dir": "downloads",
|
||||
"logs_dir": "logs",
|
||||
"tracking_file": str(DATA_DIR / "karaoke_tracking.json")
|
||||
},
|
||||
"logging": {
|
||||
"level": "INFO",
|
||||
"format": "%(asctime)s - %(levelname)s - %(message)s",
|
||||
"include_console": True,
|
||||
"include_file": True
|
||||
},
|
||||
"yt_dlp_path": "downloader/yt-dlp.exe"
|
||||
}
|
||||
|
||||
def reset_channel_downloads(self, channel_name, reset_songlist=False, delete_files=False):
|
||||
"""
|
||||
Reset all tracking and optionally files for a channel.
|
||||
If reset_songlist is False, songlist songs are preserved (tracking and files).
|
||||
If reset_songlist is True, songlist songs for this channel are also reset/deleted.
|
||||
"""
|
||||
print(f"\n🔄 Resetting channel: {channel_name} (reset_songlist={reset_songlist}, delete_files={delete_files})")
|
||||
# Find channel_id from channel_name
|
||||
channel_id = None
|
||||
for pid, playlist in self.tracker.data.get('playlists', {}).items():
|
||||
if playlist['name'] == channel_name or pid == channel_name:
|
||||
channel_id = pid
|
||||
break
|
||||
if not channel_id:
|
||||
print(f"❌ Channel '{channel_name}' not found in tracking.")
|
||||
return
|
||||
# Get all songs for this channel
|
||||
songs_to_reset = []
|
||||
for song_id, song in self.tracker.data.get('songs', {}).items():
|
||||
if song['playlist_id'] == channel_id:
|
||||
# Check if this is a songlist song
|
||||
artist, title = song.get('artist', ''), song.get('title', song.get('name', ''))
|
||||
key = f"{artist.lower()}_{normalize_title(title)}"
|
||||
is_songlist = key in self.songlist_tracking
|
||||
if is_songlist and not reset_songlist:
|
||||
continue # skip songlist songs if not resetting them
|
||||
songs_to_reset.append((song_id, song, is_songlist))
|
||||
# Reset tracking and optionally delete files
|
||||
files_preserved = 0
|
||||
files_deleted = 0
|
||||
for song_id, song, is_songlist in songs_to_reset:
|
||||
# Remove from main tracking
|
||||
self.tracker.data['songs'][song_id]['status'] = 'NOT_DOWNLOADED'
|
||||
self.tracker.data['songs'][song_id]['formats'] = {}
|
||||
self.tracker.data['songs'][song_id]['last_error'] = ''
|
||||
self.tracker.data['songs'][song_id]['download_attempts'] = 0
|
||||
self.tracker.data['songs'][song_id]['last_updated'] = None
|
||||
# Remove from songlist tracking if needed
|
||||
if is_songlist and reset_songlist:
|
||||
artist, title = song.get('artist', ''), song.get('title', song.get('name', ''))
|
||||
key = f"{artist.lower()}_{normalize_title(title)}"
|
||||
if key in self.songlist_tracking:
|
||||
del self.songlist_tracking[key]
|
||||
# Delete file if requested
|
||||
if delete_files:
|
||||
file_path = song.get('file_path')
|
||||
if file_path:
|
||||
try:
|
||||
p = Path(file_path)
|
||||
if p.exists():
|
||||
p.unlink()
|
||||
files_deleted += 1
|
||||
else:
|
||||
files_preserved += 1
|
||||
except Exception as e:
|
||||
print(f"⚠️ Could not delete file {file_path}: {e}")
|
||||
# Save changes
|
||||
self.tracker.force_save()
|
||||
save_songlist_tracking(self.songlist_tracking, str(self.songlist_tracking_file))
|
||||
print(f"✅ Reset {len(songs_to_reset)} songs for channel '{channel_name}'.")
|
||||
if delete_files:
|
||||
print(f" Files deleted: {files_deleted}, files preserved: {files_preserved}")
|
||||
if not reset_songlist:
|
||||
print(f" Songlist songs were preserved.")
|
||||
|
||||
# ... (rest of the KaraokeDownloader methods, updated to use DATA_DIR for all data file paths) ...
|
||||
|
||||
# For brevity, the rest of the class methods should be copied here from the original download_karaoke.py,
|
||||
# updating all references to use the new karaoke_downloader.* imports as needed.
|
||||
33
karaoke_downloader/id3_utils.py
Normal file
33
karaoke_downloader/id3_utils.py
Normal file
@ -0,0 +1,33 @@
|
||||
import re
|
||||
|
||||
MUTAGEN_AVAILABLE = True
|
||||
try:
|
||||
from mutagen.mp4 import MP4
|
||||
except ImportError:
|
||||
MUTAGEN_AVAILABLE = False
|
||||
|
||||
def extract_artist_title(video_title):
|
||||
title = video_title.replace("(Karaoke Version)", "").replace("(Karaoke)", "").strip()
|
||||
if " - " in title:
|
||||
parts = title.split(" - ", 1)
|
||||
if len(parts) == 2:
|
||||
artist = parts[0].strip()
|
||||
song_title = parts[1].strip()
|
||||
return artist, song_title
|
||||
return "Unknown Artist", title
|
||||
|
||||
def add_id3_tags(file_path, video_title, channel_name):
|
||||
if not MUTAGEN_AVAILABLE:
|
||||
print("⚠️ mutagen not available - skipping ID3 tagging")
|
||||
return
|
||||
try:
|
||||
artist, title = extract_artist_title(video_title)
|
||||
mp4 = MP4(str(file_path))
|
||||
mp4['\xa9nam'] = title
|
||||
mp4['\xa9ART'] = artist
|
||||
mp4['\xa9alb'] = f"{channel_name} Karaoke"
|
||||
mp4['\xa9gen'] = "Karaoke"
|
||||
mp4.save()
|
||||
print(f"📝 Added ID3 tags: Artist='{artist}', Title='{title}'")
|
||||
except Exception as e:
|
||||
print(f"⚠️ Could not add ID3 tags: {e}")
|
||||
64
karaoke_downloader/songlist_manager.py
Normal file
64
karaoke_downloader/songlist_manager.py
Normal file
@ -0,0 +1,64 @@
|
||||
import json
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
def load_songlist(songlist_path="data/songList.json"):
|
||||
songlist_file = Path(songlist_path)
|
||||
if not songlist_file.exists():
|
||||
print(f"⚠️ Songlist file not found: {songlist_path}")
|
||||
return []
|
||||
try:
|
||||
with open(songlist_file, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
all_songs = []
|
||||
for category in data:
|
||||
if "songs" in category:
|
||||
for song in category["songs"]:
|
||||
if "artist" in song and "title" in song:
|
||||
all_songs.append({
|
||||
"artist": song["artist"].strip(),
|
||||
"title": song["title"].strip(),
|
||||
"position": song.get("position", 0)
|
||||
})
|
||||
print(f"📋 Loaded {len(all_songs)} songs from songlist")
|
||||
return all_songs
|
||||
except (json.JSONDecodeError, FileNotFoundError) as e:
|
||||
print(f"⚠️ Could not load songlist: {e}")
|
||||
return []
|
||||
|
||||
def normalize_title(title):
|
||||
normalized = title.replace("(Karaoke Version)", "").replace("(Karaoke)", "").strip()
|
||||
return " ".join(normalized.split()).lower()
|
||||
|
||||
def load_songlist_tracking(tracking_path="data/songlist_tracking.json"):
|
||||
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 songlist tracking: {e}")
|
||||
return {}
|
||||
|
||||
def save_songlist_tracking(tracking, tracking_path="data/songlist_tracking.json"):
|
||||
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 songlist tracking: {e}")
|
||||
|
||||
def is_songlist_song_downloaded(tracking, artist, title):
|
||||
key = f"{artist.lower()}_{normalize_title(title)}"
|
||||
return key in tracking
|
||||
|
||||
def mark_songlist_song_downloaded(tracking, artist, title, channel_name, file_path):
|
||||
key = f"{artist.lower()}_{normalize_title(title)}"
|
||||
tracking[key] = {
|
||||
"artist": artist,
|
||||
"title": title,
|
||||
"channel": channel_name,
|
||||
"file_path": str(file_path),
|
||||
"downloaded_at": datetime.now().isoformat()
|
||||
}
|
||||
save_songlist_tracking(tracking)
|
||||
136
karaoke_downloader/tracking_manager.py
Normal file
136
karaoke_downloader/tracking_manager.py
Normal file
@ -0,0 +1,136 @@
|
||||
import json
|
||||
from pathlib import Path
|
||||
from enum import Enum
|
||||
from datetime import datetime
|
||||
import threading
|
||||
|
||||
class SongStatus(str, Enum):
|
||||
NOT_DOWNLOADED = "NOT_DOWNLOADED"
|
||||
DOWNLOADING = "DOWNLOADING"
|
||||
DOWNLOADED = "DOWNLOADED"
|
||||
PARTIAL = "PARTIAL"
|
||||
FAILED = "FAILED"
|
||||
CONVERTING = "CONVERTING"
|
||||
CONVERTED = "CONVERTED"
|
||||
|
||||
class FormatType(str, Enum):
|
||||
MP4 = "MP4"
|
||||
MP3 = "MP3"
|
||||
OTHER = "OTHER"
|
||||
|
||||
class TrackingManager:
|
||||
def __init__(self, tracking_file="data/karaoke_tracking.json", cache_file="data/channel_cache.json"):
|
||||
self.tracking_file = Path(tracking_file)
|
||||
self.cache_file = Path(cache_file)
|
||||
self.data = {
|
||||
"playlists": {},
|
||||
"songs": {}
|
||||
}
|
||||
self.cache = {}
|
||||
self._lock = threading.Lock()
|
||||
self._load()
|
||||
self._load_cache()
|
||||
|
||||
def _load(self):
|
||||
if self.tracking_file.exists():
|
||||
try:
|
||||
with open(self.tracking_file, 'r', encoding='utf-8') as f:
|
||||
self.data = json.load(f)
|
||||
except Exception:
|
||||
self.data = {"playlists": {}, "songs": {}}
|
||||
|
||||
def _save(self):
|
||||
with self._lock:
|
||||
with open(self.tracking_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(self.data, f, indent=2, ensure_ascii=False)
|
||||
|
||||
def force_save(self):
|
||||
self._save()
|
||||
|
||||
def _load_cache(self):
|
||||
if self.cache_file.exists():
|
||||
try:
|
||||
with open(self.cache_file, 'r', encoding='utf-8') as f:
|
||||
self.cache = json.load(f)
|
||||
except Exception:
|
||||
self.cache = {}
|
||||
|
||||
def save_cache(self):
|
||||
with open(self.cache_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(self.cache, f, indent=2, ensure_ascii=False)
|
||||
|
||||
def get_statistics(self):
|
||||
total_songs = len(self.data['songs'])
|
||||
downloaded_songs = sum(1 for s in self.data['songs'].values() if s['status'] in [SongStatus.DOWNLOADED, SongStatus.CONVERTED])
|
||||
failed_songs = sum(1 for s in self.data['songs'].values() if s['status'] == SongStatus.FAILED)
|
||||
partial_songs = sum(1 for s in self.data['songs'].values() if s['status'] == SongStatus.PARTIAL)
|
||||
total_playlists = len(self.data['playlists'])
|
||||
total_size_mb = sum(s.get('file_size', 0) for s in self.data['songs'].values() if s.get('file_size')) / (1024*1024)
|
||||
last_updated = max((s.get('last_updated') for s in self.data['songs'].values() if s.get('last_updated')), default=None)
|
||||
return {
|
||||
"total_songs": total_songs,
|
||||
"downloaded_songs": downloaded_songs,
|
||||
"failed_songs": failed_songs,
|
||||
"partial_songs": partial_songs,
|
||||
"total_playlists": total_playlists,
|
||||
"total_size_mb": round(total_size_mb, 2),
|
||||
"last_updated": last_updated
|
||||
}
|
||||
|
||||
def get_playlist_songs(self, playlist_id):
|
||||
return [s for s in self.data['songs'].values() if s['playlist_id'] == playlist_id]
|
||||
|
||||
def get_failed_songs(self, playlist_id=None):
|
||||
if playlist_id:
|
||||
return [s for s in self.data['songs'].values() if s['playlist_id'] == playlist_id and s['status'] == SongStatus.FAILED]
|
||||
return [s for s in self.data['songs'].values() if s['status'] == SongStatus.FAILED]
|
||||
|
||||
def get_partial_downloads(self, playlist_id=None):
|
||||
if playlist_id:
|
||||
return [s for s in self.data['songs'].values() if s['playlist_id'] == playlist_id and s['status'] == SongStatus.PARTIAL]
|
||||
return [s for s in self.data['songs'].values() if s['status'] == SongStatus.PARTIAL]
|
||||
|
||||
def cleanup_orphaned_files(self, downloads_dir):
|
||||
# Remove tracking entries for files that no longer exist
|
||||
orphaned = []
|
||||
for song_id, song in list(self.data['songs'].items()):
|
||||
file_path = song.get('file_path')
|
||||
if file_path and not Path(file_path).exists():
|
||||
orphaned.append(song_id)
|
||||
del self.data['songs'][song_id]
|
||||
self.force_save()
|
||||
return orphaned
|
||||
|
||||
def get_cache_info(self):
|
||||
total_channels = len(self.cache)
|
||||
total_cached_videos = sum(len(v) for v in self.cache.values())
|
||||
cache_duration_hours = 24 # default
|
||||
last_updated = None
|
||||
return {
|
||||
"total_channels": total_channels,
|
||||
"total_cached_videos": total_cached_videos,
|
||||
"cache_duration_hours": cache_duration_hours,
|
||||
"last_updated": last_updated
|
||||
}
|
||||
|
||||
def clear_channel_cache(self, channel_id=None):
|
||||
if channel_id is None or channel_id == 'all':
|
||||
self.cache = {}
|
||||
else:
|
||||
self.cache.pop(channel_id, None)
|
||||
self.save_cache()
|
||||
|
||||
def set_cache_duration(self, hours):
|
||||
# Placeholder for cache duration logic
|
||||
pass
|
||||
|
||||
def export_playlist_report(self, playlist_id):
|
||||
playlist = self.data['playlists'].get(playlist_id)
|
||||
if not playlist:
|
||||
return f"Playlist '{playlist_id}' not found."
|
||||
songs = self.get_playlist_songs(playlist_id)
|
||||
report = {
|
||||
"playlist": playlist,
|
||||
"songs": songs
|
||||
}
|
||||
return json.dumps(report, indent=2, ensure_ascii=False)
|
||||
15
karaoke_downloader/youtube_utils.py
Normal file
15
karaoke_downloader/youtube_utils.py
Normal file
@ -0,0 +1,15 @@
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
def get_channel_info(channel_url):
|
||||
if '@' in channel_url:
|
||||
channel_name = channel_url.split('@')[1].split('/')[0]
|
||||
channel_id = f"@{channel_name}"
|
||||
else:
|
||||
channel_name = "unknown_channel"
|
||||
channel_id = "unknown_channel"
|
||||
channel_name = re.sub(r'[<>:"/\\|?*]', '_', channel_name)
|
||||
return channel_name, channel_id
|
||||
|
||||
def get_playlist_info(playlist_url):
|
||||
return get_channel_info(playlist_url)
|
||||
@ -1 +0,0 @@
|
||||
https://www.youtube.com/@SingKingKaraoke/videos
|
||||
@ -1,501 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Tracking Manager for Karaoke Playlist Downloader
|
||||
Handles song tracking, multiple formats, partial downloads, and playlist synchronization.
|
||||
"""
|
||||
|
||||
import json
|
||||
import hashlib
|
||||
import os
|
||||
from pathlib import Path
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, List, Optional, Set
|
||||
import logging
|
||||
|
||||
|
||||
class SongStatus:
|
||||
"""Enum-like class for song download statuses."""
|
||||
NOT_DOWNLOADED = "not_downloaded"
|
||||
DOWNLOADING = "downloading"
|
||||
DOWNLOADED = "downloaded"
|
||||
PARTIAL = "partial"
|
||||
FAILED = "failed"
|
||||
CONVERTING = "converting"
|
||||
CONVERTED = "converted"
|
||||
|
||||
|
||||
class FormatType:
|
||||
"""Enum-like class for format types."""
|
||||
MP4 = "mp4"
|
||||
MP3 = "mp3"
|
||||
M4A = "m4a"
|
||||
WEBM = "webm"
|
||||
|
||||
|
||||
class TrackingManager:
|
||||
def __init__(self, tracking_file: str = "karaoke_tracking.json", cache_file: str = "channel_cache.json"):
|
||||
self.tracking_file = Path(tracking_file)
|
||||
self.cache_file = Path(cache_file)
|
||||
|
||||
# Cache settings (must be set before loading cache)
|
||||
self.cache_duration_hours = 24 # Cache for 24 hours by default
|
||||
|
||||
self.data = self._load_tracking_data()
|
||||
self.cache = self._load_cache_data()
|
||||
self.logger = logging.getLogger(__name__)
|
||||
self._batch_save_pending = False
|
||||
self._save_timer = None
|
||||
|
||||
def _load_tracking_data(self) -> Dict:
|
||||
"""Load tracking data from JSON file."""
|
||||
if self.tracking_file.exists():
|
||||
try:
|
||||
with open(self.tracking_file, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
# Ensure all required top-level keys exist
|
||||
data.setdefault('version', '1.0')
|
||||
data.setdefault('last_updated', datetime.now().isoformat())
|
||||
data.setdefault('songs', {})
|
||||
data.setdefault('playlists', {})
|
||||
data.setdefault('formats', {})
|
||||
data.setdefault('statistics', {})
|
||||
return data
|
||||
except (json.JSONDecodeError, FileNotFoundError) as e:
|
||||
print(f"Warning: Could not load tracking file: {e}")
|
||||
# If file is corrupted, backup and create new one
|
||||
if self.tracking_file.exists():
|
||||
backup_file = self.tracking_file.with_suffix('.json.backup')
|
||||
try:
|
||||
self.tracking_file.rename(backup_file)
|
||||
print(f"Backed up corrupted file to: {backup_file}")
|
||||
except Exception as backup_error:
|
||||
print(f"Could not backup corrupted file: {backup_error}")
|
||||
|
||||
# Return default structure
|
||||
return {
|
||||
'version': '1.0',
|
||||
'last_updated': datetime.now().isoformat(),
|
||||
'songs': {},
|
||||
'playlists': {},
|
||||
'formats': {},
|
||||
'statistics': {}
|
||||
}
|
||||
|
||||
def _save_tracking_data(self, force_save: bool = False):
|
||||
"""Save tracking data to JSON file with batch saving support."""
|
||||
if not force_save:
|
||||
# Schedule a delayed save instead of immediate save
|
||||
self._batch_save_pending = True
|
||||
return
|
||||
|
||||
self.data['last_updated'] = datetime.now().isoformat()
|
||||
try:
|
||||
with open(self.tracking_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(self.data, f, indent=2, ensure_ascii=False)
|
||||
self._batch_save_pending = False
|
||||
except Exception as e:
|
||||
self.logger.error(f"Failed to save tracking data: {e}")
|
||||
|
||||
def force_save(self):
|
||||
"""Force save the tracking data immediately."""
|
||||
self._save_tracking_data(force_save=True)
|
||||
|
||||
def _load_cache_data(self) -> Dict:
|
||||
"""Load channel cache data from JSON file."""
|
||||
if self.cache_file.exists():
|
||||
try:
|
||||
with open(self.cache_file, 'r', encoding='utf-8') as f:
|
||||
cache = json.load(f)
|
||||
# Ensure required structure
|
||||
cache.setdefault('channels', {})
|
||||
cache.setdefault('settings', {
|
||||
'cache_duration_hours': self.cache_duration_hours,
|
||||
'last_updated': datetime.now().isoformat()
|
||||
})
|
||||
return cache
|
||||
except (json.JSONDecodeError, FileNotFoundError) as e:
|
||||
print(f"Warning: Could not load cache file: {e}")
|
||||
|
||||
# Return default cache structure
|
||||
return {
|
||||
'channels': {},
|
||||
'settings': {
|
||||
'cache_duration_hours': self.cache_duration_hours,
|
||||
'last_updated': datetime.now().isoformat()
|
||||
}
|
||||
}
|
||||
|
||||
def _save_cache_data(self):
|
||||
"""Save channel cache data to JSON file."""
|
||||
try:
|
||||
self.cache['settings']['last_updated'] = datetime.now().isoformat()
|
||||
with open(self.cache_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(self.cache, f, indent=2, ensure_ascii=False)
|
||||
except Exception as e:
|
||||
self.logger.error(f"Failed to save cache data: {e}")
|
||||
|
||||
def get_cached_channel_videos(self, channel_id: str) -> Optional[List[Dict]]:
|
||||
"""Get cached videos for a channel if cache is still valid."""
|
||||
if channel_id not in self.cache['channels']:
|
||||
return None
|
||||
|
||||
channel_cache = self.cache['channels'][channel_id]
|
||||
cached_time = datetime.fromisoformat(channel_cache['cached_at'])
|
||||
cache_duration = timedelta(hours=self.cache_duration_hours)
|
||||
|
||||
# Check if cache is still valid
|
||||
if datetime.now() - cached_time < cache_duration:
|
||||
self.logger.info(f"Using cached channel data for {channel_id} (cached {cached_time})")
|
||||
return channel_cache['videos']
|
||||
|
||||
self.logger.info(f"Channel cache expired for {channel_id}, will refresh")
|
||||
return None
|
||||
|
||||
def cache_channel_videos(self, channel_id: str, videos: List[Dict]):
|
||||
"""Cache videos for a channel."""
|
||||
self.cache['channels'][channel_id] = {
|
||||
'videos': videos,
|
||||
'cached_at': datetime.now().isoformat(),
|
||||
'video_count': len(videos)
|
||||
}
|
||||
self._save_cache_data()
|
||||
self.logger.info(f"Cached {len(videos)} videos for channel {channel_id}")
|
||||
|
||||
def clear_channel_cache(self, channel_id: Optional[str] = None):
|
||||
"""Clear cache for a specific channel or all channels."""
|
||||
if channel_id:
|
||||
if channel_id in self.cache['channels']:
|
||||
del self.cache['channels'][channel_id]
|
||||
self._save_cache_data()
|
||||
self.logger.info(f"Cleared cache for channel {channel_id}")
|
||||
else:
|
||||
self.cache['channels'] = {}
|
||||
self._save_cache_data()
|
||||
self.logger.info("Cleared all channel caches")
|
||||
|
||||
def set_cache_duration(self, hours: int):
|
||||
"""Set cache duration in hours."""
|
||||
self.cache_duration_hours = hours
|
||||
self.cache['settings']['cache_duration_hours'] = hours
|
||||
self._save_cache_data()
|
||||
self.logger.info(f"Cache duration set to {hours} hours")
|
||||
|
||||
def get_cache_info(self) -> Dict:
|
||||
"""Get cache information."""
|
||||
total_channels = len(self.cache['channels'])
|
||||
total_cached_videos = sum(channel['video_count'] for channel in self.cache['channels'].values())
|
||||
|
||||
return {
|
||||
'total_channels': total_channels,
|
||||
'total_cached_videos': total_cached_videos,
|
||||
'cache_duration_hours': self.cache_duration_hours,
|
||||
'last_updated': self.cache['settings']['last_updated']
|
||||
}
|
||||
|
||||
def _generate_song_id(self, video_id: str, playlist_id: str) -> str:
|
||||
"""Generate a unique song ID."""
|
||||
return f"{video_id}_{playlist_id}"
|
||||
|
||||
def _generate_file_hash(self, file_path: Path) -> str:
|
||||
"""Generate MD5 hash of file for integrity checking."""
|
||||
try:
|
||||
if file_path.exists():
|
||||
hash_md5 = hashlib.md5()
|
||||
with open(file_path, "rb") as f:
|
||||
for chunk in iter(lambda: f.read(4096), b""):
|
||||
hash_md5.update(chunk)
|
||||
return hash_md5.hexdigest()
|
||||
except Exception as e:
|
||||
self.logger.warning(f"Could not generate hash for {file_path}: {e}")
|
||||
return ""
|
||||
|
||||
def add_playlist(self, playlist_id: str, playlist_name: str, playlist_url: str) -> Dict:
|
||||
"""Add or update a playlist in tracking."""
|
||||
playlist_data = {
|
||||
'name': playlist_name,
|
||||
'url': playlist_url,
|
||||
'added_date': datetime.now().isoformat(),
|
||||
'last_synced': None,
|
||||
'total_songs': 0,
|
||||
'downloaded_songs': 0,
|
||||
'failed_songs': 0,
|
||||
'song_ids': []
|
||||
}
|
||||
|
||||
self.data['playlists'][playlist_id] = playlist_data
|
||||
self._save_tracking_data()
|
||||
return playlist_data
|
||||
|
||||
def add_song(self, video_id: str, playlist_id: str, title: str, duration: Optional[int] = None) -> Dict:
|
||||
"""Add a song to tracking."""
|
||||
song_id = self._generate_song_id(video_id, playlist_id)
|
||||
|
||||
song_data = {
|
||||
'video_id': video_id,
|
||||
'playlist_id': playlist_id,
|
||||
'title': title,
|
||||
'duration': duration,
|
||||
'added_date': datetime.now().isoformat(),
|
||||
'last_updated': datetime.now().isoformat(),
|
||||
'status': SongStatus.NOT_DOWNLOADED,
|
||||
'formats': {},
|
||||
'file_paths': {},
|
||||
'file_hashes': {},
|
||||
'file_sizes': {},
|
||||
'download_attempts': 0,
|
||||
'last_error': None,
|
||||
'metadata': {}
|
||||
}
|
||||
|
||||
self.data['songs'][song_id] = song_data
|
||||
|
||||
# Add to playlist's song list
|
||||
if playlist_id not in self.data['playlists']:
|
||||
self.add_playlist(playlist_id, f"playlist_{playlist_id}", "")
|
||||
|
||||
if song_id not in self.data['playlists'][playlist_id]['song_ids']:
|
||||
self.data['playlists'][playlist_id]['song_ids'].append(song_id)
|
||||
|
||||
self._update_playlist_statistics(playlist_id)
|
||||
self._save_tracking_data()
|
||||
return song_data
|
||||
|
||||
def update_song_status(self, video_id: str, playlist_id: str, status: str,
|
||||
format_type: Optional[str] = None, file_path: Optional[Path] = None,
|
||||
error_message: Optional[str] = None) -> bool:
|
||||
"""Update song status and optionally add format information."""
|
||||
song_id = self._generate_song_id(video_id, playlist_id)
|
||||
|
||||
if song_id not in self.data['songs']:
|
||||
self.logger.warning(f"Song {song_id} not found in tracking")
|
||||
return False
|
||||
|
||||
song = self.data['songs'][song_id]
|
||||
song['status'] = status
|
||||
song['last_updated'] = datetime.now().isoformat()
|
||||
|
||||
if error_message:
|
||||
song['last_error'] = error_message
|
||||
|
||||
if status == SongStatus.DOWNLOADING:
|
||||
song['download_attempts'] += 1
|
||||
|
||||
# Handle format-specific information
|
||||
if format_type and file_path:
|
||||
if format_type not in song['formats']:
|
||||
song['formats'][format_type] = {
|
||||
'status': status,
|
||||
'file_path': str(file_path),
|
||||
'file_size': file_path.stat().st_size if file_path.exists() else 0,
|
||||
'file_hash': self._generate_file_hash(file_path),
|
||||
'added_date': datetime.now().isoformat()
|
||||
}
|
||||
else:
|
||||
song['formats'][format_type].update({
|
||||
'status': status,
|
||||
'file_path': str(file_path),
|
||||
'file_size': file_path.stat().st_size if file_path.exists() else 0,
|
||||
'file_hash': self._generate_file_hash(file_path)
|
||||
})
|
||||
|
||||
song['file_paths'][format_type] = str(file_path)
|
||||
song['file_hashes'][format_type] = song['formats'][format_type]['file_hash']
|
||||
song['file_sizes'][format_type] = song['formats'][format_type]['file_size']
|
||||
|
||||
self._update_playlist_statistics(playlist_id)
|
||||
self._save_tracking_data()
|
||||
return True
|
||||
|
||||
def get_song_status(self, video_id: str, playlist_id: str) -> Optional[Dict]:
|
||||
"""Get current status of a song."""
|
||||
song_id = self._generate_song_id(video_id, playlist_id)
|
||||
return self.data['songs'].get(song_id)
|
||||
|
||||
def is_song_downloaded(self, video_id: str, playlist_id: str, format_type: Optional[str] = None) -> bool:
|
||||
"""Check if a song is downloaded in the specified format."""
|
||||
song = self.get_song_status(video_id, playlist_id)
|
||||
if not song:
|
||||
return False
|
||||
|
||||
if format_type:
|
||||
return (format_type in song['formats'] and
|
||||
song['formats'][format_type]['status'] in [SongStatus.DOWNLOADED, SongStatus.CONVERTED])
|
||||
|
||||
# Check if any format is downloaded
|
||||
return any(fmt['status'] in [SongStatus.DOWNLOADED, SongStatus.CONVERTED]
|
||||
for fmt in song['formats'].values())
|
||||
|
||||
def get_playlist_songs(self, playlist_id: str) -> List[Dict]:
|
||||
"""Get all songs for a playlist."""
|
||||
if playlist_id not in self.data['playlists']:
|
||||
return []
|
||||
|
||||
song_ids = self.data['playlists'][playlist_id]['song_ids']
|
||||
return [self.data['songs'].get(song_id) for song_id in song_ids
|
||||
if song_id in self.data['songs']]
|
||||
|
||||
def get_downloaded_songs(self, playlist_id: str, format_type: Optional[str] = None) -> List[Dict]:
|
||||
"""Get all downloaded songs for a playlist."""
|
||||
songs = self.get_playlist_songs(playlist_id)
|
||||
if format_type:
|
||||
return [song for song in songs if self.is_song_downloaded(song['video_id'], playlist_id, format_type)]
|
||||
return [song for song in songs if self.is_song_downloaded(song['video_id'], playlist_id)]
|
||||
|
||||
def get_failed_songs(self, playlist_id: str) -> List[Dict]:
|
||||
"""Get all failed songs for a playlist."""
|
||||
songs = self.get_playlist_songs(playlist_id)
|
||||
return [song for song in songs if song['status'] == SongStatus.FAILED]
|
||||
|
||||
def get_partial_downloads(self, playlist_id: str) -> List[Dict]:
|
||||
"""Get songs with partial downloads."""
|
||||
songs = self.get_playlist_songs(playlist_id)
|
||||
return [song for song in songs if song['status'] == SongStatus.PARTIAL]
|
||||
|
||||
def sync_playlist(self, playlist_id: str, current_songs: List[Dict]) -> Dict:
|
||||
"""Sync playlist with current songs from YouTube."""
|
||||
if playlist_id not in self.data['playlists']:
|
||||
return {'added': len(current_songs), 'removed': 0, 'total': len(current_songs)}
|
||||
|
||||
existing_song_ids = set(self.data['playlists'][playlist_id]['song_ids'])
|
||||
current_song_ids = set()
|
||||
|
||||
added_songs = 0
|
||||
for song_info in current_songs:
|
||||
song_id = self._generate_song_id(song_info['video_id'], playlist_id)
|
||||
current_song_ids.add(song_id)
|
||||
|
||||
if song_id not in existing_song_ids:
|
||||
self.add_song(
|
||||
song_info['video_id'],
|
||||
playlist_id,
|
||||
song_info['title'],
|
||||
song_info.get('duration')
|
||||
)
|
||||
added_songs += 1
|
||||
|
||||
# Remove songs that no longer exist in playlist
|
||||
removed_songs = len(existing_song_ids - current_song_ids)
|
||||
for song_id in existing_song_ids - current_song_ids:
|
||||
if song_id in self.data['songs']:
|
||||
del self.data['songs'][song_id]
|
||||
|
||||
# Update playlist sync time
|
||||
self.data['playlists'][playlist_id]['last_synced'] = datetime.now().isoformat()
|
||||
self._update_playlist_statistics(playlist_id)
|
||||
|
||||
# Force save after batch operations
|
||||
self.force_save()
|
||||
|
||||
return {
|
||||
'added': added_songs,
|
||||
'removed': removed_songs,
|
||||
'total': len(current_songs)
|
||||
}
|
||||
|
||||
def _update_playlist_statistics(self, playlist_id: str):
|
||||
"""Update playlist statistics."""
|
||||
if playlist_id not in self.data['playlists']:
|
||||
return
|
||||
|
||||
songs = self.get_playlist_songs(playlist_id)
|
||||
playlist = self.data['playlists'][playlist_id]
|
||||
|
||||
playlist['total_songs'] = len(songs)
|
||||
playlist['downloaded_songs'] = len([s for s in songs if self.is_song_downloaded(s['video_id'], playlist_id)])
|
||||
playlist['failed_songs'] = len([s for s in songs if s['status'] == SongStatus.FAILED])
|
||||
|
||||
def get_statistics(self) -> Dict:
|
||||
"""Get overall statistics."""
|
||||
total_songs = len(self.data['songs'])
|
||||
total_playlists = len(self.data['playlists'])
|
||||
|
||||
downloaded_songs = sum(1 for song in self.data['songs'].values()
|
||||
if self.is_song_downloaded(song['video_id'], song['playlist_id']))
|
||||
|
||||
failed_songs = sum(1 for song in self.data['songs'].values()
|
||||
if song['status'] == SongStatus.FAILED)
|
||||
|
||||
partial_songs = sum(1 for song in self.data['songs'].values()
|
||||
if song['status'] == SongStatus.PARTIAL)
|
||||
|
||||
total_size = 0
|
||||
for song in self.data['songs'].values():
|
||||
for size in song['file_sizes'].values():
|
||||
total_size += size
|
||||
|
||||
return {
|
||||
'total_songs': total_songs,
|
||||
'total_playlists': total_playlists,
|
||||
'downloaded_songs': downloaded_songs,
|
||||
'failed_songs': failed_songs,
|
||||
'partial_songs': partial_songs,
|
||||
'total_size_bytes': total_size,
|
||||
'total_size_mb': round(total_size / (1024 * 1024), 2),
|
||||
'last_updated': self.data['last_updated']
|
||||
}
|
||||
|
||||
def cleanup_orphaned_files(self, downloads_dir: Path) -> List[str]:
|
||||
"""Remove tracking entries for files that no longer exist."""
|
||||
orphaned_songs = []
|
||||
|
||||
for song_id, song in self.data['songs'].items():
|
||||
for format_type, file_path in song['file_paths'].items():
|
||||
if not Path(file_path).exists():
|
||||
# Remove the format entry
|
||||
if format_type in song['formats']:
|
||||
del song['formats'][format_type]
|
||||
if format_type in song['file_paths']:
|
||||
del song['file_paths'][format_type]
|
||||
if format_type in song['file_hashes']:
|
||||
del song['file_hashes'][format_type]
|
||||
if format_type in song['file_sizes']:
|
||||
del song['file_sizes'][format_type]
|
||||
|
||||
# Update status if no formats remain
|
||||
if not song['formats']:
|
||||
song['status'] = SongStatus.NOT_DOWNLOADED
|
||||
|
||||
orphaned_songs.append(song_id)
|
||||
|
||||
self._save_tracking_data()
|
||||
return orphaned_songs
|
||||
|
||||
def export_playlist_report(self, playlist_id: str, output_file: str = None) -> str:
|
||||
"""Export a detailed report for a playlist."""
|
||||
if playlist_id not in self.data['playlists']:
|
||||
return "Playlist not found"
|
||||
|
||||
playlist = self.data['playlists'][playlist_id]
|
||||
songs = self.get_playlist_songs(playlist_id)
|
||||
|
||||
report = f"""
|
||||
Karaoke Playlist Report
|
||||
======================
|
||||
Playlist: {playlist['name']}
|
||||
URL: {playlist['url']}
|
||||
Total Songs: {playlist['total_songs']}
|
||||
Downloaded: {playlist['downloaded_songs']}
|
||||
Failed: {playlist['failed_songs']}
|
||||
Last Synced: {playlist['last_synced'] or 'Never'}
|
||||
|
||||
Song Details:
|
||||
"""
|
||||
|
||||
for song in songs:
|
||||
status_icon = {
|
||||
SongStatus.DOWNLOADED: "✅",
|
||||
SongStatus.CONVERTED: "✅",
|
||||
SongStatus.DOWNLOADING: "⏳",
|
||||
SongStatus.PARTIAL: "⚠️",
|
||||
SongStatus.FAILED: "❌",
|
||||
SongStatus.NOT_DOWNLOADED: "⏸️"
|
||||
}.get(song['status'], "❓")
|
||||
|
||||
formats = ", ".join(song['formats'].keys()) if song['formats'] else "None"
|
||||
report += f"{status_icon} {song['title']} ({formats})\n"
|
||||
|
||||
if output_file:
|
||||
with open(output_file, 'w', encoding='utf-8') as f:
|
||||
f.write(report)
|
||||
|
||||
return report
|
||||
Loading…
Reference in New Issue
Block a user