refactor: move files to git_monitor/ directory and add run.py

This commit is contained in:
2026-03-06 19:30:27 +01:00
parent d2556bf656
commit 1b6c7dcc01
9 changed files with 13 additions and 0 deletions

37
git_monitor/config.py Normal file
View File

@@ -0,0 +1,37 @@
import json
import os
CONFIG_FILE = "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()

27
git_monitor/logger.py Normal file
View File

@@ -0,0 +1,27 @@
import logging
import os
import sys
def setup_logger():
log_file = "git_monitor.log"
# Ensure log is in the application directory
log_path = os.path.abspath(log_file)
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_path)
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()