Merge pull request 'v1.0.1' (#4) from v1.0.1 into main

Reviewed-on: #4

.gitignore

@@ -1,10 +1,8 @@
 app.log
 Settings/settings.json
-Configs/Eltex MES2424 AC - Сеть FTTB 2G, доп.txt
-Configs/конфиг доп 3750-52 с айпи 172.17.141.133 .txt
-DALL·E 2024-12-29 01.01.02 - Square vector logo_ A clean and minimalistic app icon for serial port management software. The design prominently features a simplified rectangular CO.ico
-test.py
+Configs/
 __pycache__/
 Firmware/1.jpg
 Firmware/2
-output/
+output/
+.venv/

README.md

@@ -0,0 +1,44 @@
+# ComConfigCopy
+
+Программа для копирования конфигураций на коммутаторы.
+
+## Описание
+
+ComConfigCopy - это утилита, разработанная для автоматизации процесса копирования конфигураций на сетевые коммутаторы. Программа предоставляет удобный графический интерфейс для управления процессом копирования и настройки параметров подключения.
+
+## Основные возможности
+
+- Копирование конфигураций на коммутаторы через COM-порт
+- Поддержка различных скоростей подключения
+- Автоматическое определение доступных COM-портов
+- Возможность сохранения и загрузки настроек
+- Автоматическое обновление через GitHub
+
+## Системные требования
+
+- Windows 7/8/10/11
+- Python 3.8 или выше
+- Доступ к COM-портам
+
+## Установка
+
+1. Скачайте последнюю версию программы из [репозитория](https://gitea.filow.ru/LowaSC/ComConfigCopy/releases)
+2. Распакуйте архив в удобное место
+3. Запустите файл `ComConfigCopy.exe`
+
+## Использование
+
+1. Выберите COM-порт из списка доступных
+2. Настройте параметры подключения (скорость, биты данных и т.д.)
+3. Выберите файл конфигурации для отправки
+4. Нажмите кнопку "Отправить" для начала процесса копирования
+
+## Контакты
+
+- Email: SPRF555@gmail.com
+- Telegram: [@LowaSC](https://t.me/LowaSC)
+- Репозиторий: [ComConfigCopy](https://gitea.filow.ru/LowaSC/ComConfigCopy)
+
+## Лицензия
+
+Этот проект распространяется под лицензией MIT. Подробности смотрите в файле [LICENSE](LICENSE). 

TFTPServer.py

@@ -0,0 +1,349 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+TFTP сервер для передачи прошивки с компьютера на коммутатор.
+
+- Создает сервер по заданному IP и порту.
+- Расшаривает папку Firmware.
+- Показывает текущее состояние сервера и статус передачи файла:
+  - кому (IP устройства),
+  - сколько осталось байт,
+  - сколько передано байт,
+  - время передачи.
+"""
+
+import os
+import socket
+import struct
+import threading
+import time
+
+class TFTPServer:
+    def __init__(self, share_folder):
+        """
+        Инициализация TFTP сервера.
+
+        :param share_folder: Путь к папке, содержащей файлы (например, папка 'Firmware')
+        """
+        self.share_folder = share_folder
+        self.log_callback = None
+        self.running = False
+        self.server_socket = None
+        self.lock = threading.Lock()
+        self.transfer_sockets = set()  # Множество для хранения всех активных сокетов передачи
+        # Словарь активных передач для мониторинга их статуса.
+        # Ключ – адрес клиента, значение – словарь с информацией о передаче.
+        self.active_transfers = {}
+
+    def set_log_callback(self, callback):
+        """
+        Установка функции обратного вызова для логирования сообщений.
+
+        :param callback: Функция, принимающая строку сообщения.
+        """
+        self.log_callback = callback
+
+    def log(self, message):
+        """
+        Функция логирования: вызывает callback (если задан) или выводит сообщение в консоль.
+        
+        :param message: Строка с сообщением для логирования.
+        """
+        if self.log_callback:
+            self.log_callback(message)
+        else:
+            print(message)
+
+    def start_server(self, ip, port):
+        """
+        Запуск TFTP сервера на указанном IP и порту.
+
+        :param ip: IP-адрес для привязки сервера.
+        :param port: Порт для TFTP сервера.
+        """
+        if self.running:
+            self.log("[WARN] Сервер уже запущен")
+            return
+
+        self.running = True
+        try:
+            self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            self.server_socket.bind((ip, port))
+            self.log(f"[INFO] TFTP сервер запущен на {ip}:{port}")
+            
+            while self.running:
+                try:
+                    self.server_socket.settimeout(1.0)
+                    data, client_addr = self.server_socket.recvfrom(2048)
+                    if data and self.running:
+                        threading.Thread(target=self.handle_request, args=(data, client_addr), daemon=True).start()
+                except socket.timeout:
+                    continue
+                except socket.error as e:
+                    if self.running:  # Логируем ошибку только если сервер еще запущен
+                        self.log(f"[ERROR] Ошибка получения данных: {str(e)}")
+                    break
+                except Exception as e:
+                    if self.running:  # Логируем ошибку только если сервер еще запущен
+                        self.log(f"[ERROR] Ошибка получения данных: {str(e)}")
+                    break
+        except Exception as e:
+            self.log(f"[ERROR] Ошибка запуска сервера: {str(e)}")
+        finally:
+            self.running = False
+            if self.server_socket:
+                try:
+                    self.server_socket.close()
+                except:
+                    pass
+                self.server_socket = None
+
+    def stop_server(self):
+        """
+        Остановка TFTP сервера.
+        """
+        if not self.running:
+            return
+
+        self.log("[INFO] Остановка TFTP сервера...")
+        self.running = False
+
+        try:
+            # Закрываем основной сокет сервера первым
+            if self.server_socket:
+                try:
+                    # Создаем временный сокет и отправляем пакет самому себе,
+                    # чтобы разблокировать recvfrom
+                    temp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+                    try:
+                        server_address = self.server_socket.getsockname()
+                        temp_socket.sendto(b'', server_address)
+                    except:
+                        pass
+                    finally:
+                        try:
+                            temp_socket.close()
+                        except:
+                            pass
+                    
+                    try:
+                        self.server_socket.close()
+                    except Exception as e:
+                        self.log(f"[WARN] Ошибка при закрытии основного сокета: {str(e)}")
+                finally:
+                    self.server_socket = None
+
+            # Закрываем все активные сокеты передачи
+            with self.lock:
+                active_sockets = list(self.transfer_sockets)
+                self.transfer_sockets.clear()
+                active_transfers = dict(self.active_transfers)
+                self.active_transfers.clear()
+
+            # Закрываем сокеты передачи после очистки множества
+            for sock in active_sockets:
+                try:
+                    if sock:
+                        sock.close()
+                except Exception as e:
+                    self.log(f"[WARN] Ошибка при закрытии сокета передачи: {str(e)}")
+
+            # Отправляем сообщения об остановке для активных передач
+            for client_addr, transfer_info in active_transfers.items():
+                try:
+                    self.send_error(client_addr, 0, "Сервер остановлен")
+                except:
+                    pass
+
+        except Exception as e:
+            self.log(f"[ERROR] Ошибка при остановке сервера: {str(e)}")
+        finally:
+            self.running = False  # Гарантируем, что флаг running будет False
+            self.log("[INFO] TFTP сервер остановлен")
+
+    def handle_request(self, data, client_addr):
+        """
+        Обработка входящего запроса от клиента.
+
+        :param data: Полученные данные (UDP-пакет).
+        :param client_addr: Адрес клиента, отправившего пакет.
+        """
+        if len(data) < 2:
+            self.log(f"[WARN] Получен некорректный пакет от {client_addr}")
+            return
+        opcode = struct.unpack("!H", data[:2])[0]
+        if opcode == 1:  # RRQ (Read Request) – запрос на чтение файла
+            self.handle_rrq(data, client_addr)
+        else:
+            self.log(f"[WARN] Неподдерживаемый запрос (опкод {opcode}) от {client_addr}")
+
+    def handle_rrq(self, data, client_addr):
+        """
+        Обработка запроса на чтение файла (RRQ).
+
+        :param data: Данные запроса.
+        :param client_addr: Адрес клиента.
+        """
+        try:
+            # RRQ формата: 2 байта опкода, затем строка имени файла, за которой следует 0,
+            # затем строка режима (например, "octet"), и завершается 0.
+            parts = data[2:].split(b'\0')
+            if len(parts) < 2:
+                self.log(f"[WARN] Некорректный RRQ пакет от {client_addr}")
+                return
+            filename = parts[0].decode('utf-8')
+            mode = parts[1].decode('utf-8').lower()
+            self.log(f"[INFO] Получен RRQ от {client_addr}: файл '{filename}', режим '{mode}'")
+            if mode != "octet":
+                self.send_error(client_addr, 0, "Поддерживается только octet режим")
+                return
+            file_path = os.path.join(self.share_folder, filename)
+            if not os.path.isfile(file_path):
+                self.send_error(client_addr, 1, "Файл не найден")
+                return
+            # Запускаем передачу файла в новом потоке.
+            threading.Thread(target=self.send_file, args=(file_path, client_addr), daemon=True).start()
+        except Exception as e:
+            self.log(f"[ERROR] Ошибка обработки RRQ: {str(e)}")
+
+    def send_error(self, client_addr, error_code, error_message):
+        """
+        Отправка сообщения об ошибке клиенту.
+
+        :param client_addr: Адрес клиента.
+        :param error_code: Код ошибки.
+        :param error_message: Текст ошибки.
+        """
+        # Формируем TFTP пакет ошибки: 2 байта опкода (5), 2 байта кода ошибки, сообщение об ошибке и завершающий 0.
+        packet = struct.pack("!HH", 5, error_code) + error_message.encode('utf-8') + b'\0'
+        self.server_socket.sendto(packet, client_addr)
+        self.log(f"[INFO] Отправлено сообщение об ошибке '{error_message}' клиенту {client_addr}")
+
+    def send_file(self, file_path, client_addr):
+        """
+        Передача файла клиенту по протоколу TFTP.
+        """
+        BLOCK_SIZE = 512
+        MAX_RETRIES = 5
+        TIMEOUT = 2.0
+        transfer_socket = None
+        try:
+            if not os.path.exists(file_path):
+                self.log(f"[ERROR] Файл '{file_path}' не существует")
+                self.send_error(client_addr, 1, "Файл не найден")
+                return
+
+            filesize = os.path.getsize(file_path)
+            if filesize == 0:
+                self.log(f"[ERROR] Файл '{file_path}' пуст")
+                self.send_error(client_addr, 0, "Файл пуст")
+                return
+
+            start_time = time.time()
+            file_basename = os.path.basename(file_path)
+            
+            # Регистрируем активную передачу
+            with self.lock:
+                self.active_transfers[client_addr] = {
+                    'filename': file_basename,
+                    'filesize': filesize,
+                    'bytes_sent': 0,
+                    'start_time': start_time
+                }
+            
+            # Создаем новый сокет для передачи данных
+            transfer_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            transfer_socket.settimeout(TIMEOUT)
+            
+            with self.lock:
+                self.transfer_sockets.add(transfer_socket)
+            
+            self.log(f"[INFO] Начало передачи файла '{file_basename}' клиенту {client_addr}. Размер файла: {filesize} байт.")
+
+            with open(file_path, 'rb') as file:
+                block_number = 1
+                last_successful_block = 0
+                
+                while True:
+                    # Читаем блок данных
+                    data = file.read(BLOCK_SIZE)
+                    
+                    # Формируем и отправляем пакет данных
+                    packet = struct.pack('!HH', 3, block_number) + data
+                    
+                    retries = 0
+                    while retries < MAX_RETRIES:
+                        try:
+                            transfer_socket.sendto(packet, client_addr)
+                            
+                            # Ожидаем подтверждение
+                            while True:
+                                try:
+                                    ack_data, ack_addr = transfer_socket.recvfrom(4)
+                                    if ack_addr == client_addr and len(ack_data) >= 4:
+                                        opcode, ack_block = struct.unpack('!HH', ack_data)
+                                        if opcode == 4:  # ACK
+                                            if ack_block == block_number:
+                                                # Успешное подтверждение
+                                                last_successful_block = block_number
+                                                bytes_sent = min((block_number * BLOCK_SIZE), filesize)
+                                                
+                                                # Обновляем информацию о прогрессе
+                                                with self.lock:
+                                                    if client_addr in self.active_transfers:
+                                                        self.active_transfers[client_addr]['bytes_sent'] = bytes_sent
+                                                
+                                                # Логируем статус каждую секунду
+                                                current_time = time.time()
+                                                if current_time - start_time >= 1.0:
+                                                    bytes_remaining = filesize - bytes_sent
+                                                    elapsed_time = current_time - start_time
+                                                    self.log(f"[STATUS] Клиент: {client_addr} | Файл: {file_basename} | "
+                                                           f"Отправлено: {bytes_sent}/{filesize} байт | "
+                                                           f"Осталось: {bytes_remaining} байт | "
+                                                           f"Время: {elapsed_time:.2f} сек.")
+                                                
+                                                break
+                                            elif ack_block < block_number:
+                                                # Получен старый ACK, игнорируем
+                                                continue
+                                except socket.timeout:
+                                    break
+                            
+                            if last_successful_block == block_number:
+                                break
+                            else:
+                                retries += 1
+                                self.log(f"[WARN] Таймаут ожидания ACK для блока {block_number} от {client_addr}. "
+                                       f"Попытка {retries + 1}.")
+                        except Exception as e:
+                            retries += 1
+                            self.log(f"[ERROR] Ошибка при передаче блока {block_number}: {str(e)}")
+                    
+                    if retries >= MAX_RETRIES:
+                        self.log(f"[ERROR] Превышено максимальное количество попыток для блока {block_number}")
+                        return
+                    
+                    block_number += 1
+                    
+                    # Если отправили меньше BLOCK_SIZE байт, это был последний блок
+                    if len(data) < BLOCK_SIZE:
+                        break
+
+            self.log(f"[INFO] Передача файла '{file_basename}' клиенту {client_addr} завершена успешно")
+
+        except Exception as e:
+            self.log(f"[ERROR] Ошибка при передаче файла: {str(e)}")
+        finally:
+            # Очищаем информацию о передаче
+            with self.lock:
+                if client_addr in self.active_transfers:
+                    del self.active_transfers[client_addr]
+                if transfer_socket in self.transfer_sockets:
+                    self.transfer_sockets.remove(transfer_socket)
+            
+            if transfer_socket:
+                try:
+                    transfer_socket.close()
+                except:
+                    pass 

about_window.py

@@ -2,19 +2,22 @@
 # -*- coding: utf-8 -*-
 
 import tkinter as tk
-from tkinter import ttk, BOTH, X, BOTTOM
+from tkinter import ttk, BOTH, X, BOTTOM, END
 import webbrowser
 
 class AboutWindow(tk.Toplevel):
     def __init__(self, parent):
         super().__init__(parent)
         self.title("О программе")
-        self.geometry("400x300")
+        self.geometry("600x500")
         self.resizable(False, False)
         
-        # Создаем фрейм
+        # Сохраняем ссылку на родительское окно
+        self.parent = parent
+        
+        # Создаем фрейм для содержимого
         about_frame = ttk.Frame(self, padding="20")
-        about_frame.pack(fill=BOTH, expand=True)
+        about_frame.pack(fill=BOTH, expand=True, padx=10, pady=10)
         
         # Заголовок
         ttk.Label(
@@ -33,7 +36,7 @@ class AboutWindow(tk.Toplevel):
         # Версия
         ttk.Label(
             about_frame,
-            text="Версия 1.0",
+            text=f"Версия {getattr(parent, 'VERSION', '1.0.0')}",
             font=("Segoe UI", 10)
         ).pack(pady=(0, 20))
         
@@ -66,7 +69,7 @@ class AboutWindow(tk.Toplevel):
         
         ttk.Label(
             contact_frame,
-            text="Email: LowaWorkMail@gmail.com"
+            text="Email: SPRF555@gmail.com"
         ).pack(anchor="w")
         
         telegram_link = ttk.Label(
@@ -80,14 +83,14 @@ class AboutWindow(tk.Toplevel):
         
         # Кнопка закрытия
         ttk.Button(
-            about_frame,
+            self,
             text="Закрыть",
             command=self.destroy
-        ).pack(side=BOTTOM, pady=(20, 0))
+        ).pack(side=BOTTOM, pady=10)
         
         # Центрируем окно
         self.center_window()
-        
+    
     def center_window(self):
         self.update_idletasks()
         width = self.winfo_width()
@@ -95,6 +98,6 @@ class AboutWindow(tk.Toplevel):
         x = (self.winfo_screenwidth() // 2) - (width // 2)
         y = (self.winfo_screenheight() // 2) - (height // 2)
         self.geometry(f"{width}x{height}+{x}+{y}")
-        
+    
     def open_url(self, url):
         webbrowser.open(url) 

requirements.txt

@@ -0,0 +1,4 @@
+tftpy>=0.8.0
+pyserial>=3.5
+requests>=2.31.0
+packaging>=23.2 

update_checker.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import logging
+import requests
+import threading
+import re
+from packaging import version
+import xml.etree.ElementTree as ET
+import html
+
+class UpdateCheckError(Exception):
+    """Исключение для ошибок проверки обновлений"""
+    pass
+
+class ReleaseType:
+    """Типы релизов"""
+    STABLE = "stable"
+    PRERELEASE = "prerelease"
+
+class UpdateChecker:
+    """Класс для проверки обновлений программы"""
+    
+    def __init__(self, current_version, repo_url, include_prereleases=False):
+        self.current_version = current_version
+        self.repo_url = repo_url
+        self.include_prereleases = include_prereleases
+        self.rss_url = f"{repo_url}/releases.rss"
+        self.release_info = None
+
+    def _clean_html(self, html_text):
+        """Очищает HTML-разметку и форматирует текст"""
+        if not html_text:
+            return ""
+        text = re.sub(r'<[^>]+>', '', html_text)
+        text = html.unescape(text)
+        text = re.sub(r'\n\s*\n', '\n\n', text)
+        return '\n'.join(line.strip() for line in text.splitlines()).strip()
+
+    def _parse_release_info(self, item):
+        """Извлекает информацию о релизе из RSS item"""
+        title = item.find('title').text if item.find('title') is not None else ''
+        link = item.find('link').text if item.find('link') is not None else ''
+        description = item.find('description').text if item.find('description') is not None else ''
+        content = item.find('{http://purl.org/rss/1.0/modules/content/}encoded')
+        content_text = content.text if content is not None else ''
+        
+        # Извлекаем версию и проверяем тип релиза из тега
+        version_match = re.search(r'/releases/tag/(?:pre-)?v?(\d+\.\d+(?:\.\d+)?)', link)
+        if not version_match:
+            return None
+            
+        version_str = version_match.group(1)
+        # Проверяем наличие префикса pre- в теге
+        is_prerelease = 'pre-' in link.lower()
+        
+        # Форматируем название релиза
+        formatted_title = title
+        if title == version_str or not title.strip():
+            # Если заголовок пустой или совпадает с версией, создаем стандартное название
+            release_type = "Пре-релиз" if is_prerelease else "Версия"
+            formatted_title = f"{release_type} {version_str}"
+        elif not re.search(version_str, title):
+            # Если версия не указана в заголовке, добавляем её
+            formatted_title = f"{title} ({version_str})"
+            
+        # Форматируем описание
+        formatted_description = self._clean_html(content_text or description)
+        if not formatted_description.strip():
+            formatted_description = "Нет описания"
+            
+        # Добавляем метку типа релиза в начало описания
+        release_type_label = "[Пре-релиз] " if is_prerelease else ""
+        formatted_description = f"{release_type_label}{formatted_description}"
+        
+        return {
+            'title': formatted_title,
+            'link': link,
+            'description': formatted_description,
+            'version': version_str,
+            'type': ReleaseType.PRERELEASE if is_prerelease else ReleaseType.STABLE
+        }
+
+    def check_updates(self, callback=None):
+        """Проверяет наличие обновлений в асинхронном режиме"""
+        def check_worker():
+            try:
+                logging.info(f"Текущая версия программы: {self.current_version}")
+                logging.info(f"Проверка пре-релизов: {self.include_prereleases}")
+                logging.info(f"Запрос RSS ленты: {self.rss_url}")
+                
+                response = requests.get(self.rss_url, timeout=10)
+                response.raise_for_status()
+                
+                root = ET.fromstring(response.content)
+                items = root.findall('.//item')
+                if not items:
+                    raise UpdateCheckError("Релизы не найдены")
+                
+                logging.info(f"Найдено {len(items)} релизов")
+                
+                latest_version = None
+                latest_info = None
+                
+                for item in items:
+                    release_info = self._parse_release_info(item)
+                    if not release_info:
+                        continue
+                        
+                    is_prerelease = release_info['type'] == ReleaseType.PRERELEASE
+                    logging.info(
+                        f"Проверка релиза: {release_info['title']}, "
+                        f"версия: {release_info['version']}, "
+                        f"тип: {'пре-релиз' if is_prerelease else 'стабильная'}"
+                    )
+                    
+                    # Пропускаем пре-релизы если они не включены
+                    if is_prerelease and not self.include_prereleases:
+                        logging.info(f"Пропуск пре-релиза: {release_info['version']}")
+                        continue
+                    
+                    # Сравниваем версии
+                    try:
+                        current_ver = version.parse(latest_version or "0.0.0")
+                        new_ver = version.parse(release_info['version'].split('-')[0])  # Убираем суффикс для сравнения
+                        
+                        if new_ver > current_ver:
+                            latest_version = release_info['version']
+                            latest_info = release_info
+                            logging.info(f"Новая версия: {latest_version}")
+                            
+                    except version.InvalidVersion as e:
+                        logging.warning(f"Некорректный формат версии {release_info['version']}: {e}")
+                        continue
+                
+                if not latest_info:
+                    raise UpdateCheckError("Не найдены подходящие версии")
+                
+                self.release_info = latest_info
+                
+                # Сравниваем с текущей версией
+                current_ver = version.parse(self.current_version)
+                latest_ver = version.parse(latest_version.split('-')[0])
+                update_available = latest_ver > current_ver
+                
+                logging.info(f"Сравнение версий: текущая {current_ver} <-> последняя {latest_ver}")
+                logging.info(f"Доступно обновление: {update_available}")
+                
+                if callback:
+                    callback(update_available, None)
+                    
+            except UpdateCheckError as e:
+                logging.error(str(e))
+                if callback:
+                    callback(False, str(e))
+            except Exception as e:
+                logging.error(f"Ошибка при проверке обновлений: {e}", exc_info=True)
+                if callback:
+                    callback(False, str(e))
+
+        threading.Thread(target=check_worker, daemon=True).start()
+    
+    def get_release_notes(self):
+        """Возвращает информацию о последнем релизе"""
+        return self.release_info