proxmox-infrastruktur/docker/samsung-tv-api/app.py

"""
Samsung TV API Service
Steuert Samsung Smart TVs (2016+) über WebSocket API
Für n8n Integration auf Proxmox
"""

import os
import logging
from flask import Flask, request, jsonify
from samsungtvws import SamsungTVWS
from wakeonlan import send_magic_packet

# Logging konfigurieren
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = Flask(__name__)

# Konfiguration aus Umgebungsvariablen
TV_IP = os.environ.get('SAMSUNG_TV_IP', '192.168.178.100')
TV_MAC = os.environ.get('SAMSUNG_TV_MAC', '')  # Optional für Wake-on-LAN
TV_NAME = os.environ.get('SAMSUNG_TV_NAME', 'n8n-proxmox')
TV_PORT = int(os.environ.get('SAMSUNG_TV_PORT', '8002'))  # 8002 für SSL, 8001 für non-SSL

# Token-Pfad für persistente Authentifizierung
TOKEN_FILE = os.environ.get('SAMSUNG_TV_TOKEN_FILE', '/data/tv-token.txt')


def get_tv_connection():
    """Erstellt eine Verbindung zum Samsung TV"""
    try:
        tv = SamsungTVWS(
            host=TV_IP,
            port=TV_PORT,
            name=TV_NAME,
            token_file=TOKEN_FILE
        )
        return tv
    except Exception as e:
        logger.error(f"TV Verbindung fehlgeschlagen: {e}")
        return None


@app.route('/health', methods=['GET'])
def health():
    """Health-Check Endpoint"""
    return jsonify({'status': 'ok', 'tv_ip': TV_IP})


@app.route('/tv/status', methods=['GET'])
def tv_status():
    """Prüft ob der TV erreichbar ist"""
    try:
        tv = get_tv_connection()
        if tv:
            # Versuche eine einfache Abfrage
            info = tv.rest_device_info()
            return jsonify({
                'status': 'online',
                'info': info
            })
    except Exception as e:
        logger.warning(f"TV nicht erreichbar: {e}")

    return jsonify({'status': 'offline', 'tv_ip': TV_IP})


@app.route('/tv/power/on', methods=['POST'])
def power_on():
    """TV einschalten via Wake-on-LAN"""
    if not TV_MAC:
        return jsonify({
            'error': 'SAMSUNG_TV_MAC nicht konfiguriert',
            'hint': 'Setze SAMSUNG_TV_MAC Umgebungsvariable'
        }), 400

    try:
        send_magic_packet(TV_MAC)
        logger.info(f"Wake-on-LAN gesendet an {TV_MAC}")
        return jsonify({'status': 'wol_sent', 'mac': TV_MAC})
    except Exception as e:
        logger.error(f"Wake-on-LAN fehlgeschlagen: {e}")
        return jsonify({'error': str(e)}), 500


@app.route('/tv/power/off', methods=['POST'])
def power_off():
    """TV ausschalten"""
    try:
        tv = get_tv_connection()
        if tv:
            tv.send_key('KEY_POWER')
            return jsonify({'status': 'ok', 'action': 'power_off'})
    except Exception as e:
        logger.error(f"Power off fehlgeschlagen: {e}")
        return jsonify({'error': str(e)}), 500

    return jsonify({'error': 'TV nicht erreichbar'}), 503


@app.route('/tv/key', methods=['POST'])
def send_key():
    """Sendet eine Taste zum TV

    Body: {"key": "KEY_VOLUP"} oder {"keys": ["KEY_VOLUP", "KEY_VOLUP"]}

    Häufige Keys:
    - KEY_POWER, KEY_POWEROFF
    - KEY_VOLUP, KEY_VOLDOWN, KEY_MUTE
    - KEY_CHUP, KEY_CHDOWN
    - KEY_UP, KEY_DOWN, KEY_LEFT, KEY_RIGHT, KEY_ENTER
    - KEY_RETURN, KEY_HOME, KEY_MENU
    - KEY_SOURCE, KEY_HDMI1, KEY_HDMI2, KEY_HDMI3
    - KEY_PLAY, KEY_PAUSE, KEY_STOP
    """
    data = request.get_json() or {}

    # Einzelne Taste oder Liste von Tasten
    keys = data.get('keys', [])
    if 'key' in data:
        keys = [data['key']]

    if not keys:
        return jsonify({
            'error': 'key oder keys Parameter erforderlich',
            'example': {'key': 'KEY_VOLUP'}
        }), 400

    try:
        tv = get_tv_connection()
        if tv:
            for key in keys:
                tv.send_key(key)
                logger.info(f"Key gesendet: {key}")
            return jsonify({'status': 'ok', 'keys_sent': keys})
    except Exception as e:
        logger.error(f"Key senden fehlgeschlagen: {e}")
        return jsonify({'error': str(e)}), 500

    return jsonify({'error': 'TV nicht erreichbar'}), 503


