mirror of
https://github.com/morgan9e/warehouse
synced 2026-04-14 00:04:08 +09:00
394 lines
11 KiB
Python
394 lines
11 KiB
Python
from gi.repository import Gio, Gtk, GLib, Gdk
|
|
from src.gtk.error_toast import ErrorToast
|
|
import subprocess, os, pathlib
|
|
import gettext
|
|
|
|
_ = gettext.gettext
|
|
home = f"{pathlib.Path.home()}"
|
|
icon_theme = Gtk.IconTheme.new()
|
|
icon_theme.add_search_path(f"{home}/.local/share/flatpak/exports/share/icons")
|
|
direction = Gtk.Image().get_direction()
|
|
|
|
|
|
class Flatpak:
|
|
def open_app(self, callback=None):
|
|
self.failed_app_run = None
|
|
|
|
def thread(*args):
|
|
if self.is_runtime:
|
|
self.failed_app_run = "error: cannot open a runtime"
|
|
try:
|
|
subprocess.run(["flatpak-spawn", "--host", "flatpak", "run", f"{self.info['ref']}"], capture_output=True, text=True, check=True)
|
|
except subprocess.CalledProcessError as cpe:
|
|
self.failed_app_run = cpe
|
|
except Exception as e:
|
|
self.failed_app_run = e
|
|
|
|
Gio.Task.new(None, None, callback).run_in_thread(thread)
|
|
|
|
def open_data(self):
|
|
if not os.path.exists(self.data_path):
|
|
return f"Path '{self.data_path}' does not exist"
|
|
try:
|
|
Gio.AppInfo.launch_default_for_uri(f"file://{self.data_path}", None)
|
|
except GLib.GError as e:
|
|
return e
|
|
|
|
def get_data_size(self, callback=None):
|
|
size = [None]
|
|
|
|
def thread(*args):
|
|
sed = "sed 's/K/ KB/; s/M/ MB/; s/G/ GB/; s/T/ TB/; s/P/ PB/;'"
|
|
size[0] = subprocess.run(["sh", "-c", f"du -sh {self.data_path} | {sed}"], capture_output=True, text=True).stdout.split("\t")[0]
|
|
|
|
def on_done(*arg):
|
|
if callback:
|
|
callback(f"~ {size[0]}")
|
|
|
|
Gio.Task.new(None, None, on_done).run_in_thread(thread)
|
|
|
|
def trash_data(self, callback=None):
|
|
try:
|
|
subprocess.run(["gio", "trash", self.data_path], capture_output=True, text=True, check=True)
|
|
except subprocess.CalledProcessError as cpe:
|
|
raise cpe
|
|
except Exception as e:
|
|
raise e
|
|
|
|
def set_mask(self, should_mask, callback=None):
|
|
self.failed_mask = None
|
|
|
|
def thread(*args):
|
|
cmd = ["flatpak-spawn", "--host", "flatpak", "mask", self.info["id"]]
|
|
installation = self.info["installation"]
|
|
if installation == "user" or installation == "system":
|
|
cmd.append(f"--{installation}")
|
|
else:
|
|
cmd.append(f"--installation={installation}")
|
|
|
|
if not should_mask:
|
|
cmd.append("--remove")
|
|
|
|
try:
|
|
subprocess.run(cmd, check=True, capture_output=True, text=True)
|
|
self.is_masked = should_mask
|
|
except subprocess.CalledProcessError as cpe:
|
|
self.failed_mask = cpe
|
|
except Exception as e:
|
|
self.failed_mask = e
|
|
|
|
Gio.Task.new(None, None, callback).run_in_thread(thread)
|
|
|
|
def set_pin(self, should_pin, callback=None):
|
|
self.failed_pin = None
|
|
if not self.is_runtime:
|
|
self.failed_pin = "Cannot pin an application"
|
|
|
|
def thread(*args):
|
|
cmd = ["flatpak-spawn", "--host", "flatpak", "pin", f"runtime/{self.info['ref']}"]
|
|
installation = self.info["installation"]
|
|
if installation == "user" or installation == "system":
|
|
cmd.append(f"--{installation}")
|
|
else:
|
|
cmd.append(f"--installation={installation}")
|
|
|
|
if not should_pin:
|
|
cmd.append("--remove")
|
|
|
|
try:
|
|
subprocess.run(cmd, check=True, capture_output=True, text=True)
|
|
except subprocess.CalledProcessError as cpe:
|
|
self.failed_pin = cpe
|
|
except Exception as e:
|
|
self.failed_mask = e
|
|
|
|
Gio.Task.new(None, None, callback).run_in_thread(thread)
|
|
|
|
def uninstall(self, callee_callback=None):
|
|
self.failed_uninstall = None
|
|
|
|
def callback(*args):
|
|
HostInfo.main_window.remove_refresh_lockout("uninstalling packages")
|
|
if not callee_callback is None:
|
|
callee_callback()
|
|
|
|
def thread(*args):
|
|
HostInfo.main_window.add_refresh_lockout("uninstalling packages")
|
|
cmd = ["flatpak-spawn", "--host", "flatpak", "uninstall", "-y", self.info["ref"]]
|
|
installation = self.info["installation"]
|
|
if installation == "system" or installation == "user":
|
|
cmd.append(f"--{installation}")
|
|
else:
|
|
cmd.append(f"--installation={installation}")
|
|
|
|
try:
|
|
subprocess.run(cmd, check=True, text=True, capture_output=True)
|
|
except subprocess.CalledProcessError as cpe:
|
|
self.failed_uninstall = cpe
|
|
except Exception as e:
|
|
self.failed_uninstall = e
|
|
|
|
Gio.Task.new(None, None, callback).run_in_thread(thread)
|
|
|
|
def get_cli_info(self):
|
|
cli_info = {}
|
|
cmd = "LC_ALL=C flatpak info "
|
|
installation = self.info["installation"]
|
|
|
|
if installation == "user":
|
|
cmd += "--user "
|
|
elif installation == "system":
|
|
cmd += "--system "
|
|
else:
|
|
cmd += f"--installation={installation} "
|
|
|
|
cmd += self.info["ref"]
|
|
try:
|
|
output = subprocess.run(["flatpak-spawn", "--host", "sh", "-c", cmd], text=True, capture_output=True).stdout
|
|
except Exception as e:
|
|
raise e
|
|
|
|
lines = output.strip().split("\n")
|
|
cli_info["description"] = ""
|
|
first = lines.pop(0)
|
|
if " - " in first:
|
|
cli_info["description"] = first.split(" - ")[1]
|
|
|
|
# Handle descriptions that contain newlines
|
|
while (line := lines.pop(0)) and not ":" in line:
|
|
if len(line) > 0:
|
|
cli_info["description"] += f" {line}"
|
|
|
|
for i, word in enumerate(lines):
|
|
if not ":" in word:
|
|
continue
|
|
|
|
word = word.strip().split(": ", 1)
|
|
if len(word) < 2:
|
|
continue
|
|
|
|
word[0] = word[0].lower()
|
|
if "installed" in word[0]:
|
|
word[1] = word[1].replace("?", " ")
|
|
cli_info[word[0]] = word[1]
|
|
|
|
self.cli_info = cli_info
|
|
return cli_info
|
|
|
|
def __init__(self, columns):
|
|
self.info = {
|
|
"name": columns[0],
|
|
"id": columns[1],
|
|
"version": columns[2],
|
|
"branch": columns[3],
|
|
"arch": columns[4],
|
|
"origin": columns[5],
|
|
"installation": columns[6],
|
|
"ref": columns[7],
|
|
"installed_size": columns[8],
|
|
"options": columns[9],
|
|
}
|
|
self.is_runtime = "runtime" in self.info["options"]
|
|
self.data_path = f"{home}/.var/app/{self.info["id"]}"
|
|
self.data_size = -1
|
|
self.cli_info = None
|
|
installation = self.info["installation"]
|
|
if len(i := installation.split(" ")) > 1:
|
|
self.info["installation"] = i[1].replace("(", "").replace(")", "")
|
|
else:
|
|
self.info["installation"] = installation
|
|
|
|
self.is_eol = "eol=" in self.info["options"]
|
|
self.dependent_runtime = None
|
|
self.failed_app_run = None
|
|
self.failed_mask = None
|
|
self.failed_uninstall = None
|
|
self.app_row = None
|
|
|
|
try:
|
|
self.is_masked = self.info["id"] in HostInfo.masks[self.info["installation"]]
|
|
except KeyError:
|
|
self.is_masked = False
|
|
|
|
try:
|
|
self.is_pinned = f"runtime/{self.info['ref']}" in HostInfo.pins[self.info["installation"]]
|
|
except KeyError:
|
|
self.is_pinned = False
|
|
|
|
try:
|
|
self.icon_path = icon_theme.lookup_icon(self.info["id"], None, 512, 1, direction, 0).get_file().get_path()
|
|
except GLib.GError as e:
|
|
print(f"Minor error in looking up icon for {self.info['id']}", e)
|
|
self.icon_path = None
|
|
|
|
|
|
class Remote:
|
|
def __init__(self, name, title, disabled):
|
|
self.name = name
|
|
self.title = title
|
|
self.disabled = disabled
|
|
if title == "" or title == "-":
|
|
self.title = name
|
|
|
|
|
|
class HostInfo:
|
|
home = home
|
|
clipboard = Gdk.Display.get_default().get_clipboard()
|
|
main_window = None
|
|
snapshots_path = f"{home}/.var/app/io.github.flattool.Warehouse/data/Snapshots/"
|
|
|
|
# Get all possible installation icon theme dirs
|
|
output = subprocess.run(
|
|
["flatpak-spawn", "--host", "flatpak", "--installations"],
|
|
text=True,
|
|
capture_output=True,
|
|
).stdout
|
|
lines = output.strip().split("\n")
|
|
for i in lines:
|
|
icon_theme.add_search_path(f"{i}/exports/share/icons")
|
|
|
|
flatpaks = []
|
|
id_to_flatpak = {}
|
|
ref_to_flatpak = {}
|
|
remotes = {}
|
|
installations = []
|
|
masks = {}
|
|
pins = {}
|
|
dependent_runtime_refs = []
|
|
|
|
@classmethod
|
|
def get_flatpaks(this, callback=None):
|
|
# Callback is a function to run after the host flatpaks are found
|
|
this.flatpaks.clear()
|
|
this.id_to_flatpak.clear()
|
|
this.ref_to_flatpak.clear()
|
|
this.remotes.clear()
|
|
this.installations.clear()
|
|
this.masks.clear()
|
|
this.pins.clear()
|
|
this.dependent_runtime_refs.clear()
|
|
|
|
def thread(task, *args):
|
|
# Remotes
|
|
def remote_info(installation):
|
|
cmd = ["flatpak-spawn", "--host", "flatpak", "remotes", "--columns=name,title,options", "--show-disabled"]
|
|
if installation == "user" or installation == "system":
|
|
cmd.append(f"--{installation}")
|
|
else:
|
|
cmd.append(f"--installation={installation}")
|
|
output = subprocess.run(
|
|
cmd,
|
|
text=True,
|
|
capture_output=True,
|
|
).stdout
|
|
lines = output.strip().split("\n")
|
|
remote_list = []
|
|
if lines[0] != "":
|
|
for line in lines:
|
|
line = line.split("\t")
|
|
remote_list.append(Remote(name=line[0], title=line[1], disabled=(len(line) == 3) and "disabled" in line[2]))
|
|
this.remotes[installation] = remote_list
|
|
|
|
# Masks
|
|
cmd = [
|
|
"flatpak-spawn",
|
|
"--host",
|
|
"flatpak",
|
|
"mask",
|
|
]
|
|
if installation == "user" or installation == "system":
|
|
cmd.append(f"--{installation}")
|
|
else:
|
|
cmd.append(f"--installation={installation}")
|
|
output = subprocess.run(
|
|
cmd,
|
|
text=True,
|
|
capture_output=True,
|
|
).stdout
|
|
lines = output.strip().replace(" ", "").split("\n")
|
|
if lines[0] != "":
|
|
this.masks[installation] = lines
|
|
|
|
# Pins
|
|
cmd = [
|
|
"flatpak-spawn",
|
|
"--host",
|
|
"flatpak",
|
|
"pin",
|
|
]
|
|
if installation == "user" or installation == "system":
|
|
cmd.append(f"--{installation}")
|
|
else:
|
|
cmd.append(f"--installation={installation}")
|
|
output = subprocess.run(
|
|
cmd,
|
|
text=True,
|
|
capture_output=True,
|
|
).stdout
|
|
lines = output.strip().replace(" ", "").split("\n")
|
|
if lines[0] != "":
|
|
this.pins[installation] = lines
|
|
|
|
try:
|
|
# Installations
|
|
# Get all config files for any extra installations
|
|
custom_install_config_path = "/run/host/etc/flatpak/installations.d"
|
|
if os.path.exists(custom_install_config_path):
|
|
for file in os.listdir(custom_install_config_path):
|
|
with open(f"{custom_install_config_path}/{file}", "r") as f:
|
|
for line in f:
|
|
if line.startswith("[Installation"):
|
|
# Get specifically the installation name itself
|
|
this.installations.append(line.replace('[Installation "', "").replace('"]', "").strip())
|
|
|
|
this.installations.append("user")
|
|
this.installations.append("system")
|
|
for i in this.installations:
|
|
remote_info(i)
|
|
|
|
# Packages
|
|
output = subprocess.run(
|
|
["flatpak-spawn", "--host", "flatpak", "list", "--columns=name,application,version,branch,arch,origin,installation,ref,size,options"],
|
|
text=True,
|
|
check=True,
|
|
capture_output=True,
|
|
).stdout
|
|
lines = output.strip().split("\n")
|
|
for i in lines:
|
|
package = Flatpak(i.split("\t"))
|
|
this.flatpaks.append(package)
|
|
this.id_to_flatpak[package.info["id"]] = package
|
|
this.ref_to_flatpak[package.info["ref"]] = package
|
|
|
|
# Dependent Runtimes
|
|
output = subprocess.run(
|
|
["flatpak-spawn", "--host", "flatpak", "list", "--columns=runtime,ref"],
|
|
text=True,
|
|
check=True,
|
|
capture_output=True,
|
|
).stdout
|
|
lines = output.split("\n")
|
|
for index, line in enumerate(lines):
|
|
split_line = line.split("\t")
|
|
if len(split_line) < 2 or split_line[0] == "":
|
|
continue
|
|
|
|
package = this.flatpaks[index]
|
|
if package.is_runtime:
|
|
continue
|
|
|
|
runtime = split_line[0]
|
|
package.dependent_runtime = this.ref_to_flatpak[runtime]
|
|
if not runtime in this.dependent_runtime_refs:
|
|
this.dependent_runtime_refs.append(runtime)
|
|
|
|
# store runtimes in sorted order
|
|
this.dependent_runtime_refs = sorted(this.dependent_runtime_refs)
|
|
|
|
except subprocess.CalledProcessError as cpe:
|
|
this.main_window.toast_overlay.add_toast(ErrorToast(_("Could not load packages"), cpe.stderr).toast)
|
|
except Exception as e:
|
|
this.main_window.toast_overlay.add_toast(ErrorToast(_("Could not load packages"), str(e)).toast)
|
|
|
|
Gio.Task.new(None, None, callback).run_in_thread(thread)
|