436 lines
No EOL
16 KiB
Python
436 lines
No EOL
16 KiB
Python
import os
|
|
import re
|
|
import sys
|
|
from PyQt6.QtWidgets import (
|
|
QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
|
|
QLabel, QLineEdit, QTextEdit, QPushButton, QTableWidget,
|
|
QTableWidgetItem, QHeaderView, QFileDialog, QMessageBox,
|
|
QDialog, QListWidget, QListWidgetItem, QAbstractItemView, QCheckBox
|
|
)
|
|
from PyQt6.QtCore import Qt, QThread, pyqtSignal
|
|
from PyQt6.QtGui import QColor
|
|
import yt_dlp
|
|
|
|
# Standard-Zielordner
|
|
DEFAULT_OUTPUT_DIR = r"\\192.168.180.78\music\Instagram"
|
|
|
|
|
|
def parse_song_input(raw_text: str, enable_insta_parser: bool = True) -> list[str]:
|
|
"""
|
|
Parst eingegebenen Text. Erkennt automatisch das 2-zeilige Instagram-Format
|
|
(Titel, Interpret, 'E'-Badges, '...') und wandelt es in 'Interpret - Titel' um.
|
|
"""
|
|
lines = [line.strip() for line in raw_text.splitlines() if line.strip()]
|
|
if not lines:
|
|
return []
|
|
|
|
if not enable_insta_parser:
|
|
return lines
|
|
|
|
# 1. Instagram-Badges und UI-Reste filtern
|
|
cleaned_lines = []
|
|
for line in lines:
|
|
# Einzelne Explicit-Markierungen überspringen
|
|
if line.upper() in ("E", "[E]", "🅴"):
|
|
continue
|
|
# Optionale UI-Zähler (z. B. "12,5 Tsd. Reels") überspringen
|
|
if re.search(r'^\d+([.,]\d+)?\s*(Tsd\.|Mio\.|k|m)?\s*(Reels|Beiträge|Videos)?$', line, re.IGNORECASE):
|
|
continue
|
|
cleaned_lines.append(line)
|
|
|
|
# 2. Prüfen, ob der Text bereits das Standardformat 'Artist - Song' nutzt
|
|
hyphen_count = sum(1 for line in cleaned_lines if " - " in line)
|
|
if hyphen_count > len(cleaned_lines) / 2:
|
|
return cleaned_lines
|
|
|
|
# 3. Instagram-Format: Zeilen paarweise zusammenfügen (Zeile 1: Titel, Zeile 2: Interpret)
|
|
parsed_queries = []
|
|
i = 0
|
|
while i < len(cleaned_lines):
|
|
if i + 1 < len(cleaned_lines):
|
|
title = cleaned_lines[i]
|
|
artist = cleaned_lines[i + 1]
|
|
|
|
# Trailing-Punkte wie '...' oder ',...' bereinigen
|
|
clean_title = re.sub(r'[,.\s]*(\.\.\.|…)$', '', title).strip()
|
|
|
|
# Zusammensetzen als "Interpret - Titel"
|
|
parsed_queries.append(f"{artist} - {clean_title}")
|
|
i += 2
|
|
else:
|
|
# Falls am Ende eine einzelne Zeile übrig bleibt
|
|
single = re.sub(r'[,.\s]*(\.\.\.|…)$', '', cleaned_lines[i]).strip()
|
|
parsed_queries.append(single)
|
|
i += 1
|
|
|
|
return parsed_queries
|
|
|
|
|
|
class DownloadWorker(QThread):
|
|
sig_status = pyqtSignal(int, str, str)
|
|
sig_success = pyqtSignal(int, str, str, str)
|
|
sig_failed = pyqtSignal(int, str)
|
|
sig_finished = pyqtSignal()
|
|
|
|
def __init__(self, items, output_dir):
|
|
super().__init__()
|
|
self.items = items
|
|
self.output_dir = output_dir
|
|
self.is_running = True
|
|
|
|
def run(self):
|
|
for row, query in self.items:
|
|
if not self.is_running:
|
|
break
|
|
|
|
self.sig_status.emit(row, "Sucht...", "YouTube-Suche läuft")
|
|
|
|
ydl_opts = {
|
|
"format": "bestaudio/best",
|
|
"outtmpl": os.path.join(self.output_dir, "%(title)s.%(ext)s"),
|
|
"writethumbnail": True, # Lädt das YouTube-Thumbnail herunter
|
|
"postprocessors": [
|
|
{
|
|
"key": "FFmpegExtractAudio",
|
|
"preferredcodec": "mp3",
|
|
"preferredquality": "192",
|
|
},
|
|
{
|
|
# Konvertiert das YouTube-WebP-Bild in echtes JPG für MP3-Kompatibilität
|
|
"key": "FFmpegThumbnailsConvertor",
|
|
"format": "jpg",
|
|
},
|
|
{"key": "FFmpegMetadata"}, # Schreibt Titel-, Artist- und Album-Tags
|
|
{
|
|
# Bettet das JPG direkt in die MP3-Datei ein und löscht das temporäre Bild
|
|
"key": "EmbedThumbnail",
|
|
},
|
|
],
|
|
"quiet": True,
|
|
"no_warnings": True,
|
|
}
|
|
|
|
try:
|
|
target = query if query.startswith("http") else f"ytsearch1:{query}"
|
|
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
self.sig_status.emit(row, "Lädt...", "Stream wird konvertiert")
|
|
info = ydl.extract_info(target, download=True)
|
|
|
|
if "entries" in info:
|
|
if not info["entries"]:
|
|
self.sig_failed.emit(row, "Keine Treffer gefunden")
|
|
continue
|
|
video = info["entries"][0]
|
|
else:
|
|
video = info
|
|
|
|
title = video.get("title", "Unbekannt")
|
|
duration_secs = video.get("duration", 0)
|
|
mins, secs = divmod(duration_secs, 60)
|
|
duration_str = f"{mins}:{secs:02d}"
|
|
|
|
self.sig_success.emit(row, title, duration_str, "MP3 fertig gespeichert")
|
|
|
|
except Exception as e:
|
|
err_msg = str(e)
|
|
if "ERROR:" in err_msg:
|
|
err_msg = err_msg.split("ERROR:")[-1].strip()
|
|
self.sig_failed.emit(row, err_msg)
|
|
|
|
self.sig_finished.emit()
|
|
|
|
def stop(self):
|
|
self.is_running = False
|
|
|
|
|
|
class AlternativeDialog(QDialog):
|
|
def __init__(self, query, parent=None):
|
|
super().__init__(parent)
|
|
self.setWindowTitle("Alternative auswählen")
|
|
self.resize(650, 420)
|
|
self.selected_url = None
|
|
self.candidates = []
|
|
|
|
layout = QVBoxLayout(self)
|
|
|
|
search_layout = QHBoxLayout()
|
|
self.edit_query = QLineEdit(query)
|
|
btn_search = QPushButton("Erneut suchen")
|
|
btn_search.clicked.connect(self.search)
|
|
search_layout.addWidget(self.edit_query)
|
|
search_layout.addWidget(btn_search)
|
|
layout.addLayout(search_layout)
|
|
|
|
self.list_widget = QListWidget()
|
|
self.list_widget.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
|
|
layout.addWidget(self.list_widget)
|
|
|
|
btn_layout = QHBoxLayout()
|
|
self.btn_select = QPushButton("Diesen Track herunterladen")
|
|
self.btn_select.setEnabled(False)
|
|
self.btn_select.clicked.connect(self.accept_selection)
|
|
|
|
btn_cancel = QPushButton("Abbrechen")
|
|
btn_cancel.clicked.connect(self.reject)
|
|
|
|
btn_layout.addStretch()
|
|
btn_layout.addWidget(btn_cancel)
|
|
btn_layout.addWidget(self.btn_select)
|
|
layout.addLayout(btn_layout)
|
|
|
|
self.list_widget.itemSelectionChanged.connect(self.on_selection_changed)
|
|
self.search()
|
|
|
|
def search(self):
|
|
query = self.edit_query.text().strip()
|
|
if not query:
|
|
return
|
|
|
|
self.list_widget.clear()
|
|
self.btn_select.setEnabled(False)
|
|
self.list_widget.addItem("Suche auf YouTube nach Top 5 Treffern...")
|
|
QApplication.processEvents()
|
|
|
|
opts = {"extract_flat": True, "quiet": True, "no_warnings": True}
|
|
try:
|
|
with yt_dlp.YoutubeDL(opts) as ydl:
|
|
res = ydl.extract_info(f"ytsearch5:{query}", download=False)
|
|
entries = res.get("entries", [])
|
|
self.list_widget.clear()
|
|
self.candidates = []
|
|
|
|
if not entries:
|
|
self.list_widget.addItem("Keine Treffer gefunden.")
|
|
return
|
|
|
|
for entry in entries:
|
|
title = entry.get("title", "Ohne Titel")
|
|
uploader = entry.get("uploader", "Unbekannter Kanal")
|
|
dur = entry.get("duration", 0)
|
|
mins, secs = divmod(dur, 60) if dur else (0, 0)
|
|
dur_str = f"{mins}:{secs:02d}"
|
|
url = entry.get("url") or f"https://www.youtube.com/watch?v={entry.get('id')}"
|
|
|
|
self.candidates.append(url)
|
|
item = QListWidgetItem(f"🎵 {title}\n 👤 Kanal: {uploader} | ⏱ {dur_str}")
|
|
self.list_widget.addItem(item)
|
|
|
|
except Exception as e:
|
|
self.list_widget.clear()
|
|
self.list_widget.addItem(f"Fehler bei der Suche: {e}")
|
|
|
|
def on_selection_changed(self):
|
|
row = self.list_widget.currentRow()
|
|
if 0 <= row < len(self.candidates):
|
|
self.btn_select.setEnabled(True)
|
|
self.selected_url = self.candidates[row]
|
|
|
|
def accept_selection(self):
|
|
if self.selected_url:
|
|
self.accept()
|
|
|
|
|
|
class MainWindow(QMainWindow):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.setWindowTitle("Instagram Batch Audio Downloader")
|
|
self.resize(1050, 700)
|
|
|
|
self.output_dir = DEFAULT_OUTPUT_DIR
|
|
os.makedirs(self.output_dir, exist_ok=True)
|
|
self.worker = None
|
|
|
|
self.init_ui()
|
|
|
|
def init_ui(self):
|
|
central_widget = QWidget()
|
|
self.setCentralWidget(central_widget)
|
|
main_layout = QVBoxLayout(central_widget)
|
|
main_layout.setSpacing(10)
|
|
|
|
# 1. Zielordner Auswahl
|
|
folder_layout = QHBoxLayout()
|
|
folder_layout.addWidget(QLabel("Zielordner:"))
|
|
self.lbl_folder = QLineEdit(self.output_dir)
|
|
self.lbl_folder.setReadOnly(True)
|
|
btn_browse = QPushButton("Ordner wählen...")
|
|
btn_browse.clicked.connect(self.choose_folder)
|
|
folder_layout.addWidget(self.lbl_folder)
|
|
folder_layout.addWidget(btn_browse)
|
|
main_layout.addLayout(folder_layout)
|
|
|
|
# 2. Eingabefeld
|
|
main_layout.addWidget(QLabel("<b>Songliste / Instagram-Kopierter Text:</b>"))
|
|
self.txt_batch_input = QTextEdit()
|
|
self.txt_batch_input.setPlaceholderText(
|
|
"Hier den kopierten Text aus Instagram (Titel & Interpret untereinander) oder Songliste einfügen..."
|
|
)
|
|
self.txt_batch_input.setFixedHeight(120)
|
|
main_layout.addWidget(self.txt_batch_input)
|
|
|
|
# Checkbox für Instagram 2-Zeilen Erkennung
|
|
self.chk_insta_mode = QCheckBox(
|
|
"Instagram-Format bereinigen (Titel & Interpret zweizeilig zusammenfügen, 'E'-Badges & '...' filtern)"
|
|
)
|
|
self.chk_insta_mode.setChecked(True)
|
|
main_layout.addWidget(self.chk_insta_mode)
|
|
|
|
# Steuerungsleiste
|
|
btn_bar = QHBoxLayout()
|
|
btn_add = QPushButton("In Warteschlange übernehmen")
|
|
btn_add.clicked.connect(self.add_songs_to_table)
|
|
|
|
self.btn_start = QPushButton("Alle Downloads starten")
|
|
self.btn_start.setStyleSheet("font-weight: bold; background-color: #2b78e4; color: white; padding: 6px;")
|
|
self.btn_start.clicked.connect(self.start_batch_download)
|
|
|
|
btn_clear = QPushButton("Tabelle leeren")
|
|
btn_clear.clicked.connect(self.clear_table)
|
|
|
|
btn_bar.addWidget(btn_add)
|
|
btn_bar.addWidget(self.btn_start)
|
|
btn_bar.addStretch()
|
|
btn_bar.addWidget(btn_clear)
|
|
main_layout.addLayout(btn_bar)
|
|
|
|
# 3. Tabelle
|
|
self.table = QTableWidget()
|
|
self.table.setColumnCount(6)
|
|
self.table.setHorizontalHeaderLabels([
|
|
"Suchbegriff (Generiert)",
|
|
"Gefundener YouTube-Titel",
|
|
"Dauer",
|
|
"Status",
|
|
"Details / Grund",
|
|
"Aktion"
|
|
])
|
|
|
|
header = self.table.horizontalHeader()
|
|
header.setSectionResizeMode(0, QHeaderView.ResizeMode.Interactive)
|
|
header.setSectionResizeMode(1, QHeaderView.ResizeMode.Stretch)
|
|
header.setSectionResizeMode(2, QHeaderView.ResizeMode.ResizeToContents)
|
|
header.setSectionResizeMode(3, QHeaderView.ResizeMode.ResizeToContents)
|
|
header.setSectionResizeMode(4, QHeaderView.ResizeMode.Interactive)
|
|
header.setSectionResizeMode(5, QHeaderView.ResizeMode.ResizeToContents)
|
|
|
|
self.table.setColumnWidth(0, 260)
|
|
self.table.setColumnWidth(4, 250)
|
|
main_layout.addWidget(self.table)
|
|
|
|
def choose_folder(self):
|
|
folder = QFileDialog.getExistingDirectory(self, "Zielordner wählen", self.output_dir)
|
|
if folder:
|
|
self.output_dir = folder
|
|
self.lbl_folder.setText(folder)
|
|
|
|
def add_songs_to_table(self):
|
|
raw_text = self.txt_batch_input.toPlainText().strip()
|
|
if not raw_text:
|
|
QMessageBox.warning(self, "Hinweis", "Bitte mindestens einen Songnamen eingeben.")
|
|
return
|
|
|
|
songs = parse_song_input(raw_text, self.chk_insta_mode.isChecked())
|
|
|
|
for song_query in songs:
|
|
row = self.table.rowCount()
|
|
self.table.insertRow(row)
|
|
|
|
item_query = QTableWidgetItem(song_query)
|
|
item_title = QTableWidgetItem("-")
|
|
item_duration = QTableWidgetItem("-")
|
|
item_status = QTableWidgetItem("Warteschlange")
|
|
item_status.setForeground(QColor("#777777"))
|
|
item_details = QTableWidgetItem("Bereit zum Download")
|
|
|
|
self.table.setItem(row, 0, item_query)
|
|
self.table.setItem(row, 1, item_title)
|
|
self.table.setItem(row, 2, item_duration)
|
|
self.table.setItem(row, 3, item_status)
|
|
self.table.setItem(row, 4, item_details)
|
|
|
|
btn_alt = QPushButton("🔍 Alternativen")
|
|
btn_alt.clicked.connect(lambda _, r=row: self.open_alternatives_dialog(r))
|
|
self.table.setCellWidget(row, 5, btn_alt)
|
|
|
|
self.txt_batch_input.clear()
|
|
|
|
def clear_table(self):
|
|
if self.worker and self.worker.isRunning():
|
|
QMessageBox.warning(self, "Achtung", "Downloads laufen noch. Bitte warten.")
|
|
return
|
|
self.table.setRowCount(0)
|
|
|
|
def start_batch_download(self):
|
|
if self.worker and self.worker.isRunning():
|
|
QMessageBox.information(self, "Info", "Downloads laufen bereits.")
|
|
return
|
|
|
|
queue = []
|
|
for r in range(self.table.rowCount()):
|
|
status_item = self.table.item(r, 3)
|
|
if status_item and status_item.text() != "Fertig":
|
|
query = self.table.item(r, 0).text()
|
|
queue.append((r, query))
|
|
|
|
if not queue:
|
|
QMessageBox.information(self, "Info", "Keine offenen Einträge in der Warteschlange.")
|
|
return
|
|
|
|
self.btn_start.setEnabled(False)
|
|
self.worker = DownloadWorker(queue, self.output_dir)
|
|
self.worker.sig_status.connect(self.on_worker_status)
|
|
self.worker.sig_success.connect(self.on_worker_success)
|
|
self.worker.sig_failed.connect(self.on_worker_failed)
|
|
self.worker.sig_finished.connect(self.on_worker_finished)
|
|
self.worker.start()
|
|
|
|
def on_worker_status(self, row, status, details):
|
|
item_status = self.table.item(row, 3)
|
|
item_status.setText(status)
|
|
item_status.setForeground(QColor("#1a73e8"))
|
|
self.table.item(row, 4).setText(details)
|
|
|
|
def on_worker_success(self, row, title, duration, details):
|
|
self.table.item(row, 1).setText(title)
|
|
self.table.item(row, 2).setText(duration)
|
|
|
|
item_status = self.table.item(row, 3)
|
|
item_status.setText("Fertig")
|
|
item_status.setForeground(QColor("#0f9d58"))
|
|
|
|
self.table.item(row, 4).setText(details)
|
|
|
|
def on_worker_failed(self, row, reason):
|
|
item_status = self.table.item(row, 3)
|
|
item_status.setText("Fehlgeschlagen")
|
|
item_status.setForeground(QColor("#d93025"))
|
|
|
|
self.table.item(row, 4).setText(reason)
|
|
|
|
def on_worker_finished(self):
|
|
self.btn_start.setEnabled(True)
|
|
QMessageBox.information(self, "Abgeschlossen", "Alle Downloads in der Warteschlange wurden verarbeitet.")
|
|
|
|
def open_alternatives_dialog(self, row):
|
|
query_item = self.table.item(row, 0)
|
|
current_query = query_item.text() if query_item else ""
|
|
|
|
dlg = AlternativeDialog(current_query, self)
|
|
if dlg.exec() == QDialog.DialogCode.Accepted and dlg.selected_url:
|
|
self.table.item(row, 3).setText("Wartet...")
|
|
self.table.item(row, 4).setText("Alternative ausgewählt, lade...")
|
|
|
|
worker = DownloadWorker([(row, dlg.selected_url)], self.output_dir)
|
|
worker.sig_status.connect(self.on_worker_status)
|
|
worker.sig_success.connect(self.on_worker_success)
|
|
worker.sig_failed.connect(self.on_worker_failed)
|
|
worker.start()
|
|
self._single_worker = worker
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app = QApplication(sys.argv)
|
|
window = MainWindow()
|
|
window.show()
|
|
sys.exit(app.exec()) |