import os import json import io import zipfile import asyncio 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 # Konfiguration PASSWORD = 'toniessindteuer' BASE_DIR = '/mnt/NAS/Toniebox/' app = Flask(__name__) app.config['SECRET_KEY'] = 'tonieflix-super-secret-key' CORS(app, supports_credentials=True, allow_headers="*") # --------------------------------------------------------- # Toniecloud API & Playwright Hilfsfunktionen # --------------------------------------------------------- async def get_tonie_api_token(username: str, password: str) -> str | None: """Loggt sich automatisiert bei Tonies ein und gibt das Access-Token zurück.""" 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: print(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"} # 1. S3 Berechtigung anfordern 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/") # 2. Datei zu S3 hochladen 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}") # 3. Tonie-Details abrufen & Kapitel aktualisieren 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)}) # Neues Kapitel anhängen 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 # --------------------------------------------------------- def get_books(single_book=None): books = [] 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) tracks = sorted([f for f in files if f.lower().endswith(('.mp3', '.ogg'))]) has_cover = any(f.lower() == 'cover.jpg' for f in files) if tracks: books.append({'name': item, 'tracks': tracks, 'has_cover': has_cover}) return books # --------------------------------------------------------- # HTML & JS Template (Frontend) # --------------------------------------------------------- TEMPLATE = """ Tonieflix Abmelden

🎬 Tonieflix

{% if single_mode %} 📚 Gesamte Bibliothek laden {% else %} 🦊 Tonie-Login {% endif %}
{% for book in books %}
{% if book.has_cover %} Cover {% else %} Kein Cover {% endif %}
{{ book.name }}
{% 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(): if request.method == 'OPTIONS' or request.path.startswith('/api/books'): return None if request.args.get('pw') == PASSWORD: return None 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') 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') # Playwright Token holen (Asynchroner Aufruf im Sync Flask) token = asyncio.run(get_tonie_api_token(email, pwd)) if token: session['tonie_token'] = token return redirect(url_for('index')) else: 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(): return render_template_string(TEMPLATE, books=get_books(), auto_play="", single_mode=False) @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) @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/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) if not book_name or not tonie_id: return jsonify({"success": False, "error": "Fehlende Parameter."}), 400 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 files = sorted([f for f in os.listdir(book_path) if f.lower().endswith(('.mp3', '.ogg'))]) if not files: return jsonify({"success": False, "error": "Keine Audiospuren im Hörbuch gefunden."}), 400 # Jede Spur nacheinander hochladen for index, track in enumerate(files): file_path = os.path.join(book_path, track) # Titel des Kapitels (Dateiname ohne Endung oder schöner Name) title = os.path.splitext(track)[0] # Beim ersten Track wird ggf. der 'delete_all_chapters' Schalter angewendet, danach hängen wir an 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 ) return jsonify({"success": True}) except Exception as e: return jsonify({"success": False, "error": str(e)}), 500 if __name__ == '__main__': app.run(host='0.0.0.0', port=5005)