#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Пример использования PrepressLab Web API с графическим интерфейсом Требования: - Python 3.9+ - PySide6: pip install PySide6 - requests: pip install requests """ import sys import os import time import zipfile import threading import tempfile from typing import Optional try: from PySide6.QtCore import Qt, QThread, Signal from PySide6.QtGui import QPainter, QPixmap from PySide6.QtWidgets import ( QApplication, QMainWindow, QWidget, QTabWidget, QVBoxLayout, QHBoxLayout, QFormLayout, QLabel, QLineEdit, QPushButton, QFileDialog, QMessageBox, QTextEdit, QComboBox, ) except ImportError: print("Ошибка: требуется установить PySide6") print("Выполните: pip install PySide6") sys.exit(1) import requests APP_PUBLIC_NAME = "PrepressLab API Client" def resource_path(relative: str = "") -> str: """Возвращает путь к ресурсу""" if getattr(sys, "frozen", False): base = sys._MEIPASS # type: ignore[attr-defined] else: base = os.path.dirname(os.path.abspath(__file__)) return os.path.join(base, relative) def show_error(parent, text: str): QMessageBox.critical(parent, "Ошибка", text) def show_info(parent, text: str): QMessageBox.information(parent, "Информация", text) # ======= Виджет рабочего стола с фоном ======= class DesktopWidget(QWidget): """Вкладка 'Рабочий стол' с фоновым изображением и заголовком.""" def __init__(self, parent=None): super().__init__(parent) # ищем background.png или background.jpg self.background: Optional[QPixmap] = None for name in ("background.png", "background.jpg"): path = resource_path(name) if os.path.isfile(path): self.background = QPixmap(path) break self.main_layout = QVBoxLayout(self) self.main_layout.setContentsMargins(0, 20, 0, 20) self.main_layout.setSpacing(15) # заголовок title_label = QLabel(APP_PUBLIC_NAME) title_label.setAlignment(Qt.AlignHCenter | Qt.AlignTop) font = title_label.font() font.setPointSize(22) font.setBold(True) title_label.setFont(font) title_label.setStyleSheet("color: white; background: transparent;") self.main_layout.addWidget(title_label) self.main_layout.addSpacing(35) def paintEvent(self, event): super().paintEvent(event) if self.background: painter = QPainter(self) painter.drawPixmap(0, 0, self.width(), self.height(), self.background) # ======= Поток для обработки задачи ======= class JobThread(QThread): """Поток для выполнения задачи API в фоне""" status_update = Signal(str) finished_signal = Signal(bool, str) def __init__(self, api_url, token, mode, folder_path, output_path): super().__init__() self.api_url = api_url self.token = token self.mode = mode self.folder_path = folder_path self.output_path = output_path self._stop = False def stop(self): self._stop = True def run(self): try: # 1. Создание ZIP self.status_update.emit("Создание ZIP архива...") zip_path = self.create_zip_from_folder(self.folder_path) if self._stop: return # 2. Создание задачи self.status_update.emit("Создание задачи на сервере...") job_id = self.create_job(zip_path) if self._stop: return self.status_update.emit(f"Задача создана: {job_id}") # 3. Ожидание завершения self.status_update.emit("Ожидание обработки...") result_name = self.wait_for_job(job_id) if self._stop: return self.status_update.emit(f"Обработка завершена: {result_name}") # 4. Скачивание результата self.status_update.emit("Скачивание результата...") self.download_result(job_id, self.output_path) if self._stop: return self.status_update.emit("Готово! Результат сохранён.") self.finished_signal.emit(True, f"Результат сохранён в: {self.output_path}") except Exception as e: self.finished_signal.emit(False, str(e)) def create_zip_from_folder(self, folder_path): """Создаёт ZIP-архив из папки с файлами""" # Используем временный файл в системной папке temp для избежания проблем с кодировкой temp_dir = tempfile.gettempdir() zip_path = os.path.join(temp_dir, "prepresslab_input.zip") # Если файл уже существует, удаляем его if os.path.exists(zip_path): try: os.remove(zip_path) except: pass try: with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zipf: for root, dirs, files in os.walk(folder_path): for file in files: file_path = os.path.join(root, file) # Используем относительный путь для избежания проблем с кодировкой try: arcname = os.path.relpath(file_path, folder_path) # Нормализуем путь для кроссплатформенности arcname = arcname.replace('\\', '/') zipf.write(file_path, arcname) except Exception as e: # Пропускаем файлы с проблемами кодировки self.status_update.emit(f"Пропущен файл (проблема кодировки): {file}") continue except Exception as e: raise Exception(f"Ошибка создания ZIP архива: {e}") return zip_path def create_job(self, zip_path): """Создаёт задачу обработки""" url = f"{self.api_url}/api/jobs" params = {"mode": self.mode} # Исправление проблемы с кодировкой: используем правильную кодировку для всех строк auth_header = f"Bearer {self.token}" headers = { "Authorization": auth_header, "Content-Type": "application/zip" } # Открываем файл в бинарном режиме и отправляем try: with open(zip_path, "rb") as f: response = requests.post( url, params=params, headers=headers, data=f, timeout=300 # Таймаут для больших файлов ) except UnicodeEncodeError as e: # Если проблема с кодировкой в пути к файлу raise Exception(f"Ошибка кодировки в пути к файлу: {e}. Убедитесь, что путь не содержит специальных символов.") if response.status_code == 200: return response.json()["id"] else: error_text = response.text try: # Пытаемся декодировать ответ if isinstance(error_text, bytes): error_text = error_text.decode('utf-8', errors='ignore') except: pass raise Exception(f"Ошибка создания задачи: {response.status_code} - {error_text}") def wait_for_job(self, job_id): """Ожидает завершения задачи""" url = f"{self.api_url}/api/jobs/{job_id}/status" auth_header = f"Bearer {self.token}" headers = {"Authorization": auth_header} while not self._stop: response = requests.get(url, headers=headers) job = response.json() status = job["status"] self.status_update.emit(f"Статус: {status}") if status == "done": return job.get("result_name", "result.zip") elif status == "error": error_msg = job.get("error", "Неизвестная ошибка") raise Exception(f"Ошибка обработки: {error_msg}") time.sleep(2) def download_result(self, job_id, output_path): """Скачивает результат обработки""" url = f"{self.api_url}/api/jobs/{job_id}/download" auth_header = f"Bearer {self.token}" headers = {"Authorization": auth_header} response = requests.get(url, headers=headers, stream=True) if response.status_code == 200: with open(output_path, "wb") as f: for chunk in response.iter_content(chunk_size=8192): if self._stop: return f.write(chunk) return True else: raise Exception(f"Ошибка скачивания: {response.status_code} - {response.text}") # ======================= Главное окно ======================= class MainWindow(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle(f"{APP_PUBLIC_NAME}") self.resize(900, 600) self.job_thread: Optional[JobThread] = None self.tabs = QTabWidget() self.setCentralWidget(self.tabs) self.desktop_tab = self._create_desktop_tab() self.settings_tab = self._create_settings_tab() self.tabs.addTab(self.desktop_tab, "Рабочий стол") self.tabs.addTab(self.settings_tab, "Настройки") # ================== ВКЛАДКА 1: РАБОЧИЙ СТОЛ ================== def _create_desktop_tab(self) -> QWidget: w = DesktopWidget() layout = w.main_layout # Поля ввода form_layout = QFormLayout() # Токен self.edit_token = QLineEdit() self.edit_token.setPlaceholderText("Введите токен API") self.edit_token.setEchoMode(QLineEdit.Password) form_layout.addRow("Токен API:", self.edit_token) # Папка с файлами self.edit_folder = QLineEdit() self.edit_folder.setPlaceholderText("Выберите папку с файлами для обработки") btn_folder_browse = QPushButton("…") btn_folder_browse.setFixedWidth(32) btn_folder_browse.clicked.connect(self.on_browse_folder) row_folder = QHBoxLayout() row_folder.addWidget(self.edit_folder) row_folder.addWidget(btn_folder_browse) form_layout.addRow("Папка с файлами:", row_folder) # Режим обработки self.combo_mode = QComboBox() self.combo_mode.addItems([ "convert", "layout_plotter", "layout_rezak", "layout_card" ]) form_layout.addRow("Режим обработки:", self.combo_mode) # Папка для результата self.edit_output = QLineEdit() self.edit_output.setPlaceholderText("Выберите папку для сохранения результата") btn_output_browse = QPushButton("…") btn_output_browse.setFixedWidth(32) btn_output_browse.clicked.connect(self.on_browse_output) row_output = QHBoxLayout() row_output.addWidget(self.edit_output) row_output.addWidget(btn_output_browse) form_layout.addRow("Папка для результата:", row_output) layout.addLayout(form_layout) layout.addSpacing(20) # Кнопка запуска def add_centered_button(text: str, slot, width: int = 260): btn = QPushButton(text) btn.setFixedWidth(width) row = QHBoxLayout() row.addStretch() row.addWidget(btn) row.addStretch() layout.addLayout(row) btn.clicked.connect(slot) return btn self.btn_start = add_centered_button( "Начать обработку", self.on_start_clicked, ) self.btn_stop = add_centered_button( "Остановить", self.on_stop_clicked, ) self.btn_stop.setEnabled(False) layout.addSpacing(20) # Логи log_label = QLabel("Лог выполнения:") layout.addWidget(log_label) self.log_text = QTextEdit() self.log_text.setReadOnly(True) self.log_text.setMaximumHeight(200) layout.addWidget(self.log_text) layout.addStretch() return w def on_browse_folder(self): folder = QFileDialog.getExistingDirectory(self, "Выберите папку с файлами") if folder: self.edit_folder.setText(folder) def on_browse_output(self): folder = QFileDialog.getExistingDirectory(self, "Выберите папку для результата") if folder: self.edit_output.setText(folder) def on_start_clicked(self): token = self.edit_token.text().strip() folder = self.edit_folder.text().strip() mode = self.combo_mode.currentText() output_dir = self.edit_output.text().strip() if not token: show_error(self, "Введите токен API") return if not folder or not os.path.isdir(folder): show_error(self, "Выберите папку с файлами") return if not output_dir: show_error(self, "Выберите папку для результата") return # Определяем URL API api_url = self.edit_api_url.text().strip() if hasattr(self, 'edit_api_url') else "https://apl.alsion.su" if not api_url: api_url = "https://apl.alsion.su" output_path = os.path.join(output_dir, "result.zip") self.log_text.clear() self.log_text.append("Начало обработки...") self.log_text.append(f"Папка: {folder}") self.log_text.append(f"Режим: {mode}") self.log_text.append("") self.btn_start.setEnabled(False) self.btn_stop.setEnabled(True) # Запуск в отдельном потоке self.job_thread = JobThread(api_url, token, mode, folder, output_path) self.job_thread.status_update.connect(self.on_status_update) self.job_thread.finished_signal.connect(self.on_job_finished) self.job_thread.start() def on_stop_clicked(self): if self.job_thread: self.job_thread.stop() self.log_text.append("Остановка...") def on_status_update(self, message: str): self.log_text.append(message) def on_job_finished(self, success: bool, message: str): self.btn_start.setEnabled(True) self.btn_stop.setEnabled(False) if success: show_info(self, message) self.log_text.append("✓ Успешно завершено!") else: show_error(self, message) self.log_text.append(f"✗ Ошибка: {message}") # ================== ВКЛАДКА 2: НАСТРОЙКИ ================== def _create_settings_tab(self) -> QWidget: w = QWidget() layout = QVBoxLayout(w) form = QFormLayout() # URL API self.edit_api_url = QLineEdit("https://apl.alsion.su") form.addRow("URL API:", self.edit_api_url) layout.addLayout(form) layout.addStretch() # Информация info_label = QLabel( "Для использования API необходимо:\n" "1. Получить токен у администратора\n" "2. Выбрать папку с файлами для обработки\n" "3. Выбрать режим обработки\n" "4. Указать папку для сохранения результата\n" "5. Нажать 'Начать обработку'" ) info_label.setWordWrap(True) layout.addWidget(info_label) return w def main(): app = QApplication(sys.argv) app.setStyle("Fusion") # Современный стиль window = MainWindow() window.show() sys.exit(app.exec()) if __name__ == "__main__": main()