1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
|
import os
import shutil
from PyQt6.QtCore import QObject, QProcess, QProcessEnvironment, pyqtSignal
HELPER = os.environ.get("RAVEOS_CENTER_HELPER", "/usr/lib/raveos-center/raveos-center-helper")
def helper_elerheto():
return os.access(HELPER, os.X_OK) and shutil.which("pkexec") is not None
def root(*args):
return ["pkexec", HELPER, *args]
class Lepes:
def __init__(self, cimke, parancs, kotelezo=True):
self.cimke = cimke
self.parancs = parancs
self.kotelezo = kotelezo
class Futtato(QObject):
sor = pyqtSignal(str)
lepes_indul = pyqtSignal(str)
kesz = pyqtSignal(bool, list)
def __init__(self, szulo=None):
super().__init__(szulo)
self.lepesek = []
self.hibasak = []
self.folyamat = None
self.utolso_kod = None
self._maradek = b""
def fut(self):
return self.folyamat is not None
def indit(self, lepesek):
if self.fut():
return
self.lepesek = list(lepesek)
self.hibasak = []
self._kovetkezo()
def _kovetkezo(self):
if not self.lepesek:
self.folyamat = None
self.kesz.emit(not self.hibasak, self.hibasak)
return
lepes = self.lepesek.pop(0)
self.aktualis = lepes
self.lepes_indul.emit(lepes.cimke)
self.sor.emit(f":: {lepes.cimke}")
folyamat = QProcess(self)
folyamat.setProcessChannelMode(QProcess.ProcessChannelMode.MergedChannels)
kornyezet = QProcessEnvironment.systemEnvironment()
kornyezet.insert("LC_ALL", "C.UTF-8")
folyamat.setProcessEnvironment(kornyezet)
folyamat.readyReadStandardOutput.connect(self._olvas)
folyamat.finished.connect(self._vege)
folyamat.errorOccurred.connect(self._inditasi_hiba)
self._maradek = b""
self.folyamat = folyamat
folyamat.start(lepes.parancs[0], lepes.parancs[1:])
def _olvas(self):
if self.folyamat is None:
return
adat = self._maradek + bytes(self.folyamat.readAllStandardOutput())
*sorok, self._maradek = adat.replace(b"\r", b"\n").split(b"\n")
for s in sorok:
szoveg = s.decode("utf-8", errors="replace").rstrip()
if szoveg:
self.sor.emit(szoveg)
def _inditasi_hiba(self, hiba):
if hiba == QProcess.ProcessError.FailedToStart:
self.sor.emit(f"[HIBA] nem indítható: {self.aktualis.parancs[0]}")
self._lezar(False)
def _vege(self, kod, allapot):
self._olvas()
if self._maradek:
self.sor.emit(self._maradek.decode("utf-8", errors="replace").rstrip())
self._maradek = b""
rendben = allapot == QProcess.ExitStatus.NormalExit and kod == 0
if not rendben and kod in (126, 127) and self.aktualis.parancs[0] == "pkexec":
self.sor.emit("[MEGSZAKÍTVA] a hitelesítés elmaradt")
self._lezar(rendben, kod)
def _lezar(self, rendben, kod=None):
if self.folyamat is None:
return
self.utolso_kod = kod
self.folyamat.deleteLater()
self.folyamat = None
if rendben:
self.sor.emit(f"[OK] {self.aktualis.cimke}")
else:
self.sor.emit(f"[HIBA{f' {kod}' if kod is not None else ''}] {self.aktualis.cimke}")
self.hibasak.append(self.aktualis.cimke)
if self.aktualis.kotelezo:
self.lepesek = []
self.folyamat = None
self._kovetkezo()
|