Compare commits

...

11 Commits

13 changed files with 350 additions and 289 deletions

2
.gitignore vendored
View File

@@ -1,3 +1,5 @@
git_monitor.log
config.json
.vscode
venv-win
moje-repo

View File

@@ -286,7 +286,7 @@ Responsible for Windows notifications.
Recommended library:
```
win10toast
win10toast or plyer
```
Main class:

45
git_monitor/config.py Normal file
View File

@@ -0,0 +1,45 @@
import json
import os
import sys
def get_app_dir():
if getattr(sys, 'frozen', False):
return os.path.dirname(sys.executable)
else:
# If in git_monitor package, go up to the project root
return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
CONFIG_FILE = os.path.join(get_app_dir(), "config.json")
class Config:
def __init__(self):
self.data = {
"repository_path": ""
}
self.load()
def load(self):
if os.path.exists(CONFIG_FILE):
try:
with open(CONFIG_FILE, "r") as f:
self.data = json.load(f)
except Exception as e:
print(f"Error loading config: {e}")
def save(self):
try:
with open(CONFIG_FILE, "w") as f:
json.dump(self.data, f, indent=4)
except Exception as e:
print(f"Error saving config: {e}")
@property
def repository_path(self):
return self.data.get("repository_path", "")
@repository_path.setter
def repository_path(self, value):
self.data["repository_path"] = value
self.save()
config = Config()

View File

@@ -0,0 +1,64 @@
import time
import os
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from logger import logger
from git_manager import git_manager
class RepositoryWatcher:
def __init__(self, path):
self.path = path
self.observer = Observer()
self.event_handler = GitEventHandler()
def start(self):
self.observer.schedule(self.event_handler, self.path, recursive=True)
self.observer.start()
logger.info(f"Started monitoring: {self.path}")
def stop(self):
self.observer.stop()
self.observer.join()
logger.info(f"Stopped monitoring: {self.path}")
class GitEventHandler(FileSystemEventHandler):
def on_modified(self, event):
if not event.is_directory:
self.handle_event(event, "modify")
def on_created(self, event):
if not event.is_directory:
self.handle_event(event, "create")
def on_deleted(self, event):
if not event.is_directory:
self.handle_event(event, "delete")
def on_moved(self, event):
if not event.is_directory:
src_name = os.path.basename(event.src_path)
dest_name = os.path.basename(event.dest_path)
action = "rename"
file_name = f"{src_name} -> {dest_name}"
self.handle_event(event, action, custom_file_name=file_name)
def handle_event(self, event, action, custom_file_name=None):
# Ignore .git directory and the log file
file_path = event.src_path if action != "rename" else event.dest_path
file_name = os.path.basename(file_path)
# Check if any part of the path is .git
if ".git" in file_path.split(os.sep) or file_name == "git_monitor.log":
return
display_name = custom_file_name if custom_file_name else file_name
logger.info(f"File event: {action} on {display_name}")
repo_root = git_manager.repo_path
if repo_root:
if custom_file_name:
# For renames, we use the custom format provided
git_manager.commit_change(action, custom_file_name)
else:
rel_path = os.path.relpath(file_path, repo_root)
git_manager.commit_change(action, rel_path)

View File

