Improved directory structure; updated readme
This commit is contained in:
parent
d4d5096f1f
commit
d99c5d2f23
11 changed files with 111 additions and 31 deletions
2
src/finch_filer/__init__.py
Normal file
2
src/finch_filer/__init__.py
Normal file
|
|
@ -0,0 +1,2 @@
|
|||
__version__ = "1.0.1"
|
||||
__author__ = "Gull"
|
||||
464
src/finch_filer/app.py
Normal file
464
src/finch_filer/app.py
Normal file
|
|
@ -0,0 +1,464 @@
|
|||
"""
|
||||
Frontend GUI that displays a list of files from a folder.
|
||||
|
||||
File actions (move, copy, or delete) can be set for generic file types.
|
||||
"""
|
||||
|
||||
__author__ = "Gull"
|
||||
|
||||
import add_root_to_sys_path
|
||||
add_root_to_sys_path.add_to_sys_path(__file__, "src")
|
||||
|
||||
import tkinter as tk
|
||||
import logging as log
|
||||
from finch_filer import file_manager as fm
|
||||
from finch_filer import utils as ut
|
||||
from tkinter import ttk
|
||||
from tkinter import messagebox as mb
|
||||
from tkinter import filedialog as fd
|
||||
|
||||
APP_NAME = "Finch Filer"
|
||||
LOG_PATH = "%/Gullbase/FinchFiler/app.log"
|
||||
RULES_PATH = "%/Gullbase/FinchFiler/file_rules_custom.json"
|
||||
|
||||
"""
|
||||
TODO:
|
||||
- Help dropdown on menu bar
|
||||
"""
|
||||
|
||||
def setup():
|
||||
"""Initializes the logging system, and a few other requirements."""
|
||||
from sys import stdout
|
||||
|
||||
log.basicConfig(**ut.get_log_config(LOG_PATH, log.DEBUG))
|
||||
log.getLogger().addHandler(log.StreamHandler(stdout)) # Console
|
||||
log.info("App started")
|
||||
|
||||
try:
|
||||
from ctypes import windll
|
||||
|
||||
myappid = "com.gullbase.finchfiler"
|
||||
windll.shell32.SetCurrentProcessExplicitAppUserModelID(myappid)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
def run():
|
||||
"""Sets up and runs the app."""
|
||||
setup()
|
||||
app = App()
|
||||
app.mainloop()
|
||||
|
||||
class App(tk.Tk):
|
||||
"""TKinter GUI and related methods."""
|
||||
def __init__(self, width=640, height=480):
|
||||
super().__init__()
|
||||
self.title(APP_NAME)
|
||||
self.option_add('*tearOff', tk.FALSE)
|
||||
self.iconbitmap(ut.parse_dir("#/icon.ico"))
|
||||
self.fm = fm.Manager(True) # Main file manager object
|
||||
self.log = list() # Running list of all logs
|
||||
self.temp = {"column": "", "reverse": False} # Temporary data
|
||||
self.current_mode = "all"
|
||||
self.custom_rules_path = RULES_PATH
|
||||
self.gui() # User interface
|
||||
self.load_config() # File rules configuration
|
||||
self.prep() # Ready state
|
||||
|
||||
# Centers the app on the screen
|
||||
screen_width = self.winfo_screenwidth()
|
||||
screen_height = self.winfo_screenheight()
|
||||
center_x = int(screen_width / 2 - width / 2)
|
||||
center_y = int(screen_height / 2 - height / 2)
|
||||
self.geometry(f"{width}x{height}+{center_x}+{center_y}")
|
||||
|
||||
def get_filemode_files(self, key, ignore_all=True):
|
||||
"""Returns a dict of files that match the given file mode key."""
|
||||
if key == "all": # Special key representing all file types
|
||||
return self.fm.filedata if not ignore_all else dict()
|
||||
|
||||
files = dict()
|
||||
for k, v in self.fm.filedata.items():
|
||||
if v["mode"] == key: files[k] = v
|
||||
|
||||
return files
|
||||
|
||||
def update_fileview(self):
|
||||
"""Updates the treeview widget in the app."""
|
||||
tree = self.v_tree.get()
|
||||
self.fm.update_file_data()
|
||||
log.info(f"Processed file view with {len(self.fm.filedata)} items")
|
||||
log.info(f"Show as tree: {str(tree)}")
|
||||
|
||||
while len(self.fileview.get_children()) > 0: # Clears all children
|
||||
self.fileview.delete(self.fileview.get_children()[-1])
|
||||
|
||||
filemodes = self.fm.filemodes
|
||||
if not tree: filemodes = {"all": {"name": "All Files"}} # Fallback
|
||||
for k, v in filemodes.items():
|
||||
files = self.get_filemode_files(k, tree)
|
||||
if len(files) > 0:
|
||||
id = self.fileview.insert("", tk.END, text=v["name"],
|
||||
open=True)
|
||||
|
||||
for i, (k, v) in enumerate(files.items()):
|
||||
values = (str(ut.format_date(v["time"])),
|
||||
str(ut.format_bytes(v["size"])))
|
||||
nid = self.fileview.insert(id, tk.END, text=k,
|
||||
values=values)
|
||||
self.fm.filedata[k]["_treeid"] = nid
|
||||
self.fm.filedata[k]["_order"] = i # Used for sorting
|
||||
|
||||
def sort_fileview_items(self, column="name"):
|
||||
"""Sorts treeview items by the given column type."""
|
||||
columns = ("name", "time", "size")
|
||||
c_index = columns.index(column) if column in columns else -1
|
||||
if c_index == -1: return
|
||||
|
||||
reverse = False
|
||||
if self.temp["column"] == column:
|
||||
self.temp["reverse"] = not self.temp["reverse"] # Toggles reverse
|
||||
reverse = self.temp["reverse"]
|
||||
|
||||
self.temp["column"] = column
|
||||
|
||||
files = self.fm.filedata
|
||||
temp = list() # Stores file data in a temporary list, for sorting
|
||||
for c1 in self.fileview.get_children():
|
||||
for c2 in self.fileview.get_children(c1):
|
||||
key = self.fileview.item(c2)["text"]
|
||||
temp.append((key, files[key]["time"], files[key]["size"]))
|
||||
temp.sort(key=lambda v: v[c_index], reverse=reverse)
|
||||
|
||||
for i, v in enumerate(temp):
|
||||
files[v[0]]["_order"] = i
|
||||
|
||||
for index, (k, _, _) in enumerate(temp):
|
||||
id = files[k]["_treeid"]
|
||||
index = files[k]["_order"]
|
||||
self.fileview.move(id, self.fileview.parent(id), index)
|
||||
|
||||
temp.clear() # Clears list after every loop
|
||||
|
||||
def load_config(self, filepath=""):
|
||||
"""Loads the file sorting rules from the given file path."""
|
||||
if filepath == "": filepath = self.custom_rules_path
|
||||
try:
|
||||
path = ut.parse_dir(filepath)
|
||||
except FileNotFoundError as error:
|
||||
log.error("Failed to get file path for loading config file!")
|
||||
print(error)
|
||||
else:
|
||||
try:
|
||||
data = ut.load_json_file(path)
|
||||
self.fm.setup_file_rules(data, True)
|
||||
self.prep()
|
||||
log.info(f"Loaded custom file rules: {path}")
|
||||
except Exception as error:
|
||||
log.error("Failed to load config file!")
|
||||
print(error)
|
||||
|
||||
def save_config(self, filepath=""):
|
||||
"""Saves the file sorting rules to the given file path."""
|
||||
if filepath == "": filepath = self.custom_rules_path
|
||||
try:
|
||||
path = ut.parse_dir(filepath)
|
||||
except FileNotFoundError as error:
|
||||
log.error("Failed to get file path for saving config file!")
|
||||
print(error)
|
||||
else:
|
||||
try:
|
||||
ut.save_json_file(path, {"filemodes": self.fm.filemodes})
|
||||
log.info(f"Saved custom file rules: {path}")
|
||||
except Exception as error:
|
||||
log.error("Failed to save config file!")
|
||||
print(error)
|
||||
|
||||
def reset_config(self):
|
||||
"""Resets the file sorting rules back to default."""
|
||||
try:
|
||||
data = ut.load_json_file(fm.Manager.RULES_PATH)
|
||||
self.fm.setup_file_rules(data, True)
|
||||
self.prep()
|
||||
log.info(f"Reloaded default file rules")
|
||||
except Exception:
|
||||
log.error("Failed to load default file rules!")
|
||||
|
||||
def update_mode_data(self, mode, key, value):
|
||||
"""Updates file mode data (key and value)."""
|
||||
if mode == "": mode = self.current_mode # Uses current mode
|
||||
if not mode in self.fm.filemodes:
|
||||
log.warning(f"File mode not found: {mode}")
|
||||
return
|
||||
data = self.fm.filemodes[mode]
|
||||
data.update({key: value})
|
||||
log.info(f"Set properties for file mode: {key} = {value}")
|
||||
|
||||
if key == "name": # Refreshes mode list
|
||||
self.v_modelist.set([v["name"] for v in self.fm.filemodes.values()])
|
||||
elif key == "extensions": # Refreshes extensions cache
|
||||
value = value.replace(" ", "") # Removes all spaces
|
||||
data.update({"extensions": value.split(",")}) # Converts to list
|
||||
self.fm.match_file_types()
|
||||
self.fm.update_file_data()
|
||||
self.update_fileview()
|
||||
|
||||
def set_rules_variables(self, mode):
|
||||
"""Updates the internal variables of the file rules (key is mode)."""
|
||||
if not mode in self.fm.filemodes:
|
||||
log.warning(f"File mode not found: {mode}")
|
||||
return
|
||||
data = self.fm.filemodes[mode]
|
||||
self.v_name.set(data["name"])
|
||||
self.v_action.set(data["action"])
|
||||
self.v_targetdir.set(data["destination"])
|
||||
self.v_extstr.set(", ".join(data["extensions"]))
|
||||
|
||||
def select_mode_from_list(self, event):
|
||||
"""Selects the current file mode to use; also updates variables."""
|
||||
key = ""
|
||||
sel = self.w_list.curselection()
|
||||
if len(sel) == 0: return
|
||||
value = self.w_list.get(sel)
|
||||
for k, v in self.fm.filemodes.items():
|
||||
if v["name"] == value:
|
||||
key = k
|
||||
break
|
||||
|
||||
if len(key) > 0: # Valid key was found
|
||||
self.current_mode = key
|
||||
self.set_rules_variables(key)
|
||||
log.debug(f"Selected file mode: {key}")
|
||||
|
||||
def ask_sort_directory(self):
|
||||
"""Prompts which file directory to process files for the given mode."""
|
||||
if not self.current_mode in self.fm.filemodes: return
|
||||
data = self.fm.filemodes[self.current_mode]
|
||||
dir = fd.askdirectory(initialdir = ut.parse_dir(data["destination"]),
|
||||
title = "Select target directory")
|
||||
if len(dir) > 0:
|
||||
self.v_targetdir.set(dir)
|
||||
self.update_mode_data("", "destination", dir)
|
||||
|
||||
def set_directory(self, dir):
|
||||
"""Sets the file directory for the app to use."""
|
||||
self.fm.set_directory(dir)
|
||||
log.debug(f"Set custom file directory: {dir}")
|
||||
self.update_fileview()
|
||||
|
||||
def ask_directory(self):
|
||||
"""Prompts which file directory for the app to use."""
|
||||
dir = fd.askdirectory(initialdir = ut.parse_dir(self.fm.dir),
|
||||
title = "Select directory")
|
||||
if len(dir) > 0:
|
||||
self.set_directory(dir)
|
||||
|
||||
def ask_load_rules(self):
|
||||
"""Prompts which file rules config for the app to use."""
|
||||
initdir = ut.parse_dir(self.fm.config_path)
|
||||
filetypes = (("Config files", "*.json"), ("All files", "*.*"))
|
||||
dir = fd.askopenfilename(initialdir = initdir,
|
||||
filetypes = filetypes,
|
||||
title = "Load config file")
|
||||
if len(dir) > 0:
|
||||
self.custom_rules_path = dir
|
||||
self.load_config(dir)
|
||||
|
||||
def ask_save_rules(self):
|
||||
"""Prompts which file name to use to save the file rules config."""
|
||||
initdir = ""
|
||||
if self.custom_rules_path == RULES_PATH:
|
||||
initdir = ut.parse_dir("~/") # Switch to home directory
|
||||
else:
|
||||
initdir = ut.parse_dir(self.custom_rules_path)
|
||||
|
||||
filetypes = (("Config files", "*.json"), ("All files", "*.*"))
|
||||
dir = fd.asksaveasfilename(initialdir = initdir,
|
||||
filetypes = filetypes,
|
||||
title = "Save config file")
|
||||
if len(dir) > 0:
|
||||
if ".json" not in dir: dir += ".json"
|
||||
self.custom_rules_path = dir
|
||||
self.save_config(dir)
|
||||
|
||||
def prep(self):
|
||||
"""Makes the app ready to use after initialization."""
|
||||
self.update_fileview()
|
||||
self.v_modelist.set([v["name"] for v in self.fm.filemodes.values()])
|
||||
if self.current_mode in self.fm.filemodes:
|
||||
self.set_rules_variables(self.current_mode)
|
||||
|
||||
def run_task(self):
|
||||
"""Runs the main file sorting task."""
|
||||
result = mb.askyesno("Task", f"Ready? {len(self.fm.filedata)} files "
|
||||
"will be affected.")
|
||||
if not result: return
|
||||
|
||||
log.info("File manager task started")
|
||||
self.fm.run_task()
|
||||
|
||||
stats = self.fm.work["stats"]
|
||||
mb.showinfo("Task",
|
||||
f"Task done! Processed {stats['total']} files in total.\n"
|
||||
+ f"Moved {stats['move']} files.\n"
|
||||
+ f"Copied {stats['copy']} files.\n"
|
||||
+ f"Deleted {stats['delete']} files.\n")
|
||||
|
||||
log.info("File manager task finished")
|
||||
self.update_fileview()
|
||||
|
||||
def gui(self):
|
||||
"""Main user interface of the app."""
|
||||
self.gui_menu()
|
||||
|
||||
self.book = ttk.Notebook(self)
|
||||
self.book.add(self.gui_files(), text="Files")
|
||||
self.book.add(self.gui_rules(), text="Rules")
|
||||
|
||||
self.book.pack(expand=True, fill=tk.BOTH)
|
||||
|
||||
def gui_menu(self):
|
||||
"""File bar of the app."""
|
||||
a_menu = tk.Menu(self)
|
||||
self.config(menu=a_menu)
|
||||
|
||||
a_menu_file = tk.Menu(a_menu)
|
||||
a_menu_file.add_command(label="Open Directory",
|
||||
command=self.ask_directory)
|
||||
a_menu_file.add_separator()
|
||||
|
||||
a_menu_file.add_command(label="Load Rules",
|
||||
command=self.ask_load_rules)
|
||||
a_menu_file.add_command(label="Save Rules",
|
||||
command=self.save_config)
|
||||
a_menu_file.add_command(label="Save Rules As...",
|
||||
command=self.ask_save_rules)
|
||||
a_menu_file.add_command(label="Reset Rules",
|
||||
command=self.reset_config)
|
||||
a_menu_file.add_separator()
|
||||
|
||||
a_menu_file.add_command(label="Exit", command=self.destroy)
|
||||
|
||||
a_menu.add_cascade(
|
||||
label="File",
|
||||
menu=a_menu_file,
|
||||
underline=0
|
||||
)
|
||||
|
||||
a_menu_settings = tk.Menu(a_menu)
|
||||
|
||||
self.v_tree = tk.BooleanVar(value=True)
|
||||
a_menu_settings.add_checkbutton(label="Show Categories",
|
||||
variable=self.v_tree,
|
||||
command=self.update_fileview)
|
||||
|
||||
a_menu.add_cascade(
|
||||
label="Settings",
|
||||
menu=a_menu_settings,
|
||||
underline=0
|
||||
)
|
||||
|
||||
|
||||
def gui_files(self):
|
||||
"""Files tab of the app."""
|
||||
frame = ttk.Frame(self.book)
|
||||
|
||||
button = ttk.Button(frame, text="Start Task", command=self.run_task)
|
||||
button.pack(side=tk.TOP, anchor=tk.W, pady=5)
|
||||
|
||||
self.fileview_columns = ("Last Modified", "Size")
|
||||
self.fileview = ttk.Treeview(frame, columns=self.fileview_columns)
|
||||
# Disabled "show=headings"
|
||||
|
||||
temp_columns = (("name", "Name",
|
||||
lambda: self.sort_fileview_items("name")),
|
||||
("time", "Last Modified",
|
||||
lambda: self.sort_fileview_items("time")),
|
||||
("size", "Size",
|
||||
lambda: self.sort_fileview_items("size")))
|
||||
|
||||
for i, v in enumerate(temp_columns):
|
||||
id = "#0" if i == 0 else v[1]
|
||||
width = 200 if i == 0 else 100
|
||||
self.fileview.heading(id, text=v[1], command=v[2])
|
||||
self.fileview.column(id, width=width, anchor=tk.W)
|
||||
|
||||
sbar = ttk.Scrollbar(frame, orient=tk.VERTICAL,
|
||||
command=self.fileview.yview)
|
||||
self.fileview.configure(yscrollcommand=sbar.set)
|
||||
|
||||
self.fileview.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
|
||||
sbar.pack(side=tk.RIGHT, fill=tk.Y)
|
||||
|
||||
return frame
|
||||
|
||||
def gui_rules(self):
|
||||
"""Rules tab of the app."""
|
||||
um = self.update_mode_data
|
||||
frame = ttk.Frame(self.book)
|
||||
|
||||
main = ttk.Frame(frame)
|
||||
main.columnconfigure(0, weight=1, uniform="column")
|
||||
main.columnconfigure(1, weight=3, uniform="column")
|
||||
|
||||
modelist = [v["name"] for v in self.fm.filemodes.values()]
|
||||
self.v_modelist = tk.Variable(value=modelist)
|
||||
self.w_list = tk.Listbox(main, listvariable=self.v_modelist,
|
||||
selectmode=tk.SINGLE)
|
||||
self.w_list.bind('<<ListboxSelect>>', self.select_mode_from_list)
|
||||
self.w_list.grid(row=0, column=0, padx=5, sticky=tk.NSEW)
|
||||
|
||||
details = ttk.Frame(main)
|
||||
|
||||
i_frame = ttk.Labelframe(details, text="Name")
|
||||
|
||||
self.v_name = tk.StringVar()
|
||||
f_name = lambda e: um("", "name", self.v_name.get())
|
||||
entry = ttk.Entry(i_frame, textvariable=self.v_name)
|
||||
entry.bind('<Return>', f_name)
|
||||
entry.pack(side=tk.LEFT)
|
||||
i_frame.pack(side=tk.TOP, anchor=tk.W, pady=5)
|
||||
|
||||
i_frame = ttk.Labelframe(details, text="Action")
|
||||
|
||||
self.v_action = tk.StringVar()
|
||||
self.v_actions = ("move", "copy", "delete", "ignore")
|
||||
f_action = lambda: um("", "action", self.v_action.get())
|
||||
for s in self.v_actions:
|
||||
str = f"{s.capitalize()} "
|
||||
radio = ttk.Radiobutton(i_frame, text=str, value=s,
|
||||
variable=self.v_action, command = f_action)
|
||||
radio.pack(side=tk.LEFT)
|
||||
i_frame.pack(side=tk.TOP, anchor=tk.W, pady=5)
|
||||
|
||||
i_frame = ttk.Labelframe(details, text="Target Directory")
|
||||
|
||||
self.v_targetdir = tk.StringVar()
|
||||
f_targetdir = lambda e: um("", "destination", self.v_targetdir.get())
|
||||
entry = ttk.Entry(i_frame, textvariable=self.v_targetdir)
|
||||
entry.bind('<Return>', f_targetdir)
|
||||
entry.pack(side=tk.TOP, fill=tk.X)
|
||||
|
||||
f_userdir = lambda: self.ask_sort_directory()
|
||||
button = ttk.Button(i_frame, text="Browse...", command=f_userdir)
|
||||
button.pack(side=tk.TOP, anchor=tk.W)
|
||||
|
||||
i_frame.pack(side=tk.TOP, fill=tk.X, pady=5)
|
||||
|
||||
i_frame = ttk.Labelframe(details, text="File Filters")
|
||||
|
||||
self.v_extstr = tk.StringVar()
|
||||
f_extstr = lambda e: um("", "extensions", self.v_extstr.get())
|
||||
entry = ttk.Entry(i_frame, textvariable=self.v_extstr)
|
||||
entry.bind('<Return>', f_extstr)
|
||||
entry.pack(side=tk.TOP, fill=tk.X)
|
||||
|
||||
i_frame.pack(side=tk.TOP, fill=tk.X, pady=5)
|
||||
|
||||
details.grid(row=0, column=1, sticky=tk.NSEW)
|
||||
|
||||
main.pack(side=tk.TOP, fill=tk.X)
|
||||
|
||||
return frame
|
||||
|
||||
if __name__ == "__main__":
|
||||
run()
|
||||
215
src/finch_filer/file_manager.py
Normal file
215
src/finch_filer/file_manager.py
Normal file
|
|
@ -0,0 +1,215 @@
|
|||
"""File manager and organizer module."""
|
||||
|
||||
__author__ = "Gull"
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import logging as log
|
||||
import send2trash as s2t
|
||||
from finch_filer import utils as ut
|
||||
|
||||
debug_mode = False
|
||||
|
||||
class Manager:
|
||||
"""File manager and organizer system."""
|
||||
# These constants are defined here, for cross module access
|
||||
SOURCE_PATH = "~/Downloads/"
|
||||
RULES_PATH = "data/file_rules_default.json"
|
||||
|
||||
"""File manager that can store and sort files in a directory."""
|
||||
def __init__(self, auto_config=False):
|
||||
self.dir = ut.parse_dir(Manager.SOURCE_PATH) # Current directory
|
||||
self.config_path = Manager.RULES_PATH # Current file rules
|
||||
self.filedata = dict() # Dictionary of files in directory
|
||||
self.filemodes = dict() # Contains file mode data
|
||||
self.filetypes = dict() # Maps file extensions to modes
|
||||
self.work = { # Work history of last task
|
||||
"files": list(),
|
||||
"stats": {"move": 0, "copy": 0, "delete": 0, "ignore": 0,
|
||||
"total": 0}
|
||||
}
|
||||
|
||||
if auto_config:
|
||||
self.set_directory(ut.parse_dir(self.dir))
|
||||
self.setup_file_rules(ut.load_json_file(self.config_path))
|
||||
|
||||
def set_directory(self, dir):
|
||||
"""Sets the working directory of the file manager."""
|
||||
if os.path.exists(dir):
|
||||
self.dir = dir
|
||||
else:
|
||||
raise FileNotFoundError(f"Directory path not found: {dir}")
|
||||
|
||||
def set_config_path(self, dir):
|
||||
"""Sets the working config path of the file manager."""
|
||||
if os.path.exists(dir):
|
||||
self.config_path = dir
|
||||
else:
|
||||
raise FileNotFoundError(f"Config path not found: {dir}")
|
||||
|
||||
def setup_file_rules(self, data, reset=False):
|
||||
"""Sets or resets file rules from the given data."""
|
||||
if not "filemodes" in data: return
|
||||
if reset: self.filemodes.clear()
|
||||
|
||||
filemodes = data["filemodes"]
|
||||
for k, v in filemodes.items():
|
||||
self.filemodes[k] = v
|
||||
|
||||
self.match_file_types()
|
||||
|
||||
def match_file_types(self):
|
||||
"""Organizes the file types dict into modes."""
|
||||
self.filetypes.clear()
|
||||
for k, v in self.filemodes.items():
|
||||
if "extensions" in v:
|
||||
for ext in v["extensions"]:
|
||||
self.filetypes[ext] = k
|
||||
|
||||
def set_type_list(self, key, values):
|
||||
"""Sets the list of extensions for the given file type."""
|
||||
if key in self.filetypes:
|
||||
self.filetypes[key] = list(values)
|
||||
|
||||
def get_file_properties(self, short_filename):
|
||||
"""Returns the time, size, mode, and full path of a single file."""
|
||||
fullpath = os.path.join(self.dir, short_filename)
|
||||
try:
|
||||
os.stat(fullpath)
|
||||
except Exception as error:
|
||||
raise error
|
||||
|
||||
_, ext = os.path.splitext(fullpath)
|
||||
ext = ext[1:].lower()
|
||||
|
||||
mode = "other"
|
||||
for k, v in self.filetypes.items():
|
||||
if ext in k: mode = v # Matched extension
|
||||
for k, v in self.filetypes.items():
|
||||
if k in short_filename: mode = v # Matched file name
|
||||
# File name has priority over extension
|
||||
|
||||
stat = os.stat(fullpath)
|
||||
return {"time": stat.st_mtime, "size": stat.st_size, "mode": mode,
|
||||
"fullpath": fullpath}
|
||||
|
||||
def update_file_data(self):
|
||||
"""Updates the data of all files in the active directory."""
|
||||
self.filedata.clear()
|
||||
self.force_update = False
|
||||
with os.scandir(self.dir) as scan:
|
||||
for entry in scan:
|
||||
if entry.is_file():
|
||||
data = self.get_file_properties(entry.name)
|
||||
self.filedata.update({entry.name: data})
|
||||
|
||||
def run_task(self):
|
||||
"""Sorts all files in the active directory, following file rules."""
|
||||
w_files = self.work["files"]
|
||||
w_stats = self.work["stats"]
|
||||
w_files.clear()
|
||||
for k in w_stats: w_stats[k] = 0
|
||||
self.match_file_types() # Updates file types dict
|
||||
log.info(f"Now processing {len(self.filedata)} files")
|
||||
if debug_mode:
|
||||
log.warning("Debug mode is enabled; all files will be ignored")
|
||||
if self.force_update:
|
||||
self.update_file_data()
|
||||
all_action = "ignore"
|
||||
if "all" in self.filemodes:
|
||||
all_action = self.filemodes["all"]["action"]
|
||||
if all_action != "ignore":
|
||||
log.info(f"Enforcing action for all files: {all_action}")
|
||||
|
||||
for v in self.filedata.values():
|
||||
rule = self.filemodes[v["mode"]]
|
||||
action = rule["action"] if all_action == "ignore" else all_action
|
||||
|
||||
if action == "move":
|
||||
src, dst = move(v["fullpath"], rule["destination"])
|
||||
w_files.append((action, src, dst))
|
||||
elif action == "copy":
|
||||
src, dst = copy(v["fullpath"], rule["destination"])
|
||||
w_files.append((action, src, dst))
|
||||
elif action == "delete":
|
||||
src = delete(v["fullpath"])
|
||||
w_files.append((action, src, "(trash)"))
|
||||
|
||||
if action in w_stats: w_stats[action] += 1
|
||||
if action != "ignore": w_stats["total"] += 1
|
||||
|
||||
if debug_mode:
|
||||
log.info(f"Simulated {w_stats['total']} files in total")
|
||||
else:
|
||||
log.info(f"Processed {w_stats['total']} files in total")
|
||||
log.info(f"Moved {w_stats['move']} files")
|
||||
log.info(f"Copied {w_stats['copy']} files")
|
||||
log.info(f"Deleted {w_stats['delete']} files")
|
||||
|
||||
def set_debug_mode(value):
|
||||
"""Enables or disables the file manager debug mode."""
|
||||
global debug_mode
|
||||
if type(value) == None:
|
||||
debug_mode = not debug_mode # Toggles the mode
|
||||
else:
|
||||
debug_mode = value
|
||||
|
||||
def move(src, dst):
|
||||
"""Moves a file from a source to a destination."""
|
||||
try:
|
||||
src, dst = ut.parse_dir(src), ut.parse_dir(dst)
|
||||
except Exception as error:
|
||||
log.error(f"Invalid source or destination: {src} -> {dst}")
|
||||
print(error)
|
||||
return src, dst
|
||||
|
||||
if not debug_mode:
|
||||
try:
|
||||
shutil.move(src, dst)
|
||||
except Exception as error:
|
||||
log.error(f"Failed to move file: {src} -> {dst}")
|
||||
print(error)
|
||||
return src, dst
|
||||
|
||||
log.info(f"Moved file: {src} > {dst}")
|
||||
return src, dst
|
||||
|
||||
def copy(src, dst):
|
||||
"""Copies a file from a source to a destination."""
|
||||
try:
|
||||
src, dst = ut.parse_dir(src), ut.parse_dir(dst)
|
||||
except Exception as error:
|
||||
log.error(f"Invalid source or destination: {src} -> {dst}")
|
||||
print(error)
|
||||
return src, dst
|
||||
|
||||
if not debug_mode:
|
||||
try:
|
||||
shutil.copy2(src, dst)
|
||||
except Exception as error:
|
||||
log.error(f"Failed to copy file: {src} -> {dst}")
|
||||
print(error)
|
||||
return src, dst
|
||||
|
||||
log.info(f"Copied file: {src} > {dst}")
|
||||
return src, dst
|
||||
|
||||
def delete(src):
|
||||
"""Moves a file to the OS trash equivalent."""
|
||||
try:
|
||||
src = ut.parse_dir(src)
|
||||
except Exception as error:
|
||||
log.error(f"Invalid source: {src}")
|
||||
print(error)
|
||||
return src
|
||||
|
||||
if not debug_mode:
|
||||
try:
|
||||
s2t.send2trash(src)
|
||||
except s2t.TrashPermissionError as error:
|
||||
log.error(f"Failed to delete file: {src}")
|
||||
print(error)
|
||||
return src
|
||||
|
||||
log.info(f"Deleted file: {src}")
|
||||
return src
|
||||
74
src/finch_filer/script.py
Normal file
74
src/finch_filer/script.py
Normal file
|
|
@ -0,0 +1,74 @@
|
|||
"""Script mode for Finch Filer."""
|
||||
|
||||
__author__ = "Gull"
|
||||
|
||||
import add_root_to_sys_path
|
||||
add_root_to_sys_path.add_to_sys_path(__file__, "src")
|
||||
|
||||
import sys
|
||||
import logging as log
|
||||
from argparse import ArgumentParser as ap
|
||||
from finch_filer import file_manager as fm
|
||||
from finch_filer import utils as ut
|
||||
|
||||
LOG_PATH = "%/Gullbase/FinchFiler/script.log"
|
||||
SOURCE_PATH = "~/Downloads"
|
||||
RULES_PATH = "%/Gullbase/FinchFiler/file_rules_custom.json"
|
||||
RULES_PATH_ALT = "data/file_rules_default.json"
|
||||
|
||||
def setup_log():
|
||||
log.basicConfig(**ut.get_log_config(LOG_PATH, log.DEBUG))
|
||||
log.getLogger().addHandler(log.StreamHandler(sys.stdout)) # Console
|
||||
log.info("Script started")
|
||||
|
||||
def get_parser():
|
||||
parser = ap(prog="Finch Filer",
|
||||
description="Moves, copies, or deletes files from a source "
|
||||
"directory.",
|
||||
add_help=True)
|
||||
parser.add_argument("-s", "--source", default=SOURCE_PATH,
|
||||
help="source directory (default: User Downloads)")
|
||||
parser.add_argument("-r", "--rules", default=RULES_PATH,
|
||||
help="config file (default: Same as app setting)")
|
||||
parser.add_argument("-m", "--mode", default="default",
|
||||
help="action for all files (move, copy, delete)")
|
||||
parser.add_argument("-d", "--debug", action="store_true",
|
||||
help="enables debug mode and ignores file actions")
|
||||
|
||||
return parser
|
||||
|
||||
def run():
|
||||
args = get_parser().parse_args()
|
||||
source = ut.parse_dir(args.source, True)
|
||||
rules = ut.parse_dir(args.rules, True)
|
||||
mode = args.mode
|
||||
if args.debug: fm.set_debug_mode(True)
|
||||
|
||||
setup_log()
|
||||
|
||||
mgr = fm.Manager()
|
||||
try:
|
||||
mgr.set_directory(source)
|
||||
except FileNotFoundError as error:
|
||||
print(error)
|
||||
return
|
||||
|
||||
try:
|
||||
mgr.setup_file_rules(ut.load_json_file(rules))
|
||||
except FileNotFoundError as error:
|
||||
try:
|
||||
rules = ut.parse_dir(RULES_PATH_ALT, True)
|
||||
mgr.setup_file_rules(ut.load_json_file(rules))
|
||||
except FileNotFoundError as error:
|
||||
print(error)
|
||||
return
|
||||
|
||||
if mode != "default" and "all" in mgr.filemodes:
|
||||
mgr.filemodes["all"]["action"] = mode
|
||||
|
||||
mgr.update_file_data()
|
||||
mgr.run_task()
|
||||
log.info("Script finished")
|
||||
|
||||
if __name__ == "__main__":
|
||||
run()
|
||||
92
src/finch_filer/utils.py
Normal file
92
src/finch_filer/utils.py
Normal file
|
|
@ -0,0 +1,92 @@
|
|||
"""Utility functions."""
|
||||
|
||||
__author__ = "Gull"
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import json
|
||||
import pathlib
|
||||
import appdirs as ad
|
||||
|
||||
KILOBYTES = 1024
|
||||
|
||||
def format_bytes(bytes):
|
||||
"""Returns a B, KB, MB, GB, or TB string from the given bytes."""
|
||||
units = {
|
||||
"B": float(1),
|
||||
"KB": float(KILOBYTES),
|
||||
"MB": float(KILOBYTES ** 2), # 1,048,576
|
||||
"GB": float(KILOBYTES ** 3), # 1,073,741,824
|
||||
"TB": float(KILOBYTES ** 4) # 1,099,511,627,776
|
||||
}
|
||||
|
||||
if bytes < units["KB"]:
|
||||
return f"{bytes} B"
|
||||
elif units["KB"] <= bytes < units["MB"]:
|
||||
return f"{(bytes / units['KB']):.2f} KB"
|
||||
elif units["MB"] <= bytes < units["GB"]:
|
||||
return f"{(bytes / units['MB']):.2f} MB"
|
||||
elif units["GB"] <= bytes < units["TB"]:
|
||||
return f"{(bytes / units['GB']):.2f} GB"
|
||||
elif units["TB"] <= bytes:
|
||||
return f"{(bytes / units['TB']):.2f} TB"
|
||||
|
||||
def format_date(timestamp):
|
||||
"""Returns a time string (ISO 8601) from the given integer timestamp."""
|
||||
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(timestamp))
|
||||
|
||||
def load_json_file(filepath):
|
||||
"""Opens a JSON file, and returns JSON data as a dict."""
|
||||
try:
|
||||
with open(filepath, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
return data
|
||||
except Exception as error:
|
||||
raise error
|
||||
|
||||
def save_json_file(filepath, data, raw=False):
|
||||
"""Saves a JSON file, given a dict."""
|
||||
try:
|
||||
with open(filepath, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, sort_keys=True, indent=None if raw else 2)
|
||||
except Exception as error:
|
||||
raise error
|
||||
|
||||
def parse_dir(path="", ignore_mkdir=False):
|
||||
"""Converts and creates a directory (can be full file path)."""
|
||||
if path[0] == "~": # Denotes the current user directory
|
||||
path = pathlib.Path(path).expanduser()
|
||||
elif path[0] == "%": # Denotes the local appdata directory
|
||||
path = pathlib.Path(ad.user_data_dir(None, False)) / path.lstrip("%/")
|
||||
elif path[0] == "#": # Denotes the project root directory
|
||||
try:
|
||||
base_path = sys._MEIPASS # For PyInstaller
|
||||
except Exception:
|
||||
base_path = os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
path = pathlib.Path(base_path).parent.parent / path.lstrip("#/")
|
||||
else:
|
||||
path = pathlib.Path(path)
|
||||
|
||||
if not ignore_mkdir:
|
||||
try:
|
||||
# This seems sloppy
|
||||
dir = str(path)
|
||||
_, ext = os.path.splitext(dir)
|
||||
if len(ext) > 0: dir = os.path.dirname(dir) #Is file
|
||||
pathlib.Path(dir).mkdir(parents=True, exist_ok=True)
|
||||
except Exception as error:
|
||||
raise error
|
||||
|
||||
# Returns as a string
|
||||
return str(path)
|
||||
|
||||
def get_log_config(path="~/", level=None):
|
||||
return {
|
||||
"filename":parse_dir(path),
|
||||
"filemode":"w",
|
||||
"level":level,
|
||||
"format":"{asctime}: [{levelname}] {message}",
|
||||
"style":"{"
|
||||
}
|
||||
Loading…
Reference in a new issue