epagris/CodeCast_Client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

- ignoring files implementation basics

Browse Source
This commit is contained in:
András Wiesner 2026-03-05 11:01:42 +01:00
parent ab24d03d3f
commit d84c76d0d3
2 changed files with 103 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
sample_settings.json Normal file
View File

@ -0,0 +1,18 @@
{
"copy-policy": {
"general": "prohibited",
"per-file": {
"allowed": [
"main.c",
"main.h",
"test1.md"
],
"prohibited": [
"echo_path.sh"
]
}
},
"ignored-paths": [
"build/sub/*"
]
}

111
sync_client.py
View File

@ -1,3 +1,4 @@
import fnmatch
import platform
import tkinter as tk
from tkinter import ttk
@ -18,12 +19,17 @@ class SyncClient:
else:
return path.removeprefix("/")
SETTINGS_DIR: str = ".codecast"
MAIN_SETTINGS_FILE: str = SETTINGS_DIR + os.sep + "settings.json"
def start(self) -> None:
_sself = self
self.load_settings()
self._event_queue = queue.Queue()
def comm_thread_routine(queue: queue.Queue):
def fs_watch_thread_routine(queue: queue.Queue):
run = True
while run:
event: FileSystemEvent = queue.get()
@ -31,8 +37,14 @@ class SyncClient:
if event.src_path == "**CLOSE**" and event.is_synthetic == True:
run = False
continue
elif event.src_path == SyncClient.MAIN_SETTINGS_FILE or event.dest_path == SyncClient.MAIN_SETTINGS_FILE:
_sself.load_settings()
rel_path = str(event.src_path).removeprefix(_sself.dir()).removeprefix(os.sep)
for pattern in self.get_ignored_paths():
if fnmatch.fnmatch(rel_path, pattern):
continue
if event.event_type == "created" or event.event_type == "modified":
_sself.upload(rel_path)
elif event.event_type == "moved":
@ -53,8 +65,8 @@ class SyncClient:
if event.src_path != _sself.dir():
_sself._event_queue.put(event)
self.__comm_thread = threading.Thread(target=comm_thread_routine, args=(self._event_queue,))
self.__comm_thread.start()
self.__watch_thread = threading.Thread(target=fs_watch_thread_routine, args=(self._event_queue,))
self.__watch_thread.start()
observer = Observer()
event_handler = FileChangeHandler()
@ -69,8 +81,9 @@ class SyncClient:
self.__url = url
self.__id = id
self.__key = key
self.__ignored_paths = set()
self.__comm_thread: threading.Thread
self.__watch_thread: threading.Thread
self._event_queue: queue.Queue
@ -90,6 +103,32 @@ class SyncClient:
return "file"
def _set_ignored_paths(self, ignored_paths: list | set) -> None:
old_ignored_paths = set(self.__ignored_paths)
new_ignored_paths = set(ignored_paths)
included = old_ignored_paths - new_ignored_paths
excluded = new_ignored_paths - old_ignored_paths
self.upload_all(included)
self.delete_all(excluded)
self.__ignored_paths = new_ignored_paths
def get_ignored_paths(self) -> set:
return self.__ignored_paths
def load_settings(self) -> None:
try:
with open(self.__dir + os.sep + self.MAIN_SETTINGS_FILE, "r") as sf:
settings = json.loads(sf.read())
self._set_ignored_paths(settings.get("ignored-paths", []))
except:
self._set_ignored_paths(set())
def upload(self, filename: str) -> bool:
print("[UPLOAD]", filename)
@ -164,42 +203,63 @@ class SyncClient:
def close(self) -> None:
self.__observer.stop()
self._event_queue.put(FileSystemEvent("**CLOSE**", is_synthetic=True))
self.__comm_thread.join()
self.__watch_thread.join()
return
def upload_all(self) -> None:
# create directories first
def diriter(path, rel_path = "", types = [ "directory", "file" ]):
listing = []
for entry in os.scandir(path):
path = (rel_path + os.sep + entry.name).removeprefix(os.sep)
name = entry.name.removeprefix(os.sep)
if entry.is_dir():
if "directory" in types:
listing.append({"type": "directory", "path": path, "name": name})
dir_listing = diriter(entry.path, rel_path=path, types=types)
listing.extend(dir_listing)
else:
if "file" in types:
listing.append({"type": "file",
"path": path,
"name": name})
return listing
def __list_files(self, types = [ "directory", "file" ], path_match: set | None = None) -> list:
def diriter(path, rel_path = "", types = [ "directory", "file" ], path_match: set | None = None):
listing = []
for entry in os.scandir(path):
path = (rel_path + os.sep + entry.name).removeprefix(os.sep)
dirs = diriter(self.dir(), types=["directory"])
if path_match is not None:
skip_entry = True
for pattern in path_match:
if fnmatch.fnmatch(path, pattern):
skip_entry = False
break
else:
skip_entry = False
if not skip_entry:
name = entry.name.removeprefix(os.sep)
if entry.is_dir():
if "directory" in types:
listing.append({"type": "directory", "path": path, "name": name})
dir_listing = diriter(entry.path, rel_path=path, types=types, path_match=path_match)
listing.extend(dir_listing)
else:
if "file" in types:
listing.append({"type": "file",
"path": path,
"name": name})
return listing
return diriter(self.dir(), types=types, path_match=path_match)
def upload_all(self, path_match: set | None = None) -> None:
# create directories first
dirs = self.__list_files(types=["directory"], path_match=path_match)
for dir in dirs:
self.upload(dir["path"])
# then upload the files
files = diriter(self.dir(), types=["file"])
files = self.__list_files(types=["file"], path_match=path_match)
for file in files:
self.upload(file["path"])
pass
def delete_all(self, path_match: set | None = None) -> None:
files_and_dirs = self.__list_files(path_match=path_match)
for entry in files_and_dirs:
self.delete(entry["path"])
# ------------------
class SyncClientGui:
@ -346,7 +406,6 @@ class SyncClientGui:
self.__sc = SyncClient(self.__dir, self.__url, self.__id, self.__key)
self.__sc.wipe()
self.__sc.upload_all()
self.__sc.start()