@app.route('/tv/volume', methods=['POST'])
def set_volume():
    """Setzt die Lautstärke

    Body: {"volume": 25} oder {"action": "up|down|mute"}
    """
    data = request.get_json() or {}

    try:
        tv = get_tv_connection()
        if not tv:
            return jsonify({'error': 'TV nicht erreichbar'}), 503

        if 'volume' in data:
            # Absolute Lautstärke (nicht alle TVs unterstützen das)
            volume = int(data['volume'])
            # Sende entsprechend viele KEY_VOLUP/DOWN
            return jsonify({
                'status': 'ok',
                'note': 'Absolute Lautstärke nicht direkt unterstützt, nutze action: up/down'
            })

        action = data.get('action', '').lower()
        key_map = {
            'up': 'KEY_VOLUP',
            'down': 'KEY_VOLDOWN',
            'mute': 'KEY_MUTE'
        }

        if action in key_map:
            tv.send_key(key_map[action])
            return jsonify({'status': 'ok', 'action': action})

        return jsonify({
            'error': 'Ungültige action',
            'valid_actions': list(key_map.keys())
        }), 400

    except Exception as e:
        logger.error(f"Volume Steuerung fehlgeschlagen: {e}")
        return jsonify({'error': str(e)}), 500


@app.route('/tv/channel', methods=['POST'])
def change_channel():
    """Wechselt den Kanal

    Body: {"action": "up|down"} oder {"number": "123"}
    """
    data = request.get_json() or {}

    try:
        tv = get_tv_connection()
        if not tv:
            return jsonify({'error': 'TV nicht erreichbar'}), 503

        action = data.get('action', '').lower()
        if action == 'up':
            tv.send_key('KEY_CHUP')
            return jsonify({'status': 'ok', 'action': 'channel_up'})
        elif action == 'down':
            tv.send_key('KEY_CHDOWN')
            return jsonify({'status': 'ok', 'action': 'channel_down'})

        # Kanal per Nummer eingeben
        number = data.get('number', '')
        if number:
            key_map = {str(i): f'KEY_{i}' for i in range(10)}
            for digit in str(number):
                if digit in key_map:
                    tv.send_key(key_map[digit])
            tv.send_key('KEY_ENTER')
            return jsonify({'status': 'ok', 'channel': number})

        return jsonify({
            'error': 'action oder number Parameter erforderlich',
            'example': {'action': 'up'}
        }), 400

    except Exception as e:
        logger.error(f"Kanal wechseln fehlgeschlagen: {e}")
        return jsonify({'error': str(e)}), 500


@app.route('/tv/source', methods=['POST'])
def change_source():
    """Wechselt die Eingangsquelle

    Body: {"source": "hdmi1|hdmi2|hdmi3|tv"}
    """
    data = request.get_json() or {}
    source = data.get('source', '').lower()

    source_map = {
        'hdmi1': 'KEY_HDMI1',
        'hdmi2': 'KEY_HDMI2',
        'hdmi3': 'KEY_HDMI3',
        'hdmi4': 'KEY_HDMI4',
        'tv': 'KEY_TV',
        'source': 'KEY_SOURCE'  # Öffnet Source-Menü
    }

    if source not in source_map:
        return jsonify({
            'error': 'Ungültige source',
            'valid_sources': list(source_map.keys())
        }), 400

    try:
        tv = get_tv_connection()
        if tv:
            tv.send_key(source_map[source])
            return jsonify({'status': 'ok', 'source': source})
    except Exception as e:
        logger.error(f"Source wechseln fehlgeschlagen: {e}")
        return jsonify({'error': str(e)}), 500

    return jsonify({'error': 'TV nicht erreichbar'}), 503


