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
|
from __future__ import annotations
from PyQt6.QtCore import QThread, pyqtSignal
import subprocess
from pathlib import Path
class GitWorker(QThread):
output = pyqtSignal(str)
error = pyqtSignal(str)
finished = pyqtSignal(bool) # True = success
def __init__(self, repo_path: Path, command: list[str], parent=None):
super().__init__(parent)
self._repo_path = repo_path
self._command = command
def run(self):
try:
result = subprocess.run(
self._command,
cwd=self._repo_path,
capture_output=True,
text=True,
)
if result.stdout:
self.output.emit(result.stdout)
if result.returncode != 0:
self.error.emit(result.stderr or f"Exit code {result.returncode}")
self.finished.emit(False)
else:
self.finished.emit(True)
except Exception as e:
self.error.emit(str(e))
self.finished.emit(False)
@staticmethod
def pull(repo_path: Path, parent=None) -> "GitWorker":
return GitWorker(repo_path, ["git", "pull"], parent)
@staticmethod
def push_master(repo_path: Path, parent=None) -> "GitWorker":
return GitWorker(repo_path, ["git", "push", "origin", "master"], parent)
@staticmethod
def push_production(repo_path: Path, parent=None) -> "GitWorker":
return GitWorker(repo_path, ["git", "push", "origin", "production"], parent)
@staticmethod
def remove_article(repo_path: Path, rel_path: str, slug: str, lang: str, parent=None) -> "GitWorker":
class _RmWorker(QThread):
output = pyqtSignal(str)
error = pyqtSignal(str)
finished = pyqtSignal(bool)
def run(self_inner):
for cmd in [
["git", "rm", "-r", rel_path],
["git", "commit", "-m", f"remove: {slug} ({lang})"],
]:
result = subprocess.run(cmd, cwd=repo_path, capture_output=True, text=True)
if result.stdout:
self_inner.output.emit(result.stdout)
if result.returncode != 0:
self_inner.error.emit(result.stderr or f"Exit code {result.returncode}")
self_inner.finished.emit(False)
return
self_inner.finished.emit(True)
return _RmWorker(parent)
@staticmethod
def restore_article(repo_path: Path, commit_hash: str, rel_path: str, parent=None) -> "GitWorker":
return GitWorker(repo_path, ["git", "checkout", commit_hash, "--", rel_path], parent)
|