I have made a program that does what the title explains. It also hashes the file to check for changes. I will be doing this for a project to keep track of files that has been changed when doing different things in that certain filetree. I need this to be done as an sqlite/python app to make information easily available regarding the files of interest. This will later be used in an TKinter GUI app and the Parent column contains the rowid for its parent folder to make it work with TKinter. I will make it able to use the Tags column for searching within the app as i progress.
I have been programming for a few years, but not much, I feel a lack in confidence and therefore I am throwing my code out here to get some review. I need to get better at OOP, efficiency and my logical sense needs an update from some bright minds. Please see the full code below, it is two files: filesystemScanner.py:
import os
import hashlib
from pathlib import Path
from mydatabasemanagerfile import MyDatabaseManager
path = "C:\\Users\Skaft\\Documents\\GitHub"
isFolder = True
fullList = []
currhash = ""
db_name = "parenttest.db"
table_name = "Windows10"
my_current_db = MyDatabaseManager(db_name, table_name)
#####################################################################
# copied from https://stackoverflow.com/questions/24937495/
# how-can-i-calculate-a-hash-for-a-filesystem-directory-using-python
#####################################################################
def md5_update_from_dir(directory, my_hash):
assert Path(directory).is_dir()
for hash_path in sorted(Path(directory).iterdir(), key=lambda p: str(p).lower()):
my_hash.update(hash_path.name.encode())
if hash_path.is_file():
with open(hash_path, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
my_hash.update(chunk)
elif hash_path.is_dir():
my_hash = md5_update_from_dir(hash_path, my_hash)
return my_hash
def md5_dir(directory):
return md5_update_from_dir(directory, hashlib.md5()).hexdigest()
####################################################
# End of copy
####################################################
def file_hasher(hashing_file):
try:
if not isFolder:
with open(hashing_file, "rb") as current:
readfile = current.read()
current_hash = hashlib.md5(readfile).hexdigest()
return current_hash
except PermissionError:
current_hash = "Could not hash due to permission"
return current_hash
for (dirpath, dirnames, filenames) in os.walk(path):
isFolder = True
full_dirpath = dirpath + "\\"
fullList.append([full_dirpath, md5_dir(full_dirpath), isFolder])
for file in filenames:
file_path = dirpath + "\\" + file
isFolder = False
fullList.append([file_path, file_hasher(file_path), isFolder])
my_current_db.initialize_db()
for file_info in fullList:
my_current_db.my_updater(file_info[0], file_info[1], file_info[2])
my_current_db.inserting_parentID()
my_current_db.inserting_suffix()
my_current_db.close_db()
And mydatabasemanagerfile.py
import sqlite3
import datetime
import pathlib
class MyDatabaseManager:
def __init__(self, db_name, table_name):
self.db_name = db_name
self.table_name = table_name
self.conn = sqlite3.connect(self.db_name)
self.cur = self.conn.cursor()
def initialize_db(self):
"""Function to open database, create if not exists"""
print("Opened database successfully")
# drop_if_exist = "DROP TABLE IF EXISTS {}".format(self.table_name)
# self.cur.execute(drop_if_exist)
create_table = '''CREATE TABLE IF NOT EXISTS {}
("Path" TEXT NOT NULL,
"Information" TEXT,
"Tags" TEXT,
"Parent" INTEGER,
"isFolder" INTEGER,
"fileExt" TEXT,
"MD5" TEXT,
"Updated" TEXT,
"See also" TEXT
)'''.format(self.table_name)
self.cur.execute(create_table)
print("Table created successfully")
def my_updater(self, path, md5, isFolder):
query_path = '''SELECT rowid, Path FROM {} WHERE Path = "{}"'''.format(self.table_name, path)
if self.conn.execute(query_path).fetchone() is None: # Test if returns None.
result_path = "Please add me" # Placeholder to specify that it needs to be inserted, it does not exist
result_hash = "Please add me"
else:
result_path = self.conn.execute(query_path).fetchone()[1]
rowid = self.conn.execute(query_path).fetchone()[0]
query_hash = '''SELECT MD5 FROM {} WHERE rowid = "{}"'''.format(self.table_name, rowid)
result_hash = self.conn.execute(query_hash).fetchone()[0]
current_time = datetime.datetime.now()
if path == result_path and md5 != result_hash:
query = '''UPDATE {} SET MD5=?, Updated=? WHERE rowid = "{}"'''.format(self.table_name, rowid)
self.cur.execute(query, (md5, current_time))
elif path != result_path:
query = '''INSERT INTO {} (Path, MD5, isFolder, Updated) VALUES (?, ?, ?, ?)'''.format(self.table_name)
self.cur.execute(query, (path, md5, isFolder, current_time))
else:
#print("Path: ", path, "\n", "result_path: ", result_path, "\n", "md5 : ", md5, "\n", "result_hash: ",
# result_hash, "\n")
pass
self.conn.commit()
def inserting_parentID(self):
query = '''SELECT rowid, Path FROM {}'''.format(self.table_name)
all_paths = self.cur.execute(query).fetchall()
for rowid, path in all_paths:
parent_path_split = path.split("\\")
if rowid == 1:
continue
if path[-1] == "\\":
parent_path_joined = "\\".join(parent_path_split[:-2]) + "\\"
parent_path_joinedads = "\\".join(parent_path_split[:-2]) + "\\"
else:
parent_path_joined = "\\".join(parent_path_split[:-1]) + "\\"
parent_query = '''SELECT rowid FROM {} WHERE Path = "{}"'''.format(self.table_name, parent_path_joined)
parent_paths = self.cur.execute(parent_query).fetchone()
update_parentID = '''UPDATE {} SET Parent=? WHERE rowid = "{}"'''.format(self.table_name, rowid)
self.cur.execute(update_parentID, (parent_paths))
self.conn.commit()
def inserting_suffix(self):
query = '''SELECT rowid, Path FROM {}'''.format(self.table_name)
all_paths = self.cur.execute(query).fetchall()
for rowid, path in all_paths:
ext = pathlib.Path(path).suffix
if ext == "":
continue
update_ext_query = '''UPDATE {} SET fileExt=? WHERE rowid = "{}"'''.format(self.table_name, rowid)
self.cur.execute(update_ext_query, (ext,))
self.conn.commit()
def close_db(self):
self.conn.close()
Any feedback is appreciated to get a more professional and standardized result. I find this a good way to learn, to view and discuss improvement on my own code.