import re import subprocess from subprocess import Popen, PIPE from selfprivacy_api.utils.huey import huey from selfprivacy_api.jobs import JobStatus, Jobs, Job COMPLETED_WITH_ERROR = ( "We are sorry, сompleted with an error, report it to support chat" ) RESULT_WAS_NOT_FOUND_ERROR = "We are sorry, garbage collection result was not found, something went wrong, report it to support chat :(" CLEAR_COMPLETED = "Cleaning completed." def run_nix_store_print_dead() -> PIPE: subprocess.run( ["nix-env", "-p", "/nix/var/nix/profiles/system", "--delete-generations old"], check=False, ) return subprocess.check_output(["nix-store", "--gc", "--print-dead"]).decode( "utf-8" ) def run_nix_collect_garbage() -> subprocess.Popen: return subprocess.Popen( ["nix-store", "--gc"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT ).stdout def parse_line(line): """ We parse the string for the presence of a final line, with the final amount of space cleared. Simply put, we're just looking for a similar string: "1537 store paths deleted, 339.84 MiB freed". """ pattern = re.compile(r"[+-]?\d+\.\d+ \w+(?= freed)") match = re.search(pattern, line) if match is None: return ( JobStatus.FINISHED, 100, COMPLETED_WITH_ERROR, RESULT_WAS_NOT_FOUND_ERROR, ) else: return ( JobStatus.FINISHED, 100, CLEAR_COMPLETED, f"{match.group(0)} have been cleared", ) def process_stream(job, stream, total_dead_packages): completed_packages = 0 prev_progress = 0 for line in stream: line = line.decode("utf-8") if "deleting '/nix/store/" in line: completed_packages += 1 percent = int((completed_packages / total_dead_packages) * 100) if percent - prev_progress >= 5: Jobs.update( job=Job, status=JobStatus.RUNNING, progress=percent, status_text="Cleaning...", ) prev_progress = percent elif "store paths deleted," in line: status = parse_line(line) Jobs.update( job=Job, status=status[0], progress=status[1], status_text=status[2], result=status[3], ) def get_dead_packages(output): dead = len(re.findall("/nix/store/", output)) percent = 0 if dead != 0: percent = 100 / dead return dead, percent @huey.task() def calculate_and_clear_dead_packages(job: Jobs): Jobs.update( job=Job, status=JobStatus.RUNNING, progress=0, status_text="Calculate the number of dead packages...", ) dead_packages, package_equal_to_percent = get_dead_packages( run_nix_store_print_dead() ) if dead_packages == 0: Jobs.update( job=Job, status=JobStatus.FINISHED, progress=100, status_text="Nothing to clear", result="System is clear", ) return Jobs.update( job=Job, status=JobStatus.RUNNING, progress=0, status_text=f"Found {dead_packages} packages to remove!", ) process_stream(job, run_nix_collect_garbage(), package_equal_to_percent) def start_nix_collect_garbage() -> Job: job = Jobs.add( type_id="maintenance.collect_nix_garbage", name="Collect garbage", description="Cleaning up unused packages", ) calculate_and_clear_dead_packages(job=job) return job