1142 lines
44 KiB
Python
1142 lines
44 KiB
Python
"""
|
||
Spotify Recorder – Embeds the Spotify desktop app (already logged in)
|
||
directly into the application window. No web player, no re-login.
|
||
Records each song as a numbered MP3 in a playlist folder.
|
||
"""
|
||
|
||
import customtkinter as ctk
|
||
import tkinter as tk
|
||
from tkinter import filedialog
|
||
import ctypes
|
||
from ctypes import wintypes
|
||
import psutil
|
||
import subprocess
|
||
import threading
|
||
import time
|
||
import os
|
||
import sys
|
||
import traceback
|
||
import datetime
|
||
from audio import AudioRecorder
|
||
|
||
# ── Win32 constants ────────────────────────────────────────────
|
||
user32 = ctypes.windll.user32
|
||
GWL_STYLE = -16
|
||
WS_CHILD = 0x40000000
|
||
WS_CAPTION = 0x00C00000
|
||
WS_THICKFRAME = 0x00040000
|
||
WS_POPUP = 0x80000000
|
||
SWP_FRAMECHANGED = 0x0020
|
||
SWP_NOMOVE = 0x0002
|
||
SWP_NOSIZE = 0x0001
|
||
SWP_NOZORDER = 0x0004
|
||
|
||
# ── Colors ─────────────────────────────────────────────────────
|
||
BG = "#0b0b0b"
|
||
CARD = "#151515"
|
||
BORDER = "#1e1e1e"
|
||
GREEN = "#1DB954"
|
||
GREEN_H = "#1ed760"
|
||
RED = "#e74c3c"
|
||
RED_H = "#ff6b6b"
|
||
ORANGE = "#f59e0b"
|
||
TXT = "#ffffff"
|
||
TXT2 = "#b3b3b3"
|
||
DIM = "#555555"
|
||
|
||
PANEL_W = 330
|
||
|
||
PAUSE_TITLES = {"spotify", "spotify premium", "spotify free"}
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════
|
||
# Spotify window helper functions
|
||
# ═══════════════════════════════════════════════════════════════
|
||
|
||
def get_spotify_pids():
|
||
pids = set()
|
||
for proc in psutil.process_iter(["pid", "name"]):
|
||
try:
|
||
if (proc.info["name"] or "").lower() == "spotify.exe":
|
||
pids.add(proc.info["pid"])
|
||
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
||
pass
|
||
return pids
|
||
|
||
|
||
def find_spotify_window():
|
||
pids = get_spotify_pids()
|
||
if not pids:
|
||
return None
|
||
best = [None]
|
||
best_area = [0]
|
||
|
||
@ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.HWND, wintypes.LPARAM)
|
||
def cb(hwnd, _lp):
|
||
if user32.IsWindowVisible(hwnd):
|
||
pid = wintypes.DWORD()
|
||
user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
|
||
if pid.value in pids:
|
||
rect = wintypes.RECT()
|
||
user32.GetWindowRect(hwnd, ctypes.byref(rect))
|
||
w = rect.right - rect.left
|
||
h = rect.bottom - rect.top
|
||
area = w * h
|
||
if area > best_area[0] and w > 300 and h > 200:
|
||
best_area[0] = area
|
||
best[0] = hwnd
|
||
return True
|
||
|
||
user32.EnumWindows(cb, 0)
|
||
return best[0]
|
||
|
||
|
||
def get_window_title(hwnd):
|
||
length = user32.GetWindowTextLengthW(hwnd)
|
||
if length > 0:
|
||
buf = ctypes.create_unicode_buffer(length + 1)
|
||
user32.GetWindowTextW(hwnd, buf, length + 1)
|
||
return buf.value
|
||
return ""
|
||
|
||
|
||
def launch_spotify():
|
||
paths = [
|
||
os.path.join(os.getenv("APPDATA", ""), "Spotify", "Spotify.exe"),
|
||
os.path.join(
|
||
os.getenv("LOCALAPPDATA", ""),
|
||
"Microsoft", "WindowsApps", "Spotify.exe",
|
||
),
|
||
]
|
||
for p in paths:
|
||
if os.path.exists(p):
|
||
subprocess.Popen([p])
|
||
return True
|
||
try:
|
||
subprocess.Popen(["spotify"])
|
||
return True
|
||
except FileNotFoundError:
|
||
pass
|
||
return False
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════
|
||
# Main application
|
||
# ═══════════════════════════════════════════════════════════════
|
||
|
||
class SpotifyRecorderApp:
|
||
|
||
def __init__(self):
|
||
try:
|
||
ctypes.windll.shcore.SetProcessDpiAwareness(1)
|
||
except Exception:
|
||
pass
|
||
|
||
ctk.set_appearance_mode("dark")
|
||
ctk.set_default_color_theme("green")
|
||
|
||
self.root = ctk.CTk()
|
||
self.root.title("Spotify Recorder")
|
||
self.root.geometry("1400x880")
|
||
self.root.minsize(1000, 620)
|
||
self.root.configure(fg_color=BG)
|
||
self.root.protocol("WM_DELETE_WINDOW", self._on_close)
|
||
|
||
# state
|
||
self.recorder = AudioRecorder()
|
||
self.spotify_hwnd = None
|
||
self.original_style = None
|
||
self.original_parent = None
|
||
self.active = False
|
||
self.is_playing = False
|
||
self.playlist_name = ""
|
||
self.session_dir = ""
|
||
self.track_number = 0
|
||
self.current_track = None
|
||
self.songs = []
|
||
self.recorded_songs = set()
|
||
self.rec_start = None
|
||
self.last_title = ""
|
||
self.output_base = os.path.join(
|
||
os.path.expanduser("~"), "Music", "SpotifyRecordings"
|
||
)
|
||
os.makedirs(self.output_base, exist_ok=True)
|
||
self.bitrate = 320
|
||
self.auto_stop_secs = 60
|
||
self.pause_since = None
|
||
self._pulse_on = True
|
||
self._alive_fails = 0
|
||
self.ad_playing = False
|
||
self._auto_record_ready = False
|
||
self.playlist_total = 0
|
||
self._user_stopped = False
|
||
self._restart_after_id = None
|
||
self._restart_fails = 0
|
||
self._lock = threading.Lock()
|
||
|
||
self._build_controls()
|
||
self._build_embed_area()
|
||
|
||
threading.Thread(target=self._attach_spotify, daemon=True).start()
|
||
self._tick_monitor()
|
||
self._tick_timer()
|
||
self._tick_level()
|
||
self._tick_pulse()
|
||
self._tick_alive()
|
||
|
||
# ── build left panel ───────────────────────────────────────
|
||
|
||
def _build_controls(self):
|
||
panel = ctk.CTkFrame(self.root, width=PANEL_W, fg_color=CARD,
|
||
corner_radius=0, border_width=0)
|
||
panel.pack(side="left", fill="y")
|
||
panel.pack_propagate(False)
|
||
|
||
pad = {"padx": 20}
|
||
|
||
# header
|
||
hdr = ctk.CTkFrame(panel, fg_color="transparent")
|
||
hdr.pack(fill="x", **pad, pady=(22, 0))
|
||
icon = ctk.CTkFrame(hdr, width=42, height=42, fg_color=GREEN,
|
||
corner_radius=12)
|
||
icon.pack(side="left")
|
||
icon.pack_propagate(False)
|
||
ctk.CTkLabel(icon, text="\u266b", font=ctk.CTkFont(size=22, weight="bold"),
|
||
text_color="white").pack(expand=True)
|
||
hcol = ctk.CTkFrame(hdr, fg_color="transparent")
|
||
hcol.pack(side="left", padx=10)
|
||
ctk.CTkLabel(hcol, text="Spotify Recorder",
|
||
font=ctk.CTkFont(size=18, weight="bold"),
|
||
text_color=TXT).pack(anchor="w")
|
||
ctk.CTkLabel(hcol, text="Desktop Audio Capture",
|
||
font=ctk.CTkFont(size=10), text_color=DIM).pack(anchor="w")
|
||
|
||
# spotify status
|
||
self.sp_label = ctk.CTkLabel(
|
||
panel, text="\u25cf Spotify wird gesucht\u2026",
|
||
font=ctk.CTkFont(size=11), text_color=DIM)
|
||
self.sp_label.pack(**pad, pady=(16, 0), anchor="w")
|
||
|
||
# separator
|
||
ctk.CTkFrame(panel, height=1, fg_color=BORDER).pack(fill="x", **pad, pady=12)
|
||
|
||
# playlist input
|
||
ctk.CTkLabel(panel, text="ORDNER / PLAYLIST",
|
||
font=ctk.CTkFont(size=9, weight="bold"),
|
||
text_color=DIM).pack(**pad, anchor="w")
|
||
self.pl_entry = ctk.CTkEntry(
|
||
panel, placeholder_text="Optional - wird auto-generiert",
|
||
font=ctk.CTkFont(size=13), height=38, corner_radius=10,
|
||
fg_color="#1a1a1a", border_color=BORDER)
|
||
self.pl_entry.pack(fill="x", **pad, pady=(6, 0))
|
||
ctk.CTkLabel(panel, text="Leer lassen = automatisch (Datum/Uhrzeit)",
|
||
font=ctk.CTkFont(size=9), text_color="#444"
|
||
).pack(**pad, anchor="w", pady=(2, 0))
|
||
|
||
# track card
|
||
tcard = ctk.CTkFrame(panel, fg_color="#111111", corner_radius=14,
|
||
border_width=1, border_color=BORDER)
|
||
tcard.pack(fill="x", **pad, pady=(14, 0))
|
||
|
||
srow = ctk.CTkFrame(tcard, fg_color="transparent")
|
||
srow.pack(fill="x", padx=14, pady=(12, 0))
|
||
self.dot_lbl = ctk.CTkLabel(srow, text="\u25cf", width=14,
|
||
font=ctk.CTkFont(size=10), text_color=DIM)
|
||
self.dot_lbl.pack(side="left")
|
||
self.stat_lbl = ctk.CTkLabel(srow, text="BEREIT",
|
||
font=ctk.CTkFont(size=10, weight="bold"),
|
||
text_color=DIM)
|
||
self.stat_lbl.pack(side="left", padx=(5, 0))
|
||
self.time_lbl = ctk.CTkLabel(srow, text="",
|
||
font=ctk.CTkFont(size=12, weight="bold"),
|
||
text_color=DIM)
|
||
self.time_lbl.pack(side="right")
|
||
|
||
ctk.CTkFrame(tcard, height=1, fg_color=BORDER).pack(fill="x", padx=14, pady=8)
|
||
|
||
self.tnum_lbl = ctk.CTkLabel(tcard, text="",
|
||
font=ctk.CTkFont(size=10, weight="bold"),
|
||
text_color=GREEN)
|
||
self.tnum_lbl.pack(padx=14, anchor="w")
|
||
self.artist_lbl = ctk.CTkLabel(tcard, text="\u2014",
|
||
font=ctk.CTkFont(size=15, weight="bold"),
|
||
text_color=TXT)
|
||
self.artist_lbl.pack(padx=14, anchor="w")
|
||
self.title_lbl = ctk.CTkLabel(tcard, text="",
|
||
font=ctk.CTkFont(size=12), text_color=TXT2)
|
||
self.title_lbl.pack(padx=14, anchor="w", pady=(1, 0))
|
||
|
||
lvl_row = ctk.CTkFrame(tcard, fg_color="transparent")
|
||
lvl_row.pack(fill="x", padx=14, pady=(12, 14))
|
||
ctk.CTkLabel(lvl_row, text="PEGEL",
|
||
font=ctk.CTkFont(size=8, weight="bold"),
|
||
text_color="#444").pack(side="left", padx=(0, 8))
|
||
self.level_bar = ctk.CTkProgressBar(
|
||
lvl_row, height=5, corner_radius=3,
|
||
fg_color="#222", progress_color=GREEN)
|
||
self.level_bar.pack(side="left", fill="x", expand=True)
|
||
self.level_bar.set(0)
|
||
|
||
# record button
|
||
self.rec_btn = ctk.CTkButton(
|
||
panel, text="\u23fa AUFNAHME STARTEN",
|
||
font=ctk.CTkFont(size=14, weight="bold"), height=48,
|
||
corner_radius=24, fg_color=GREEN, hover_color=GREEN_H,
|
||
command=self._toggle)
|
||
self.rec_btn.pack(fill="x", **pad, pady=(14, 0))
|
||
|
||
# song list
|
||
ctk.CTkFrame(panel, height=1, fg_color=BORDER).pack(fill="x", **pad, pady=12)
|
||
sl_hdr = ctk.CTkFrame(panel, fg_color="transparent")
|
||
sl_hdr.pack(fill="x", **pad)
|
||
ctk.CTkLabel(sl_hdr, text="AUFGENOMMEN",
|
||
font=ctk.CTkFont(size=9, weight="bold"),
|
||
text_color=DIM).pack(side="left")
|
||
self.cnt_lbl = ctk.CTkLabel(sl_hdr, text="0 Songs",
|
||
font=ctk.CTkFont(size=9, weight="bold"),
|
||
text_color=DIM)
|
||
self.cnt_lbl.pack(side="right")
|
||
|
||
self.song_box = ctk.CTkTextbox(
|
||
panel, font=ctk.CTkFont(size=11), fg_color=CARD,
|
||
text_color=TXT2, border_width=0, corner_radius=0,
|
||
state="disabled", height=120, activate_scrollbars=True)
|
||
self.song_box.pack(fill="both", expand=True, **pad, pady=(6, 0))
|
||
|
||
# settings
|
||
ctk.CTkFrame(panel, height=1, fg_color=BORDER).pack(fill="x", **pad, pady=10)
|
||
ctk.CTkLabel(panel, text="EINSTELLUNGEN",
|
||
font=ctk.CTkFont(size=9, weight="bold"),
|
||
text_color=DIM).pack(**pad, anchor="w")
|
||
|
||
fr = ctk.CTkFrame(panel, fg_color="transparent")
|
||
fr.pack(fill="x", **pad, pady=(6, 0))
|
||
ctk.CTkLabel(fr, text="Ordner", font=ctk.CTkFont(size=12),
|
||
text_color=TXT2).pack(side="left")
|
||
ctk.CTkButton(fr, text="Aendern", width=65, height=26,
|
||
corner_radius=8, font=ctk.CTkFont(size=11),
|
||
fg_color="#2a2a2a", hover_color="#383838",
|
||
command=self._choose_folder).pack(side="right")
|
||
ctk.CTkButton(fr, text="\U0001f4c2 Oeffnen", width=80, height=26,
|
||
corner_radius=8, font=ctk.CTkFont(size=11),
|
||
fg_color="#2a2a2a", hover_color="#383838",
|
||
command=self._open_folder).pack(side="right", padx=(0, 6))
|
||
self.folder_lbl = ctk.CTkLabel(
|
||
panel, text=self.output_base, font=ctk.CTkFont(size=9),
|
||
text_color=GREEN, wraplength=PANEL_W - 50, anchor="w")
|
||
self.folder_lbl.pack(**pad, anchor="w", pady=(2, 0))
|
||
|
||
br_row = ctk.CTkFrame(panel, fg_color="transparent")
|
||
br_row.pack(fill="x", **pad, pady=(6, 0))
|
||
ctk.CTkLabel(br_row, text="Qualitaet", font=ctk.CTkFont(size=12),
|
||
text_color=TXT2).pack(side="left")
|
||
self.br_var = ctk.StringVar(value="320 kbps")
|
||
ctk.CTkOptionMenu(
|
||
br_row, values=["128 kbps", "192 kbps", "256 kbps", "320 kbps"],
|
||
variable=self.br_var, width=110, height=26,
|
||
font=ctk.CTkFont(size=11),
|
||
fg_color="#2a2a2a", button_color="#333", button_hover_color="#444",
|
||
command=self._set_bitrate
|
||
).pack(side="right")
|
||
|
||
as_row = ctk.CTkFrame(panel, fg_color="transparent")
|
||
as_row.pack(fill="x", **pad, pady=(8, 4))
|
||
ctk.CTkLabel(as_row, text="Auto-Stop", font=ctk.CTkFont(size=12),
|
||
text_color=TXT2).pack(side="left")
|
||
self.as_var = ctk.StringVar(value="60 Sek.")
|
||
ctk.CTkOptionMenu(
|
||
as_row, values=["15 Sek.", "30 Sek.", "60 Sek.", "120 Sek.", "Aus"],
|
||
variable=self.as_var, width=110, height=26,
|
||
font=ctk.CTkFont(size=11),
|
||
fg_color="#2a2a2a", button_color="#333", button_hover_color="#444",
|
||
command=self._set_auto_stop
|
||
).pack(side="right")
|
||
ctk.CTkLabel(panel, text="Stoppt Aufnahme wenn Playlist endet",
|
||
font=ctk.CTkFont(size=9), text_color="#444"
|
||
).pack(**pad, anchor="w", pady=(0, 6))
|
||
|
||
# total stats
|
||
ctk.CTkFrame(panel, height=1, fg_color=BORDER).pack(fill="x", **pad, pady=(0, 6))
|
||
self.total_lbl = ctk.CTkLabel(
|
||
panel, text="",
|
||
font=ctk.CTkFont(size=10), text_color=DIM)
|
||
self.total_lbl.pack(**pad, anchor="w", pady=(0, 10))
|
||
self._refresh_total_stats()
|
||
|
||
# ── build right embed area ─────────────────────────────────
|
||
|
||
def _build_embed_area(self):
|
||
self.embed_outer = ctk.CTkFrame(self.root, fg_color="#000000",
|
||
corner_radius=0)
|
||
self.embed_outer.pack(side="right", fill="both", expand=True)
|
||
|
||
self.embed_frame = tk.Frame(self.embed_outer, bg="#000000")
|
||
self.embed_frame.pack(fill="both", expand=True)
|
||
|
||
self.embed_msg = ctk.CTkLabel(
|
||
self.embed_outer,
|
||
text="\u23f3 Spotify wird gesucht\u2026\n\n"
|
||
"Stelle sicher, dass die Spotify Desktop-App\n"
|
||
"geoeffnet ist (bereits eingeloggt).",
|
||
font=ctk.CTkFont(size=15), text_color="#444",
|
||
justify="center")
|
||
self.embed_msg.place(relx=0.5, rely=0.5, anchor="center")
|
||
|
||
self.embed_frame.bind("<Configure>", self._on_embed_resize)
|
||
|
||
# ── Spotify embedding ──────────────────────────────────────
|
||
|
||
def _attach_spotify(self):
|
||
for _ in range(30):
|
||
hwnd = find_spotify_window()
|
||
if hwnd:
|
||
self.root.after(100, lambda h=hwnd: self._embed(h))
|
||
return
|
||
time.sleep(1)
|
||
|
||
if not get_spotify_pids():
|
||
self.root.after(0, lambda: self.embed_msg.configure(
|
||
text="\u26a0 Spotify nicht gefunden.\n\n"
|
||
"Starte die Spotify Desktop-App\n"
|
||
"und klicke dann hier.",
|
||
))
|
||
launched = launch_spotify()
|
||
if launched:
|
||
for _ in range(20):
|
||
hwnd = find_spotify_window()
|
||
if hwnd:
|
||
self.root.after(100, lambda h=hwnd: self._embed(h))
|
||
return
|
||
time.sleep(1)
|
||
|
||
self.root.after(0, lambda: self._update_sp_status(False))
|
||
|
||
def _embed(self, hwnd):
|
||
self.embed_frame.update_idletasks()
|
||
container = self.embed_frame.winfo_id()
|
||
|
||
self.original_style = user32.GetWindowLongW(hwnd, GWL_STYLE)
|
||
self.original_parent = user32.GetParent(hwnd)
|
||
self.spotify_hwnd = hwnd
|
||
|
||
user32.SetParent(hwnd, container)
|
||
|
||
style = self.original_style
|
||
style &= ~WS_CAPTION
|
||
style &= ~WS_THICKFRAME
|
||
style &= ~WS_POPUP
|
||
style |= WS_CHILD
|
||
user32.SetWindowLongW(hwnd, GWL_STYLE, style)
|
||
|
||
user32.SetWindowPos(hwnd, 0, 0, 0, 0, 0,
|
||
SWP_FRAMECHANGED | SWP_NOMOVE | SWP_NOSIZE | SWP_NOZORDER)
|
||
|
||
w = self.embed_frame.winfo_width()
|
||
h = self.embed_frame.winfo_height()
|
||
user32.MoveWindow(hwnd, 0, 0, max(w, 400), max(h, 300), True)
|
||
|
||
self.embed_msg.place_forget()
|
||
self._update_sp_status(True)
|
||
|
||
def _on_embed_resize(self, event):
|
||
if self.spotify_hwnd and user32.IsWindow(self.spotify_hwnd):
|
||
user32.MoveWindow(self.spotify_hwnd, 0, 0,
|
||
event.width, event.height, True)
|
||
|
||
def _restore_spotify(self):
|
||
if self.spotify_hwnd and self.original_style is not None:
|
||
try:
|
||
if user32.IsWindow(self.spotify_hwnd):
|
||
user32.SetParent(self.spotify_hwnd, self.original_parent or 0)
|
||
user32.SetWindowLongW(self.spotify_hwnd, GWL_STYLE,
|
||
self.original_style)
|
||
user32.SetWindowPos(self.spotify_hwnd, 0, 0, 0, 0, 0,
|
||
SWP_FRAMECHANGED | SWP_NOMOVE |
|
||
SWP_NOSIZE | SWP_NOZORDER)
|
||
user32.ShowWindow(self.spotify_hwnd, 5)
|
||
except Exception:
|
||
pass
|
||
self.spotify_hwnd = None
|
||
|
||
def _update_sp_status(self, connected):
|
||
if connected:
|
||
self.sp_label.configure(
|
||
text="\u25cf Spotify verbunden", text_color=GREEN)
|
||
else:
|
||
self.sp_label.configure(
|
||
text="\u25cf Spotify nicht verbunden", text_color=RED)
|
||
|
||
# ── recording control ──────────────────────────────────────
|
||
|
||
def _toggle(self):
|
||
if self.active:
|
||
self._user_stopped = True
|
||
self._stop()
|
||
else:
|
||
self._start()
|
||
|
||
def _start(self):
|
||
if self._restart_after_id:
|
||
self.root.after_cancel(self._restart_after_id)
|
||
self._restart_after_id = None
|
||
|
||
if not self.spotify_hwnd or not user32.IsWindow(self.spotify_hwnd):
|
||
self._set_status("SPOTIFY FEHLT", RED)
|
||
return
|
||
|
||
self._user_stopped = False
|
||
user_name = self.pl_entry.get().strip()
|
||
self.rec_btn.configure(state="disabled")
|
||
self._set_status("WIRD GESTARTET\u2026", GREEN)
|
||
threading.Thread(
|
||
target=self._start_async, args=(user_name,), daemon=True
|
||
).start()
|
||
|
||
def _start_async(self, user_name):
|
||
if not user_name:
|
||
detected = self._try_detect_playlist()
|
||
name = detected or datetime.datetime.now().strftime(
|
||
"Spotify_%Y-%m-%d_%H-%M")
|
||
else:
|
||
name = user_name
|
||
|
||
total = self._try_detect_track_count()
|
||
|
||
title = get_window_title(self.spotify_hwnd)
|
||
if not title or " - " not in title:
|
||
self._send_play()
|
||
time.sleep(2)
|
||
|
||
self.root.after(0, lambda n=name, t=total: self._start_finish(n, t))
|
||
|
||
def _start_finish(self, name, total=0):
|
||
if not self.spotify_hwnd or not user32.IsWindow(self.spotify_hwnd):
|
||
self._set_status("SPOTIFY FEHLT", RED)
|
||
self.rec_btn.configure(state="normal")
|
||
return
|
||
|
||
self.pl_entry.configure(state="normal")
|
||
self.pl_entry.delete(0, "end")
|
||
self.pl_entry.insert(0, name)
|
||
|
||
safe = name
|
||
for ch in '<>:"/\\|?*':
|
||
safe = safe.replace(ch, "_")
|
||
self.playlist_name = name
|
||
self.session_dir = os.path.join(self.output_base, safe)
|
||
os.makedirs(self.session_dir, exist_ok=True)
|
||
|
||
try:
|
||
self.recorder.start()
|
||
except RuntimeError as e:
|
||
self._set_status(str(e)[:40], RED)
|
||
self.rec_btn.configure(state="normal")
|
||
return
|
||
|
||
self.recorded_songs = self._scan_existing_songs()
|
||
existing = len(self.recorded_songs)
|
||
|
||
self.active = True
|
||
if total > 0:
|
||
self.playlist_total = total
|
||
self.track_number = existing
|
||
self.current_track = None
|
||
self.songs = []
|
||
self.last_title = ""
|
||
self.pause_since = None
|
||
|
||
skipping = False
|
||
title = get_window_title(self.spotify_hwnd)
|
||
if title and " - " in title:
|
||
parts = title.split(" - ", 1)
|
||
artist, song = parts[0].strip(), parts[1].strip()
|
||
self.last_title = title
|
||
self.is_playing = True
|
||
|
||
if self._song_key(artist, song) in self.recorded_songs:
|
||
skipping = True
|
||
self.artist_lbl.configure(text=artist)
|
||
self.title_lbl.configure(
|
||
text=f"\u2714 {song}")
|
||
self.tnum_lbl.configure(text="BEREITS AUFGENOMMEN")
|
||
else:
|
||
self.track_number += 1
|
||
self.current_track = {"artist": artist, "title": song}
|
||
self._update_track_ui()
|
||
self.rec_start = time.time()
|
||
|
||
if self.playlist_total > 0:
|
||
self.cnt_lbl.configure(
|
||
text=f"{existing} / {self.playlist_total} Songs")
|
||
elif existing > 0:
|
||
self.cnt_lbl.configure(text=f"{existing} bereits vorhanden")
|
||
|
||
self.rec_btn.configure(
|
||
text="\u23f9 AUFNAHME STOPPEN",
|
||
fg_color=RED, hover_color=RED_H, state="normal")
|
||
self.pl_entry.configure(state="disabled")
|
||
if skipping:
|
||
self._set_status("\u23ed WARTE AUF NEUEN SONG", ORANGE)
|
||
else:
|
||
self._set_status("AUFNAHME", GREEN)
|
||
|
||
def _send_play(self):
|
||
"""Send media play/pause key to start Spotify playback."""
|
||
user32.keybd_event(0xB3, 0, 0, 0)
|
||
time.sleep(0.05)
|
||
user32.keybd_event(0xB3, 0, 2, 0)
|
||
|
||
def _try_detect_playlist(self):
|
||
"""Best-effort: detect playlist name via PowerShell UI Automation."""
|
||
if not self.spotify_hwnd:
|
||
return None
|
||
try:
|
||
hwnd_val = int(self.spotify_hwnd)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
ps = (
|
||
"$ErrorActionPreference='SilentlyContinue';"
|
||
"Add-Type -AssemblyName UIAutomationClient;"
|
||
"Add-Type -AssemblyName UIAutomationTypes;"
|
||
"$r=[System.Windows.Automation.AutomationElement]"
|
||
f"::FromHandle([IntPtr]::new({hwnd_val}));"
|
||
"if($r){"
|
||
"$c=[System.Windows.Automation.PropertyCondition]::new("
|
||
"[System.Windows.Automation.AutomationElement]"
|
||
"::ControlTypeProperty,"
|
||
"[System.Windows.Automation.ControlType]::Hyperlink);"
|
||
"$ls=$r.FindAll("
|
||
"[System.Windows.Automation.TreeScope]::Descendants,$c);"
|
||
"foreach($l in $ls){"
|
||
"$n=$l.Current.Name;"
|
||
"$b=$l.Current.BoundingRectangle;"
|
||
"if($n -and $n.Length -ge 2 -and $b.Width -gt 0){"
|
||
"'{0}|{1}' -f [int]$b.Bottom,$n"
|
||
"}}}"
|
||
)
|
||
|
||
try:
|
||
r = subprocess.run(
|
||
["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass",
|
||
"-Command", ps],
|
||
capture_output=True, text=True, timeout=8,
|
||
creationflags=0x08000000,
|
||
)
|
||
raw = [ln.strip() for ln in r.stdout.split("\n")
|
||
if ln.strip() and "|" in ln]
|
||
except Exception:
|
||
return None
|
||
|
||
if not raw:
|
||
return None
|
||
|
||
entries = []
|
||
for line in raw:
|
||
pos, _, nm = line.partition("|")
|
||
try:
|
||
entries.append((int(pos), nm))
|
||
except ValueError:
|
||
pass
|
||
|
||
if not entries:
|
||
return None
|
||
|
||
title = get_window_title(self.spotify_hwnd)
|
||
skip = {
|
||
"spotify", "home", "search", "your library", "library",
|
||
"play", "pause", "next", "previous", "shuffle", "repeat",
|
||
"queue", "volume", "connect to a device", "lyrics",
|
||
"settings", "profile", "close", "minimize", "maximize",
|
||
"now playing view", "premium", "free", "install app",
|
||
"create playlist", "liked songs", "your episodes",
|
||
"what's new", "open in desktop app", "download",
|
||
"go back", "go forward", "resize",
|
||
}
|
||
if title and " - " in title:
|
||
parts = title.split(" - ", 1)
|
||
skip.add(parts[0].strip().lower())
|
||
skip.add(parts[1].strip().lower())
|
||
|
||
entries.sort(key=lambda x: x[0], reverse=True)
|
||
|
||
for _, nm in entries:
|
||
if nm.lower() not in skip and len(nm) >= 2:
|
||
return nm
|
||
|
||
return None
|
||
|
||
def _try_detect_track_count(self):
|
||
"""Best-effort: detect number of tracks in current playlist/album."""
|
||
if not self.spotify_hwnd:
|
||
return 0
|
||
try:
|
||
hwnd_val = int(self.spotify_hwnd)
|
||
except (TypeError, ValueError):
|
||
return 0
|
||
|
||
ps = (
|
||
"$ErrorActionPreference='SilentlyContinue';"
|
||
"Add-Type -AssemblyName UIAutomationClient;"
|
||
"Add-Type -AssemblyName UIAutomationTypes;"
|
||
"$r=[System.Windows.Automation.AutomationElement]"
|
||
f"::FromHandle([IntPtr]::new({hwnd_val}));"
|
||
"if($r){{"
|
||
"$c=[System.Windows.Automation.PropertyCondition]::new("
|
||
"[System.Windows.Automation.AutomationElement]"
|
||
"::ControlTypeProperty,"
|
||
"[System.Windows.Automation.ControlType]::Text);"
|
||
"$ts=$r.FindAll("
|
||
"[System.Windows.Automation.TreeScope]::Descendants,$c);"
|
||
"foreach($t in $ts){{"
|
||
"$n=$t.Current.Name;"
|
||
"if($n -match '(\\d+)\\s*(songs?|titel|tracks?|titres?|morceaux"
|
||
"|canciones|brani)'){{$matches[1];break}}"
|
||
"}}}}"
|
||
)
|
||
|
||
try:
|
||
r = subprocess.run(
|
||
["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass",
|
||
"-Command", ps],
|
||
capture_output=True, text=True, timeout=10,
|
||
creationflags=0x08000000,
|
||
)
|
||
out = r.stdout.strip()
|
||
if out:
|
||
return int(out.split()[0])
|
||
except Exception:
|
||
pass
|
||
return 0
|
||
|
||
def _stop(self, finished=False):
|
||
self.active = False
|
||
self.pause_since = None
|
||
with self._lock:
|
||
frames = self.recorder.harvest_frames()
|
||
if frames and self.current_track and self.track_number > 0:
|
||
self._save(frames, self.current_track, self.track_number)
|
||
self.recorder.stop()
|
||
self.current_track = None
|
||
self.rec_start = None
|
||
|
||
self.rec_btn.configure(text="\u23fa AUFNAHME STARTEN",
|
||
fg_color=GREEN, hover_color=GREEN_H)
|
||
self.pl_entry.configure(state="normal")
|
||
n = len(self.songs)
|
||
if self.playlist_total > 0:
|
||
info = f"{self.track_number}/{self.playlist_total} Songs"
|
||
else:
|
||
info = f"{n} Songs aufgenommen"
|
||
|
||
not_complete = (self.playlist_total > 0
|
||
and self.track_number < self.playlist_total)
|
||
should_restart = not_complete and not self._user_stopped
|
||
|
||
if should_restart and self._restart_fails < 5:
|
||
self._set_status(f"{info} \u2013 NEUSTART\u2026", ORANGE)
|
||
self._restart_countdown(5)
|
||
elif finished:
|
||
self._set_status(f"FERTIG \u2013 {info}", GREEN)
|
||
else:
|
||
self._set_status(
|
||
f"GESTOPPT \u2013 {info}" if n else "BEREIT",
|
||
GREEN if n else DIM)
|
||
self._user_stopped = False
|
||
self.level_bar.set(0)
|
||
|
||
def _restart_countdown(self, secs):
|
||
"""Count down and then auto-restart recording."""
|
||
if secs <= 0:
|
||
self._auto_restart()
|
||
return
|
||
if self._user_stopped or self.active:
|
||
self._restart_after_id = None
|
||
return
|
||
info = f"{self.track_number}/{self.playlist_total}"
|
||
self._set_status(
|
||
f"{info} \u2013 NEUSTART IN {secs}s\u2026", ORANGE)
|
||
self._restart_after_id = self.root.after(
|
||
1000, lambda: self._restart_countdown(secs - 1))
|
||
|
||
def _auto_restart(self):
|
||
"""Automatically restart recording to continue the playlist."""
|
||
self._restart_after_id = None
|
||
if self.active or self._user_stopped:
|
||
return
|
||
if (self.playlist_total > 0
|
||
and self.track_number < self.playlist_total):
|
||
prev_count = self.track_number
|
||
self._start()
|
||
self.root.after(
|
||
15000, lambda p=prev_count: self._check_restart_progress(p))
|
||
|
||
def _check_restart_progress(self, prev_count):
|
||
"""Increment fail counter if no new songs were recorded."""
|
||
if self.track_number <= prev_count and self.active:
|
||
self._restart_fails += 1
|
||
else:
|
||
self._restart_fails = 0
|
||
|
||
# ── track monitoring ───────────────────────────────────────
|
||
|
||
def _tick_monitor(self):
|
||
if self.spotify_hwnd and user32.IsWindow(self.spotify_hwnd):
|
||
try:
|
||
title = get_window_title(self.spotify_hwnd)
|
||
except Exception:
|
||
title = ""
|
||
|
||
if self.active and self.current_track is None and self.is_playing:
|
||
self.recorder.harvest_frames()
|
||
|
||
if not title:
|
||
self.root.after(500, self._tick_monitor)
|
||
return
|
||
|
||
if " - " in title:
|
||
self.ad_playing = False
|
||
if not self.is_playing:
|
||
self.is_playing = True
|
||
self.pause_since = None
|
||
if self.active:
|
||
self.recorder.resume()
|
||
if self.current_track is not None:
|
||
self._set_status("AUFNAHME", GREEN)
|
||
elif self._auto_record_ready:
|
||
self._start()
|
||
if not self._auto_record_ready:
|
||
self._auto_record_ready = True
|
||
|
||
if title != self.last_title and self.active:
|
||
self._handle_track_change(title)
|
||
elif title != self.last_title:
|
||
parts = title.split(" - ", 1)
|
||
self.artist_lbl.configure(text=parts[0].strip())
|
||
self.title_lbl.configure(text=parts[1].strip())
|
||
self.last_title = title
|
||
|
||
else:
|
||
if not self._auto_record_ready:
|
||
self._auto_record_ready = True
|
||
if title.lower().strip() in PAUSE_TITLES:
|
||
self.ad_playing = False
|
||
if self.is_playing:
|
||
self.is_playing = False
|
||
if self.active:
|
||
self.recorder.pause()
|
||
self.pause_since = time.time()
|
||
self._set_status("PAUSIERT", ORANGE)
|
||
else:
|
||
if not self.ad_playing:
|
||
self.ad_playing = True
|
||
if self.active and self.current_track \
|
||
and self.track_number > 0:
|
||
with self._lock:
|
||
frames = self.recorder.harvest_frames()
|
||
if frames:
|
||
num = self.track_number
|
||
trk = dict(self.current_track)
|
||
threading.Thread(
|
||
target=self._save,
|
||
args=(frames, trk, num),
|
||
daemon=True).start()
|
||
self.current_track = None
|
||
if self.active:
|
||
self._set_status("\U0001f507 WERBUNG", ORANGE)
|
||
self.artist_lbl.configure(
|
||
text="Werbung erkannt")
|
||
self.title_lbl.configure(
|
||
text="wird nicht aufgenommen")
|
||
self.tnum_lbl.configure(text="")
|
||
if self.active:
|
||
self.recorder.harvest_frames()
|
||
self.pause_since = None
|
||
|
||
if (self.active and self.pause_since
|
||
and self.auto_stop_secs > 0):
|
||
all_done = (self.playlist_total > 0
|
||
and self.track_number >= self.playlist_total)
|
||
timeout = self.auto_stop_secs
|
||
if self.playlist_total > 0 and not all_done:
|
||
timeout = max(timeout, 300)
|
||
elapsed = time.time() - self.pause_since
|
||
remaining = timeout - elapsed
|
||
if remaining <= 0:
|
||
self._stop(finished=True)
|
||
else:
|
||
if all_done:
|
||
lbl = f"Fertig! Stop {int(remaining)}s"
|
||
elif self.playlist_total > 0:
|
||
lbl = (f"{self.track_number}/{self.playlist_total}"
|
||
f" \u23f8 {int(remaining)}s")
|
||
else:
|
||
lbl = f"Stop {int(remaining)}s"
|
||
self.time_lbl.configure(
|
||
text=lbl, text_color=ORANGE)
|
||
|
||
self.root.after(500, self._tick_monitor)
|
||
|
||
def _handle_track_change(self, new_title):
|
||
parts = new_title.split(" - ", 1)
|
||
new_artist, new_song = parts[0].strip(), parts[1].strip()
|
||
|
||
key = self._song_key(new_artist, new_song)
|
||
if key in self.recorded_songs:
|
||
with self._lock:
|
||
self.recorder.harvest_frames()
|
||
self.current_track = None
|
||
self.artist_lbl.configure(text=new_artist)
|
||
self.title_lbl.configure(text=f"\u2714 {new_song}")
|
||
self.tnum_lbl.configure(text="BEREITS AUFGENOMMEN")
|
||
self._set_status("\u23ed WARTE AUF NEUEN SONG", ORANGE)
|
||
self.rec_start = time.time()
|
||
return
|
||
|
||
new_track = {"artist": new_artist, "title": new_song}
|
||
with self._lock:
|
||
frames = self.recorder.harvest_frames()
|
||
if frames and self.current_track and self.track_number > 0:
|
||
num = self.track_number
|
||
trk = dict(self.current_track)
|
||
threading.Thread(
|
||
target=self._save, args=(frames, trk, num), daemon=True
|
||
).start()
|
||
self.track_number += 1
|
||
self.current_track = new_track
|
||
self.rec_start = time.time()
|
||
|
||
self._update_track_ui()
|
||
self._set_status("AUFNAHME", GREEN)
|
||
|
||
# ── duplicate detection ─────────────────────────────────────
|
||
|
||
@staticmethod
|
||
def _song_key(artist, title):
|
||
return f"{artist}\n{title}".lower().strip()
|
||
|
||
def _scan_existing_songs(self):
|
||
"""Scan session directory for already-recorded songs (by filename)."""
|
||
found = set()
|
||
if not self.session_dir or not os.path.isdir(self.session_dir):
|
||
return found
|
||
for f in os.listdir(self.session_dir):
|
||
if not f.lower().endswith(".mp3"):
|
||
continue
|
||
name = f[:-4]
|
||
parts = name.split(" - ", 2)
|
||
if len(parts) >= 3:
|
||
found.add(self._song_key(parts[1], parts[2]))
|
||
return found
|
||
|
||
# ── save ───────────────────────────────────────────────────
|
||
|
||
def _save(self, frames, track, number):
|
||
artist = track.get("artist", "")
|
||
title = track.get("title", "")
|
||
if not artist and not title:
|
||
return
|
||
|
||
parts = [f"{number:02d}"]
|
||
if artist:
|
||
parts.append(artist)
|
||
if title:
|
||
parts.append(title)
|
||
safe = " - ".join(parts)
|
||
for ch in '<>:"/\\|?*':
|
||
safe = safe.replace(ch, "_")
|
||
|
||
path = os.path.join(self.session_dir, f"{safe}.mp3")
|
||
n = 1
|
||
base = path[:-4]
|
||
while os.path.exists(path):
|
||
path = f"{base} ({n}).mp3"
|
||
n += 1
|
||
|
||
ok = self.recorder.save_mp3(
|
||
frames, path, title=title, artist=artist,
|
||
bitrate=self.bitrate, album=self.playlist_name,
|
||
track_num=number,
|
||
)
|
||
if ok:
|
||
self.recorded_songs.add(self._song_key(artist, title))
|
||
name = os.path.basename(path)
|
||
self.songs.append(name)
|
||
self.root.after(0, lambda: self._add_song(name))
|
||
self.root.after(0, self._refresh_total_stats)
|
||
|
||
def _add_song(self, name):
|
||
self.song_box.configure(state="normal")
|
||
self.song_box.insert("end", f"\u2713 {name}\n")
|
||
self.song_box.see("end")
|
||
self.song_box.configure(state="disabled")
|
||
if self.playlist_total > 0:
|
||
self.cnt_lbl.configure(
|
||
text=f"{self.track_number} / {self.playlist_total} Songs")
|
||
else:
|
||
self.cnt_lbl.configure(text=f"{len(self.songs)} Songs")
|
||
|
||
# ── UI helpers ─────────────────────────────────────────────
|
||
|
||
def _update_track_ui(self):
|
||
if self.current_track:
|
||
self.artist_lbl.configure(text=self.current_track["artist"])
|
||
self.title_lbl.configure(text=self.current_track["title"])
|
||
if self.playlist_total > 0:
|
||
self.tnum_lbl.configure(
|
||
text=f"TRACK #{self.track_number:02d}"
|
||
f" / {self.playlist_total}")
|
||
else:
|
||
self.tnum_lbl.configure(
|
||
text=f"TRACK #{self.track_number:02d}")
|
||
|
||
def _set_status(self, text, color):
|
||
self.stat_lbl.configure(text=text, text_color=color)
|
||
self.dot_lbl.configure(text_color=color)
|
||
|
||
def _tick_timer(self):
|
||
if self.active and self.rec_start:
|
||
s = int(time.time() - self.rec_start)
|
||
m, s = divmod(s, 60)
|
||
self.time_lbl.configure(text=f"{m:02d}:{s:02d}", text_color=GREEN)
|
||
elif not self.active:
|
||
self.time_lbl.configure(text="", text_color=DIM)
|
||
self.root.after(500, self._tick_timer)
|
||
|
||
def _tick_level(self):
|
||
if self.active:
|
||
lvl = self.recorder.get_level()
|
||
smooth = self.level_bar.get() * 0.35 + lvl * 0.65
|
||
self.level_bar.set(smooth)
|
||
self.level_bar.configure(
|
||
progress_color=RED if lvl > 0.85 else GREEN)
|
||
else:
|
||
cur = self.level_bar.get()
|
||
if cur > 0.01:
|
||
self.level_bar.set(cur * 0.5)
|
||
else:
|
||
self.level_bar.set(0)
|
||
self.root.after(80, self._tick_level)
|
||
|
||
def _tick_pulse(self):
|
||
if self.active:
|
||
self._pulse_on = not self._pulse_on
|
||
self.dot_lbl.configure(
|
||
text_color=GREEN if self._pulse_on else "#14532d")
|
||
self.root.after(650, self._tick_pulse)
|
||
|
||
def _tick_alive(self):
|
||
if self.spotify_hwnd:
|
||
if user32.IsWindow(self.spotify_hwnd):
|
||
self._alive_fails = 0
|
||
else:
|
||
self._alive_fails += 1
|
||
new_hwnd = find_spotify_window()
|
||
if new_hwnd:
|
||
self.spotify_hwnd = new_hwnd
|
||
self._alive_fails = 0
|
||
else:
|
||
not_done = (self.playlist_total > 0
|
||
and self.track_number < self.playlist_total)
|
||
threshold = 10 if not_done else 4
|
||
if self._alive_fails >= threshold:
|
||
self.spotify_hwnd = None
|
||
self._update_sp_status(False)
|
||
self.embed_msg.configure(
|
||
text="\u26a0 Spotify wurde geschlossen.\n\n"
|
||
"Starte Spotify neu und\n"
|
||
"starte die App erneut.")
|
||
self.embed_msg.place(relx=0.5, rely=0.5,
|
||
anchor="center")
|
||
if self.active:
|
||
self._stop()
|
||
self.root.after(3000, self._tick_alive)
|
||
|
||
# ── statistics ─────────────────────────────────────────────
|
||
|
||
def _count_total_mp3s(self):
|
||
"""Count all MP3 files recursively in output_base."""
|
||
total = 0
|
||
for root, _dirs, files in os.walk(self.output_base):
|
||
for f in files:
|
||
if f.lower().endswith(".mp3"):
|
||
total += 1
|
||
return total
|
||
|
||
def _refresh_total_stats(self):
|
||
n = self._count_total_mp3s()
|
||
folders = 0
|
||
try:
|
||
for entry in os.scandir(self.output_base):
|
||
if entry.is_dir():
|
||
folders += 1
|
||
except OSError:
|
||
pass
|
||
self.total_lbl.configure(
|
||
text=f"\U0001f4ca Gesamt: {n} Songs in {folders} Playlists")
|
||
|
||
# ── settings ───────────────────────────────────────────────
|
||
|
||
def _choose_folder(self):
|
||
p = filedialog.askdirectory(initialdir=self.output_base)
|
||
if p:
|
||
self.output_base = p
|
||
self.folder_lbl.configure(text=p)
|
||
self._refresh_total_stats()
|
||
|
||
def _open_folder(self):
|
||
folder = self.session_dir if self.session_dir and os.path.isdir(
|
||
self.session_dir) else self.output_base
|
||
os.startfile(folder)
|
||
|
||
def _set_bitrate(self, val):
|
||
self.bitrate = int(val.split()[0])
|
||
|
||
def _set_auto_stop(self, val):
|
||
if val == "Aus":
|
||
self.auto_stop_secs = 0
|
||
else:
|
||
self.auto_stop_secs = int(val.split()[0])
|
||
|
||
# ── cleanup ────────────────────────────────────────────────
|
||
|
||
def _on_close(self):
|
||
self._user_stopped = True
|
||
if self._restart_after_id:
|
||
self.root.after_cancel(self._restart_after_id)
|
||
self._restart_after_id = None
|
||
if self.active:
|
||
self._stop()
|
||
self._restore_spotify()
|
||
self.recorder.cleanup()
|
||
self.root.destroy()
|
||
|
||
def run(self):
|
||
self.root.mainloop()
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════
|
||
# Entry point
|
||
# ═══════════════════════════════════════════════════════════════
|
||
|
||
def main():
|
||
app = SpotifyRecorderApp()
|
||
app.run()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except Exception:
|
||
import tkinter as _tk
|
||
from tkinter import messagebox as _mb
|
||
_r = _tk.Tk()
|
||
_r.withdraw()
|
||
_mb.showerror(
|
||
"Spotify Recorder",
|
||
f"Fehler:\n\n{traceback.format_exc()}\n\n"
|
||
f"pip install -r requirements.txt",
|
||
)
|
||
_r.destroy()
|
||
sys.exit(1)
|