Files
auto-git/main.py
2026-03-06 18:57:01 +01:00

135 lines
5.3 KiB
Python

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()