Add headless task runner and CLI entrypoint

ok-oldking · ok-oldking · commit a4c15e47aa47 · 2026-05-16T22:26:32.000+08:00
Introduce headless execution support and a CLI for running one-off tasks without the GUI. Adds HeadlessApp, OK.headless_app, OK.run_task and run_task helper to select tasks by index, name, class or instance, plus safer wait/join and exit handling. Integrates argument parsing (parse_arguments_to_map) and switches TaskManager to use the headless app when appropriate; enforces ASCII install-path failure in headless mode. StartController.do_start now returns a boolean success value and TaskManager avoids initializing the debug file watcher in headless mode. Add ok.cli module and ok.__main__ plus a console_scripts entry in setup.py to expose the ok command.
diff --git a/ok/__init__.py b/ok/__init__.py
@@ -28,7 +28,7 @@
 from ok.util.config import Config, ConfigOption
 from ok.util.handler import Handler, ExitEvent
 from ok.util.logger import config_logger, Logger
-from ok.util.process import check_mutex, get_first_gpu_free_memory_mib
+from ok.util.process import check_mutex, get_first_gpu_free_memory_mib, parse_arguments_to_map
 from ok.util.file import get_path_relative_to_exe, install_path_isascii
 from ok.util.window import windows_graphics_available
 from ok.device.intercation import DoNothingInteraction, BaseInteraction, BrowserInteraction, PostMessageInteraction, \
@@ -238,6 +238,57 @@ def handle_sigint(signum, frame):
 
         sys.exit(self.app.exec())
 
+
+class HeadlessApp:
+    """Small app facade for running tasks without creating any UI windows."""
+
+    def __init__(self, config, exit_event=None):
+        og.exit_event = exit_event
+        og.handler = Handler(exit_event, 'global')
+        self.config = config
+        self.debug = config.get('debug', False)
+        self.headless = True
+        self.global_config = og.global_config
+        self.icon = None
+        self.exit_event = exit_event
+        self.po_translation = None
+        self.to_translate = None
+
+        from ok.gui.common.config import cfg
+        self.locale = cfg.get(cfg.language).value
+        from ok.gui.StartController import StartController
+        self.start_controller = StartController(self.config, exit_event)
+
+        og.app = self
+        if my_app := self.config.get('my_app'):
+            og.my_app = init_class_by_name(my_app[0], my_app[1], exit_event)
+        logger.debug('init headless app end')
+
+    def tr(self, key):
+        if not key:
+            return key
+        if ok_tr := QCoreApplication.translate("app", key):
+            if ok_tr != key:
+                return ok_tr
+        if self.po_translation is None:
+            locale_name = self.locale.name()
+            try:
+                from ok.gui.i18n.GettextTranslator import get_translations
+                self.po_translation = get_translations(locale_name)
+                self.po_translation.install()
+                logger.info(f'headless translation installed for {locale_name}')
+            except:
+                logger.error(f'install headless translations error for {locale_name}')
+                self.po_translation = "Failed"
+        if self.po_translation != 'Failed':
+            return self.po_translation.gettext(key)
+        return key
+
+    def quit(self):
+        if self.exit_event:
+            self.exit_event.set()
+
+
 def get_my_id():
     mac = uuid.getnode()
     value_with_salt = 'mac123:' + str(mac)
@@ -294,8 +345,10 @@ def __init__(self, config):
         logger.info(
             f"pyappify  app_version:{pyappify.app_version}, app_profile:{pyappify.app_profile}, pyappify_version:{pyappify.pyappify_version} pyappify_upgradeable:{pyappify.pyappify_upgradeable}, pyappify_executable:{pyappify.pyappify_executable}")
         config['debug'] = config.get("debug", False)
+        self.args = parse_arguments_to_map()
         self.task_executor = None
         self._app = None
+        self._headless_app = None
         self.debug = config['debug']
         self.global_config = GlobalConfig(config.get('global_configs'))
         windows_config = config.get('windows')
@@ -337,9 +390,21 @@ def app(self):
             self._app = App(self.config, self.task_executor, self.exit_event)
         return self._app
 
+    @property
+    def headless_app(self):
+        if self._headless_app is None:
+            self._headless_app = HeadlessApp(self.config, self.exit_event)
+        return self._headless_app
+
+    def should_init_task_manager_headless(self):
+        return not self.config.get("use_gui") or self.args.get('task', 0) > 0
+
     def start(self):
         logger.info(f'OK start id:{id(self)} pid:{os.getpid()}')
         try:
+            if self.args.get('task', 0) > 0:
+                self.run_task(self.args.get('task'))
+                return
             if self.config.get("use_gui"):
                 if not self.init_error:
                     self.app.show_main_window()
@@ -374,9 +439,76 @@ def start(self):
         except Exception as e:
             logger.error("start error", e)
             self.exit_event.set()
-            if self.app:
+            if self._app or self._headless_app:
                 self.quit()
 
