LowaSC/ComConfigCopy
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Remove standalone modules after application refactoring

Browse Source 
- Delete about_window.py, ComConfigCopy.py, TFTPServer.py, and update_checker.py
- These modules have been integrated into the main application structure
- Cleanup of redundant files following previous refactoring efforts
This commit is contained in:
Lowa
2025-02-16 23:54:59 +03:00
parent cb5329ddb7
commit 57d173e00e
4 changed files with 0 additions and 1901 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1274
ComConfigCopy.py
View File

File diff suppressed because it is too large Load Diff

349
TFTPServer.py
View File

@@ -1,349 +0,0 @@
#!/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

103
about_window.py
View File

@@ -1,103 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import tkinter as tk
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("600x500")
self.resizable(False, False)
# Сохраняем ссылку на родительское окно
self.parent = parent
# Создаем фрейм для содержимого
about_frame = ttk.Frame(self, padding="20")
about_frame.pack(fill=BOTH, expand=True, padx=10, pady=10)
# Заголовок
ttk.Label(
about_frame,
text="ComConfigCopy",
font=("Segoe UI", 16, "bold")
).pack(pady=(0, 10))
# Описание
ttk.Label(
about_frame,
text="Программа для копирования конфигураций на коммутаторы",
wraplength=350
).pack(pady=(0, 20))
# Версия
ttk.Label(
about_frame,
text=f"Версия {getattr(parent, 'VERSION', '1.0.0')}",
font=("Segoe UI", 10)
).pack(pady=(0, 20))
# Контактная информация
contact_frame = ttk.Frame(about_frame)
contact_frame.pack(fill=X, pady=(0, 20))
# Исходный код
ttk.Label(
contact_frame,
text="Исходный код:",
font=("Segoe UI", 10, "bold")
).pack(anchor="w")
source_link = ttk.Label(
contact_frame,
text="https://gitea.filow.ru/LowaSC/ComConfigCopy",
cursor="hand2",
foreground="blue"
)
source_link.pack(anchor="w")
source_link.bind("<Button-1>", lambda e: self.open_url("https://gitea.filow.ru/LowaSC/ComConfigCopy"))
# Контакты
ttk.Label(
contact_frame,
text="\nКонтакты:",
font=("Segoe UI", 10, "bold")
).pack(anchor="w")
ttk.Label(
contact_frame,
text="Email: LowaWorkMail@gmail.com"
).pack(anchor="w")
telegram_link = ttk.Label(
contact_frame,
text="Telegram: @LowaSC",
cursor="hand2",
foreground="blue"
)
telegram_link.pack(anchor="w")
telegram_link.bind("<Button-1>", lambda e: self.open_url("https://t.me/LowaSC"))
# Кнопка закрытия
ttk.Button(
self,
text="Закрыть",
command=self.destroy
).pack(side=BOTTOM, pady=10)
# Центрируем окно
self.center_window()
def center_window(self):
self.update_idletasks()
width = self.winfo_width()
height = self.winfo_height()
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)

175
update_checker.py
View File

@@ -1,175 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import json
import logging
import requests
import threading
from packaging import version
class UpdateCheckError(Exception):
"""Исключение для ошибок проверки обновлений"""
pass
class UpdateChecker:
"""Класс для проверки обновлений программы"""
def __init__(self, current_version, repo_url):
self.current_version = current_version
self.repo_url = repo_url
# Формируем базовый URL API
self.api_url = repo_url.replace("gitea.filow.ru", "gitea.filow.ru/api/v1/repos/LowaSC/ComConfigCopy")
self._update_available = False
self._latest_version = None
self._latest_release = None
self._error = None
self._changelog = None
def get_changelog(self, callback=None):
"""
Получение changelog из репозитория.
:param callback: Функция обратного вызова, которая будет вызвана после получения changelog
"""
def fetch():
try:
# Пытаемся получить CHANGELOG.md из репозитория
response = requests.get(f"{self.api_url}/contents/CHANGELOG.md", timeout=10)
response.raise_for_status()
content = response.json()
if "content" in content:
import base64
changelog_content = base64.b64decode(content["content"]).decode("utf-8")
self._changelog = changelog_content
self._error = None
else:
raise UpdateCheckError("Не удалось получить содержимое CHANGELOG.md")
except requests.RequestException as e:
error_msg = f"Ошибка получения changelog: {e}"
logging.error(error_msg, exc_info=True)
self._error = error_msg
self._changelog = None
except Exception as e:
error_msg = f"Неизвестная ошибка при получении changelog: {e}"
logging.error(error_msg, exc_info=True)
self._error = error_msg
self._changelog = None
finally:
if callback:
callback(self._changelog, self._error)
# Запускаем получение в отдельном потоке
threading.Thread(target=fetch, daemon=True).start()
def check_updates(self, callback=None):
"""
Проверка наличия обновлений.
:param callback: Функция обратного вызова, которая будет вызвана после проверки
"""
def check():
try:
response = requests.get(f"{self.api_url}/releases", timeout=10)
response.raise_for_status()
releases = response.json()
if not releases:
raise UpdateCheckError("Не найдено релизов в репозитории")
latest_release = releases[0]
latest_version = latest_release.get("tag_name", "").lstrip("v")
if not latest_version:
raise UpdateCheckError("Не удалось определить версию последнего релиза")
try:
if version.parse(latest_version) > version.parse(self.current_version):
self._update_available = True
self._latest_version = latest_version
self._latest_release = latest_release
logging.info(f"Доступно обновление: {latest_version}")
else:
logging.info("Обновления не требуются")
except version.InvalidVersion as e:
raise UpdateCheckError(f"Некорректный формат версии: {e}")
self._error = None
except requests.RequestException as e:
error_msg = f"Ошибка сетевого подключения: {e}"
logging.error(error_msg, exc_info=True)
self._error = error_msg
except UpdateCheckError as e:
logging.error(str(e), exc_info=True)
self._error = str(e)
except Exception as e:
error_msg = f"Неизвестная ошибка при проверке обновлений: {e}"
logging.error(error_msg, exc_info=True)
self._error = error_msg
finally:
if callback:
callback(self._update_available, self._error)
@property
def update_available(self):
"""Доступно ли обновление"""
return self._update_available
@property
def latest_version(self):
"""Последняя доступная версия"""
return self._latest_version
@property
def error(self):
"""Последняя ошибка при проверке обновлений"""
return self._error
@property
def changelog(self):
"""Текущий changelog"""
return self._changelog
def get_release_notes(self):
"""Получение информации о последнем релизе"""
if self._latest_release:
return {
"version": self._latest_version,
"description": self._latest_release.get("body", ""),
"download_url": self._latest_release.get("assets", [{}])[0].get("browser_download_url", "")
}
return None
def get_releases(self, callback=None):
"""
Получение списка релизов из репозитория.
:param callback: Функция обратного вызова, которая будет вызвана после получения списка релизов
"""
def fetch():
try:
response = requests.get(f"{self.api_url}/releases", timeout=10)
response.raise_for_status()
releases = response.json()
if not releases:
raise UpdateCheckError("Не найдено релизов в репозитории")
self._error = None
if callback:
callback(releases, None)
except requests.RequestException as e:
error_msg = f"Ошибка сетевого подключения: {e}"
logging.error(error_msg, exc_info=True)
self._error = error_msg
if callback:
callback(None, error_msg)
except Exception as e:
error_msg = f"Ошибка при получении списка релизов: {e}"
logging.error(error_msg, exc_info=True)
self._error = error_msg
if callback:
callback(None, error_msg)
# Запускаем получение в отдельном потоке
threading.Thread(target=fetch, daemon=True).start()