SSH重构: 1.优化ui_page_task模块

2.新增remote_plan_service

Signed-off-by: wuqiongjin <suchinfinity@qq.com>

Author: wuqiongjin

src/extend/remote_plan_service.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+from typing import Callable
+
+from src.extend.remote_linux_runner import RemoteLinuxRunner
+from src.extend.ssh_client import SshClient, SshConfig
+from src.utils.const import WebPath, Key
+from src.utils.utils import Utils
+
+
+class RemotePlanService:
+    def __init__(self, ssh_cfg_getter: Callable[[], SshConfig | None]):
+        self._ssh_cfg_getter = ssh_cfg_getter
+
+    def _ensure_runner_ready(self, remote: RemoteLinuxRunner) -> tuple[bool, str | None]:
+        version = Utils.get_app_version_from_config_json(default="")
+        if not version:
+            return False, "无法获取版本号"
+
+        url = WebPath.LinuxRunnerDownloadUrlTemplate.format(version=version)
+        ok2, err = remote.ensure_installed_from_url(version=version, url=url)
+        if not ok2:
+            return False, err or "远端安装 linux-runner 失败"
+
+        code, out, err2 = remote.set_current(version)
+        if code != 0:
+            return False, (err2 or out or "").strip() or "远端 set_current 失败"
+
+        return True, None
+
+    def cron_create(self, task: dict) -> tuple[bool, str | None]:
+        cfg = self._ssh_cfg_getter()
+        if not cfg:
+            return False, "SSH配置缺失"
+
+        with SshClient(cfg) as ssh:
+            remote = RemoteLinuxRunner(ssh)
+            ok2, err = self._ensure_runner_ready(remote)
+            if not ok2:
+                return False, err
+
+            code, out, err2 = remote.cron_create(task)
+            if code != 0:
+                return False, (err2 or out or "").strip() or "远端创建 crontab 失败"
+
+        return True, None
+
+    def cron_delete(self, task_names) -> tuple[bool, str | None]:
+        cfg = self._ssh_cfg_getter()
+        if not cfg:
+            return False, "SSH配置缺失"
+
+        if task_names is None:
+            return False, "task_names为空"
+
+        if isinstance(task_names, str):
+            task_names = [task_names]
+
+        with SshClient(cfg) as ssh:
+            remote = RemoteLinuxRunner(ssh)
+            ok2, err = self._ensure_runner_ready(remote)
+            if not ok2:
+                return False, err
+
+            for name in task_names:
+                if not name:
+                    continue
+                code, out, err2 = remote.cron_delete(name)
+                if code != 0:
+                    return False, (err2 or out or "").strip() or "远端删除 crontab 失败"
+
+        return True, None
+
+    @staticmethod
+    def task_names_from_plan(task: dict):
+        plan_name = task.get(Key.SystemPlanName)
+        if task.get(Key.TriggerType) == Key.Multiple and isinstance(plan_name, dict):
+            return [plan_name.get(k) for k in plan_name]
+        return plan_name

ui/main_window/auto_clock_window.py

@@ -12,6 +12,7 @@
 from src.extend.ssh_client import SshClient, SshConfig
 from src.store.settings_store import SettingsStore
 from src.store.data_store import LocalDataStore, RemoteDataStore, IDataStore
+from src.extend.remote_plan_service import RemotePlanService
 
 
 class AutoClockWindow(MainWindow):
@@ -41,6 +42,8 @@ def __init__(self):
         self._remote_data_root_abs = None
         self._remote_ssh_cfg: SshConfig | None = None
 
+        self._remote_plan_service: RemotePlanService | None = None
+
         self._data_store: IDataStore = LocalDataStore()
 
         self.load_data_json()
@@ -129,6 +132,8 @@ def connect_remote_and_reload(self) -> tuple[bool, str | None]:
             self._remote_data_root_abs = store.remote_data_root_abs
             self._remote_ssh_cfg = cfg
             self._data_store = store
+
+            self._remote_plan_service = RemotePlanService(lambda: self._remote_ssh_cfg)
             self.load_data_json()
             self._reload_pages()
             self.refresh_ssh_status()
@@ -139,6 +144,7 @@ def connect_remote_and_reload(self) -> tuple[bool, str | None]:
             self._remote_home_dir = None
             self._remote_data_root_abs = None
             self._remote_ssh_cfg = None
+            self._remote_plan_service = None
             self.refresh_ssh_status()
             return False, str(e)
 
@@ -149,6 +155,8 @@ def disconnect_remote_and_reload(self):
         self._remote_data_root_abs = None
         self._remote_ssh_cfg = None
 
+        self._remote_plan_service = None
+
         self._data_store = LocalDataStore()
 
         AppPath.clear_remote()
@@ -312,6 +320,10 @@ def sync_remote_tasks_json(self) -> bool:
     def data_store(self) -> IDataStore:
         return self._data_store
 
+    @property
+    def remote_plan_service(self) -> RemotePlanService | None:
+        return self._remote_plan_service
+
     def write_tasks_list(self, tasks) -> bool:
         return self._data_store.write_tasks(tasks)
 

ui/pages/ui_page_task.py

@@ -14,8 +14,6 @@
 from ui.custom.custom_widget import TaskListWidget
 from ui.template.ui_page import PageContent, Container
 from src.ui.ui_system_plan import SystemPlanDialog
