A find and replace script written in Python. (Works with Unix Systems only)

find_and_replace.py

#!/usr/bin/env python3
from argparse import ArgumentParser
from multiprocessing import cpu_count
from os import getcwd, listdir, path, system
from queue import Queue
from threading import Thread
from typing import List

_FILEPATHS = Queue(maxsize=cpu_count() - 1)


def find_and_replace(find: str, replace: str, directories: List[str], ignore=None, recursive=False):
    """
    Parameters
    ----------
    find: str
        The string that will be replaced.

    replace: str
        The replacement string.

    directories: List[str]
        A list of directory paths that contain the target files.
        (default is [{cwd}])

    ignore: List[str], optional
        A list of directory paths to ignore.
        (default is None)

    recursive: bool, optional
        A flag that indicates if sub-directories should be recursively searched.
        (default is False)
    """

    n_workers = cpu_count() - 1
    workers = list()

    # Starting worker threads
    for _ in range(n_workers):
        thread = Thread(target=_worker, args=(find, replace))
        thread.start()
        workers.append(thread)

    # Queuing filepaths
    for directory in directories:
        for filepath in _filepath_generator(directory, ignore, recursive):
            _FILEPATHS.put(filepath)

    # Sending stop commands
    for _ in range(n_workers):
        _FILEPATHS.put(None)

    # Joining threads
    for thread in workers:
        thread.join()


def _worker(find: str, replace: str) -> None:
    while True:
        filepath = _FILEPATHS.get()
        if filepath is None:
            break
        if not _is_text_file(filepath):
            continue
        text = ""
        try:
            file_io = open(filepath, "r")
            text = file_io.read()
        except UnicodeDecodeError:
            text = str(text.encode("utf-8").strip())
        finally:
            file_io.close()
        if text.find(find) > -1:
            text = text.replace(find, replace)
            with open(filepath, "w") as file_io:
                file_io.write(text)
        _FILEPATHS.task_done()


def _is_text_file(filepath: str) -> bool:
    file_info = system(f"file {filepath}")
    return file_info.find("text") > -1


def _filepath_generator(directory: str, ignore: List[str], recursive: bool) -> str:
    for item_name in listdir(directory):
        if item_name in ignore:
            continue
        item_path = path.join(directory, item_name)
        if path.isdir(item_path) and recursive:
            yield from _filepath_generator(item_path, ignore, True)
        elif path.isfile(item_path):
            yield item_path
        else:
            print(f"[WARN] Unknown item found: {item_path}")


def __parse_arguments() -> 'args':
    parser = ArgumentParser(description="A CLI find & replace tool")

    help_str = "The string that will be replaced"
    parser.add_argument("search", type=str, help=help_str)

    help_str = "The replacement string"
    parser.add_argument("replace", type=str, help=help_str)

    help_str = "A flag that indicates if sub-directories should be recursively searched"
    parser.add_argument("-r", "--recursive",
                        action="store_true", help=help_str)

    help_str = "The path of directories that contain the target files"
    parser.add_argument("-d", "--directories", required=False, type=str, nargs="+", help=help_str)

    help_str = "The name of directories and files to ignore"
    parser.add_argument("-i", "--ignore", required=False, type=str, nargs="+", help=help_str)

    return parser.parse_args()


def _find_and_replace():
    args = __parse_arguments()
    find_and_replace(
        args.search,
        args.replace,
        (args.directories or [getcwd()]),
        (args.ignore or [".git"]),
        (args.recursive or False)
    )
    exit(0)


if __name__ == "__main__":
    _find_and_replace()