@@ -0,0 +1,60 @@
import os
from git import Repo, exc
from logger import logger
from notifier import notifier
class GitManager:
def __init__(self, repo_path=None):
self.repo_path = repo_path
self.repo = None
if repo_path:
self.load_repository(repo_path)
def load_repository(self, path):
if self.is_git_repository(path):
try:
self.repo = Repo(path)
self.repo_path = path
logger.info(f"Loaded repository: {path}")
return True
except Exception as e:
logger.error(f"Failed to load repository {path}: {e}")
notifier.notify("Git Error", f"Failed to load repository: {e}")
else:
logger.warning(f"{path} is not a valid Git repository.")
notifier.notify("Git Error", f"{path} is not a valid Git repository.")
return False
def is_git_repository(self, path):
if not path or not os.path.isdir(path):
return False
try:
Repo(path).git_dir
return True
except (exc.InvalidGitRepositoryError, exc.NoSuchPathError):
return False
def commit_change(self, action, file_name):
if not self.repo:
logger.error("No repository loaded for commit.")
return False
commit_msg = f"{action}: {file_name}"
try:
# Stage all changes (simple approach for the CNC operator use-case)
self.repo.git.add(A=True)
# Check if there are changes to commit
if self.repo.is_dirty(untracked_files=True):
self.repo.index.commit(commit_msg)
logger.info(f"Committed: {commit_msg}")
notifier.notify("Git Monitor", commit_msg + " committed")
return True
else:
logger.info(f"No changes to commit for: {file_name}")
return False
except Exception as e:
logger.error(f"Git commit failed: {e}")
notifier.notify("Git Error", f"Commit failed: {e}")
return False
git_manager = GitManager()

32
git_monitor/logger.py Normal file
View File

@@ -0,0 +1,32 @@
import logging
import os
import sys
def setup_logger():
# Find application directory
if getattr(sys, 'frozen', False):
app_dir = os.path.dirname(sys.executable)
else:
# If in a package, go up one level to the root
app_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
log_file = os.path.join(app_dir, "git_monitor.log")
logger = logging.getLogger("GitMonitor")
logger.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
# File handler
file_handler = logging.FileHandler(log_file)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
# Also log to stdout for development (though it will be --noconsole eventually)
stream_handler = logging.StreamHandler(sys.stdout)
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)
return logger
logger = setup_logger()

19
git_monitor/main.py Normal file
View File

@@ -0,0 +1,19 @@
from logger import logger
from tray_app import TrayApp
class Application:
def __init__(self):
logger.info("Application starting...")
self.tray_app = TrayApp()
def run(self):
try:
self.tray_app.run()
except KeyboardInterrupt:
logger.info("Application stopped by user.")
except Exception as e:
logger.error(f"Application error: {e}")
if __name__ == "__main__":
app = Application()
app.run()

26
git_monitor/notifier.py Normal file
View File

@@ -0,0 +1,26 @@
from logger import logger
try:
from plyer import notification
PLYER_AVAILABLE = True
except ImportError:
logger.warning("plyer not found. Notifications will be printed to stdout.")
PLYER_AVAILABLE = False
class Notifier:
def notify(self, title, message):
logger.info(f"Notification: {title} - {message}")
if PLYER_AVAILABLE:
try:
notification.notify(
title=title,
message=message,
app_name="Git Monitor",
# timeout=10
)
except Exception as e:
logger.error(f"Error showing notification: {e}")
else:
print(f"[{title}] {message}")
notifier = Notifier()

View File

@@ -0,0 +1,6 @@
watchdog
GitPython
pystray
Pillow
plyer
win10toast

82
git_monitor/tray_app.py Normal file
View File