@app.route('/tv/apps', methods=['GET'])
def list_apps():
    """Listet installierte Apps auf"""
    try:
        tv = get_tv_connection()
        if tv:
            apps = tv.app_list()
            return jsonify({'status': 'ok', 'apps': apps})
    except Exception as e:
        logger.error(f"App-Liste abrufen fehlgeschlagen: {e}")
        return jsonify({'error': str(e)}), 500

    return jsonify({'error': 'TV nicht erreichbar'}), 503


@app.route('/tv/app/open', methods=['POST'])
def open_app():
    """Startet eine App

    Body: {"app_id": "Netflix"} oder {"app_id": "3201907018807"}

    Bekannte App-IDs:
    - Netflix: 3201907018807
    - YouTube: 111299001912
    - Prime Video: 3201910019365
    - Disney+: 3201901017640
    """
    data = request.get_json() or {}
    app_id = data.get('app_id')

    if not app_id:
        return jsonify({
            'error': 'app_id Parameter erforderlich',
            'hint': 'Nutze GET /tv/apps für eine Liste'
        }), 400

    try:
        tv = get_tv_connection()
        if tv:
            tv.run_app(app_id)
            return jsonify({'status': 'ok', 'app': app_id})
    except Exception as e:
        logger.error(f"App starten fehlgeschlagen: {e}")
        return jsonify({'error': str(e)}), 500

    return jsonify({'error': 'TV nicht erreichbar'}), 503


@app.route('/tv/browser', methods=['POST'])
def open_browser():
    """Öffnet eine URL im TV-Browser

    Body: {"url": "https://example.com"}
    """
    data = request.get_json() or {}
    url = data.get('url')

    if not url:
        return jsonify({'error': 'url Parameter erforderlich'}), 400

    try:
        tv = get_tv_connection()
        if tv:
            tv.open_browser(url)
            return jsonify({'status': 'ok', 'url': url})
    except Exception as e:
        logger.error(f"Browser öffnen fehlgeschlagen: {e}")
        return jsonify({'error': str(e)}), 500

    return jsonify({'error': 'TV nicht erreichbar'}), 503


# API Dokumentation
@app.route('/', methods=['GET'])
def api_docs():
    """API Übersicht"""
    return jsonify({
        'service': 'Samsung TV API',
        'tv_ip': TV_IP,
        'endpoints': {
            'GET /health': 'Health-Check',
            'GET /tv/status': 'TV Status prüfen',
            'POST /tv/power/on': 'TV einschalten (Wake-on-LAN)',
            'POST /tv/power/off': 'TV ausschalten',
            'POST /tv/key': 'Taste senden {"key": "KEY_VOLUP"}',
            'POST /tv/volume': 'Lautstärke {"action": "up|down|mute"}',
            'POST /tv/channel': 'Kanal {"action": "up|down"} oder {"number": "123"}',
            'POST /tv/source': 'Quelle {"source": "hdmi1|hdmi2|tv"}',
            'GET /tv/apps': 'Installierte Apps auflisten',
            'POST /tv/app/open': 'App starten {"app_id": "Netflix"}',
            'POST /tv/browser': 'URL öffnen {"url": "https://..."}',
        },
        'common_keys': [
            'KEY_POWER', 'KEY_VOLUP', 'KEY_VOLDOWN', 'KEY_MUTE',
            'KEY_CHUP', 'KEY_CHDOWN', 'KEY_UP', 'KEY_DOWN',
            'KEY_LEFT', 'KEY_RIGHT', 'KEY_ENTER', 'KEY_RETURN',
            'KEY_HOME', 'KEY_MENU', 'KEY_SOURCE',
            'KEY_PLAY', 'KEY_PAUSE', 'KEY_STOP'
        ]
    })


if __name__ == '__main__':
    logger.info(f"Samsung TV API startet - TV IP: {TV_IP}")
    app.run(host='0.0.0.0', port=5000, debug=True)