cfg_vcs/main.py - master - qoopdata/IT-Planet_Pro_SberLinux

cfg_vcs/main.py
165 строк6 KB
Kirill Bezuglyi
solutions
17 ноя 2024, 23:43
17 ноя 2024, 23:432b02cf3
from datetime import datetime
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler


import argparse
import difflib
import getpass
import git
import hashlib
import os
import sys
import time
import yaml


class ConfigFileEventHandler(FileSystemEventHandler):
    def __init__(self, version_manager):
        self.version_manager = version_manager

    def on_modified(self, event):
        if not event.is_directory:
            self.version_manager.save_version(event.src_path)
            print(f'file modified: {event.src_path}')

    def on_created(self, event):
        if not event.is_directory:
            self.version_manager.save_version(event.src_path)
            print(f'file created: {event.src_path}')

    def on_deleted(self, event):
        if not event.is_directory:
            self.version_manager.handle_deletion(event.src_path)
            print(f'file deleted: {event.src_path}')

class VersionManager:
    def __init__(self, storage_dir):
        self.storage_dir = storage_dir
        self.repo_dir = Path(storage_dir).parent
        os.makedirs(self.storage_dir, exist_ok=True)

        if not (self.repo_dir / ".git").exists():
            self.repo = git.Repo.init(self.repo_dir)
        else:
            self.repo = git.Repo(self.repo_dir)

    def _hash_content(self, content):
        return hashlib.sha256(content.encode()).hexdigest()

    def save_version(self, file_path):
        if not os.path.exists(file_path):
            return
        
        with open(file_path, 'r') as f:
            content = f.read()

        metadata = {
            'time': datetime.now().isoformat(),
            'author': getpass.getuser(),
            'description': 'Auto-saved change',
            'hash': self._hash_content(content),
            'content': content,
        }
        version_file = os.path.join(self.storage_dir, os.path.basename(file_path).replace('.', '_') + '.yaml')
        with open(version_file, 'a') as vf:
            yaml.dump([{file_path: metadata}], vf)
            print(f'file save: {file_path}')

    def handle_deletion(self, file_path):
        metadata = {
            'time': datetime.now().isoformat(),
            'author': getpass.getuser(),
            'description': 'File deleted',
        }
        version_file = os.path.join(self.storage_dir, os.path.basename(file_path).replace('.', '_') + '.yaml')
        with open(version_file, 'a') as vf:
            yaml.dump([{file_path: metadata}], vf)

    def rollback(self, file_path, target_time):
        version_file = os.path.join(self.storage_dir, os.path.basename(file_path).replace('.', '_') + '.yaml')
        if not os.path.exists(version_file):
            return False

        with open(version_file, 'r') as vf:
            versions = yaml.safe_load(vf)

        for entry in reversed(versions):
            if entry[file_path]['time'] <= target_time:
                with open(file_path, 'w') as f:
                    f.write(entry[file_path]['content'])
                print(f'file rollback: {file_path}')
                return True
        return False

    def commit_changes(self, message):
        self.repo.git.add('.')
        self.repo.index.commit(message)
        print(f'committed changes: {message}')

    def get_change_history(self):
        try:
            for commit in self.repo.iter_commits():
                print(f'commit {commit.hexsha[:7]} - {commit.message.strip()} - {commit.author.name} - {commit.committed_datetime}')
        except ValueError:
            print(f'no commit history yet')

    def rollback_to_commit(self, commit_id):
        try:
            self.repo.git.checkout(commit_id)
            print(f'rolled back to commit: {commit_id}')
        except git.exc.GitCommandError as e:
            print(f'failed to rollback: {e}')

def compare_versions(file_path, version1, version2):
    with open(version1, 'r') as v1, open(version2, 'r') as v2:
        diff = difflib.unified_diff(
            v1.readlines(),
            v2.readlines(),
            fromfile='Version 1',
            tofile='Version 2',
        )
    return ''.join(diff)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Configuration File Versioning Tool')
    parser.add_argument('action', choices=['start', 'commit', 'history', 'rollback'], help='Action to perform')
    parser.add_argument('--message', help='Commit message for saving changes')
    parser.add_argument('--commit_id', help='Commit ID to rollback to')

    args = parser.parse_args()
    storage_dir = './versions'
    version_manager = VersionManager(storage_dir)

    if args.action == 'start':
        event_handler = ConfigFileEventHandler(version_manager)
        observer = Observer()
        observer.schedule(event_handler, path='./config_files', recursive=True)

        print('program started')
        try:
            observer.start()
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            observer.stop()
        observer.join()
        print('program finished')

    elif args.action == 'commit':
        if not args.message:
            print('Error: Commit message is required for "commit" action.')
        else:
            version_manager.commit_changes(args.message)

    elif args.action == 'history':
        print('Change history:')
        version_manager.get_change_history()

    elif args.action == 'rollback':
        if not args.commit_id:
            print('Error: Commit ID is required for "rollback" action.')
        else:
            version_manager.rollback_to_commit(args.commit_id)