309 lines
12 KiB
Python
309 lines
12 KiB
Python
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_()) |