@@ -0,0 +1,82 @@
import os
import threading
import tkinter as tk
from tkinter import filedialog
from PIL import Image, ImageDraw
import pystray
from pystray import MenuItem as item
from logger import logger
from config import config
from git_manager import git_manager
from file_watcher import RepositoryWatcher
from notifier import notifier
class TrayApp:
def __init__(self):
self.icon = None
self.watcher = None
self.root = tk.Tk()
self.root.withdraw() # Hide main tkinter window
# Load last used repo if exists
last_repo = config.repository_path
if last_repo:
if git_manager.load_repository(last_repo):
self.start_monitoring(last_repo)
def create_icon(self):
# Create a simple icon
width = 64
height = 64
image = Image.new('RGB', (width, height), color=(73, 109, 137))
dc = ImageDraw.Draw(image)
dc.rectangle([width // 4, height // 4, width * 3 // 4, height * 3 // 4], fill=(255, 255, 255))
menu = (
item('Git Monitor', lambda: None, enabled=False),
pystray.Menu.SEPARATOR,
item('Select Repository', self.select_repository),
item('Exit', self.exit_app),
)
self.icon = pystray.Icon("GitMonitor", image, "Git Monitor", menu)
def select_repository(self, icon=None, item=None):
# Open folder dialog in a separate thread or use the hidden root
# tkinter dialogs need to run in the main thread or with care
repo_path = filedialog.askdirectory(title="Select Git Repository Folder")
if repo_path:
logger.info(f"User selected repository: {repo_path}")
if git_manager.load_repository(repo_path):
config.repository_path = repo_path
self.start_monitoring(repo_path)
def start_monitoring(self, path):
if self.watcher:
self.stop_monitoring()
self.watcher = RepositoryWatcher(path)
self.watcher.start()
notifier.notify("Git Monitor", f"Started monitoring: {os.path.basename(path)}")
def stop_monitoring(self):
if self.watcher:
self.watcher.stop()
self.watcher = None
notifier.notify("Git Monitor", "Stopped monitoring")
def exit_app(self, icon=None, item=None):
logger.info("Exiting application...")
self.stop_monitoring()
if self.icon:
self.icon.stop()
self.root.quit()
os._exit(0) # Ensure all threads terminate
def run(self):
self.create_icon()
self.icon.run()
if __name__ == "__main__":
app = TrayApp()
app.run()

135
main.py
View File

@@ -1,135 +0,0 @@
import time
import os
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from git import Repo
# KONFIGURACJA
PATH_TO_REPO = './moje-repo' # Ścieżka do Twojego katalogu git
COMMIT_MESSAGE = "Automatyczny commit: zmiana w pliku"
import pathspec
from watchdog.events import FileSystemEventHandler
class GitAutoCommitHandler(FileSystemEventHandler):
def __init__(self, repo_path):
self.repo_path = os.path.abspath(repo_path)
self.repo = Repo(self.repo_path)
self.spec = self._load_gitignore()
# Słownik do przechowywania czasu ostatniego commitu dla każdego pliku
self.last_committed = {}
self.debounce_seconds = 0.5 # Minimalny odstęp między commitami dla jednego pliku
def _load_gitignore(self):
gitignore_path = os.path.join(self.repo_path, '.gitignore')
if os.path.exists(gitignore_path):
with open(gitignore_path, 'r') as f:
# Tworzymy wzorzec dopasowania na podstawie linii w pliku
return pathspec.PathSpec.from_lines('gitwildmatch', f)
return None
def _is_ignored(self, file_path):
# Sprawdzamy ścieżkę względem głównego folderu repozytorium
relative_path = os.path.relpath(file_path, self.repo_path)
# Zawsze ignoruj folder .git
if ".git" in relative_path.split(os.sep):
return True
if self.spec and self.spec.match_file(relative_path):
return True
return False
def on_modified(self, event):
if not event.is_directory and not self._is_ignored(event.src_path):
self._process_event(event.src_path, "update-file")
def on_created(self, event):
if not event.is_directory and not self._is_ignored(event.src_path):
self._process_event(event.src_path, "new-file")
def on_deleted(self, event):
if not event.is_directory:
self._process_event(event.src_path, "remove-file")
def on_moved(self, event):
if not event.is_directory:
# Sprawdzamy, czy nowa lokalizacja nie jest ignorowana
if self._is_ignored(event.dest_path):
# Jeśli przenieśliśmy plik do folderu ignorowanego, traktujemy to jak usunięcie
self._process_event(event.src_path, "remove-file")
else:
self._process_rename(event.src_path, event.dest_path)
def _process_event(self, file_path, action_label):
if self._is_ignored(file_path):
return
current_time = time.time()
last_time = self.last_committed.get(file_path, 0)
# Sprawdź, czy minęło dość czasu od ostatniego commitu tego pliku
if current_time - last_time > self.debounce_seconds:
self.last_committed[file_path] = current_time
self._commit(file_path, action_label)
def _process_rename(self, old_path, new_path):
current_time = time.time()
# Używamy nowej ścieżki jako klucza do debounce
last_time = self.last_committed.get(new_path, 0)
if current_time - last_time > self.debounce_seconds:
self.last_committed[new_path] = current_time
try:
old_rel = os.path.relpath(old_path, self.repo_path)
new_rel = os.path.relpath(new_path, self.repo_path)
# Git sam wykrywa rename, jeśli dodamy oba pliki (usunięty i nowy)
# Ale możemy to zrobić jawnie dla czystości:
self.repo.index.remove([old_rel], working_tree=False)
self.repo.index.add([new_rel])
if self.repo.index.diff("HEAD"):
self.repo.index.commit(f"rename-file: {old_rel} -> {new_rel}")
print(f"🔄 RENAME: {old_rel} -> {new_rel}")
except Exception as e:
print(f"⚠️ Błąd podczas zmiany nazwy: {e}")
def _commit(self, file_path, action_label):
try:
relative_path = os.path.relpath(file_path, self.repo_path)
if action_label == "remove-file":
self.repo.index.remove([relative_path])
else:
self.repo.index.add([relative_path])
if self.repo.index.diff("HEAD"):
self.repo.index.commit(f"{action_label}: {relative_path}")
print(f"{action_label}: {relative_path}")
# if self.repo.is_dirty(path=relative_path) or relative_path in self.repo.untracked_files:
# self.repo.index.add([relative_path])
# self.repo.index.commit(f"Auto-commit: {relative_path}")
# print(f"✅ Zapisano: {relative_path}")
except Exception as e:
print(f"❌ Błąd: {e}")
if __name__ == "__main__":
if not os.path.exists(os.path.join(PATH_TO_REPO, ".git")):
print("Błąd: Wskazany katalog nie jest repozytorium Gita!")
else:
event_handler = GitAutoCommitHandler(PATH_TO_REPO)
observer = Observer()
observer.schedule(event_handler, PATH_TO_REPO, recursive=True)
print(f"🚀 Śledzenie katalogu: {PATH_TO_REPO}...")
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()

13
run.py Normal file
View File

@@ -0,0 +1,13 @@
import sys
import os
# Ensure the git_monitor directory is in the search path
current_dir = os.path.dirname(os.path.abspath(__file__))
git_monitor_dir = os.path.join(current_dir, "git_monitor")
sys.path.insert(0, git_monitor_dir)
from main import Application
if __name__ == "__main__":
app = Application()
app.run()

View File

@@ -1,153 +0,0 @@
import os
import time
import threading
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from git import Repo
import pathspec
from plyer import notification
from pystray import Icon, Menu, MenuItem
from PIL import Image, ImageDraw
import logging
# Logi będą zapisywane w tym samym folderze co skrypt
log_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "debug_log.txt")
logging.basicConfig(
filename=log_path,
level=logging.DEBUG,
format='%(asctime)s - %(levelname)s - %(message)s'
)
# --- TWOJA KLASA GIT (Z POPRZEDNICH KROKÓW) ---
class GitAutoCommitHandler(FileSystemEventHandler):
def __init__(self, repo_path):
self.repo_path = os.path.abspath(repo_path)
self.repo = Repo(self.repo_path)
self.spec = self._load_gitignore()
self.last_committed = {}
self.debounce_seconds = 3.0
def _load_gitignore(self):
gitignore_path = os.path.join(self.repo_path, '.gitignore')
if os.path.exists(gitignore_path):
with open(gitignore_path, 'r') as f:
return pathspec.PathSpec.from_lines('gitwildmatch', f)
return None
def _is_ignored(self, file_path):
abs_file_path = os.path.abspath(file_path)
relative_path = os.path.relpath(abs_file_path, self.repo_path)
if ".git" in relative_path.split(os.sep):
logging.debug(f"Ignorowanie (folder .git): {relative_path}")
return True
return self.spec.match_file(relative_path) if self.spec else False
def on_created(self, event):
if not event.is_directory: self._process_event(event.src_path, "new-file")
def on_modified(self, event):
if not event.is_directory: self._process_event(event.src_path, "update-file")
def on_deleted(self, event):
if not event.is_directory: self._process_event(event.src_path, "remove-file")
def on_moved(self, event):
if not event.is_directory: self._process_rename(event.src_path, event.dest_path)
def _process_event(self, file_path, action):
abs_path = os.path.abspath(file_path)
if self._is_ignored(abs_path):
return
logging.debug(f"Zdarzenie: {action} - {file_path}")
now = time.time()
if now - self.last_committed.get(abs_path, 0) > self.debounce_seconds:
self.last_committed[abs_path] = now
self._commit(abs_path, action)
def _process_rename(self, old, new):
if self._is_ignored(new): return
rel_old, rel_new = os.path.relpath(old, self.repo_path), os.path.relpath(new, self.repo_path)
self.repo.index.remove([rel_old], working_tree=False)
self.repo.index.add([rel_new])
self._finalize_commit(f"rename-file: {rel_old} -> {rel_new}", rel_new)
def _commit(self, file_path, action):
logging.debug(f"Commitowanie: {action} - {file_path}")
rel_path = os.path.relpath(file_path, self.repo_path)
if action == "remove-file":
self.repo.index.remove([rel_path])
else:
self.repo.index.add([rel_path])
self._finalize_commit(f"{action}: {rel_path}", rel_path)
# def _finalize_commit(self, msg, display_name):
# if self.repo.index.diff("HEAD"):
# self.repo.index.commit(msg)
# # POWIADOMIENIE WINDOWS
# notification.notify(
# title="Git Auto-Commit",
# message=f"Zapisano zmianę: {display_name}",
# app_name="Git Watcher",
# timeout=3
# )
def _finalize_commit(self, msg, display_name):
try:
if self.repo.is_dirty(untracked_files=True): # Dokładniejsze sprawdzenie
self.repo.index.commit(msg)
logging.info(f"✅ COMMIT SUCCESS: {msg}")
# Próba powiadomienia z łapaniem błędów
try:
notification.notify(
title="Git Auto-Commit",
message=f"Zapisano: {display_name}",
timeout=5
)
except Exception as e:
logging.error(f"Błąd powiadomienia: {e}")
except Exception as e:
logging.error(f"Błąd finalizacji commitu: {e}")
# --- LOGIKA TRAY I URUCHAMIANIA ---
def create_image():
# Generuje prostą ikonę (niebieskie koło), jeśli nie masz pliku .ico
width, height = 64, 64
image = Image.new('RGB', (width, height), (255, 255, 255))
dc = ImageDraw.Draw(image)
dc.ellipse([10, 10, 54, 54], fill=(0, 122, 204))
return image
def run_watcher(path, stop_event):
event_handler = GitAutoCommitHandler(path)
observer = Observer()
observer.schedule(event_handler, path, recursive=True)
observer.start()
while not stop_event.is_set():
time.sleep(1)
observer.stop()
observer.join()
def main():
logging.info("Aplikacja wystartowała")
repo_path = "./moje-repo" # Zmień na swoją ścieżkę
logging.info("Repozytorium: %s", repo_path)
stop_event = threading.Event()
# Uruchomienie obserwatora w osobnym wątku
watcher_thread = threading.Thread(target=run_watcher, args=(repo_path, stop_event))
watcher_thread.daemon = True
watcher_thread.start()
# Menu zasobnika
def on_quit(icon, item):
stop_event.set()
icon.stop()
icon = Icon("GitWatcher", create_image(), menu=Menu(
MenuItem(f"Śledzę: {os.path.abspath(repo_path)}", None, enabled=False),
MenuItem("Zakończ", on_quit)
))
icon.run()
if __name__ == "__main__":
main()