+    def run_task(self, task=1):
+        """
+        Run a one-time task without showing the main UI.
+
+        Args:
+            task: 1-based one-time task index, task name, task class, or task instance.
+        """
+        task = self.get_onetime_task(task)
+        logger.info(f'run task without ui: {task.name}')
+        started = self.headless_app.start_controller.do_start(task, exit_after=True)
+        if not started:
+            raise RuntimeError(f'Start task failed: {task.name}')
+        self.wait_task(task)
+        return True
+
+    def get_onetime_task(self, task):
+        if isinstance(task, int):
+            task_index = task - 1
+            if task_index < 0 or task_index >= len(self.task_executor.onetime_tasks):
+                raise IndexError(
+                    f'Task index {task} is out of range. Available range is 1-{len(self.task_executor.onetime_tasks)}')
+            return self.task_executor.onetime_tasks[task_index]
+        if isinstance(task, str):
+            normalized_task = task.lower()
+            exact_matches = [
+                candidate for candidate in self.task_executor.onetime_tasks
+                if candidate.name.lower() == normalized_task or candidate.__class__.__name__.lower() == normalized_task
+            ]
+            if len(exact_matches) == 1:
+                return exact_matches[0]
+            if len(exact_matches) > 1:
+                names = ', '.join(candidate.__class__.__name__ for candidate in exact_matches)
+                raise ValueError(f'Multiple tasks matched "{task}": {names}')
+
+            partial_matches = [
+                candidate for candidate in self.task_executor.onetime_tasks
+                if normalized_task in candidate.name.lower()
+                or normalized_task in candidate.__class__.__name__.lower()
+            ]
+            if len(partial_matches) == 1:
+                return partial_matches[0]
+            if len(partial_matches) > 1:
+                names = ', '.join(candidate.__class__.__name__ for candidate in partial_matches)
+                raise ValueError(f'Multiple tasks matched "{task}": {names}')
+
+            for candidate in self.task_executor.onetime_tasks:
+                if candidate.name == task or candidate.__class__.__name__ == task:
+                    return candidate
+            raise ValueError(f'Task not found: {task}')
+        if isinstance(task, type):
+            for candidate in self.task_executor.onetime_tasks:
+                if isinstance(candidate, task):
+                    return candidate
+            if not issubclass(task, BaseTask):
+                raise ValueError(f'Task class must inherit BaseTask: {task}')
+            if issubclass(task, TriggerTask):
+                raise ValueError(f'run_task only supports one-time BaseTask classes, got TriggerTask: {task}')
+            task_instance = task(executor=self.task_executor, app=self.headless_app)
+            task_instance.after_init(executor=self.task_executor, scene=self.task_executor.scene)
+            task_instance.post_init()
+            self.task_executor.onetime_tasks.append(task_instance)
+            logger.info(f'created headless task from class: {task_instance}')
+            return task_instance
+        if task in self.task_executor.onetime_tasks:
+            return task
+        raise ValueError(f'Unsupported one-time task selector: {task}')
+
     def do_init(self):
         logger.info(f"do_init, config: {self.config}")
         self.init_device_manager()
@@ -401,9 +533,12 @@ def do_init(self):
             ocr_target_height = ocr.get('target_height', 0)
             if not isascii:
                 logger.info(f'show_path_ascii_error')
-                self.app.show_path_ascii_error(path)
                 self.init_error = True
-                self.app.exec()
+                if self.should_init_task_manager_headless():
+                    raise RuntimeError(f'Install dir {path} must be an English path, move to another path.')
+                else:
+                    self.app.show_path_ascii_error(path)
+                    self.app.exec()
                 return False
 
         self.task_executor = TaskExecutor(self.device_manager, exit_event=self.exit_event,
@@ -413,17 +548,29 @@ def do_init(self):
                                           global_config=self.global_config, ocr_target_height=ocr_target_height,
                                           config=self.config)
         from ok.gui.tasks.TaskManger import TaskManager
