bdlg2023-sms-back/script/KdeConnect.py

import subprocess
import logging
import re

from pydantic import BaseModel


class InvalidArgument(Exception):
    pass


class InvalidDevice(Exception):
    pass


class KDEConnectError(Exception):
    pass


class KDEDevice(BaseModel):
    id: str
    name: str
    paired: bool
    reachable: bool


class KDEConnect:
    device: KDEDevice

    def __init__(self, device_name=None, device_id=None):
        if device_name is None and device_id is None:
            raise InvalidArgument("device_name or device_id need to be provided")
        try:
            # Verify kdeconnect is available
            KDEConnect.run_kde_command(["-v"])
        except FileNotFoundError:
            raise KDEConnectError("KDEconnect-cli is not available on this computer")

        # verify the device exist
        devices = KDEConnect.get_device_list()
        device = None
        for d in devices:
            if (d.name == device_name) or (d.id == device_id):
                device = d
                break
        if device is None:
            raise InvalidDevice(f"No device found with id {device_id} or name {device_name}")

        if not device.paired:
            raise InvalidDevice(f"Device {device.name} is not paired with this computer")

        if not device.reachable:
            raise InvalidDevice(f"Device {device.name} is not reachable by this computer")

        self.device = device

    def send_sms(self, phone_number: str, sms_content: str):
        # Compact phone number by removing space
        phone = re.sub("[^+0-9]", "", phone_number)
        command = [
            "-d",
            self.device.id,
            "--send-sms",
            sms_content,
            "--destination",
            phone,
        ]
        KDEConnect.run_kde_command(command)

    @staticmethod
    def run_kde_command(args: list[str]) -> str:
        command: list[str] = ["kdeconnect-cli"]
        command.extend(args)
        process: subprocess.CompletedProcess = subprocess.run(command, capture_output=True)

        subprocess_response = process.stdout.decode("utf-8")
        logging.debug("Subprocess call : \n>> " + " ".join(command) + "\n" + subprocess_response)
        if process.stderr != b"":
            error_msg: str = process.stderr.decode("utf-8")
            if error_msg.endswith("devices found\n"):
                logging.info(error_msg)
            else:
                raise KDEConnectError(error_msg)

        return subprocess_response

    @staticmethod
    def get_device_list() -> list[KDEDevice]:
        """Return the list of devices known by kde connect of this computer"""
        KDEConnect.run_kde_command(["--refresh"])
        output: list[KDEDevice] = []
        device_entries: list[str] = KDEConnect.run_kde_command(["-l"]).strip().split("\n")
        for device_str in device_entries:
            if device_str == "":
                continue
            result = re.search("^- ([^:]+): ([a-z0-9_]+)", device_str)
            reachable = "reachable" in device_str
            paired = "paired" in device_str

            device = KDEDevice(
                name=result.group(1),
                id=result.group(2),
                reachable=reachable,
                paired=paired,
            )
            output.append(device)
        return output


if __name__ == "__main__":
    for d in KDEConnect.get_device_list():
        print(d.name, d.reachable)