task3/main.py - master - georgiy/work

task3/main.py
267 строк9 KB
Georgiy Yankovskiy
task3
17 ноя 2024, 23:36
17 ноя 2024, 23:36bbf7be0
import os
import sqlite3
import time
from datetime import datetime
from shutil import copyfile
from subprocess import run, PIPE
import sys
import threading
import queue
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler


class ConfigTracker(FileSystemEventHandler):
    def __init__(self, path_to_track, database_path='db.sqlite'):
        self.path_to_track = path_to_track
        self.database_path = database_path
        try:
            self.user = os.getlogin()
        except Exception as e:
            self.user = "test"
        self.init_db()
        self.command_queue = queue.Queue()
        self.observer_thread = None
        self.input_thread = None

    def init_db(self):
        # print("init_db")
        try:
            conn = sqlite3.connect(self.database_path)
            cursor = conn.cursor()
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS versions (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    filename TEXT,
                    timestamp DATETIME,
                    author TEXT,
                    description TEXT,
                    content BLOB
                )
            ''')
            conn.commit()
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS committed (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    commit_name TEXT,
                    timestamp DATETIME
                )
            ''')
            conn.commit()
            conn.close()
        except Exception as e:
            print(e)
            pass

    def on_modified(self, event):
        # print("on_modified", event)
        try:
            if not event.is_directory:
                # print(f'Изменён файл: {event.src_path}')
                self.save_version(event.src_path)
        except Exception as e:
            print(e)

        # print(f"on_any_event File modified: {event.src_path}", event)

    def save_version(self, file_path):
        # print("save_version", file_path)
        try:
            with open(file_path, 'rb') as f:
                content = f.read()

            conn = sqlite3.connect(self.database_path)
            cursor = conn.cursor()
            cursor.execute('''INSERT INTO versions(filename, timestamp, author, content)
                              VALUES(?, ?, ?, ?)''',
                           (os.path.basename(file_path), datetime.now(), self.user, content))
            conn.commit()
            conn.close()
        except Exception as e:
            # print(e)
            pass

    def save_commit(self, commit_name):
        # print("save_version", file_path)
        try:
            conn = sqlite3.connect(self.database_path)
            cursor = conn.cursor()
            cursor.execute('''INSERT INTO committed(commit_name, timestamp)
                              VALUES(?, ?)''',
                           (commit_name, datetime.now()))
            conn.commit()
            conn.close()
        except Exception as e:
            print(e)
            pass

    def compare_versions(self, file_name, version_id_1, version_id_2):
        # print("compare_versions", file_name, version_id_1, version_id_2)
        try:
            conn = sqlite3.connect(self.database_path)
            cursor = conn.cursor()

            cursor.execute('''SELECT content FROM versions WHERE filename=? AND id IN (?, ?)''',
                           (file_name, version_id_1, version_id_2))
            results = cursor.fetchall()

            if len(results) == 2:
                diff_result = run(['diff'], input=results[0][0], stdout=PIPE, stderr=PIPE)
                return diff_result.stdout.decode() + '\n' + diff_result.stderr.decode()
            else:
                return 'Не найдено достаточно версий для сравнения.'

            conn.close()
        except Exception as e:
            pass
            # print(e)

    def revert_to_version(self, file_name, version_id):
        # print("revert_to_version", file_name, version_id)
        try:
            conn = sqlite3.connect(self.database_path)
            cursor = conn.cursor()

            cursor.execute('''SELECT content FROM versions WHERE filename=? AND id=?''',
                           (file_name, version_id))
            result = cursor.fetchone()

            if result:
                with open(os.path.join(self.path_to_track, file_name), 'wb') as f:
                    f.write(result[0])
                print(f'Файл успешно восстановлен до версии {version_id}.')
            else:
                print(f'Версия {version_id} не найдена.')

            conn.close()
        except Exception as e:
            pass
            # print(e)

    def get_history(self, file_name):
        # print("get_history", file_name)
        try:
            conn = sqlite3.connect(self.database_path)
            cursor = conn.cursor()

            cursor.execute('''SELECT id, timestamp, author, description FROM versions WHERE filename=? ORDER BY id DESC''',
                           (file_name,))
            rows = cursor.fetchall()

            for row in rows:
                print(f'{row[0]} | {row[1]} | {row[2]} | Комментарий: файл изменен')

            conn.close()
        except Exception as e:
            pass
            # print(e)

    def get_history_commits(self):
        # print("get_history", file_name)
        try:
            conn = sqlite3.connect(self.database_path)
            cursor = conn.cursor()

            cursor.execute('''SELECT id, timestamp, commit_name FROM committed ORDER BY id DESC''',
                           ())
            rows = cursor.fetchall()

            for row in rows:
                print(f'{row[0]} | {row[1]} | {row[2]}')

            conn.close()
        except Exception as e:
            print(e)
            pass

    def commit_changes(self, file_name, message=''):
        # print("commit_changes", file_name, message)
        try:
            current_file_path = os.path.join(self.path_to_track, file_name)
            new_file_path = os.path.join(self.path_to_track, '.versions', file_name + '.v' + str(int(datetime.now().timestamp())))

            try:
                os.makedirs(os.path.dirname(new_file_path), exist_ok=True)
                copyfile(current_file_path, new_file_path)
                print(f'Изменения зафиксированы в {new_file_path}.')
            except Exception as e:
                print(f'Ошибка при фиксации изменений: {e}')

            self.save_version(current_file_path, message)
        except Exception as e:
            # print(e)
            pass

    def start_observing(self):
        observer = Observer()
        observer.schedule(self, self.path_to_track, recursive=True)
        observer.start()
        self.observer_thread = threading.Thread(target=self.run_observer, args=(observer,))
        self.observer_thread.start()

    def stop_observing(self):
        if self.observer_thread is not None:
            self.observer_thread.join()

    def run_observer(self, observer):
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            observer.stop()
        observer.join()

    def process_commands(self):
        while True:
            try:
                command = input("Введите команду (revert, history, commit_history, commit, ctrl+c "
                                "несколько раз для выхода): ").strip()
                if command == 'revert':
                    file_name = input("Введите имя файла: ")
                    version_id = int(input("Введите ID: "))
                    self.revert_to_version(file_name, version_id)

                if command == 'history':
                    file_name = input("Введите имя файла: ")
                    self.get_history(file_name)

                if command == 'commit_history':
                    self.get_history_commits()

                if command == 'commit':
                    commit_name = input("Введите идентификатор коммита (название, ID или текст уникальный): ")
                    self.save_commit(commit_name)

            except Exception as ignore:
                pass

    def start_input_processing(self):
        self.input_thread = threading.Thread(target=self.process_commands)
        self.input_thread.start()

    def stop_input_processing(self):
        if self.input_thread is not None:
            self.input_thread.join()


def main():
    if len(sys.argv) != 2:
        print("Использование: python config_tracker.py <путь_к_каталогу>")
        sys.exit(1)

    path_to_track = sys.argv[1]

    tracker = ConfigTracker(path_to_track)
    tracker.start_observing()
    tracker.start_input_processing()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        tracker.stop_observing()
        tracker.stop_input_processing()


if __name__ == "__main__":
    main()