1 files changed, 150 insertions, 0 deletions
diff --git a/gamechestcli/gamechest/runners/remove.py b/gamechestcli/gamechest/runners/remove.py
new file mode 100644
index 0000000..99c4247
--- /dev/null
+++ b/gamechestcli/gamechest/runners/remove.py
@@ -0,0 +1,150 @@
+import os
+import re
+import subprocess
+import threading
+
+import humanfriendly
+
+from ..structures import Progress
+from .runnerbase import RunnerBase, neutral_locale_variables
+
+
+class Remove(RunnerBase):
+
+    _progress_re = re.compile(r'^\s*(\S+)\s+\[([^]]+)/s\]\s+ETA\s+(\S+)\s*$')
+
+    def __init__(self, path):
+        self.last_progress = Progress(linemode=True)
+        self.path = path
+        self.rm_proc = None
+        self.pv_proc = None
+        self.pipe = os.pipe()
+        self.read_fd = open(self.pipe[0], 'r', encoding='utf8')
+        #self.read_fd = open(self.pipe[0], 'rb', buffering=0)
+        self._proc_started = False
+        self._filescount = 0
+        self._counting_quit_event = threading.Event()
+        self._counting_thread = threading.Thread(
+            target=self._counting_worker,
+            args=(self._counting_quit_event, path),
+        )
+        self._counting_thread.start()
+
+    def _start_proc(self):
+        common_parameters = dict(
+            encoding='utf8',
+            env={**os.environ,
+                 **neutral_locale_variables,
+                 },
+        )
+        self.rm_proc = subprocess.Popen(
+            [
+                'rm',
+                '--verbose',
+                '--recursive',
+                '--one-file-system',
+                '--preserve-root=all',
+                '--interactive=never',
+                '--',
+                self.path,
+            ],
+        #    [  # only for testing
+        #        'testtools/fake-rm',
+        #        self.path,
+        #    ],
+            stdout=subprocess.PIPE,
+            **common_parameters,
+        )
+        self.pv_proc = subprocess.Popen(
+            [
+                'pv',
+                '--force',
+                '--size', str(self._filescount),
+                '--format', '%b %a %e',
+                '--line-mode',
+            ],
+            stdin=self.rm_proc.stdout,
+            stdout=subprocess.DEVNULL,
+            stderr=self.pipe[1],
+            **common_parameters,
+        )
+        # close child's pipe fd on parent's side or selectors will hang if
+        # process closes.
+        os.close(self.pipe[1])
+        self._proc_started = True
+
+    def _counting_worker(self, event, path):
+        # as counting files for a deep directory can take some time, make it
+        # interruptible with a thread and event for cancelling.
+        # using functools.reduce(x+1+len(y[2]), os.walk) would have been much
+        # cleaner, but the code gets harder to read with the addition of the
+        # event checking. Fallback to accumalator and iterations.
+        total = 0
+        for _, _, files in os.walk(path):
+            total += 1 + len(files)
+            if event.is_set():
+                break
+        # set filescount with calculated total (even then interrupted, as it
+        # would be a better estimation than nothing).
+        if os.path.isfile(path):
+            self._filescount = 1
+        else:
+            self._filescount = total
+        # start next processes (if event was not set)
+        if not event.is_set():
+            self._start_proc()
+
+    def get_read_fd(self):
+        return self.read_fd
+
+    def progress_read(self):
+        if not self._proc_started:
+            return self.last_progress
+        line = self.read_fd.readline()
+        #line = self.read_fd.readline().decode('utf8')
+        if match := self._progress_re.search(line):
+            written_bytes = humanfriendly.parse_size(match.group(1))
+            self.last_progress = Progress(
+                linemode=True,
+                nbbytes=written_bytes,
+                percent=int(100 * written_bytes / self._filescount),
+                speed=humanfriendly.parse_size(match.group(2)),
+                eta=match.group(3),
+            )
+        return self.last_progress
+
+    def terminate(self):
+        self._counting_quit_event.set()
+        if self._proc_started:
+            for proc in (self.rm_proc, self.pv_proc):
+                proc.terminate()
+
+    def poll(self):
+        if not self._proc_started:
+            return None
+        return self.pv_proc.poll()
+
+    def close(self):
+        self._counting_thread.join()
+        if self._proc_started:
+            for proc in (self.rm_proc, self.pv_proc):
+                proc.wait()
+        self.read_fd.close()
+
+
+if __name__ == '__main__':
+    import sys
+    import contextlib
+    import time
+    import selectors
+
+    with contextlib.ExitStack() as stack:
+        stack.enter_context(contextlib.suppress(KeyboardInterrupt))
+        runner = stack.enter_context(Remove(sys.argv[1]))
+        selector = stack.enter_context(selectors.DefaultSelector())
+        selector.register(runner.get_read_fd(), selectors.EVENT_READ)
+        while (rc := runner.poll()) is None:
+            if selector.select():
+                print(runner.progress_read())
+        print('ended with code:', rc)
+    print('test main ended')