"""Connect to registry dialog.""" import socket import time import requests from PySide6.QtWidgets import ( QDialog, QVBoxLayout, QLabel, QPushButton, QHBoxLayout, QLineEdit, QProgressBar, QTextEdit ) from PySide6.QtCore import Qt, QThread, Signal, QTimer from ...config import load_config, save_config, get_registry_url, set_registry_token class PairingWorker(QThread): """Background worker for pairing flow.""" success = Signal(str) # token error = Signal(str) status = Signal(str) def __init__(self, username: str): super().__init__() self.username = username self.hostname = socket.gethostname() self._stop = False def stop(self): self._stop = True def run(self): try: registry_url = get_registry_url() # Remove trailing /api/v1 if present to get base URL base_url = registry_url.rstrip("/") if base_url.endswith("/api/v1"): base_url = base_url[:-7] pairing_url = f"{base_url}/api/v1/pairing/check/{self.username}" self.status.emit(f"Connecting to {base_url}...") # Poll for pairing status max_attempts = 150 # 5 minutes at 2-second intervals attempt = 0 while attempt < max_attempts: if self._stop: return attempt += 1 try: response = requests.get( pairing_url, params={"hostname": self.hostname}, timeout=10 ) if response.status_code == 200: data = response.json() status = data.get("data", {}).get("status") if status == "connected": token = data.get("data", {}).get("token") if token: self.success.emit(token) return else: self.error.emit("Pairing completed but no token received") return elif status == "expired": self.error.emit("Pairing request expired. Please try again.") return elif status == "pending": self.status.emit(f"Pairing pending, approve in web UI... ({attempt})") elif status == "not_found": self.status.emit(f"Waiting for web UI pairing... ({attempt})") else: self.status.emit(f"Status: {status} ({attempt})") elif response.status_code == 404: self.status.emit(f"Waiting for web approval... ({attempt})") else: self.status.emit(f"Server returned {response.status_code}") except requests.exceptions.RequestException: self.status.emit(f"Network error, retrying... ({attempt})") time.sleep(2) self.error.emit("Timed out waiting for approval") except Exception as e: self.error.emit(str(e)) class ConnectDialog(QDialog): """Dialog for connecting to registry.""" def __init__(self, parent): super().__init__(parent) self.setWindowTitle("Connect to Registry") self.setMinimumWidth(450) self._worker = None self._setup_ui() def _setup_ui(self): """Set up the UI.""" layout = QVBoxLayout(self) layout.setSpacing(16) # Title title = QLabel("Connect Your Account") title.setStyleSheet("font-size: 16px; font-weight: 600;") layout.addWidget(title) # Instructions self.instructions = QLabel( "To publish tools, connect your CmdForge account.\n\n" "1. Enter your CmdForge username below\n" "2. Click Connect\n" "3. Go to cmdforge.brrd.tech/dashboard/connected-apps\n" "4. Click 'Connect New App' and approve this device" ) self.instructions.setWordWrap(True) layout.addWidget(self.instructions) # Username input username_layout = QHBoxLayout() username_label = QLabel("Username:") self.username_input = QLineEdit() self.username_input.setPlaceholderText("your-username") self.username_input.returnPressed.connect(self._start_pairing) username_layout.addWidget(username_label) username_layout.addWidget(self.username_input, 1) layout.addLayout(username_layout) # Device info hostname = socket.gethostname() device_label = QLabel(f"Device: {hostname}") device_label.setStyleSheet("color: #718096; font-size: 12px;") layout.addWidget(device_label) # Status self.status_label = QLabel("") self.status_label.setStyleSheet("color: #718096;") layout.addWidget(self.status_label) # Log area for debugging self.log_area = QTextEdit() self.log_area.setReadOnly(True) self.log_area.setMaximumHeight(120) self.log_area.setStyleSheet("font-family: monospace; font-size: 11px; background: #f7fafc;") self.log_area.hide() layout.addWidget(self.log_area) # Progress self.progress = QProgressBar() self.progress.setRange(0, 0) # Indeterminate self.progress.hide() layout.addWidget(self.progress) # Buttons buttons = QHBoxLayout() buttons.addStretch() self.btn_cancel = QPushButton("Cancel") self.btn_cancel.setObjectName("secondary") self.btn_cancel.clicked.connect(self._cancel) buttons.addWidget(self.btn_cancel) self.btn_connect = QPushButton("Connect") self.btn_connect.clicked.connect(self._start_pairing) buttons.addWidget(self.btn_connect) layout.addLayout(buttons) def _start_pairing(self): """Start the pairing flow.""" username = self.username_input.text().strip() if not username: self.status_label.setText("Please enter your username") self.status_label.setStyleSheet("color: #e53e3e;") return self.btn_connect.setEnabled(False) self.username_input.setEnabled(False) self.progress.show() self.log_area.clear() self.log_area.show() self.status_label.setStyleSheet("color: #718096;") self._worker = PairingWorker(username) self._worker.success.connect(self._on_success) self._worker.error.connect(self._on_error) self._worker.status.connect(self._on_status) self._worker.start() def _on_success(self, token: str): """Handle successful pairing.""" set_registry_token(token) self.progress.hide() self.status_label.setText("Connected successfully!") self.status_label.setStyleSheet("color: #38a169; font-weight: 600;") QTimer.singleShot(1000, self.accept) def _on_error(self, error: str): """Handle error.""" self.progress.hide() self.btn_connect.setEnabled(True) self.username_input.setEnabled(True) self.status_label.setText(f"Error: {error}") self.status_label.setStyleSheet("color: #e53e3e;") def _on_status(self, status: str): """Handle status update.""" self.status_label.setText(status) # Also append to log for debugging self.log_area.append(status) def _cancel(self): """Cancel pairing.""" if self._worker: self._worker.stop() self._worker.wait(1000) self.reject() def closeEvent(self, event): """Handle close.""" if self._worker: self._worker.stop() self._worker.wait(1000) super().closeEvent(event)