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_())