diff --git a/Upgrade b/Upgrade new file mode 100644 index 0000000..0f357b8 --- /dev/null +++ b/Upgrade @@ -0,0 +1,309 @@ +import sys +import json +import os +import logging +import time +from datetime import timedelta +from typing import Dict, Optional +from cryptography.fernet import Fernet +from PyQt5.QtWidgets import (QApplication, QWidget, QVBoxLayout, QLabel, + QLineEdit, QPushButton, QMessageBox, QInputDialog) +from PyQt5.QtCore import QThread, pyqtSignal, QObject +from PyQt5.QtGui import QClipboard +from exchangelib import Q, DELEGATE, Account, Credentials, Configuration, CalendarItem, EWSDateTime, EWSTimeZone +from mattermostdriver import Driver + +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') + +# Конфигурационные константы +CONFIG = { + 'CHECK_INTERVAL': 10, # seconds + 'NOTIFICATION_MINUTES': 15, + 'CREDENTIALS_FILE': 'credentials.json', + 'TIMEZONE': 'Europe/Moscow', + 'NOTIFICATION_COOLDOWN': 1800 # 30 minutes in seconds +} + +MOSCOW_TZ = EWSTimeZone(CONFIG['TIMEZONE']) +last_notification_times: Dict[str, EWSDateTime] = {} + +class EncryptionManager: + """Класс для управления шифрованием данных""" + def __init__(self, key: Optional[bytes] = None): + self.key = key or Fernet.generate_key() + + def encrypt(self, data: str) -> str: + return Fernet(self.key).encrypt(data.encode()).decode() + + def decrypt(self, encrypted_data: str) -> str: + return Fernet(self.key).decrypt(encrypted_data.encode()).decode() + + @property + def key_str(self) -> str: + return self.key.decode() + +class CredentialsManager(QObject): + """Класс для управления учетными данными""" + def __init__(self, encryption: EncryptionManager): + super().__init__() + self.encryption = encryption + + def load_credentials(self) -> Dict[str, str]: + """Загрузка и расшифровка учетных данных""" + if not os.path.exists(CONFIG['CREDENTIALS_FILE']): + return {} + + with open(CONFIG['CREDENTIALS_FILE'], 'r') as f: + return {k: self.encryption.decrypt(v) + for k, v in json.load(f).items()} + + def save_credentials(self, data: Dict[str, str]) -> None: + """Шифрование и сохранение учетных данных""" + encrypted = {k: self.encryption.encrypt(v) for k, v in data.items()} + with open(CONFIG['CREDENTIALS_FILE'], 'w') as f: + json.dump(encrypted, f) + +class MattermostClient: + """Клиент для работы с Mattermost""" + def __init__(self, credentials: Dict[str, str]): + self.driver = Driver({ + 'url': credentials['mattermost_url'], + 'token': credentials['mattermost_token'], + 'scheme': 'https', + 'port': 443, + 'headers': {'User-Agent': credentials['user_agent']} + }) + self.driver.login() + self.channel_id = credentials['mattermost_channel_id'] + + def send_message(self, message: str) -> bool: + """Отправка сообщения в Mattermost""" + try: + self.driver.posts.create_post({ + 'channel_id': self.channel_id, + 'message': message + }) + return True + except Exception as e: + logging.error(f"Mattermost error: {str(e)}") + return False + +class OutlookCalendarManager: + """Менеджер для работы с календарем Outlook""" + def __init__(self, credentials: Dict[str, str]): + self.account = Account( + primary_smtp_address=credentials['email'], + config=Configuration( + server='mail.tinkoff.ru', + credentials=Credentials( + username=f"{credentials['domain']}\\{credentials['username']}", + password=credentials['password'] + ) + ), + autodiscover=False, + access_type=DELEGATE + ) + + def get_upcoming_events(self, minutes: int) -> list: + """Получение предстоящих событий""" + now = EWSDateTime.now(tz=MOSCOW_TZ) + end = now + timedelta(minutes=minutes) + return list(self.account.calendar.view(start=now, end=end)) + +class NotificationService: + """Сервис управления уведомлениями""" + @staticmethod + def should_notify(event_id: str) -> bool: + """Проверка необходимости отправки уведомления""" + last_time = last_notification_times.get(event_id) + if not last_time: + return True + + cooldown = (EWSDateTime.now(tz=MOSCOW_TZ) - last_time).total_seconds() + return cooldown >= CONFIG['NOTIFICATION_COOLDOWN'] + +class ScriptThread(QThread): + finished = pyqtSignal() + error = pyqtSignal(str) + + def __init__(self, credentials: Dict[str, str], encryption: EncryptionManager): + super().__init__() + self.credentials = credentials + self.encryption = encryption + self._running = True + + def run(self): + while self._running: + try: + self.check_calendar() + except Exception as e: + self.error.emit(str(e)) + time.sleep(CONFIG['CHECK_INTERVAL']) + + def check_calendar(self): + """Основная логика проверки календаря""" + calendar = OutlookCalendarManager(self.credentials) + events = calendar.get_upcoming_events(CONFIG['NOTIFICATION_MINUTES']) + + mm_client = MattermostClient(self.credentials) + + for event in events: + if not isinstance(event, CalendarItem): + continue + + event_id = f"{event.subject}_{event.start.strftime('%Y%m%d%H%M')}" + if NotificationService.should_notify(event_id): + message = self.create_message(event) + if mm_client.send_message(message): + last_notification_times[event_id] = EWSDateTime.now(tz=MOSCOW_TZ) + + def create_message(self, event: CalendarItem) -> str: + """Формирование сообщения для Mattermost""" + start = event.start.astimezone(MOSCOW_TZ).strftime('%Y-%m-%d %H:%M:%S') + end = event.end.astimezone(MOSCOW_TZ).strftime('%Y-%m-%d %H:%M:%S') + return ( + f"@all Внимание! Скоро начнется встреча:\n" + f"Тема: {event.subject}\n" + f"Начало: {start}\n" + f"Окончание: {end}\n" + f"Место: {event.location or 'Не указано'}\n" + ) + + def stop(self): + self._running = False + +class LoginWindow(QWidget): + """Главное окно приложения""" + FIELDS = { + 'username': 'Логин:', + 'password': 'Пароль:', + 'domain': 'Домен:', + 'email': 'Email:', + 'mattermost_token': 'Токен Mattermost:', + 'mattermost_url': 'URL Mattermost:', + 'mattermost_channel_id': 'ID канала Mattermost:', + 'user_agent': 'User Agent:', + } + + def __init__(self): + super().__init__() + self.encryption: Optional[EncryptionManager] = None + self.credentials_manager: Optional[CredentialsManager] = None + self.script_thread: Optional[ScriptThread] = None + + self.init_encryption() + self.init_ui() + + def init_encryption(self): + """Инициализация системы шифрования""" + if os.path.exists(CONFIG['CREDENTIALS_FILE']): + self.handle_existing_credentials() + else: + self.handle_new_credentials() + + def handle_existing_credentials(self): + """Обработка существующих учетных данных""" + key, ok = QInputDialog.getText(self, 'Ввод ключа', 'Введите ваш ключ шифрования:') + if ok and key: + self.encryption = EncryptionManager(key.encode()) + self.credentials_manager = CredentialsManager(self.encryption) + else: + sys.exit() + + def handle_new_credentials(self): + """Генерация новых учетных данных""" + self.encryption = EncryptionManager() + self.show_key_warning() + self.credentials_manager = CredentialsManager(self.encryption) + + def show_key_warning(self): + """Отображение предупреждения о новом ключе""" + msg = QMessageBox() + msg.setWindowTitle('Новый ключ') + msg.setText(f'Ваш ключ шифрования: {self.encryption.key_str}') + + copy_btn = QPushButton('Копировать ключ') + copy_btn.clicked.connect(lambda: QApplication.clipboard().setText(self.encryption.key_str)) + + msg.addButton(copy_btn, QMessageBox.ActionRole) + msg.exec_() + + def init_ui(self): + """Инициализация пользовательского интерфейса""" + layout = QVBoxLayout() + self.inputs = {} + + for field, label in self.FIELDS.items(): + self.inputs[field] = QLineEdit() + if 'password' in field or 'token' in field: + self.inputs[field].setEchoMode(QLineEdit.Password) + + layout.addWidget(QLabel(label)) + layout.addWidget(self.inputs[field]) + + self.load_credentials() + + self.start_btn = QPushButton('Старт') + self.start_btn.clicked.connect(self.start_script) + + self.stop_btn = QPushButton('Стоп') + self.stop_btn.setEnabled(False) + self.stop_btn.clicked.connect(self.stop_script) + + layout.addWidget(self.start_btn) + layout.addWidget(self.stop_btn) + + self.setLayout(layout) + self.setWindowTitle('Управление уведомлениями') + self.setGeometry(300, 300, 400, 600) + self.show() + + def load_credentials(self): + """Загрузка учетных данных в форму""" + try: + credentials = self.credentials_manager.load_credentials() + for field in self.FIELDS: + if field in credentials: + self.inputs[field].setText(credentials[field]) + except Exception as e: + logging.error(f"Ошибка загрузки учетных данных: {e}") + + def validate_inputs(self) -> bool: + """Валидация введенных данных""" + return all(self.inputs[field].text().strip() for field in self.FIELDS) + + def start_script(self): + """Запуск основного скрипта""" + if not self.validate_inputs(): + QMessageBox.warning(self, 'Ошибка', 'Все поля должны быть заполнены!') + return + + credentials = {field: self.inputs[field].text() + for field in self.FIELDS} + self.credentials_manager.save_credentials(credentials) + + self.script_thread = ScriptThread(credentials, self.encryption) + self.script_thread.error.connect(self.show_error) + self.script_thread.start() + + self.start_btn.setEnabled(False) + self.stop_btn.setEnabled(True) + + def stop_script(self): + """Остановка скрипта""" + if self.script_thread and self.script_thread.isRunning(): + self.script_thread.stop() + self.script_thread.quit() + self.script_thread.wait() + + self.start_btn.setEnabled(True) + self.stop_btn.setEnabled(False) + + def show_error(self, message: str): + """Отображение ошибок""" + QMessageBox.critical(self, 'Ошибка', message) + +if __name__ == '__main__': + app = QApplication(sys.argv) + window = LoginWindow() + sys.exit(app.exec_()) \ No newline at end of file