308 lines
13 KiB
Python
308 lines
13 KiB
Python
from PySide6.QtCore import QStandardPaths, QDataStream, QByteArray, QIODevice, Signal, Qt, QTimer, QCryptographicHash
|
|
from PySide6.QtGui import QPalette, QColor
|
|
from PySide6.QtNetwork import QLocalServer, QLocalSocket
|
|
from PySide6.QtWidgets import QApplication
|
|
|
|
import qasync
|
|
import sys
|
|
import asyncio
|
|
import os
|
|
import platform
|
|
import argparse
|
|
import hashlib
|
|
import random
|
|
import string
|
|
import base64
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
from src.logs import configure_logging
|
|
from windows.main_window import MainWindow
|
|
|
|
|
|
def apply_dark_theme(app):
|
|
app.setStyle("Fusion")
|
|
|
|
dark_palette = QPalette()
|
|
|
|
# Utilisation de gris plus clairs pour les principaux éléments
|
|
dark_palette.setColor(QPalette.ColorRole.Window, QColor(90, 90, 95)) # Barre de titre plus claire
|
|
dark_palette.setColor(QPalette.ColorRole.WindowText, QColor(255, 255, 255))
|
|
dark_palette.setColor(QPalette.ColorRole.Base, QColor(60, 60, 65)) # Zone de texte plus claire
|
|
dark_palette.setColor(QPalette.ColorRole.AlternateBase, QColor(80, 80, 85))
|
|
dark_palette.setColor(QPalette.ColorRole.ToolTipBase, QColor(70, 70, 75))
|
|
dark_palette.setColor(QPalette.ColorRole.ToolTipText, QColor(255, 255, 255))
|
|
dark_palette.setColor(QPalette.ColorRole.Text, QColor(255, 255, 255))
|
|
dark_palette.setColor(QPalette.ColorRole.Button, QColor(75, 75, 80)) # Boutons plus clairs
|
|
dark_palette.setColor(QPalette.ColorRole.ButtonText, QColor(255, 255, 255))
|
|
dark_palette.setColor(QPalette.ColorRole.BrightText, QColor(255, 0, 0))
|
|
|
|
# Les autres paramètres restent inchangés
|
|
dark_palette.setColor(QPalette.ColorRole.Highlight, QColor(61, 142, 201))
|
|
dark_palette.setColor(QPalette.ColorRole.HighlightedText, QColor(255, 255, 255))
|
|
dark_palette.setColor(QPalette.ColorRole.Link, QColor(77, 166, 230))
|
|
dark_palette.setColor(QPalette.ColorRole.LinkVisited, QColor(120, 120, 250))
|
|
dark_palette.setColor(QPalette.ColorGroup.Disabled, QPalette.ColorRole.WindowText, QColor(150, 150, 150))
|
|
dark_palette.setColor(QPalette.ColorGroup.Disabled, QPalette.ColorRole.Text, QColor(150, 150, 150))
|
|
dark_palette.setColor(QPalette.ColorGroup.Disabled, QPalette.ColorRole.ButtonText, QColor(150, 150, 150))
|
|
|
|
app.setPalette(dark_palette)
|
|
|
|
|
|
class SingleApplication(QApplication):
|
|
# Signal émis lorsque des fichiers sont reçus d'une instance secondaire
|
|
files_received = Signal(list)
|
|
|
|
def __init__(self, app_id, args):
|
|
super().__init__(args)
|
|
|
|
# Configuration du logger
|
|
self.logger = logging.getLogger(__name__)
|
|
self.logger.debug("Initialisation de SingleApplication")
|
|
|
|
self.app_id = app_id
|
|
self.logger.debug(f"ID de l'application: {app_id}")
|
|
|
|
self.shared_key = hashlib.sha256(app_id.encode()).hexdigest()[:16]
|
|
self.logger.debug(f"Clé partagée générée: {self.shared_key}")
|
|
|
|
self.server = None
|
|
self.is_primary_instance = self.try_connect_to_primary()
|
|
|
|
if self.is_primary_instance:
|
|
# C'est la première instance, on crée un serveur local
|
|
self.logger.info("Instance primaire détectée, création du serveur local")
|
|
self.server = QLocalServer()
|
|
self.server.newConnection.connect(self.handle_new_connection)
|
|
self.logger.debug("Signal newConnection connecté")
|
|
|
|
if not self.server.listen(self.app_id):
|
|
self.logger.warning(f"Échec de l'écoute sur {self.app_id}, tentative de suppression du serveur existant")
|
|
# En cas d'erreur (serveur déjà existant mais zombie), on le supprime et on réessaie
|
|
QLocalServer.removeServer(self.app_id)
|
|
if self.server.listen(self.app_id):
|
|
self.logger.info(f"Serveur local créé avec succès après suppression de l'ancien")
|
|
else:
|
|
self.logger.error(f"Impossible de créer le serveur local même après suppression")
|
|
else:
|
|
self.logger.info(f"Serveur local créé avec succès")
|
|
else:
|
|
self.logger.info("Instance secondaire détectée, fermeture de l'application")
|
|
QTimer.singleShot(0, self.quit)
|
|
|
|
def encrypt_data(self, data_str):
|
|
"""
|
|
Méthode simple pour brouiller les données
|
|
|
|
Args:
|
|
data_str (str): Données à chiffrer
|
|
|
|
Returns:
|
|
str: Données chiffrées en base64
|
|
"""
|
|
self.logger.debug(f"Chiffrement des données (longueur: {len(data_str)})")
|
|
|
|
# Générer une "nonce" aléatoire pour éviter que les mêmes données produisent le même résultat
|
|
nonce = ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(8))
|
|
self.logger.debug(f"Nonce générée: {nonce}")
|
|
|
|
# Combiner la nonce, la clé et les données
|
|
combined = nonce + self.shared_key + data_str
|
|
self.logger.debug("Données combinées avec nonce et clé partagée")
|
|
|
|
# Utiliser SHA-256 pour obtenir un hash
|
|
hash_obj = QCryptographicHash(QCryptographicHash.Algorithm.Sha256)
|
|
hash_obj.addData(combined.encode())
|
|
signature = hash_obj.result().toHex().data().decode()[:16]
|
|
self.logger.debug(f"Signature générée: {signature}")
|
|
|
|
# Encoder le tout en base64
|
|
encoded = base64.b64encode((nonce + signature + data_str).encode()).decode()
|
|
self.logger.debug(f"Données encodées en base64 (longueur: {len(encoded)})")
|
|
return encoded
|
|
|
|
def decrypt_data(self, encoded_str):
|
|
"""
|
|
Déchiffre les données et vérifie leur intégrité
|
|
|
|
Args:
|
|
encoded_str (str): Données chiffrées en base64
|
|
|
|
Returns:
|
|
str ou None: Données déchiffrées ou None en cas d'erreur
|
|
"""
|
|
self.logger.debug(f"Déchiffrement des données (longueur: {len(encoded_str)})")
|
|
|
|
try:
|
|
# Décoder de base64
|
|
decoded = base64.b64decode(encoded_str.encode()).decode()
|
|
self.logger.debug("Données décodées de base64")
|
|
|
|
# Extraire nonce, signature et données
|
|
nonce = decoded[:8]
|
|
signature = decoded[8:24]
|
|
data_str = decoded[24:]
|
|
self.logger.debug(f"Nonce extraite: {nonce}, signature: {signature}")
|
|
|
|
# Vérifier la signature
|
|
combined = nonce + self.shared_key + data_str
|
|
hash_obj = QCryptographicHash(QCryptographicHash.Algorithm.Sha256)
|
|
hash_obj.addData(combined.encode())
|
|
expected_signature = hash_obj.result().toHex().data().decode()[:16]
|
|
self.logger.debug(f"Signature attendue: {expected_signature}")
|
|
|
|
if signature != expected_signature:
|
|
self.logger.warning("Signature invalide, données potentiellement corrompues ou falsifiées")
|
|
return None
|
|
|
|
self.logger.debug(f"Données déchiffrées avec succès (longueur: {len(data_str)})")
|
|
return data_str
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Erreur lors du déchiffrement: {e}", exc_info=True)
|
|
return None
|
|
|
|
def try_connect_to_primary(self):
|
|
"""
|
|
Essaie de se connecter à l'instance primaire de l'application
|
|
|
|
Returns:
|
|
bool: True si c'est l'instance primaire, False sinon
|
|
"""
|
|
self.logger.debug(f"Tentative de connexion à l'instance primaire: {self.app_id}")
|
|
socket = QLocalSocket()
|
|
socket.connectToServer(self.app_id, QIODevice.OpenModeFlag.WriteOnly)
|
|
|
|
if socket.waitForConnected(500):
|
|
self.logger.info("Connexion établie avec l'instance primaire")
|
|
|
|
# Récupérer les arguments pour les envoyer à l'instance primaire
|
|
args = sys.argv[1:] if len(sys.argv) > 1 else []
|
|
self.logger.debug(f"Arguments à transmettre: {args}")
|
|
|
|
encrypt_args = self.encrypt_data(";".join(args))
|
|
self.logger.debug("Arguments chiffrés pour transmission")
|
|
|
|
# Envoyer les arguments à l'instance primaire
|
|
stream = QDataStream(socket)
|
|
stream.writeQString(encrypt_args)
|
|
socket.flush()
|
|
self.logger.debug("Données envoyées à l'instance primaire")
|
|
|
|
if socket.waitForBytesWritten(1000):
|
|
self.logger.debug("Données écrites avec succès")
|
|
else:
|
|
self.logger.warning("Délai d'attente dépassé pour l'écriture des données")
|
|
|
|
socket.disconnectFromServer()
|
|
self.logger.debug("Déconnexion du serveur")
|
|
|
|
QTimer.singleShot(0, self.quit)
|
|
self.logger.info("Instance secondaire, programmation de la fermeture")
|
|
|
|
return False # Ce n'est pas l'instance primaire
|
|
|
|
self.logger.info("Aucune instance primaire trouvée, devenant l'instance primaire")
|
|
return True # C'est l'instance primaire
|
|
|
|
def handle_new_connection(self):
|
|
"""
|
|
Gère une nouvelle connexion d'une instance secondaire
|
|
"""
|
|
self.logger.info("Nouvelle connexion d'une instance secondaire reçue")
|
|
socket = self.server.nextPendingConnection()
|
|
|
|
if socket.waitForReadyRead(2000):
|
|
self.logger.debug("Données disponibles pour lecture")
|
|
stream = QDataStream(socket)
|
|
|
|
encrypted_args = stream.readQString()
|
|
self.logger.debug(f"Arguments chiffrés reçus (longueur: {len(encrypted_args)})")
|
|
|
|
args_str = self.decrypt_data(encrypted_args)
|
|
if args_str:
|
|
self.logger.debug(f"Arguments déchiffrés: {args_str}")
|
|
|
|
# Émettre un signal pour informer l'application des fichiers à ouvrir
|
|
args = args_str.split(";") if args_str else []
|
|
if args:
|
|
self.logger.info(f"Émission du signal files_received avec {len(args)} arguments")
|
|
self.files_received.emit(args)
|
|
else:
|
|
self.logger.debug("Aucun argument à traiter")
|
|
else:
|
|
self.logger.warning("Échec du déchiffrement des arguments")
|
|
else:
|
|
self.logger.warning("Délai d'attente dépassé pour la lecture des données")
|
|
|
|
socket.disconnectFromServer()
|
|
self.logger.debug("Déconnexion du client")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Analyser les arguments de la ligne de commande
|
|
parser = argparse.ArgumentParser(description='Application PySide6', allow_abbrev=False)
|
|
parser.add_argument('--dev', action='store_true', help='Active le mode développement avec logs')
|
|
parser.add_argument('files', nargs='*', help='Fichiers à ouvrir')
|
|
args, unknown = parser.parse_known_args()
|
|
|
|
# Configurer le logging en fonction du mode
|
|
configure_logging(args.dev)
|
|
|
|
# Créer un logger pour le module principal
|
|
logger = logging.getLogger(__name__)
|
|
logger.info("Démarrage de l'application")
|
|
|
|
# Configuration des variables d'environnement pour QtWebEngine
|
|
os.environ["QTWEBENGINE_CHROMIUM_FLAGS"] = "--enable-gpu-rasterization --ignore-gpu-blocklist"
|
|
logger.debug("Flags Chromium configurés pour l'accélération GPU")
|
|
|
|
if args.dev:
|
|
os.environ["QTWEBENGINE_REMOTE_DEBUGGING"] = "4000"
|
|
logger.info("Mode développement activé avec débogage distant sur le port 4000")
|
|
|
|
# Création de l'application
|
|
app_id = "OxAPP25"
|
|
logger.debug(f"ID de l'application: {app_id}")
|
|
|
|
app = SingleApplication(app_id, sys.argv)
|
|
apply_dark_theme(app)
|
|
logger.debug("Instance SingleApplication créée")
|
|
|
|
if not app.is_primary_instance:
|
|
logger.info("Instance secondaire détectée, fermeture de l'application")
|
|
sys.exit(0)
|
|
|
|
# Configuration de la boucle d'événements asyncio
|
|
logger.debug("Configuration de la boucle d'événements asyncio")
|
|
event_loop = qasync.QEventLoop(app)
|
|
asyncio.set_event_loop(event_loop)
|
|
|
|
app_close_event = asyncio.Event()
|
|
app.aboutToQuit.connect(app_close_event.set)
|
|
logger.debug("Signal aboutToQuit connecté à l'événement de fermeture")
|
|
|
|
# Création de la fenêtre principale
|
|
logger.info("Création de la fenêtre principale")
|
|
window = MainWindow()
|
|
|
|
# Connecter le signal de fichiers reçus à une méthode de traitement
|
|
app.files_received.connect(window.handle_files)
|
|
logger.debug("Signal files_received connecté")
|
|
|
|
# Traitement des fichiers passés en arguments
|
|
if args.files:
|
|
logger.info(f"Traitement de {len(args.files)} fichiers passés en arguments")
|
|
window.handle_files(args.files)
|
|
|
|
# Affichage de la fenêtre
|
|
window.show()
|
|
logger.info("Fenêtre principale affichée")
|
|
|
|
# Exécution de la boucle d'événements
|
|
logger.debug("Démarrage de la boucle d'événements")
|
|
with event_loop:
|
|
event_loop.run_until_complete(app_close_event.wait())
|
|
|
|
logger.info("Application terminée")
|