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("Songliste / Instagram-Kopierter Text:")) 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())