-        og.task_manager = TaskManager(task_executor=self.task_executor, app=self.app,
+        task_app = self.headless_app if self.should_init_task_manager_headless() else self.app
+        og.task_manager = TaskManager(task_executor=self.task_executor, app=task_app,
                                       onetime_tasks=self.config.get('onetime_tasks', []),
                                       trigger_tasks=self.config.get('trigger_tasks', []),
                                       scene=self.config.get('scene'))
         og.executor = self.task_executor
         logger.info(f"do_init, end")
         return True
 
-    def wait_task(self):
-        while not self.exit_event.is_set():
-            time.sleep(1)
+    def wait_task(self, task=None):
+        try:
+            while not self.exit_event.is_set():
+                if task is not None and not task.enabled and self.task_executor.current_task is not task:
+                    logger.info(f'task finished without ui: {task.name}')
+                    break
+                time.sleep(1)
+        except KeyboardInterrupt:
+            logger.info("Keyboard interrupt received, exiting script.")
+        finally:
+            if task is not None:
+                self.exit_event.set()
+                if self.task_executor.thread and self.task_executor.thread != threading.current_thread():
+                    self.task_executor.thread.join(timeout=10)
 
     def console_handler(self, event):
         import win32con
@@ -450,6 +597,8 @@ def quit(self):
         self.exit_event.set()
         if self._app:
             self._app.quit()
+        if self._headless_app:
+            self._headless_app.quit()
 
     def init_device_manager(self):
         if self.device_manager is None:
@@ -458,6 +607,23 @@ def init_device_manager(self):
             og.device_manager = self.device_manager
 
 
+def run_task(config, task=1, debug=False):
+    """
+    Convenience entrypoint for scripts that only need to run one task.
+
+    Example:
+        from ok import run_task
+        from src.config import config
+
+        if __name__ == "__main__":
+            run_task(config, task=1)
+    """
+    headless_config = dict(config)
+    headless_config["use_gui"] = False
+    headless_config["debug"] = debug
+    return OK(headless_config).run_task(task)
+
+
 class BaseScene:
     def reset(self):
         pass
diff --git a/ok/__main__.py b/ok/__main__.py
@@ -0,0 +1,5 @@
+from ok.cli import main
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/ok/cli.py b/ok/cli.py
@@ -0,0 +1,83 @@
+import argparse
+import importlib
+import os
+import sys
+
+
+DEFAULT_CONFIG_TARGETS = ("src.config:config", "config:config")
+
+
+def _split_target(target):
+    if ":" in target:
+        module_name, attr_name = target.split(":", 1)
+    else:
+        module_name, attr_name = target, "config"
+    if not module_name or not attr_name:
+        raise ValueError(f"Invalid config target: {target}")
+    return module_name, attr_name
+
+
+def load_config(target=None):
+    cwd = os.getcwd()
+    if cwd not in sys.path:
+        sys.path.insert(0, cwd)
+
+    targets = (target,) if target else DEFAULT_CONFIG_TARGETS
+    errors = []
+    for item in targets:
+        module_name, attr_name = _split_target(item)
+        try:
+            module = importlib.import_module(module_name)
+            return getattr(module, attr_name)
+        except Exception as e:
+            errors.append(f"{item}: {e}")
+
+    details = "\n".join(errors)
+    raise RuntimeError(
+        "Could not load config. Tried:\n"
+        f"{details}\n"
+        "Use --config module:attribute to point to your project config."
+    )
+
+
+def run_task_command(args):
+    task = args.task_name or args.task
+    if not task:
+        raise ValueError("Task name is required. Use: ok run_task TaskName")
+
+    from ok import run_task
+
+    config = load_config(args.config)
+    run_task(config, task=task, debug=args.debug)
+    return 0
+
+
+def build_parser():
+    parser = argparse.ArgumentParser(prog="ok")
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    run_task_parser = subparsers.add_parser("run_task", help="Run a task without starting the UI")
+    run_task_parser.add_argument("task_name", nargs="?", help="Task name or class name to match")
+    run_task_parser.add_argument("-t", "--task", help="Task name or class name to match")
+    run_task_parser.add_argument("-d", "--debug", action="store_true", help="Run with debug mode enabled")
+    run_task_parser.add_argument(
+        "-c",
+        "--config",
+        help="Config import target, for example src.config:config or config:config",
+    )
+    run_task_parser.set_defaults(func=run_task_command)
+
+    return parser
+
+
+def main(argv=None):
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    try:
+        return args.func(args)
+    except Exception as e:
+        parser.exit(1, f"ok: error: {e}\n")
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/ok/gui/StartController.py b/ok/gui/StartController.py
@@ -33,12 +33,12 @@ def do_start(self, task=None, exit_after=False):
         except Exception as e:
             logger.error(f'do_start do_refresh exception: {e}', e)
             communicate.starting_emulator.emit(True, self.tr(str(e)), 0)
-            return
+            return False
 
         try:
             if self.start_exe:
                 if not self.start_device():
-                    return
+                    return False
             else:
                 logger.info('windows.start_exe is False, skip start_device')
 
@@ -66,9 +66,11 @@ def add_task_to_enable(enable_task):
 
             og.executor.start()
             communicate.starting_emulator.emit(True, None, 0)
+            return True
         except Exception as e:
             logger.error(f'do_start exception: {e}', e)
             communicate.starting_emulator.emit(True, self.tr(f'Start failed: {e}'), 0)
+            return False
 
     def start_device(self):
         device = og.device_manager.get_preferred_device()
diff --git a/ok/gui/tasks/TaskManger.py b/ok/gui/tasks/TaskManger.py
@@ -42,7 +42,7 @@ def __init__(self, task_executor, app, trigger_tasks=[], onetime_tasks=[], scene
             task.post_init()
         self.load_user_tasks()
         self.load_imported_tasks()
-        if self.debug or self.custom_tasks_enabled:
+        if (self.debug or self.custom_tasks_enabled) and not getattr(app, 'headless', False):
             self._init_debug_file_watcher()
 
     def init_tasks(self, task_classes):
diff --git a/setup.py b/setup.py
@@ -43,6 +43,11 @@
         'pycaw==20240210',
         'mouse==0.7.1'
     ],
+    entry_points={
+        'console_scripts': [
+            'ok=ok.cli:main',
+        ],
+    },
     python_requires='==3.12.*',
     zip_safe=False,
 )
