import re from PyQt6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QScrollArea, QTextEdit, QPushButton, QLabel, QSizePolicy, QStackedWidget, QFrame, QApplication, QListWidget, QListWidgetItem, QMenu, QStyledItemDelegate, QStyle, QStyleOptionViewItem ) from PyQt6.QtCore import Qt, QThread, QSize, QTimer, pyqtSignal from PyQt6.QtGui import QResizeEvent, QAction, QPainter from app.core.main_manager import MainManager, NotificationType from app.core.llm_worker import LLMWorker from typing import Optional _LATEX = { "\\alpha": "α", "\\beta": "β", "\\gamma": "γ", "\\delta": "δ", "\\epsilon": "ε", "\\zeta": "ζ", "\\eta": "η", "\\theta": "θ", "\\iota": "ι", "\\kappa": "κ", "\\lambda": "λ", "\\mu": "μ", "\\nu": "ν", "\\xi": "ξ", "\\omicron": "ο", "\\pi": "π", "\\rho": "ρ", "\\sigma": "σ", "\\tau": "τ", "\\upsilon": "υ", "\\phi": "φ", "\\chi": "χ", "\\psi": "ψ", "\\omega": "ω", "\\Gamma": "Γ", "\\Delta": "Δ", "\\Theta": "Θ", "\\Lambda": "Λ", "\\Xi": "Ξ", "\\Pi": "Π", "\\Sigma": "Σ", "\\Phi": "Φ", "\\Psi": "Ψ", "\\Omega": "Ω", "\\nabla": "∇", "\\partial": "∂", "\\cdot": "·", "\\times": "×", "\\infty": "∞", "\\rightarrow": "→", "\\Rightarrow": "⇒", "\\left(": "(", "\\right)": ")", "\\left[": "[", "\\right]": "]", } def _latex_to_text(s: str) -> str: s = re.sub(r"\\text\{([^}]*)\}", r"\1", s) s = re.sub(r"\\mathbf\{([a-zA-Z])\}", lambda m: chr(ord('𝐚') - ord('a') + ord(m.group(1))) if 'a' <= m.group(1) <= 'z' else chr(ord('𝐀') - ord('A') + ord(m.group(1))) if 'A' <= m.group(1) <= 'Z' else m.group(1), s) s = re.sub(r"\\frac\{([^}]*)\}\{([^}]*)\}", r"\1/\2", s) for cmd, char in _LATEX.items(): s = s.replace(cmd, char) s = s.replace("\\,", " ").replace("\\;", " ").replace("\\!", "") s = s.replace("{", "").replace("}", "") return s def _inline(text: str) -> str: ph, n = {}, [0] def save(m): k = f"\x00_{n[0]}\x00"; n[0] += 1 ph[k] = m.group(0); return k text = re.sub(r"`([^`]+)`", save, text) text = re.sub(r"\$\$(.+?)\$\$", save, text) text = re.sub(r"\$(.+?)\$", save, text) text = re.sub(r"\*\*\*([^*]+)\*\*\*", r"\1", text) text = re.sub(r"\*\*([^*]+)\*\*", r"\1", text) text = re.sub(r"\*([^*]+)\*", r"\1", text) for k, v in ph.items(): if v.startswith("`"): text = text.replace(k, f"{v[1:-1]}") elif v.startswith("$$"): text = text.replace(k, f"{_latex_to_text(v[2:-2])}") else: text = text.replace(k, f"{_latex_to_text(v[1:-1])}") return text def md_to_html(text: str) -> str: text = text.replace("&", "amp;").replace("<", "lt;").replace(">", "gt;") lines = text.split("\n") parts, buf, mode = [], [], "normal" def flush_buf(): nonlocal buf if not buf: return if mode == "code": parts.append(f"
{''.join(buf)}
") elif mode == "math": parts.append(f"
{_latex_to_text(chr(10).join(buf))}
") buf = [] for line in lines: s = line.strip() if mode == "code": if s.startswith("```"): flush_buf() mode = "normal" else: buf.append(line + "\n") continue if s.startswith("```"): flush_buf() mode = "code" continue if mode == "math": if s.startswith("$$"): flush_buf() mode = "normal" else: buf.append(s) continue if s.startswith("$$"): flush_buf() rest = s[2:] j = rest.find("$$") if j != -1: parts.append(f"
{_latex_to_text(rest[:j])}
") else: mode = "math" continue if s == "---" or s == "___" or s == "***": parts.append("
") continue m = re.match(r"^(#{1,3})\s+(.+)$", s) if m: parts.append(f"{_inline(m.group(2))}") continue if re.match(r"^[-*]\s", s) or re.match(r"^\d+\.\s", s): c = re.sub(r"^[-*]\s+", "", s) c = re.sub(r"^\d+\.\s+", "", c) if parts and parts[-1] != "") return "".join(parts).replace("amp;", "&").replace("lt;", "<").replace("gt;", ">").replace("&amp;", "&") class MessageBubble(QFrame): def __init__(self, text: str, is_user: bool, theme_manager, parent=None): super().__init__(parent) self.is_user = is_user self.theme_manager = theme_manager self._raw_text = text self.setObjectName("chat_bubble") self.setSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed) self._max_allowed = 400 self.label = QLabel() self.label.setWordWrap(True) self.label.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Minimum) self.label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse) layout = QVBoxLayout(self) layout.setContentsMargins(14, 10, 14, 10) layout.addWidget(self.label) self._apply_style() self.set_text(text) def _apply_style(self): theme = self.theme_manager.current_theme bg = theme.get_color("primary_color") if self.is_user else "transparent" txt = "#FFFFFF" if self.is_user else theme.get_color("text_color") self.setStyleSheet(f""" QFrame#chat_bubble {{ background-color: {bg}; border-radius: 14px; }} """) self.label.setStyleSheet(f""" background-color: transparent; color: {txt}; font-size: 14px; """) def set_text(self, text: str): self._raw_text = text self.label.setText(md_to_html(text)) self._fit() def _fit(self, text: str = ""): self.label.setMaximumWidth(self._max_allowed) sh = self.label.sizeHint() cw = self.width() fw = sh.width() + 28 if text: fw = max(fw, cw) fw = min(fw, self._max_allowed + 28) self.setFixedSize(fw, sh.height() + 20) self.updateGeometry() p = self.parentWidget() if p: p.updateGeometry() def set_text(self, text: str): self._raw_text = text self.label.setText(md_to_html(text)) self._fit() def update_text(self, text: str): self._raw_text = text self.label.setText(md_to_html(text)) self._fit(text) def set_max_allowed(self, px: int): self._max_allowed = px - 28 self._fit() class ElideDelegate(QStyledItemDelegate): def paint(self, painter, option, index): if option.state & QStyle.StateFlag.State_Editing: super().paint(painter, option, index) return opt = QStyleOptionViewItem(option) self.initStyleOption(opt, index) text = opt.text if text: vw = option.widget.viewport().width() if option.widget else option.rect.width() opt.text = painter.fontMetrics().elidedText( text, Qt.TextElideMode.ElideRight, vw - 12 ) QApplication.style().drawControl( QStyle.ControlElement.CE_ItemViewItem, opt, painter ) def updateEditorGeometry(self, editor, option, index): r = option.rect if option.widget: r.setWidth(option.widget.viewport().width()) editor.setGeometry(r) class ChatMessagesView(QWidget): send_requested = pyqtSignal(str) stop_requested = pyqtSignal() def __init__(self, theme_manager, parent=None): super().__init__(parent) self.theme_manager = theme_manager self._bubbles: list[MessageBubble] = [] self._has_messages = False main_layout = QVBoxLayout(self) main_layout.setContentsMargins(0, 0, 0, 0) main_layout.setSpacing(0) btn_size = 38 self.input_container = QWidget() self.input_container.setObjectName("chat_input_container") outer_layout = QHBoxLayout(self.input_container) outer_layout.setContentsMargins(100, 10, 100, 18) outer_layout.setSpacing(6) self.input_wrapper = QWidget() self.input_wrapper.setObjectName("chat_input_wrapper") wrapper_layout = QHBoxLayout(self.input_wrapper) wrapper_layout.setContentsMargins(0, 0, 0, 0) self.input_edit = QTextEdit() self.input_edit.setObjectName("chat_input_field") self.input_edit.setPlaceholderText("Posez votre question...") self.input_edit.setFixedHeight(btn_size) self.input_edit.document().setDocumentMargin(3) self.input_edit.setViewportMargins(18, 6, 6, 6) self.input_edit.setSizePolicy( QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed ) self.input_edit.setVerticalScrollBarPolicy( Qt.ScrollBarPolicy.ScrollBarAlwaysOff ) self.input_edit.installEventFilter(self) wrapper_layout.addWidget(self.input_edit) self.send_button = QPushButton("\u2191") self.send_button.setObjectName("chat_send_btn") self.send_button.setFixedSize(QSize(btn_size, btn_size)) self.send_button.setCursor(Qt.CursorShape.PointingHandCursor) self.send_button.clicked.connect(self._on_send_clicked) self.stop_button = QPushButton("\u25A0") self.stop_button.setObjectName("chat_stop_btn") self.stop_button.setFixedSize(QSize(btn_size, btn_size)) self.stop_button.setCursor(Qt.CursorShape.PointingHandCursor) self.stop_button.clicked.connect(self._on_stop_clicked) self.stop_button.hide() outer_layout.addWidget(self.input_wrapper, 1) outer_layout.addWidget(self.send_button, 0) outer_layout.addWidget(self.stop_button, 0) self.mode_stack = QStackedWidget() self.mode_stack.setObjectName("chat_mode_stack") self.welcome_page = QWidget() self.welcome_page.setObjectName("welcome_page") self.welcome_layout = QVBoxLayout(self.welcome_page) self.welcome_layout.setContentsMargins(0, 0, 0, 0) self.welcome_layout.addStretch() self.greeting_label = QLabel("Comment puis-je vous aider ?") self.greeting_label.setObjectName("greeting_label") self.greeting_label.setAlignment(Qt.AlignmentFlag.AlignCenter) self.welcome_layout.addWidget(self.greeting_label, 0, Qt.AlignmentFlag.AlignCenter) self.welcome_layout.addSpacing(24) # input_container added here via _enter_welcome_mode(), expands full width self.welcome_layout.addStretch() self.chat_page = QWidget() self.chat_page.setObjectName("chat_page") self.chat_layout = QVBoxLayout(self.chat_page) self.chat_layout.setContentsMargins(0, 0, 0, 0) self.chat_layout.setSpacing(0) self.scroll_area = QScrollArea() self.scroll_area.setWidgetResizable(True) self.scroll_area.setHorizontalScrollBarPolicy( Qt.ScrollBarPolicy.ScrollBarAlwaysOff ) self.scroll_area.setVerticalScrollBarPolicy( Qt.ScrollBarPolicy.ScrollBarAlwaysOff ) self.scroll_area.setObjectName("chat_scroll") self.messages_container = QWidget() self.messages_container.setObjectName("messages_container") self.messages_layout = QVBoxLayout(self.messages_container) self.messages_layout.setAlignment(Qt.AlignmentFlag.AlignTop) self.messages_layout.setSpacing(6) self.messages_layout.setContentsMargins(100, 20, 100, 20) self.scroll_area.setWidget(self.messages_container) self.chat_layout.addWidget(self.scroll_area, 1) self.mode_stack.addWidget(self.welcome_page) self.mode_stack.addWidget(self.chat_page) main_layout.addWidget(self.mode_stack) self._enter_welcome_mode() def _enter_welcome_mode(self): self.input_container.setParent(None) self.welcome_layout.insertWidget(3, self.input_container, 0) self.mode_stack.setCurrentIndex(0) self.scroll_area.setVerticalScrollBarPolicy( Qt.ScrollBarPolicy.ScrollBarAlwaysOff ) def _enter_chat_mode(self): self.input_container.setParent(None) self.chat_layout.addWidget(self.input_container, 0) self.mode_stack.setCurrentIndex(1) self.scroll_area.setVerticalScrollBarPolicy( Qt.ScrollBarPolicy.ScrollBarAsNeeded ) def _max_allowed_width(self) -> int: return int(self.width() * 0.72) def add_bubble(self, text: str, is_user: bool) -> MessageBubble: if not self._has_messages: self._has_messages = True self._enter_chat_mode() bubble = MessageBubble(text, is_user, self.theme_manager) bubble._max_allowed = self._max_allowed_width() - 28 bubble._fit() self._bubbles.append(bubble) row = QWidget() row.setStyleSheet("background-color: transparent;") row.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Minimum) row_layout = QHBoxLayout(row) row_layout.setContentsMargins(0, 0, 0, 0) align = Qt.AlignmentFlag.AlignRight if is_user else Qt.AlignmentFlag.AlignLeft row_layout.addWidget(bubble, 0, align | Qt.AlignmentFlag.AlignTop) self.messages_layout.addWidget(row) self._scroll_to_bottom() return bubble def clear_messages(self): self._bubbles.clear() self._has_messages = False for i in reversed(range(self.messages_layout.count())): item = self.messages_layout.takeAt(i) w = item.widget() if w: w.deleteLater() self._enter_welcome_mode() def apply_theme_to_bubbles(self): for b in self._bubbles: b._apply_style() def _scroll_to_bottom(self): sb = self.scroll_area.verticalScrollBar() sb.setValue(sb.maximum()) QApplication.processEvents() def resizeEvent(self, event): super().resizeEvent(event) m = self._max_allowed_width() - 28 for b in self._bubbles: b._max_allowed = m b._fit() def eventFilter(self, obj, event): if obj is self.input_edit and event.type() == event.Type.KeyPress: if event.key() == Qt.Key.Key_Return and not ( event.modifiers() & Qt.KeyboardModifier.ShiftModifier ): self._on_send_clicked() return True return super().eventFilter(obj, event) def _on_send_clicked(self): self.send_requested.emit(self.input_edit.toPlainText().strip()) def _on_stop_clicked(self): self.stop_requested.emit() class ChatSession: def __init__(self, title: str = "Nouveau Chat"): self.title = title self.messages: list[tuple[str, bool]] = [] self.llm_history: list[dict] = [] self.view: Optional[ChatMessagesView] = None class ChatWindow(QWidget): _send_request = pyqtSignal(str) def __init__(self, parent: Optional[QWidget] = None): super().__init__(parent) self.main_manager = MainManager.get_instance() self.language_manager = self.main_manager.get_language_manager() self.settings_manager = self.main_manager.get_settings_manager() self.theme_manager = self.main_manager.get_theme_manager() self.observer_manager = self.main_manager.get_observer_manager() self.observer_manager.subscribe(NotificationType.LANGUAGE, self.update_language) self.observer_manager.subscribe(NotificationType.THEME, self._on_theme_changed) self.sessions: list[ChatSession] = [] self._current_index: int = -1 self._streaming = False self._finished = False self._session_counter = 0 self.thread: Optional[QThread] = None self.worker: Optional[LLMWorker] = None self._streaming_bubble: Optional[MessageBubble] = None self.setup_ui() QTimer.singleShot(0, self._init_llm) QTimer.singleShot(0, self._new_session) def _get_model_path(self) -> str: return self.settings_manager.get_model_path() def _init_llm(self): model_path = self._get_model_path() if not model_path: self.status_label.setText("Aucun modèle configuré") self.status_label.show() return self.status_label.show() self.thread = QThread() self.worker = LLMWorker(model_path) self.worker.moveToThread(self.thread) self.worker.token_received.connect(self._on_token) self.worker.message_finished.connect(self._on_message_finished) self.worker.error_occurred.connect(self._on_llm_error) self.worker.model_loaded.connect(self._on_model_loaded) self.worker.model_info.connect(self.status_label.setText) self.worker.model_loaded.connect(lambda: self.status_label.hide()) self._send_request.connect(self.worker.send_message) self.thread.started.connect(self.worker.load_model) self.thread.start() def setup_ui(self): layout = QVBoxLayout(self) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(0) self.chat_stack = QStackedWidget() self.chat_stack.setObjectName("chat_stack") layout.addWidget(self.chat_stack, 1) self.status_label = QLabel( self.language_manager.get_text("chat_loading") ) self.status_label.setObjectName("status_label") self.status_label.setAlignment(Qt.AlignmentFlag.AlignCenter) self.status_label.setFixedHeight(28) self.status_label.hide() layout.addWidget(self.status_label) self.chat_list = QListWidget() self.chat_list.setObjectName("chat_list") self.chat_list.setSpacing(2) self.chat_list.setItemDelegate(ElideDelegate()) self.chat_list.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff) self.chat_list.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu) self.chat_list.customContextMenuRequested.connect(self._on_list_context_menu) self.chat_list.currentRowChanged.connect(self._switch_to_session) self.chat_list.itemChanged.connect(self._on_item_renamed) self.new_chat_btn = QPushButton("+ Nouveau Chat") self.new_chat_btn.setObjectName("sidebar_new_btn") self.new_chat_btn.clicked.connect(self._new_session) def get_sidebar_widget(self) -> QWidget: container = QWidget() container.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding) layout = QVBoxLayout(container) layout.setContentsMargins(8, 4, 8, 4) layout.setSpacing(4) layout.addWidget(self.chat_list, 1) layout.addWidget(self.new_chat_btn) return container def _new_session(self): self._session_counter += 1 session = ChatSession(title=f"Chat {self._session_counter}") view = ChatMessagesView(self.theme_manager) view.send_requested.connect(self._on_send_requested) view.stop_requested.connect(self._stop_generation) session.view = view self.sessions.append(session) self.chat_stack.addWidget(view) item = QListWidgetItem(session.title) item.setFlags(item.flags() | Qt.ItemFlag.ItemIsEditable) self.chat_list.addItem(item) self._current_index = len(self.sessions) - 1 self.chat_list.setCurrentRow(self._current_index) self.chat_stack.setCurrentWidget(view) view.input_edit.setFocus() def _switch_to_session(self, index: int): if index < 0 or index >= len(self.sessions) or index == self._current_index: return self._current_index = index session = self.sessions[index] if session.view: self.chat_stack.setCurrentWidget(session.view) session.view.input_edit.setFocus() def _close_session(self, index: int): if index < 0 or index >= len(self.sessions) or len(self.sessions) <= 1: return self.chat_list.blockSignals(True) self.chat_list.takeItem(index) self.chat_list.blockSignals(False) session = self.sessions.pop(index) if session.view: self.chat_stack.removeWidget(session.view) session.view.deleteLater() remaining = len(self.sessions) if remaining > 0: new_idx = min(index, remaining - 1) self._current_index = -1 self.chat_list.setCurrentRow(-1) self.chat_list.setCurrentRow(new_idx) def _on_item_renamed(self, item: QListWidgetItem): row = self.chat_list.row(item) if 0 <= row < len(self.sessions): self.sessions[row].title = item.text() def _on_list_context_menu(self, pos): item = self.chat_list.itemAt(pos) if not item: return index = self.chat_list.row(item) menu = QMenu() rename_action = QAction("Renommer", self) rename_action.triggered.connect(lambda: self.chat_list.editItem(item)) menu.addAction(rename_action) if len(self.sessions) > 1: delete_action = QAction("Supprimer", self) delete_action.triggered.connect(lambda: self._close_session(index)) menu.addAction(delete_action) menu.exec(self.chat_list.mapToGlobal(pos)) def current_session(self) -> Optional[ChatSession]: if 0 <= self._current_index < len(self.sessions): return self.sessions[self._current_index] return None def _on_send_requested(self, text: str): text = text.strip() if not text or self._streaming: return session = self.current_session() if not session or not session.view: return view = session.view view.input_edit.clear() view.add_bubble(text, True) session.messages.append((text, True)) if not self.worker: view.add_bubble( self.language_manager.get_text("chat_loading"), False ) return self._streaming = True self._finished = False view.send_button.hide() view.stop_button.show() self.status_label.setText("Génération...") self.status_label.show() bubble = view.add_bubble("", False) self._streaming_bubble = bubble self._send_request.emit(text) def _on_token(self, token: str): session = self.current_session() if not session or not session.view: return if self._streaming_bubble: raw = self._streaming_bubble._raw_text + token self._streaming_bubble.update_text(raw) session.view._scroll_to_bottom() def _on_message_finished(self): if self._finished: return self._finished = True self._streaming = False session = self.current_session() if session and session.view: if self._streaming_bubble: t = self._streaming_bubble._raw_text self._streaming_bubble.set_text(t) session.messages.append((t, False)) self._streaming_bubble = None session.view.send_button.show() session.view.stop_button.hide() session.view.input_edit.setFocus() self.status_label.hide() def _on_llm_error(self, error_msg: str): self._streaming = False self._streaming_bubble = None session = self.current_session() if session and session.view: session.view.send_button.show() session.view.stop_button.hide() session.view.input_edit.setFocus() self.status_label.setText(f"Erreur : {error_msg}") self.status_label.show() def _stop_generation(self): if self.worker: self.worker.stop() session = self.current_session() if session and session.view: if self._streaming_bubble: t = self._streaming_bubble._raw_text if not t.strip(): self._streaming_bubble.set_text( self.language_manager.get_text("chat_cancelled") ) else: self._streaming_bubble.set_text(t) session.messages.append((t, False)) self._streaming_bubble = None session.view.send_button.show() session.view.stop_button.hide() self._finished = True self._streaming = False self.status_label.hide() def _on_model_loaded(self): self.status_label.hide() session = self.current_session() if session and session.view: session.view.input_edit.setFocus() def _on_theme_changed(self): for s in self.sessions: if s.view: s.view.apply_theme_to_bubbles() def update_language(self): for s in self.sessions: if s.view: s.view.input_edit.setPlaceholderText( self.language_manager.get_text("chat_placeholder") ) def cleanup(self): if self.worker: self.worker.stop() if self.thread and self.thread.isRunning(): self.thread.quit() self.thread.wait()