Migrate Docker containers to dedicated VM
Architecture changes: - Created VM 100 "docker-services" (Debian 12 Cloud Image) - 10GB RAM, 6 Cores, 50GB system disk - Separate 100GB LVM data volume for service data - WireGuard moved from host to VM (10.0.0.2) - All containers migrated and running Updated documentation to reflect new architecture 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Martin Eckardt
367
docker/samsung-tv-api/app.py
Normal file
367
docker/samsung-tv-api/app.py
Normal file
@@ -0,0 +1,367 @@
|
||||
"""
|
||||
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)
|
||||
Reference in New Issue
Block a user