SDH-QuickLaunch-Anatase/main.py

import os, json, base64, ssl, certifi
from pathlib import Path
from json import dumps as jsonDumps
from itertools import chain

import decky_plugin

from subprocess import Popen, PIPE

from urllib.error import HTTPError
from urllib.request import urlopen, Request

FLATPAK_RUNNER = "flatpak" # or explicitly: /app/bin/flatpak
HOST_RUNNER = "hrun" # or explicitly: /app/bin/hrun

confdir = os.environ["DECKY_PLUGIN_SETTINGS_DIR"]

send_buffer = []


def split_string(string):
    return [
        string[i : i + 1024 * 60] for i in range(0, len(string), 1024 * 60)
    ]  # every 60KB


class Plugin:
    async def get_flatpaks(self):
        def list_flatpaks(cmd):
            flatpaks = []

            clean_env = os.environ.copy()
            clean_env["LD_LIBRARY_PATH"] = ""
            with Popen(cmd, stdout=PIPE, stderr=None, env=clean_env, text=True) as p:
                for line in p.stdout:
                    if '\t' not in line:
                        continue

                    name, app_id = line.strip().split('\t', 1)

                    name = name.strip()
                    app_id = app_id.strip()
                    if not name or not app_id:
                        continue

                    flatpaks.append({
                        "id": app_id,
                        "name": name,
                        "exec": f"{FLATPAK_RUNNER} run {app_id}"
                    })

            return flatpaks

        global_flatpaks = list_flatpaks([
            'flatpak', 'list', '--app', '--columns=name,application'
        ])

        local_flatpaks = list_flatpaks([
            'runuser', '-l', decky_plugin.DECKY_USER, '-c',
            'flatpak list --app --columns=name,application'
        ])

        seen = set()
        unique_flatpaks = []

        for flatpak in global_flatpaks + local_flatpaks:
            if flatpak["id"] not in seen:
                seen.add(flatpak["id"])
                unique_flatpaks.append(flatpak)

        sorted_items = sorted(unique_flatpaks, key=lambda x: x.get("name", "").lower())

        return jsonDumps(sorted_items)

    async def get_desktops(self):
        packages = []
        for desktopFile in chain(
            Path("/usr/share/applications").glob("*.desktop"),
            Path(f"{decky_plugin.DECKY_USER_HOME}/.local/share/applications/").glob("*.desktop"),
            Path("/var/usrlocal/share/applications").glob("*.desktop"),
        ):
            if not desktopFile.is_file():
                continue
            with open(desktopFile) as f:
                package = {}
                foundName = foundExec = False
                for line in f:
                    line = line.strip()
                    if line.startswith("Name="):
                        foundName = True
                        package["name"] = line[5:]
                    elif line.startswith("Exec="):
                        foundExec = True
                        execTarget = line[5:].split(" ", 1)
                        decky_plugin.logger.info("- Exec path: " + str(execTarget))
                        runner = FLATPAK_RUNNER if execTarget[0].endswith("flatpak") else HOST_RUNNER
                        package["exec"] = f"{runner} {execTarget[1]}"
                    if foundName and foundExec:
                        decky_plugin.logger.info("+ Desktop app: " + package["name"])
                        packages.append(package)
                        break

        return jsonDumps(packages)

    async def get_DECKY_USER_HOME(self):
        return decky_plugin.DECKY_USER_HOME

    async def get_config(self):
        with open(os.path.join(confdir, "config.json"), "r") as f:
            return json.load(f)

    async def set_config_value(self, key, value):
        config = json.load(open(os.path.join(confdir, "config.json")))
        config[key] = value
        with open(os.path.join(confdir, "config.json"), "w") as f:
            json.dump(config, f)

        return config

    async def get_id(self):
        with open(os.path.join(confdir, "scid.txt"), "r") as sc:
            id = sc.read()
            try:
                id = int(id)
                return id
            except ValueError:
                return -1

    async def set_id(self, id):
        with open(os.path.join(confdir, "scid.txt"), "w") as sc:
            sc.write(str(id))

    async def get_req_imgb64(self, url):
        global send_buffer
        if len(send_buffer) != 0:
            return

        req = Request(url)

        req.add_header("User-Agent", "SDH-QuickLaunch")

        try:
            content = urlopen(
                req, context=ssl.create_default_context(cafile=certifi.where())
            ).read()
            img = base64.b64encode(content).decode("ascii")
            send_buffer = split_string(img)
            new_chunk = send_buffer.pop(0)
            return {"data": new_chunk, "is_last": len(send_buffer) == 0}
        except HTTPError:
            decky_plugin.logger.error("HTTPError while requesting " + url)
            pass

    async def receive_next_chunk(self):
        global send_buffer
        new_chunk = send_buffer.pop(0)
        return {"data": new_chunk, "is_last": len(send_buffer) == 0}

    async def _main(self):
        decky_plugin.logger.info("Loading plugin")

        try:
            os.mkdir(confdir)
        except FileExistsError:
            pass

        try:
            sc = open(os.path.join(confdir, "scid.txt"), "x")
            sc.close()
        except FileExistsError:
            pass
        try:
            sc = open(os.path.join(confdir, "config.json"), "x")
            sc.write("{}")
            sc.close()
        except FileExistsError:
            pass

        decky_plugin.logger.info("Plugin loaded")

    async def _unload(self):
        pass