~gpanders/pushbroom

93b0d2d2a44a729c86a563c169d747b94d6f86df — Greg Anders 2 years ago 54fbbe3
Mild refactoring to improve pylint score
2 files changed, 60 insertions(+), 43 deletions(-)

M src/pushbroom/console.py
M src/pushbroom/sweep.py
M src/pushbroom/console.py => src/pushbroom/console.py +40 -24
@@ 11,6 11,37 @@ from pushbroom import sweep, __version__

def run():
    """Run pushbroom"""
    args = parse_args()

    setup_logging(args)

    config = read_config(args)

    logging.info("Starting pushbroom")
    for section in config.sections():
        path = config.get(section, "path")
        fullpath = os.path.abspath(os.path.expanduser(path))
        if not os.path.isdir(fullpath):
            logging.error("No such directory: %s", fullpath)
        else:
            num_days = config.getint(section, "numdays")
            trash = config.get(section, "trash", fallback=None)
            ignore = config.get(section, "ignore", fallback="").split(",")
            ignore_re = re.compile("|".join([fnmatch.translate(x) for x in ignore]))
            match = config.get(section, "match", fallback="*").split(",")
            match_re = re.compile("|".join([fnmatch.translate(x) for x in match]))
            dry_run = args.dry_run

            if trash:
                trash = os.path.abspath(os.path.expanduser(trash))
                if not os.path.isdir(trash):
                    logging.error("No such directory %s", trash)

            sweep(section, fullpath, num_days, ignore_re, match_re, trash, dry_run)


def parse_args():
    """Parse command line arguments"""
    parser = argparse.ArgumentParser(description="Clean up your filesystem.")
    parser.add_argument("-c", "--config", type=str, help="path to config file")
    parser.add_argument("-v", "--verbose", action="store_true", help="verbose output")


@@ 27,9 58,11 @@ def run():
        help="show what would be done without actually doing anything",
    )

    args = parser.parse_args()
    return parser.parse_args()


    # Set up logging
def setup_logging(args):
    """Set up logging"""
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)
    ch = logging.StreamHandler()


@@ 55,6 88,9 @@ def run():

    logger.addHandler(ch)


def read_config(args):
    """Find and read configuration file"""
    if not args.config:
        # Look under XDG_CONFIG_HOME first, then look for ~/.pushbroomrc
        args.config = os.path.join(


@@ 70,27 106,7 @@ def run():
        with open(args.config, "r") as f:
            config.read_file(f)
    except FileNotFoundError:
        logging.error("Configuration file {} not found".format(args.config))
        logging.error("Configuration file %s not found", args.config)
        sys.exit(1)

    logging.info("Starting pushbroom")
    for section in config.sections():
        path = config.get(section, "path")
        fullpath = os.path.abspath(os.path.expanduser(path))
        if not os.path.isdir(fullpath):
            logging.error("No such directory: %s", fullpath)
        else:
            num_days = config.getint(section, "numdays")
            trash = config.get(section, "trash", fallback=None)
            ignore = config.get(section, "ignore", fallback="").split(",")
            ignore_re = re.compile("|".join([fnmatch.translate(x) for x in ignore]))
            match = config.get(section, "match", fallback="*").split(",")
            match_re = re.compile("|".join([fnmatch.translate(x) for x in match]))
            dry_run = args.dry_run

            if trash:
                trash = os.path.abspath(os.path.expanduser(trash))
                if not os.path.isdir(trash):
                    logging.error("No such directory %s", trash)

            sweep(section, fullpath, num_days, ignore_re, match_re, trash, dry_run)
    return config

M src/pushbroom/sweep.py => src/pushbroom/sweep.py +20 -19
@@ 7,35 7,36 @@ SECONDS_PER_DAY = 24 * 60 * 60


def sweep(name, path, num_days, ignore, match, trash, dry_run):
    # pylint: disable = too-many-arguments
    """Remove old files from a directory

    :path: Path to remove files from
    :name:     Name of the section being cleaned
    :path:     Path to remove files from
    :num_days: Remove files older than this many days
    :ignore: Regular expression pattern of paths to ignore
    :match: Regular expression pattern of paths to remove
    :trash: If set, move files to this directory instead of deleting them
    :dry_run: Only show what would happen without actually doing anything
    :ignore:   Regular expression pattern of paths to ignore
    :match:    Regular expression pattern of paths to remove
    :trash:    If set, move files to this directory instead of deleting them
    :dry_run:  Only show what would happen without actually doing anything

    """
    logging.info("Sweeping %s", name)
    now = time.time()
    num_seconds = num_days * SECONDS_PER_DAY
    thresh = now - num_seconds
    thresh = time.time() - num_seconds
    for root, dirs, files in os.walk(path):
        dirs[:] = [d for d in dirs if re.match(match, d) and not re.match(ignore, d)]

        files = [f for f in files if re.match(match, f) and not re.match(ignore, f)]
        for file in files:
            fpath = os.path.join(root, file)
            try:
                if os.stat(fpath).st_mtime < thresh:
                    if trash:
                        logging.info("Moving %s to %s", fpath, trash)
                        if not dry_run:
                            os.rename(fpath, os.path.join(trash, file))
                    else:
                        logging.info("Deleting %s", fpath)
                        if not dry_run:
                            os.remove(fpath)
            except FileNotFoundError:
                pass
            if not os.path.exists(fpath):
                continue

            if os.stat(fpath).st_mtime < thresh:
                if trash:
                    logging.info("Moving %s to %s", fpath, trash)
                    if not dry_run:
                        os.rename(fpath, os.path.join(trash, file))
                else:
                    logging.info("Deleting %s", fpath)
                    if not dry_run:
                        os.remove(fpath)