import os import json import io import zipfile import asyncio import logging from pathlib import Path from flask import Flask, render_template_string, request, session, redirect, url_for, send_from_directory, send_file, jsonify from flask_cors import CORS from playwright.async_api import async_playwright import requests import mutagen from mutagen.mp3 import MP3 from mutagen.oggvorbis import OggVorbis import config # Konfiguration PASSWORD = config.PASSWORD BASE_DIR = config.BASE_DIR JSON_DB_PATH = getattr(config, 'JSON_DB_PATH', 'toniebox.json') # Logging Setup logging.basicConfig( filename='toniecloud.log', level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S', force=True ) logger = logging.getLogger('tonieflix') # Werkzeug-Zugriffslogs stummschalten logging.getLogger('werkzeug').setLevel(logging.WARNING) app = Flask(__name__) app.config['SECRET_KEY'] = config.SECRET_KEY CORS(app, supports_credentials=True, allow_headers="*") # --------------------------------------------------------- # Hilfsfunktion: Exakte Track-Dauer ermitteln (Fehlertolerant) # --------------------------------------------------------- def get_track_duration(file_path: str) -> float: try: # Generischer Mutagen-Fallback, der mit MP3 und OGG/Vorbis/Opus umgehen kann audio = mutagen.File(file_path) if audio and hasattr(audio, 'info') and hasattr(audio.info, 'length'): return audio.info.length except Exception as e: logger.warning(f"Konnte Dauer für {file_path} nicht auslesen: {e}") log_to_db("WARNUNG", f"Konnte Dauer für {file_path} nicht auslesen: {e}") return 0.0 # --------------------------------------------------------- # JSON Datenbank verarbeiten # --------------------------------------------------------- def load_metadata(): metadata = {} if os.path.exists(JSON_DB_PATH): try: with open(JSON_DB_PATH, 'r', encoding='utf-8') as f: raw_data = json.load(f) for item in raw_data: if item.get('type') == 'table' and item.get('name') == 'toniebox': for row in item.get('data', []): title = row.get('title') if not title: continue title_key = title.lower().strip() if title_key not in metadata: metadata[title_key] = {'genres': set(), 'formats': set(), 'agegroups': set()} if row.get('genre'): for g in str(row['genre']).split('|'): clean_g = g.strip() if clean_g: metadata[title_key]['genres'].add(clean_g) if row.get('format'): for fmt in str(row['format']).split('|'): clean_fmt = fmt.strip() if clean_fmt: metadata[title_key]['formats'].add(clean_fmt) if row.get('agegroup'): for age in str(row['agegroup']).split('|'): clean_age = age.strip() if clean_age: metadata[title_key]['agegroups'].add(clean_age) except Exception as e: logger.error(f"Fehler beim Laden der JSON-Datenbank: {e}") log_to_db("WARNUNG", f"Fehler beim Laden der JSON-Datenbank: {e}") for key, val in metadata.items(): val['genres'] = list(val['genres']) val['formats'] = list(val['formats']) val['agegroups'] = list(val['agegroups']) return metadata # --------------------------------------------------------- # Toniecloud API & Playwright Hilfsfunktionen # --------------------------------------------------------- def get_tonie_api_token(username: str, password: str) -> str | None: token_url = "https://login.tonies.com/auth/realms/tonies/protocol/openid-connect/token" payload = { "grant_type": "password", "client_id": "my-tonies", "username": username, "password": password, "scope": "openid" } headers = { "Content-Type": "application/x-www-form-urlencoded" } try: resp = requests.post(token_url, data=payload, headers=headers, timeout=10) if resp.status_code == 200: data = resp.json() return data.get("access_token") else: logger.error(f"Fehler bei Token-Anfrage: {resp.status_code} - {resp.text}") log_to_db("FEHLER", f"Fehler bei Token-Anfrage: {resp.status_code} - {resp.text}") return None except Exception as e: logger.error(f"Exception beim Token-Abruf: {e}") log_to_db("FEHLER", f"Exception beim Token-Abruf: {e}") return None async def get_tonie_api_token_playwright(username: str, password: str) -> str | None: async with async_playwright() as p: browser = await p.chromium.launch(headless=True) context = await browser.new_context(viewport={'width': 1280, 'height': 800}) page = await context.new_page() login_url = ( "https://login.tonies.com/auth/realms/tonies/protocol/openid-connect/auth?" "client_id=my-tonies&redirect_uri=https%3A%2F%2Fmy.tonies.com%2Flogin%3Fredirect%3D%252F&" "state=24115267-a4f0-4033-a2d4-bef68c83b07d&response_mode=fragment&" "response_type=code&scope=openid&nonce=ee64d559-2b1c-4f4d-a01f-40d84c4dced1&" "ui_locales=de&code_challenge=_lFACYXyUjTQEnb3JiqUWQ7xu4zA50xMNyXhhp3uUe8&" "code_challenge_method=S256" ) try: await page.goto(login_url) try: await page.wait_for_selector("[data-testid='email-toggle']", timeout=5000) await page.click("[data-testid='email-toggle']") except Exception: pass await page.wait_for_timeout(1000) await page.wait_for_selector("input[id='username'], input[name='username']", timeout=5000) await page.type("input[id='username'], input[name='username']", username, delay=30) await page.type("input[id='password'], input[name='password']", password, delay=30) await page.click("input[type='submit'], button[type='submit']") await page.wait_for_timeout(5000) access_token = await page.evaluate("() => window.localStorage.getItem('authorization')") await browser.close() return access_token except Exception as e: logger.error(f"Playwright Login-Fehler: {e}") log_to_db("FEHLER", f"Playwright Login-Fehler: {e}") await browser.close() return None def get_household_id(access_token: str) -> str: headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json", "User-Agent": "Mozilla/5.0"} resp = requests.get("https://api.tonie.cloud/v2/households", headers=headers) if resp.status_code != 200: raise Exception(f"Fehler beim Laden der Haushalte: {resp.status_code} - {resp.text}") households = resp.json() household_list = households if isinstance(households, list) else households.get("households", []) if not household_list: raise Exception("Kein Haushalt gefunden.") return household_list[0]["id"] def get_creative_tonies(access_token: str, household_id: str) -> list[dict]: headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json", "User-Agent": "Mozilla/5.0"} resp = requests.get(f"https://api.tonie.cloud/v2/households/{household_id}/creativetonies", headers=headers) if resp.status_code != 200: raise Exception(f"Fehler beim Laden der Kreativ-Tonies: {resp.status_code} - {resp.text}") raw_tonies = resp.json() if isinstance(raw_tonies, dict): raw_tonies = raw_tonies.get("creativeTonies", raw_tonies.get("items", [])) return [{"id": t.get("id"), "name": t.get("name"), "imageUrl": t.get("imageUrl")} for t in raw_tonies] def upload_audio_to_creative_tonie(access_token: str, household_id: str, tonie_id: str, file_path: str, title: str, delete_all_chapters: bool): headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json", "User-Agent": "Mozilla/5.0"} upload_req_resp = requests.post("https://api.tonie.cloud/v2/file", json={"headers": {}}, headers=headers) if upload_req_resp.status_code != 200: raise Exception(f"Upload-Request fehlgeschlagen: {upload_req_resp.text}") amazon_data = upload_req_resp.json() s3_request = amazon_data.get("request", {}) fields = s3_request.get("fields", {}) file_id = amazon_data.get("fileId") or amazon_data.get("fileID") or amazon_data.get("file") s3_url = s3_request.get("url", "https://bxn-toniecloud-prod-upload.s3.amazonaws.com/") if not Path(file_path).exists(): raise FileNotFoundError(f"Datei {file_path} nicht gefunden.") multipart_fields = { "key": fields.get("key"), "x-amz-algorithm": fields.get("x-amz-algorithm"), "x-amz-credential": fields.get("x-amz-credential"), "x-amz-date": fields.get("x-amz-date"), "policy": fields.get("policy"), "x-amz-signature": fields.get("x-amz-signature"), "x-amz-security-token": fields.get("x-amz-security-token"), } with open(file_path, "rb") as f: files = {"file": (fields.get("key"), f, "audio/mpeg")} s3_resp = requests.post(s3_url, data=multipart_fields, files=files) if s3_resp.status_code not in (200, 204): raise Exception(f"S3-Upload fehlgeschlagen: {s3_resp.text}") tonie_detail_resp = requests.get(f"https://api.tonie.cloud/v2/households/{household_id}/creativetonies/{tonie_id}", headers=headers) if tonie_detail_resp.status_code != 200: raise Exception(f"Konnte Tonie-Details nicht abrufen: {tonie_detail_resp.text}") tonie_data = tonie_detail_resp.json() cleaned_chapters = [] if not delete_all_chapters: for ch in tonie_data.get("chapters", []): c_id = ch.get("id") c_file = ch.get("file") c_title = ch.get("title", "") if isinstance(c_file, dict): c_file = c_file.get("id") or c_file.get("file") if c_id and c_file and str(c_file).lower() != "none": cleaned_chapters.append({"id": str(c_id), "file": str(c_file), "title": str(c_title)}) cleaned_chapters.append({ "id": str(fields.get("key")), "file": str(file_id), "title": str(title) }) update_payload = { "name": tonie_data.get("name"), "imageUrl": tonie_data.get("imageUrl"), "chapters": cleaned_chapters } update_resp = requests.put( f"https://api.tonie.cloud/v2/households/{household_id}/creativetonies/{tonie_id}", json=update_payload, headers=headers ) if update_resp.status_code not in (200, 204): raise Exception(f"Fehler beim Speichern der Kapitel: {update_resp.text}") # --------------------------------------------------------- # Bibliothek abrufen (Nur Grundstruktur, superschnell) # --------------------------------------------------------- def get_books(single_book=None): books = [] metadata_lookup = load_metadata() if os.path.exists(BASE_DIR): items_to_check = [single_book] if single_book else sorted(os.listdir(BASE_DIR)) for item in items_to_check: item_path = os.path.join(BASE_DIR, item) if os.path.isdir(item_path): files = os.listdir(item_path) track_files = sorted([f for f in files if f.lower().endswith(('.mp3', '.ogg'))]) has_cover = any(f.lower() == 'cover.jpg' for f in files) if track_files: item_key = item.lower().strip() meta = metadata_lookup.get(item_key, {'genres': [], 'formats': [], 'agegroups': []}) books.append({ 'name': item, 'tracks': track_files, 'has_cover': has_cover, 'genres': meta['genres'], 'formats': meta['formats'], 'agegroups': meta['agegroups'] }) return books # --------------------------------------------------------- # HTML & JS Template (Frontend) # --------------------------------------------------------- TEMPLATE = """ Tonieflix Abmelden

