gctools/status_client/path_info.py

from datetime import datetime
import os
import zipfile
from pathlib import Path
from os import path
import psutil


class PathInfo:
    root_dir = ""

    ignore = []
    file_list = []
    opened_files = {}

    def ignore_list(self):
        gitignore = self.root_dir + "\\.gitignore"
        if not path.exists(gitignore):
            pass
        with open(gitignore, "r") as f:
            for line in f.readlines():
                line = line.strip().replace("/", "\\").lower()
                if line[:1] == "*":
                    if line[-1] == "*":
                        line = line[1:-1]
                    else:
                        line = line[1:] + "\\"
                else:
                    line = "\\" + line + "\\"
                self.ignore.append(line)

    def ignored(self, filename):
        rel_filename = "\\" + str(filename).replace(self.root_dir, "").lower() + "\\"

        for e in self.ignore:
            if e in rel_filename:
                return True
        return False

    def check_dir(self, current_dir):
        if self.root_dir == "":
            self.root_dir = current_dir
            self.opened_files = self.process_handles()
            current_dir = Path(current_dir)

        for entry in current_dir.glob("*"):
            if entry.is_dir():
                self.check_dir(entry)
            elif not self.ignored(entry):
                self.file_list.append(self.file_info(entry))

    def file_info(self, filename: Path):
        st = filename.stat()
        readable = "J" if os.access(filename, os.R_OK) else "N"
        writable = "J" if os.access(filename, os.W_OK) else "N"
        handle = self.opened_files.get(str(filename), "")
        blocked = "J" if handle != "" else "N"
        file_info = [
            str(filename),
            str(st.st_size),
            datetime.fromtimestamp(st.st_ctime).isoformat(timespec="seconds"),
            datetime.fromtimestamp(st.st_mtime).isoformat(timespec="seconds"),
            readable,
            writable,
            blocked,
            handle,
        ]
        return file_info

    def write_logfile(self, logfile):
        with open(logfile, "w") as fwh:
            infos = [";".join(line) for line in self.file_list]
            fwh.write("name;size;ctime;mtime;read;write;blocked;process\n")
            fwh.write("\n".join(infos))

    def zip_to_file(self, zip_file):
        with zipfile.ZipFile(zip_file, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as zip:
            for e in self.backup_list:
                zip.write(e)

    def process_handles(self):
        files = {}
        for proc in psutil.process_iter():
            try:
                for item in proc.open_files():
                    files[item.path] = proc.name()
            except Exception:
                pass
        return files


if __name__ == "__main__":
    ti = PathInfo()
    ti.check_dir("C:\\GlobalCube")
    ti.write_logfile("C:\\GlobalCube\\Tasks\\logs\\path_info.csv")
    # print(backup_list[:10])
    # ti.zip_to_file('C:\\GAPS_Autosys\\Test.zip')