Code

import asyncio
import telnetlib3
import re

TELNET_HOST = '127.0.0.1'
TELNET_PORT = 50600
STATUS_COMMAND = 'status'  # ohne \n

async def main():
    print("[DEBUG] Starte Script")
    try:
        reader, writer = await telnetlib3.open_connection(
            TELNET_HOST, TELNET_PORT, encoding=None  # Bytes lesen
        )
    except Exception as e:
        print(f"[ERROR] Verbindung fehlgeschlagen: {e}")
        return

    # Willkommensnachricht lesen bis PL>
    try:
        welcome_bytes = await reader.readuntil(b'PL>')
        welcome = welcome_bytes.decode(errors='ignore')
        print("[DEBUG] Willkommen:\n", welcome)
    except Exception as e:
        print(f"[ERROR] Fehler beim Lesen des Willkommens: {e}")
        writer.close()
        await writer.wait_closed()
        return

    # Status-Befehl senden wie ein Mensch (Enter am Ende)
    print("[DEBUG] Sende Befehl:", STATUS_COMMAND)
    writer.write(STATUS_COMMAND.encode('ascii') + b'\r\n')
    await asyncio.sleep(0.1)  # kurze Pause, damit Antwort beginnt

    # Lese alles bis PL> (Prompt) – hier kommt die Tabelle
    try:
        raw_bytes = await reader.readuntil(b'PL>')
        raw_output = raw_bytes.decode(errors='ignore')
        print("[DEBUG] Roh-Output:\n", raw_output)
    except Exception as e:
        print(f"[ERROR] Fehler beim Lesen der Tabelle: {e}")
        writer.close()
        await writer.wait_closed()
        return

    # Tabelle extrahieren
    table_lines = []
    table_started = False
    for line in raw_output.splitlines():
        if re.match(r'\s*id\s*\|\s*dir\s*\|\s*state', line):
            table_started = True
            continue
        if table_started:
            if line.strip() == '' or line.startswith('PL>'):
                break
            table_lines.append(line)

    print("[DEBUG] Gefundene Tabellenzeilen:")
    for l in table_lines:
        print(l)

    # Tabelle parsen
    calls = []
    for line in table_lines:
        parts = [p.strip() for p in line.split('|')]
        if len(parts) >= 6:
            call = {
                'id': parts[0].lstrip('*'),  # Stern entfernen
                'dir': parts[1],
                'state': parts[2],
                'number': parts[3],
                'duration': parts[4],
                'codecs': parts[5],
            }
            calls.append(call)

    print("\n[RESULT] Calls gefunden:")
    active_call = None
    ringing_call = None
    for c in calls:
        print(f"ID: {c['id']}, State: {c['state']}, Number: {c['number']}, Duration: {c['duration']}, Codecs: {c['codecs']}")
        if c['state'] == 'active':
            active_call = c['id']
        elif c['state'] == 'ring':
            ringing_call = c['id']

    # Calls verwalten
    if active_call:
        print(f"\n[ACTION] Beende Active Call: {active_call}")
        writer.write(f"HookOn {active_call}\r\n".encode('ascii'))
        response = await reader.readuntil(b'PL>')
        print("[DEBUG] Antwort HookOn:\n", response.decode(errors='ignore'))

    if ringing_call:
        print(f"[ACTION] Nimm Ringing Call an: {ringing_call}")
        writer.write(f"HookOff {ringing_call}\r\n".encode('ascii'))
        response = await reader.readuntil(b'PL>')
        print("[DEBUG] Antwort HookOff:\n", response.decode(errors='ignore'))

    writer.close()
    await writer.wait_closed()
    print("\n[RESULT] Script beendet.")

if __name__ == "__main__":
    asyncio.run(main())