🎬 Tonieflix

{% if single_mode %} 📚 Gesamte Bibliothek laden {% else %} 🦊 Tonie-Login {% endif %}
{% if not single_mode %}
{% endif %}
{% for book in books %}
{% if book.has_cover %} Cover {% else %} Kein Cover {% endif %}
{{ book.name }}
{% if book.genres or book.formats or book.agegroups %}
{{ book.formats|join(', ') }} • {{ book.agegroups|join(', ') }}
{{ book.genres|join(', ') }}
{% endif %}
{% else %}

Keine Hörbücher gefunden.

{% endfor %}
×

Kreativ-Tonie auswählen

Lade Kreativ-Tonies...

Lade Hörbuch auf Kreativ-Tonie hoch...

Bitte Fenster nicht schließen.

""" LOGIN_TEMPLATE = """ Login - Tonieflix

🎬 Tonieflix

""" TONIE_LOGIN_TEMPLATE = """ Tonie-Login - Tonieflix

🦊 Tonies Account

Gib deine Zugangsdaten für my.tonies.com ein:

{% if error %}

{{ error }}

{% endif %}
""" # --------------------------------------------------------- # Flask Routen # --------------------------------------------------------- @app.before_request def require_login(): # CORS Preflight oder öffentliche API-Routen ignorieren if request.method == 'OPTIONS' or request.path.startswith('/api/books'): return None # Passwort über URL-Parameter abfangen (?pw=... oder ?password=...) url_pw = request.args.get('pw') or request.args.get('password') if url_pw and url_pw == PASSWORD: session['logged_in'] = True # Sauberer Redirect ohne Passwort-Parameter in der URL-Historie: args = request.args.copy() args.pop('pw', None) args.pop('password', None) clean_url = request.path + ('?' + args.to_encoded_string() if args else '') return redirect(clean_url) # Prüfen, ob eingeloggt oder auf Login/Statics unterwegs if request.endpoint not in ['login', 'static'] and not session.get('logged_in'): return redirect(url_for('login', next=request.full_path if request.full_path != '/' else None)) @app.route('/login', methods=['GET', 'POST']) def login(): next_url = request.args.get('next') or url_for('index') # Schneller Login per GET-Parameter direkt auf /login url_pw = request.args.get('pw') or request.args.get('password') if url_pw == PASSWORD: session['logged_in'] = True return redirect(next_url) if request.method == 'POST': if request.form.get('password') == PASSWORD: session['logged_in'] = True return redirect(next_url) else: return "Falsches Passwort!", 403 return render_template_string(LOGIN_TEMPLATE) @app.route('/tonie-login', methods=['GET', 'POST']) def tonie_login(): error = None if request.method == 'POST': email = request.form.get('tonie_email') pwd = request.form.get('tonie_password') token = get_tonie_api_token(email, pwd) if token: session['tonie_token'] = token session['tonie_email'] = email logger.info(f"Toniecloud-Login erfolgreich für E-Mail: {email} (IP: {request.remote_addr})") log_to_db("INFO", f"Toniecloud-Login erfolgreich für E-Mail: {email} (IP: {request.remote_addr})") return redirect(url_for('index')) else: logger.warning(f"Toniecloud-Login fehlgeschlagen für E-Mail: {email} (IP: {request.remote_addr})") log_to_db("WARNUNG", f"Toniecloud-Login fehlgeschlagen für E-Mail: {email} (IP: {request.remote_addr})") error = "Login fehlgeschlagen. Bitte Zugangsdaten prüfen." return render_template_string(TONIE_LOGIN_TEMPLATE, error=error) @app.route('/logout') def logout(): session.clear() return redirect(url_for('login')) @app.route('/') def index(): books = get_books() all_genres = set() all_formats = set() all_agegroups = set() for b in books: all_genres.update(b['genres']) all_formats.update(b['formats']) all_agegroups.update(b['agegroups']) return render_template_string( TEMPLATE, books=books, auto_play="", single_mode=False, all_genres=sorted(list(all_genres)), all_formats=sorted(list(all_formats)), all_agegroups=sorted(list(all_agegroups)) ) @app.route('/play/') def play_book(book_name): return render_template_string(TEMPLATE, books=get_books(single_book=book_name), auto_play=book_name, single_mode=True, all_genres=[], all_formats=[], all_agegroups=[]) @app.route('/cover/') def serve_cover(book_name): return send_from_directory(os.path.join(BASE_DIR, book_name), 'cover.jpg') @app.route('/stream//') def serve_audio(book_name, track_name): return send_from_directory(os.path.join(BASE_DIR, book_name), track_name) @app.route('/download/') def download_book(book_name): book_path = os.path.join(BASE_DIR, book_name) if not os.path.isdir(book_path): return "Hörbuch nicht gefunden", 404 memory_file = io.BytesIO() with zipfile.ZipFile(memory_file, 'w', zipfile.ZIP_STORED) as zf: for root, dirs, files in os.walk(book_path): for file in files: zf.write(os.path.join(root, file), arcname=file) memory_file.seek(0) return send_file(memory_file, download_name=f"{book_name}.zip", as_attachment=True, mimetype='application/zip') # --- API Endpunkte für Tonies --- @app.route('/api/tonies') def api_tonies(): token = session.get('tonie_token') if not token: return jsonify({"error": "Nicht bei Tonies eingeloggt."}), 401 try: household_id = get_household_id(token) tonies = get_creative_tonies(token, household_id) return jsonify(tonies) except Exception as e: return jsonify({"error": str(e)}), 500 @app.route('/api/book-details') def api_book_details(): book_name = request.args.get('book_name') if not book_name: return jsonify({"error": "Kein Hörbuch angegeben."}), 400 book_path = os.path.join(BASE_DIR, book_name) if not os.path.isdir(book_path): return jsonify({"error": "Hörbuch-Ordner nicht gefunden."}), 404 try: files = sorted([f for f in os.listdir(book_path) if f.lower().endswith(('.mp3', '.ogg'))]) tracks_data = [] for f in files: f_path = os.path.join(book_path, f) duration = get_track_duration(f_path) tracks_data.append({"name": f, "duration": duration}) return jsonify({"tracks": tracks_data}) except Exception as e: return jsonify({"error": str(e)}), 500 @app.route('/api/upload-to-tonie', methods=['POST']) def api_upload_to_tonie(): token = session.get('tonie_token') if not token: return jsonify({"success": False, "error": "Nicht eingeloggt."}), 401 data = request.json book_name = data.get('book_name') tonie_id = data.get('tonie_id') delete_all = data.get('delete_all', False) selected_tracks = data.get('tracks', []) user_email = session.get('tonie_email', 'Unbekannt') if not book_name or not tonie_id or not selected_tracks: return jsonify({"success": False, "error": "Fehlende Parameter."}), 400 logger.info(f"Start Upload von '{book_name}' ({len(selected_tracks)} Kapitel) auf Tonie {tonie_id} durch Benutzer: {user_email} (IP: {request.remote_addr})") log_to_db("INFO", f"Start Upload von '{book_name}' ({len(selected_tracks)} Kapitel) auf Tonie {tonie_id} durch Benutzer: {user_email} (IP: {request.remote_addr})") try: household_id = get_household_id(token) book_path = os.path.join(BASE_DIR, book_name) if not os.path.isdir(book_path): return jsonify({"success": False, "error": "Hörbuch-Ordner nicht gefunden."}), 404 for index, track in enumerate(selected_tracks): file_path = os.path.join(book_path, track) title = os.path.splitext(track)[0] should_delete = delete_all if index == 0 else False upload_audio_to_creative_tonie( access_token=token, household_id=household_id, tonie_id=tonie_id, file_path=file_path, title=title, delete_all_chapters=should_delete ) logger.info(f"Upload von '{book_name}' auf Tonie {tonie_id} durch {user_email} erfolgreich abgeschlossen.") log_to_db("INFO", f"Upload von '{book_name}' auf Tonie {tonie_id} durch {user_email} erfolgreich abgeschlossen.") return jsonify({"success": True}) except Exception as e: logger.error(f"Fehler beim Upload von '{book_name}' durch {user_email}: {str(e)}") log_to_db("FEHLER", f"Fehler beim Upload von '{book_name}' durch {user_email}: {str(e)}") return jsonify({"success": False, "error": str(e)}), 500 def log_to_db(typ: str, logsatz: str): try: database = "toniecloud" url = f"http://192.168.189.118:8000/{database}/execute" # 1. SQL-Injection / Syntax-Fehler bei Quotes verhindern (' -> '') safe_typ = str(typ).replace("'", "''") safe_logsatz = str(logsatz).replace("'", "''") payload = { "aufrufer" : "tonieflix", "sql": f"INSERT INTO flixlog (typ, logsatz) VALUES ('{safe_typ}', '{safe_logsatz}')" } # 2. Timeout (z.B. 2 Sekunden), damit das Hauptprogramm nicht hängen bleibt response = requests.post(url, json=payload, timeout=2.0) if response.status_code != 200: print(f"⚠️ Log fehlgeschlagen (Code {response.status_code}): {response.text}") except requests.exceptions.Timeout: print("⚠️ Log-Timeout: Datenbank-Server antwortet nicht.") except Exception as e: print(f"⚠️ Log-Fehler: {str(e)}") if __name__ == '__main__': log_to_db("INFO", "Starte Tonieflix") app.run(host='0.0.0.0', port=5005)