toniecloud/tonieapi.py

import asyncio
from pathlib import Path
from playwright.async_api import async_playwright
import requests

async def get_tonie_api_token(username: str, password: str) -> tuple[str | None, str | None]:
    """Loggt sich automatisiert bei Tonies ein und gibt Access- und Refresh-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"
        )

        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=50)
        await page.type("input[id='password'], input[name='password']", password, delay=50)
        await page.click("input[type='submit'], button[type='submit']")

        await page.wait_for_timeout(5000)

        access_token = await page.evaluate("() => window.localStorage.getItem('authorization')")
        refresh_token = await page.evaluate("() => window.localStorage.getItem('refreshToken')")

        await browser.close()
        return access_token, refresh_token


def get_household_id(access_token: str) -> str:
    """Ermittelt die Haushalts-ID des Accounts."""
    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 im Account gefunden.")
    return household_list[0]["id"]


def get_creative_tonies(access_token: str, household_id: str) -> list[dict]:
    """Ruft alle Kreativ-Tonies ab."""
    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"), "chapters": t.get("chapters", [])} 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):
    """Lädt eine Audiodatei hoch und fügt sie als Kapitel an den angegebenen Kreativ-Tonie an."""
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "User-Agent": "Mozilla/5.0"
    }

    # Step 1: Upload-Berechtigung bei Toniecloud anfordern
    print("Fordere S3-Upload-Berechtigung an...")
    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.status_code}): {upload_req_resp.text}")
    
    amazon_data = upload_req_resp.json()
    print("DEBUG S3 Response:", amazon_data) # Hilft beim Debuggen falls es immer noch zickt
    s3_request = amazon_data.get("request", {})
    fields = s3_request.get("fields", {})
    file_id = amazon_data.get("fileId")
    s3_url = s3_request.get("url", "https://bxn-toniecloud-prod-upload.s3.amazonaws.com/")

    # Step 2: Datei an Amazon S3 hochladen (multipart/form-data)
    print(f"Lade Datei '{file_path}' zu S3 hoch...")
    if not Path(file_path).exists():
        raise FileNotFoundError(f"Die Datei '{file_path}' wurde 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 mit Status {s3_resp.status_code}: {s3_resp.text}")

    print("S3-Upload erfolgreich.")

    # Step 3: Kapitelkonfiguration aktualisieren / an den Tonie anfügen
    print("Füge Kapitel zum Kreativ-Tonie hinzu...")
    
    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", "")
        
            # Falls file ein Dictionary ist oder leer/None, direkt verwerfen
            if isinstance(c_file, dict):
                c_file = c_file.get("id") or c_file.get("file")

            # Streng prüfen, ob ein echter String vorhanden ist (kein None, kein leerer String, kein "none")
            if c_id and c_file and str(c_file).strip() 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
    new_chapter = {
        "id": str(file_id),
        "file": str(file_id),
        "title": str(title)
    }
    cleaned_chapters.append(new_chapter)

    # Payload für den Update-Endpunkt zusammenbauen
    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 auf dem Tonie: {update_resp.status_code} - {update_resp.text}")

    print("Kapitel erfolgreich zum Kreativ-Tonie hinzugefügt!")

if __name__ == "__main__":
    import json

    USER = "sandromormile@gmail.com"
    PASS = "Sandro92@tonies"

    print("Starte automatisierten Login...")
    access_token, refresh_token = asyncio.run(get_tonie_api_token(USER, PASS))

    if access_token:
        print("Login erfolgreich.\n")
        household_id = get_household_id(access_token)
        
        # Beispiel 1: Alle Kreativ-Tonies auflisten
        tonies = get_creative_tonies(access_token, household_id)
        print(f"Gefundene Kreativ-Tonies: {len(tonies)}")
        print(json.dumps(tonies, indent=4, ensure_ascii=False))

        # Beispiel 2: Datei auf den ersten Kreativ-Tonie in der Liste hochladen (falls vorhanden)
        if tonies:
            target_tonie = tonies[1]
            print(f"\nLade Datei auf Kreativ-Tonie '{target_tonie['name']}' (ID: {target_tonie['id']}) hoch...")
            
            try:
                upload_audio_to_creative_tonie(
                    access_token=access_token,
                    household_id=household_id,
                    tonie_id=target_tonie["id"],
                    file_path="beispiel.mp3",
                    title="beispiel.mp3",
                    delete_all_chapters=True
                )
            except Exception as e:
                print(f"Upload fehlgeschlagen: {e}")
    else:
        print("Login fehlgeschlagen: Kein Token erhalten.")