-from src.extend.ssh_client import SshClient, SshConfig
-from src.extend.remote_linux_runner import RemoteLinuxRunner
 from src.utils.const import WebPath
 
 # 根据操作系统导入相应的模块
@@ -100,6 +98,15 @@ def _write_tasks(self, tasks) -> bool:
         except Exception:
             return False
 
+    def _get_remote_plan_service(self):
+        try:
+            w = self.window()
+            if hasattr(w, "remote_plan_service"):
+                return w.remote_plan_service
+        except Exception:
+            pass
+        return None
+
     def _read_config(self) -> dict:
         try:
             w = self.window()
@@ -250,33 +257,10 @@ def do_create_plan(self):
                                 MessageBox("请先在 Settings -> SSH Settings 点击“连接”，连接成功后再创建远端计划任务")
                                 return
 
-                            host = str(self._get_cfg(Key.SshHost, "") or "").strip()
-                            username = str(self._get_cfg(Key.SshUsername, "") or "").strip()
-                            password = str(self._get_cfg(Key.SshPassword, "") or "")
-                            use_pkey = bool(self._get_cfg(Key.SshUsePrivateKey, False))
-                            pkey_path = str(self._get_cfg(Key.SshPrivateKeyPath, "") or "").strip()
-                            version = Utils.get_app_version_from_config_json(default="")
-                            if not host or not username or not version:
-                                raise Exception("SSH配置不完整或无法获取版本号")
-                            url = WebPath.LinuxRunnerDownloadUrlTemplate.format(version=version)
-                            ssh_cfg = SshConfig(
-                                host=host,
-                                username=username,
-                                password=None if use_pkey else password,
-                                pkey_path=pkey_path if use_pkey else None,
-                            )
-                            with SshClient(ssh_cfg) as ssh:
-                                remote = RemoteLinuxRunner(ssh)
-                                ok2, err = remote.ensure_installed_from_url(version=version, url=url)
-                                if not ok2:
-                                    raise Exception(err or "远端安装 linux-runner 失败")
-                                code, out, err2 = remote.set_current(version)
-                                if code != 0:
-                                    raise Exception((err2 or out or "").strip() or "远端 set_current 失败")
-                                code, out, err2 = remote.cron_create(task)
-                                if code != 0:
-                                    raise Exception((err2 or out or "").strip() or "远端创建 crontab 失败")
-                            ok, error = True, None
+                            svc = self._get_remote_plan_service()
+                            if svc is None:
+                                raise Exception("远端服务未初始化，请重新连接SSH")
+                            ok, error = svc.cron_create(task)
                         else:
                             ok, error = create_task(task)
                     elif platform.system() == "Linux":
@@ -331,39 +315,12 @@ def delete_system_plan(self):
                         MessageBox("请先在 Settings -> SSH Settings 点击“连接”，连接成功后再删除远端计划任务")
                         return
 
-                    host = str(self._get_cfg(Key.SshHost, "") or "").strip()
-                    username = str(self._get_cfg(Key.SshUsername, "") or "").strip()
-                    password = str(self._get_cfg(Key.SshPassword, "") or "")
-                    use_pkey = bool(self._get_cfg(Key.SshUsePrivateKey, False))
-                    pkey_path = str(self._get_cfg(Key.SshPrivateKeyPath, "") or "").strip()
-                    version = Utils.get_app_version_from_config_json(default="")
-                    if not host or not username or not version:
-                        raise Exception("SSH配置不完整或无法获取版本号")
-                    url = WebPath.LinuxRunnerDownloadUrlTemplate.format(version=version)
-                    ssh_cfg = SshConfig(
-                        host=host,
-                        username=username,
-                        password=None if use_pkey else password,
-                        pkey_path=pkey_path if use_pkey else None,
-                    )
-                    with SshClient(ssh_cfg) as ssh:
-                        remote = RemoteLinuxRunner(ssh)
-                        ok2, err = remote.ensure_installed_from_url(version=version, url=url)
-                        if not ok2:
-                            raise Exception(err or "远端安装 linux-runner 失败")
-                        code, out, err2 = remote.set_current(version)
-                        if code != 0:
-                            raise Exception((err2 or out or "").strip() or "远端 set_current 失败")
-
-                        if delete_task[Key.TriggerType] == Key.Multiple and isinstance(plan_name, dict):
-                            for task_name in plan_name:
-                                code, out, err2 = remote.cron_delete(plan_name.get(task_name))
-                                if code != 0:
-                                    raise Exception((err2 or out or "").strip() or "远端删除 crontab 失败")
-                        else:
-                            code, out, err2 = remote.cron_delete(plan_name)
-                            if code != 0:
-                                raise Exception((err2 or out or "").strip() or "远端删除 crontab 失败")
+                    svc = self._get_remote_plan_service()
+                    if svc is None:
+                        raise Exception("远端服务未初始化，请重新连接SSH")
+
+                    names = svc.task_names_from_plan(delete_task)
+                    ok, error = svc.cron_delete(names)
                 else:
                     if delete_task[Key.TriggerType] == Key.Multiple and isinstance(plan_name, dict):
                         for task_name in plan_name: