"""Abstract parent class of guds modules"""

[0] Fix | Delete

from abc import abstractmethod

[1] Fix | Delete

import logging

[2] Fix | Delete

from os import walk

[3] Fix | Delete

from pathlib import Path

[4] Fix | Delete

import shlex

[5] Fix | Delete

import shutil

[6] Fix | Delete

import subprocess

[7] Fix | Delete

[8] Fix | Delete

[9] Fix | Delete

class ModuleBase:

[10] Fix | Delete

def __init__(self, dry_run: bool, logger: logging.Logger):

[11] Fix | Delete

self.dry_run = dry_run

[12] Fix | Delete

self.logger = logger

[13] Fix | Delete

[14] Fix | Delete

@abstractmethod

[15] Fix | Delete

def run_module(self, homedir: str) -> dict:

[16] Fix | Delete

raise NotImplementedError

[17] Fix | Delete

[18] Fix | Delete

@classmethod

[19] Fix | Delete

def calc_bytes(cls, size_check: Path) -> int:

[20] Fix | Delete

if size_check.is_file():

[21] Fix | Delete

return size_check.stat().st_size

[22] Fix | Delete

total_size = 0

[23] Fix | Delete

for root, _, files in walk(size_check):

[24] Fix | Delete

for name in files:

[25] Fix | Delete

filename = Path(root, name)

[26] Fix | Delete

try:

[27] Fix | Delete

total_size += filename.stat().st_size

[28] Fix | Delete

except OSError:

[29] Fix | Delete

pass

[30] Fix | Delete

return total_size

[31] Fix | Delete

[32] Fix | Delete

@classmethod

[33] Fix | Delete

def get_size(cls, size_check: Path) -> str:

[34] Fix | Delete

"""Find the size of provided directory or file"""

[35] Fix | Delete

total_size = cls.calc_bytes(size_check)

[36] Fix | Delete

for suffix in ['B', 'KiB', 'MiB', 'GiB', 'TiB']:

[37] Fix | Delete

if total_size < 1024.0:

[38] Fix | Delete

return f"{total_size:3.1f} {suffix}"

[39] Fix | Delete

total_size /= 1024.0

[40] Fix | Delete

return f"{total_size:3.1f} PiB"

[41] Fix | Delete

[42] Fix | Delete

def find(

[43] Fix | Delete

self,

[44] Fix | Delete

homedir: str,

[45] Fix | Delete

/,

[46] Fix | Delete

*,

[47] Fix | Delete

maxdepth=None,

[48] Fix | Delete

regex=None,

[49] Fix | Delete

prune_mail=True,

[50] Fix | Delete

**kwargs,

[51] Fix | Delete

) -> list[Path]:

[52] Fix | Delete

"""returns each line of output from the command as an item in a list"""

[53] Fix | Delete

cmd = ['find']

[54] Fix | Delete

if regex is not None:

[55] Fix | Delete

cmd.append('-O3')

[56] Fix | Delete

cmd.append(homedir)

[57] Fix | Delete

# ensure maxdepth is set first, if specified

[58] Fix | Delete

if maxdepth is not None:

[59] Fix | Delete

cmd.extend(['-maxdepth', str(maxdepth)])

[60] Fix | Delete

elif prune_mail:

[61] Fix | Delete

# optimization: prune and exclude ~/mail

[62] Fix | Delete

cmd.extend(['-not', '(', '-path', f"{homedir}/mail", '-prune', ')'])

[63] Fix | Delete

# regextype must be set before regex

[64] Fix | Delete

if regex is not None:

[65] Fix | Delete

cmd.extend(['-regextype', 'posix-extended', '-regex', regex])

[66] Fix | Delete

for key, val in kwargs.items():

[67] Fix | Delete

cmd.extend([f'-{key}', str(val)])

[68] Fix | Delete

cmd.append('-print0')

[69] Fix | Delete

self.logger.debug(shlex.join(cmd))

[70] Fix | Delete

ret = subprocess.run(

[71] Fix | Delete

cmd, stdout=subprocess.PIPE, encoding='utf-8', check=False

[72] Fix | Delete

)

[73] Fix | Delete

return [Path(x) for x in ret.stdout.split('\0') if x]

[74] Fix | Delete

[75] Fix | Delete

def delete_items(self, backuplist: list[Path]):

[76] Fix | Delete

for backup in backuplist:

[77] Fix | Delete

self.logger.info("Removing %s (%s)", backup, self.get_size(backup))

[78] Fix | Delete

if self.dry_run:

[79] Fix | Delete

continue

[80] Fix | Delete

if backup.is_file():

[81] Fix | Delete

backup.unlink()

[82] Fix | Delete

elif backup.is_dir():

[83] Fix | Delete

shutil.rmtree(str(backup))

[84] Fix | Delete

[85] Fix | Delete