429 lines
19 KiB
Python
429 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Literal
|
|
|
|
from PySide6.QtCore import Qt
|
|
from PySide6.QtWidgets import (
|
|
QApplication,
|
|
QDialog,
|
|
QHBoxLayout,
|
|
QLabel,
|
|
QMessageBox,
|
|
QPushButton,
|
|
QVBoxLayout,
|
|
)
|
|
|
|
|
|
class SingBoxRuntimeProfilesMixin:
|
|
def on_singbox_profile_edit_dialog(self, cid: str = "") -> None:
|
|
def work():
|
|
target = str(cid or "").strip() or self._selected_transport_engine_id()
|
|
if not target:
|
|
raise RuntimeError("Select a transport engine first")
|
|
if not self._select_transport_engine_by_id(target):
|
|
raise RuntimeError(f"Transport engine '{target}' not found")
|
|
|
|
self._sync_selected_singbox_profile_link(silent=True)
|
|
self._load_singbox_editor_for_selected(silent=False)
|
|
client = self._selected_transport_client()
|
|
pid = self._selected_singbox_profile_id()
|
|
if client is None or not pid:
|
|
raise RuntimeError("Select a SingBox profile first")
|
|
|
|
profile_name = self.ent_singbox_proto_name.text().strip() or str(getattr(client, "name", "") or pid).strip()
|
|
host_layout = self.grp_singbox_profile_settings.layout()
|
|
if host_layout is None:
|
|
raise RuntimeError("internal layout is unavailable")
|
|
editor = self.grp_singbox_proto_editor
|
|
insert_at = host_layout.indexOf(editor)
|
|
if insert_at >= 0:
|
|
host_layout.removeWidget(editor)
|
|
|
|
moved = False
|
|
dlg = QDialog(self)
|
|
dlg.setModal(True)
|
|
dlg.setWindowTitle(f"Edit SingBox profile: {profile_name}")
|
|
dlg.resize(860, 680)
|
|
dlg_layout = QVBoxLayout(dlg)
|
|
try:
|
|
hint = QLabel("Edit protocol fields and save draft. Use profile card menu for Run/Delete.")
|
|
hint.setStyleSheet("color: gray;")
|
|
dlg_layout.addWidget(hint)
|
|
|
|
editor.setTitle(f"{self._singbox_editor_default_title} · {profile_name}")
|
|
editor.setParent(dlg)
|
|
editor.setVisible(True)
|
|
moved = True
|
|
dlg_layout.addWidget(editor, stretch=1)
|
|
|
|
actions = QHBoxLayout()
|
|
btn_save = QPushButton("Save draft")
|
|
btn_close = QPushButton("Close")
|
|
actions.addWidget(btn_save)
|
|
actions.addStretch(1)
|
|
actions.addWidget(btn_close)
|
|
dlg_layout.addLayout(actions)
|
|
|
|
def save_draft_clicked() -> None:
|
|
try:
|
|
selected_client, _eid, selected_pid = self._selected_singbox_profile_context()
|
|
saved = self._save_singbox_editor_draft(selected_client, profile_id=selected_pid)
|
|
line = (saved.pretty_text or "").strip() or f"save profile {selected_pid}"
|
|
self._append_transport_log(f"[profile] {line}")
|
|
self.ctrl.log_gui(f"[singbox-profile] {line}")
|
|
self.lbl_transport_engine_meta.setText(f"Engine: profile {selected_pid} draft saved")
|
|
self.lbl_transport_engine_meta.setStyleSheet("color: green;")
|
|
self._render_singbox_profile_cards()
|
|
self._sync_singbox_profile_card_selection(self._selected_transport_engine_id())
|
|
QMessageBox.information(dlg, "SingBox profile", line)
|
|
except Exception as e:
|
|
QMessageBox.critical(dlg, "SingBox profile save error", str(e))
|
|
|
|
btn_save.clicked.connect(save_draft_clicked)
|
|
btn_close.clicked.connect(dlg.accept)
|
|
dlg.exec()
|
|
finally:
|
|
if moved:
|
|
dlg_layout.removeWidget(editor)
|
|
editor.setParent(self.grp_singbox_profile_settings)
|
|
editor.setTitle(self._singbox_editor_default_title)
|
|
if insert_at >= 0:
|
|
host_layout.insertWidget(insert_at, editor)
|
|
else:
|
|
host_layout.addWidget(editor)
|
|
editor.setVisible(False)
|
|
|
|
self._safe(work, title="SingBox profile edit error")
|
|
|
|
def on_transport_engine_action(
|
|
self,
|
|
action: Literal["provision", "start", "stop", "restart"],
|
|
) -> None:
|
|
def work():
|
|
cid = self._selected_transport_engine_id()
|
|
if not cid:
|
|
raise RuntimeError("Select a transport engine first")
|
|
|
|
self.lbl_transport_engine_meta.setText(f"Engine: {action} {cid}...")
|
|
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
|
|
QApplication.processEvents()
|
|
|
|
if action == "start":
|
|
selected_client = self._selected_transport_client()
|
|
if selected_client is not None and str(getattr(selected_client, "kind", "") or "").strip().lower() == "singbox":
|
|
_client, _eid, pid = self._selected_singbox_profile_context()
|
|
self.lbl_transport_engine_meta.setText(f"Engine: preparing profile {pid} for start...")
|
|
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
|
|
QApplication.processEvents()
|
|
pre = self.ctrl.singbox_profile_apply_action(
|
|
pid,
|
|
client_id=cid,
|
|
restart=False,
|
|
skip_runtime=True,
|
|
check_binary=True,
|
|
client=selected_client,
|
|
)
|
|
pre_line = (pre.pretty_text or "").strip() or f"apply profile {pid}"
|
|
self._append_transport_log(f"[profile] {pre_line}")
|
|
self.ctrl.log_gui(f"[singbox-profile] {pre_line}")
|
|
if not pre.ok:
|
|
raise RuntimeError(f"profile preflight failed: {pre_line}")
|
|
|
|
ok, msg = self._apply_transport_switch_policy(cid)
|
|
self._append_transport_log(f"[switch] {msg}")
|
|
self.ctrl.log_gui(f"[transport-switch] {msg}")
|
|
if not ok:
|
|
if "canceled by user" in msg.lower():
|
|
self.refresh_transport_engines(silent=True)
|
|
return
|
|
raise RuntimeError(msg)
|
|
|
|
res = self.ctrl.transport_client_action(cid, action if action != "start" else "start")
|
|
line = (res.pretty_text or "").strip() or f"{action} {cid}"
|
|
self._append_transport_log(f"[engine] {line}")
|
|
self.ctrl.log_gui(f"[transport-engine] {line}")
|
|
if not res.ok:
|
|
raise RuntimeError(line)
|
|
|
|
self.refresh_transport_engines(silent=True)
|
|
self.refresh_status_tab()
|
|
|
|
self._safe(work, title="Transport engine error")
|
|
|
|
def on_transport_engine_delete(self, cid: str = "") -> None:
|
|
def work():
|
|
target = str(cid or "").strip() or self._selected_transport_engine_id()
|
|
if not target:
|
|
raise RuntimeError("Select a transport engine first")
|
|
if not self._select_transport_engine_by_id(target):
|
|
raise RuntimeError(f"Transport engine '{target}' not found")
|
|
|
|
ans = QMessageBox.question(
|
|
self,
|
|
"Delete transport profile",
|
|
(
|
|
f"Delete profile '{target}'?\n\n"
|
|
"The client configuration and related runtime artifacts will be removed."
|
|
),
|
|
QMessageBox.Yes | QMessageBox.No,
|
|
QMessageBox.No,
|
|
)
|
|
if ans != QMessageBox.Yes:
|
|
return
|
|
|
|
self.lbl_transport_engine_meta.setText(f"Engine: deleting {target}...")
|
|
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
|
|
QApplication.processEvents()
|
|
|
|
res = self.ctrl.transport_client_delete_action(target, force=False, cleanup=True)
|
|
if not res.ok and "force=true" in (res.pretty_text or "").lower():
|
|
force_ans = QMessageBox.question(
|
|
self,
|
|
"Profile is referenced",
|
|
(
|
|
"This profile is referenced by current transport policy.\n"
|
|
"Force delete anyway?"
|
|
),
|
|
QMessageBox.Yes | QMessageBox.No,
|
|
QMessageBox.No,
|
|
)
|
|
if force_ans == QMessageBox.Yes:
|
|
res = self.ctrl.transport_client_delete_action(target, force=True, cleanup=True)
|
|
else:
|
|
self._append_transport_log(f"[engine] delete {target}: canceled by user")
|
|
self.ctrl.log_gui(f"[transport-engine] delete {target}: canceled by user")
|
|
return
|
|
|
|
line = (res.pretty_text or "").strip() or f"delete {target}"
|
|
self._append_transport_log(f"[engine] {line}")
|
|
self.ctrl.log_gui(f"[transport-engine] {line}")
|
|
if not res.ok:
|
|
raise RuntimeError(line)
|
|
|
|
self.refresh_transport_engines(silent=True)
|
|
self.refresh_status_tab()
|
|
|
|
self._safe(work, title="Transport engine delete error")
|
|
|
|
def on_transport_policy_rollback(self) -> None:
|
|
def work():
|
|
self.lbl_transport_engine_meta.setText("Engine: rollback policy...")
|
|
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
|
|
QApplication.processEvents()
|
|
|
|
res = self.ctrl.transport_policy_rollback_action()
|
|
line = (res.pretty_text or "").strip() or "policy rollback"
|
|
self._append_transport_log(f"[switch] {line}")
|
|
self.ctrl.log_gui(f"[transport-switch] {line}")
|
|
if not res.ok:
|
|
raise RuntimeError(line)
|
|
|
|
self.refresh_transport_engines(silent=True)
|
|
self.refresh_status_tab()
|
|
|
|
self._safe(work, title="Transport rollback error")
|
|
|
|
def on_toggle_singbox_profile_settings(self, checked: bool = False) -> None:
|
|
if checked and self.btn_singbox_toggle_global_defaults.isChecked():
|
|
self.btn_singbox_toggle_global_defaults.setChecked(False)
|
|
self._apply_singbox_compact_visibility()
|
|
self._save_ui_preferences()
|
|
|
|
def on_toggle_singbox_global_defaults(self, checked: bool = False) -> None:
|
|
if checked and self.btn_singbox_toggle_profile_settings.isChecked():
|
|
self.btn_singbox_toggle_profile_settings.setChecked(False)
|
|
self._apply_singbox_compact_visibility()
|
|
self._save_ui_preferences()
|
|
|
|
def on_toggle_singbox_activity(self, _checked: bool = False) -> None:
|
|
self._apply_singbox_compact_visibility()
|
|
self._save_ui_preferences()
|
|
|
|
def on_singbox_profile_scope_changed(self, _state: int = 0) -> None:
|
|
self._apply_singbox_profile_controls()
|
|
self._save_ui_preferences()
|
|
self._update_transport_engine_view()
|
|
|
|
def on_singbox_global_defaults_changed(self, _index: int = 0) -> None:
|
|
self._refresh_singbox_profile_effective()
|
|
self._save_ui_preferences()
|
|
self._update_transport_engine_view()
|
|
|
|
def on_singbox_global_save(self) -> None:
|
|
def work():
|
|
self._save_ui_preferences()
|
|
route, dns, killswitch = self._effective_singbox_policy()
|
|
msg = (
|
|
"Global defaults saved: "
|
|
f"routing={self._singbox_value_label('routing', route)}, "
|
|
f"dns={self._singbox_value_label('dns', dns)}, "
|
|
f"kill-switch={self._singbox_value_label('killswitch', killswitch)}"
|
|
)
|
|
self._append_transport_log(f"[profile] {msg}")
|
|
self.ctrl.log_gui(f"[singbox-settings] {msg}")
|
|
self._safe(work, title="SingBox settings error")
|
|
|
|
def on_singbox_profile_save(self) -> None:
|
|
def work():
|
|
client, eid, pid = self._selected_singbox_profile_context()
|
|
self._save_ui_preferences()
|
|
self.lbl_transport_engine_meta.setText(f"Engine: saving draft for {pid}...")
|
|
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
|
|
QApplication.processEvents()
|
|
|
|
saved = self._save_singbox_editor_draft(client, profile_id=pid)
|
|
save_line = (saved.pretty_text or "").strip() or f"save profile {pid}"
|
|
self._append_transport_log(f"[profile] {save_line}")
|
|
self.ctrl.log_gui(f"[singbox-profile] {save_line}")
|
|
|
|
route, dns, killswitch = self._effective_singbox_policy()
|
|
msg = (
|
|
f"profile settings saved for {eid}: "
|
|
f"routing={self._singbox_value_label('routing', route)}, "
|
|
f"dns={self._singbox_value_label('dns', dns)}, "
|
|
f"kill-switch={self._singbox_value_label('killswitch', killswitch)}"
|
|
)
|
|
self._append_transport_log(f"[profile] {msg}")
|
|
self.ctrl.log_gui(f"[singbox-profile] {msg}")
|
|
self.lbl_transport_engine_meta.setText(f"Engine: profile {pid} draft saved")
|
|
self.lbl_transport_engine_meta.setStyleSheet("color: green;")
|
|
self.refresh_transport_engines(silent=True)
|
|
self._safe(work, title="SingBox profile save error")
|
|
|
|
def _selected_singbox_profile_context(self):
|
|
client = self._selected_transport_client()
|
|
eid = self._selected_transport_engine_id()
|
|
pid = self._selected_singbox_profile_id()
|
|
if not eid or client is None:
|
|
raise RuntimeError("Select a transport engine first")
|
|
if not pid:
|
|
raise RuntimeError("Select a SingBox profile first")
|
|
return client, eid, pid
|
|
|
|
def _run_singbox_profile_action(
|
|
self,
|
|
*,
|
|
verb: str,
|
|
runner,
|
|
refresh_status: bool = False,
|
|
sync_draft: bool = False,
|
|
) -> None:
|
|
client, eid, pid = self._selected_singbox_profile_context()
|
|
if sync_draft:
|
|
self.lbl_transport_engine_meta.setText(f"Engine: syncing draft for {pid}...")
|
|
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
|
|
QApplication.processEvents()
|
|
saved = self._save_singbox_editor_draft(client, profile_id=pid)
|
|
save_line = (saved.pretty_text or "").strip() or f"save profile {pid}"
|
|
self._append_transport_log(f"[profile] {save_line}")
|
|
self.ctrl.log_gui(f"[singbox-profile] {save_line}")
|
|
|
|
self.lbl_transport_engine_meta.setText(f"Engine: {verb} profile {pid}...")
|
|
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
|
|
QApplication.processEvents()
|
|
|
|
res = runner(client, eid, pid)
|
|
line = (res.pretty_text or "").strip() or f"{verb} profile {pid}"
|
|
self._append_transport_log(f"[profile] {line}")
|
|
self.ctrl.log_gui(f"[singbox-profile] {line}")
|
|
|
|
if res.ok:
|
|
self.lbl_transport_engine_meta.setText(f"Engine: profile {pid} {verb} done")
|
|
self.lbl_transport_engine_meta.setStyleSheet("color: green;")
|
|
else:
|
|
self.lbl_transport_engine_meta.setText(f"Engine: profile {pid} {verb} failed")
|
|
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
|
|
|
|
self.refresh_transport_engines(silent=True)
|
|
if refresh_status:
|
|
self.refresh_status_tab()
|
|
|
|
def on_singbox_profile_preview(self) -> None:
|
|
self._safe(
|
|
lambda: self._run_singbox_profile_action(
|
|
verb="previewing",
|
|
runner=lambda client, _eid, pid: self.ctrl.singbox_profile_render_preview_action(
|
|
pid,
|
|
check_binary=None,
|
|
persist=False,
|
|
client=client,
|
|
),
|
|
refresh_status=False,
|
|
sync_draft=True,
|
|
),
|
|
title="SingBox profile preview error",
|
|
)
|
|
|
|
def on_singbox_profile_validate(self) -> None:
|
|
self._safe(
|
|
lambda: self._run_singbox_profile_action(
|
|
verb="validating",
|
|
runner=lambda client, _eid, pid: self.ctrl.singbox_profile_validate_action(
|
|
pid,
|
|
client=client,
|
|
),
|
|
refresh_status=False,
|
|
sync_draft=True,
|
|
),
|
|
title="SingBox profile validate error",
|
|
)
|
|
|
|
def on_singbox_profile_apply(self) -> None:
|
|
self._safe(
|
|
lambda: self._run_singbox_profile_action(
|
|
verb="applying",
|
|
runner=lambda client, eid, pid: self.ctrl.singbox_profile_apply_action(
|
|
pid,
|
|
client_id=eid,
|
|
restart=True,
|
|
skip_runtime=False,
|
|
client=client,
|
|
),
|
|
refresh_status=True,
|
|
sync_draft=True,
|
|
),
|
|
title="SingBox profile apply error",
|
|
)
|
|
|
|
def on_singbox_profile_rollback(self) -> None:
|
|
self._safe(
|
|
lambda: self._run_singbox_profile_action(
|
|
verb="rolling back",
|
|
runner=lambda client, eid, pid: self.ctrl.singbox_profile_rollback_action(
|
|
pid,
|
|
client_id=eid,
|
|
restart=True,
|
|
skip_runtime=False,
|
|
client=client,
|
|
),
|
|
refresh_status=True,
|
|
),
|
|
title="SingBox profile rollback error",
|
|
)
|
|
|
|
def on_singbox_profile_history(self) -> None:
|
|
def work():
|
|
client, _eid, pid = self._selected_singbox_profile_context()
|
|
self.lbl_transport_engine_meta.setText(f"Engine: loading history for {pid}...")
|
|
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
|
|
QApplication.processEvents()
|
|
|
|
lines = self.ctrl.singbox_profile_history_lines(pid, limit=20, client=client)
|
|
if not lines:
|
|
line = f"history profile {pid}: no entries"
|
|
self._append_transport_log(f"[history] {line}")
|
|
self.ctrl.log_gui(f"[singbox-profile] {line}")
|
|
self.lbl_transport_engine_meta.setText(f"Engine: history {pid} is empty")
|
|
self.lbl_transport_engine_meta.setStyleSheet("color: gray;")
|
|
return
|
|
|
|
header = f"history profile {pid}: {len(lines)} entries"
|
|
self._append_transport_log(f"[history] {header}")
|
|
self.ctrl.log_gui(f"[singbox-profile] {header}")
|
|
for ln in lines:
|
|
self._append_transport_log(f"[history] {ln}")
|
|
self.lbl_transport_engine_meta.setText(f"Engine: history loaded for {pid}")
|
|
self.lbl_transport_engine_meta.setStyleSheet("color: green;")
|
|
|
|
self._safe(work, title="SingBox profile history error")
|