cli.py

#!/usr/bin/env python3
# PawFlow CLI

"""
Point d'entree principal de PawFlow.
Usage:
    python cli.py run <flow.json> [--input <file>] [--verbose]
    python cli.py validate <flow.json>
    python cli.py list-tasks
    python cli.py info <flow.json>
    python cli.py serve [--host] [--port] [--reload]
    python cli.py gui [--host] [--port] [--headless]
    python cli.py import <file> [-o output]
    python cli.py cluster status [--api-url]
"""

import argparse
import json
import logging
import os
import sys
import time
from pathlib import Path

# Enregistrer toutes les tasks avant le parsing
import tasks  # noqa: F401 - declenche register_all_tasks()

from core import FlowFile, TaskFactory
from engine.parser import FlowParser, FlowValidator
from engine.continuous_executor import ContinuousFlowExecutor
from engine.provenance import ProvenanceRepository

# Secondary commands live in cli_commands.py to keep this file under 800 lines;
# re-exported here so `cli.cmd_*` stays importable (invariant: import stability).
from cli_commands import (  # noqa: E402,F401
    cmd_admin_user,
    cmd_cluster,
    cmd_re_embed,
    cmd_triggers,
)


def cmd_run(args):
    """Executer un flow depuis un fichier JSON."""
    flow_path = Path(args.flow)
    if not flow_path.exists():
        print(f"ERREUR: Fichier introuvable: {flow_path}")
        return 1

    # Configurer le logging
    level = logging.DEBUG if args.verbose else logging.INFO
    logging.basicConfig(
        level=level,
        format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
        datefmt='%H:%M:%S'
    )
    logger = logging.getLogger('PawFlow')

    # Parser le flow
    logger.info(f"Chargement du flow: {flow_path}")
    try:
        flow = FlowParser.parse_from_file(str(flow_path))
    except Exception as e:
        print(f"ERREUR: Impossible de parser le flow: {e}")
        return 1

    logger.info(f"Flow: {flow.name} (id={flow.id})")
    logger.info(f"Tasks: {list(flow.tasks.keys())}")
    logger.info(f"Relations: {len(flow.relations)}")

    # Valider
    errors = FlowValidator.validate(flow, strict=False)
    if errors:
        for err in errors:
            logger.warning(f"Validation: {err}")

    # Creer les FlowFiles d'entree
    input_flowfiles = []
    if args.input:
        for input_path in args.input:
            p = Path(input_path)
            if not p.exists():
                print(f"ERREUR: Fichier d'entree introuvable: {p}")
                return 1
            content = p.read_bytes()
            ff = FlowFile(
                content=content,
                attributes={
                    'filename': p.name,
                    'path': str(p.parent),
                    'absolute.path': str(p.resolve()),
                    'fileSize': str(len(content)),
                }
            )
            input_flowfiles.append(ff)
            logger.info(f"Input: {p.name} ({len(content)} bytes)")
    else:
        # FlowFile vide par defaut
        input_flowfiles = [FlowFile(content=b'', attributes={'filename': 'stdin'})]
        logger.info("Pas d'input specifie, utilisation d'un FlowFile vide")

    # Provenance
    repo = ProvenanceRepository() if args.provenance else None

    # Parse --param overrides
    param_overrides = {}
    if args.param:
        for p in args.param:
            if '=' not in p:
                print(f"ERREUR: Format invalide pour --param: '{p}' (attendu KEY=VALUE)")
                return 1
            k, v = p.split('=', 1)
            param_overrides[k.strip()] = v.strip()
        logger.info(f"Parameter overrides: {param_overrides}")

    # Execute (batch mode via ContinuousFlowExecutor)
    logger.info("--- Debut de l'execution ---")
    start = time.time()
    result = ContinuousFlowExecutor.run_batch(
        flow,
        input_flowfiles=input_flowfiles,
        parameters=param_overrides if param_overrides else None,
        max_workers=args.workers,
        max_retries=args.retries,
        timeout=args.timeout or 300,
        provenance=repo,
    )
    elapsed = time.time() - start

    # Afficher le resultat
    print()
    if result.success:
        print(f"SUCCES - {flow.name}")
    else:
        print(f"ECHEC - {flow.name}")

    print(f"  Duree: {elapsed:.3f}s")
    print(f"  FlowFiles en sortie: {len(result.output_flowfiles)}")

    if result.errors:
        print("  Erreurs:")
        for err in result.errors:
            print(f"    - {err.get('error', 'unknown')}")

    # Stats par task
    if result.task_statistics:
        print("  Statistiques par tache:")
        for tid, stats in result.task_statistics.items():
            print(f"    {tid} ({stats.task_type}): "
                  f"{stats.success_count} ok, {stats.error_count} err, "
                  f"{stats.avg_duration_ms:.1f}ms avg")

    # Provenance
    if repo and repo.size() > 0:
        prov = repo.to_dict()
        print(f"  Provenance: {prov['total_events']} evenements")
        for etype, count in prov['events_by_type'].items():
            print(f"    {etype}: {count}")

    # Output
    if args.output_dir:
        out_dir = Path(args.output_dir)
        out_dir.mkdir(parents=True, exist_ok=True)
        for i, ff in enumerate(result.output_flowfiles):
            fname = ff.get_attribute('filename', f'output_{i}.dat')
            out_path = out_dir / fname
            out_path.write_bytes(ff.content)
            logger.info(f"Output: {out_path}")
        print(f"  Fichiers ecrits dans: {out_dir}")

    return 0 if result.success else 1


def cmd_validate(args):
    """Valider un flow JSON."""
    flow_path = Path(args.flow)
    if not flow_path.exists():
        print(f"ERREUR: Fichier introuvable: {flow_path}")
        return 1

    try:
        flow = FlowParser.parse_from_file(str(flow_path))
        errors = FlowValidator.validate(flow, strict=False)
    except Exception as e:
        print(f"ERREUR: {e}")
        return 1

    if errors:
        print(f"VALIDATION ECHOUEE pour {flow.name}:")
        for err in errors:
            print(f"  - {err}")
        return 1
    else:
        print(f"VALIDE: {flow.name} ({len(flow.tasks)} tasks, {len(flow.relations)} relations)")
        return 0


def cmd_list_tasks(args):
    """Lister toutes les tasks disponibles."""
    print("Tasks disponibles:")
    print(f"{'TYPE':<25} {'NOM':<30} {'VERSION'}")
    print("-" * 65)
    for task_type in sorted(TaskFactory.list_types()):
        cls = TaskFactory.get(task_type)
        print(f"{cls.TYPE:<25} {cls.NAME:<30} {cls.VERSION}")


def cmd_info(args):
    """Afficher les infos d'un flow."""
    flow_path = Path(args.flow)
    if not flow_path.exists():
        print(f"ERREUR: Fichier introuvable: {flow_path}")
        return 1

    try:
        flow = FlowParser.parse_from_file(str(flow_path))
    except Exception as e:
        print(f"ERREUR: {e}")
        return 1

    print(f"Flow: {flow.name}")
    print(f"  ID: {flow.id}")
    print(f"  Version: {flow.version}")
    print(f"  Description: {flow.description}")
    print(f"  Tasks ({len(flow.tasks)}):")
    for tid, task in flow.tasks.items():
        print(f"    - {tid} (type={task.TYPE})")
    print(f"  Relations ({len(flow.relations)}):")
    for rel in flow.relations:
        print(f"    - {rel['from']} -> {rel['to']}")
    return 0



def _log_startup_urls(logger, host: str, port: int, install_complete: bool) -> None:
    """Log only the URL that is valid for the current startup phase."""
    base_url = f"https://{host}:{port}"
    if install_complete:
        logger.info("  Chat:    %s/chat", base_url)
    else:
        logger.info("  Install: %s/install", base_url)


def cmd_start(args):
    """Start PawFlow server.

    The PawFlow server (chat, agents, HTTP listener) runs in the main process.
    Admin GUI is served natively at /admin via the pawflow-admin flow.
    """
    import signal

    # Configure logging
    from core.server_logging import configure_server_logging
    configure_server_logging(logging.INFO)
    logger = logging.getLogger("pawflow")

    try:
        from core.cli_workspace_mounts import set_workspace_mount_mode
        _mount_arg = getattr(args, "workspace_mount", None)
        if _mount_arg is not None:
            _mount_mode = set_workspace_mount_mode(_mount_arg)
            logger.info("CLI workspace fallback mount mode: %s", _mount_mode)
    except Exception as _wm_err:
        logger.warning("CLI workspace mount mode setup failed: %s", _wm_err)

    # 1. Register tasks and restore flows in the main process
    # Cleanup orphan Docker containers from previous server run
    from core.docker_utils import get_server_id, kill_containers
    _srv_id = get_server_id()
    _killed = kill_containers(_srv_id)
    if _killed:
        logger.info("Cleaned up %d orphan Docker container(s) from previous run", _killed)

    # Initialise the capability-auth store before any HTTP/WS handler can
    # reach it. The JSON store persists across restarts so VNC/terminal/
    # code-server/port-forward sessions a user opened before the restart
    # still match their capability tokens after.
    from core.capability_auth import init_db as _init_capabilities
    from core.paths import CAPABILITIES_FILE
    _init_capabilities(CAPABILITIES_FILE)

    logger.info("Registering tasks...")
    from tasks import register_all_tasks
    register_all_tasks()

    try:
        from core.install_bootstrap import ensure_install_bootstrap, is_install_complete
        ensure_install_bootstrap(port=int(args.port))
    except Exception as _ib_err:
        logger.error("Install bootstrap setup failed: %s", _ib_err, exc_info=True)
        raise

    logger.info("Restoring deployed flows...")
    from core.executor_registry import ExecutorRegistry
    er = ExecutorRegistry.get_instance()
    er.restore_from_disk()
    n = er.count()
    logger.info(f"PawFlow server ready — {n} flow(s) restored")
    _log_startup_urls(logger, args.host, int(args.port), is_install_complete())

    # Startup security report. In production mode
    # (PAWFLOW_ENV=production / PAWFLOW_PUBLIC_MODE=true) this raises
    # SystemExit if a critical setting is unsafe (weak default key,
    # fail-open approval, ...). Always logs the snapshot.
    try:
        from core.security_report import build_report, enforce
        enforce(build_report())
    except SystemExit:
        raise
    except Exception as _se:
        logger.warning("security report failed: %s", _se, exc_info=True)

    # Boot recovery: scan every conv/{agent}/pending.jsonl and wake agents
    # that had undelivered messages when the previous process died. Without
    # this, a message enqueued just before a crash would sit on disk with
    # nobody polling for it.
    try:
        from core.pending_queue import PendingQueue
        from tasks.ai.agent_loop import AgentLoopTask
        _recovered = PendingQueue.all_nonempty()
        if _recovered:
            logger.info(f"[boot-recovery] {len(_recovered)} agent(s) have "
                        f"pending messages — scheduling wake")
            for conv_id, agent, count in _recovered:
                logger.info(f"[boot-recovery] waking {conv_id[:8]}/{agent or '_shared'} "
                            f"({count} pending)")
                AgentLoopTask.wake_agent(
                    conv_id, agent,
                    reason=f"[boot-recovery] {count} pending msg(s)",
                    delay=2.0,
                )
    except Exception as _re:
        logger.warning(f"[boot-recovery] scan failed: {_re}")

    # 2. Keep main thread alive, handle Ctrl+C gracefully
    _shutting_down = False

    def _shutdown(sig, frame):
        nonlocal _shutting_down
        if _shutting_down:
            # Second Ctrl+C → force kill immediately
            logger.info("Force kill.")
            _kill_spawned_docker_containers()
            os._exit(1)
        _shutting_down = True
        logger.info("Shutting down...")

        # Stop executors in a thread with timeout
        import threading
        def _stop_all():
            try:
                reg = ExecutorRegistry.get_instance()
                for eid in list(reg._executors.keys()):
                    try:
                        reg._executors[eid].stop()
                    except Exception:
                        pass  # nosec B110
            except Exception:
                pass  # nosec B110
        t = threading.Thread(target=_stop_all, daemon=True)
        t.start()
        t.join(timeout=3)
        # Reap BEFORE the writer drain. `docker stop` gives 10s by default,
        # then SIGKILL: a reap queued behind a 20s drain simply never runs,
        # which is how a stopped server used to leave its containers up.
        # Killing them cannot lose a message -- executors are already stopped
        # and PawFlow's own rows are written by the in-process writer below.
        _kill_spawned_docker_containers()
        # Drain the ConversationWriter FIFO BEFORE os._exit. The writer
        # runs on a daemon thread; os._exit kills it instantly, and any
        # message still in the queue is gone. Executors are stopped first
        # so no new enqueues can race the drain.
        try:
            from core.conversation_writer import ConversationWriter
            if not ConversationWriter.shutdown_all(wait_timeout=20.0):
                logger.error(
                    "ConversationWriter drain INCOMPLETE - some messages "
                    "were not persisted before shutdown")
        except Exception as _cw_err:
            logger.error("ConversationWriter drain failed: %s", _cw_err,
                         exc_info=True)
        os._exit(0)


    def _kill_spawned_docker_containers():
        """Hard-kill all containers this PawFlow server spawned.

        Authoritative pass: every container carries this server's
        `org.pawflow.server-id` label, stamped at spawn (see
        `core.docker_utils.pawflow_container_labels`). The name-prefix pass
        that follows only catches containers started by an OLDER build, from
        before the label existed -- a name list has to be updated by hand for
        each new family, and two of them (interactive Claude Code, the
        Antigravity observer) were never added, which is why they survived
        `docker stop` and had to be removed by hand.

        Scoped to this server id on purpose: several PawFlow servers can share
        one Docker host, and each may only reap its own.
        """
        for _pool_mod, _pool_cls in (
            ("core.claude_code_pool", "ClaudeCodePool"),
            ("core.codex_pool", "CodexPool"),
            ("core.gemini_pool", "GeminiPool"),
        ):
            try:
                import importlib as _importlib
                getattr(_importlib.import_module(_pool_mod), _pool_cls).instance().shutdown()
            except Exception:
                pass  # nosec B110
        try:
            import subprocess as _sp  # nosec B404
            from core.docker_utils import docker_cmd
            from core.docker_utils import (LEGACY_REAP_FORMAT,
                                           PAWFLOW_SERVER_LABEL,
                                           get_server_id, legacy_reap_ids)
            # Pass 1 — the label. One filter, every family, now and later.
            try:
                _sel = f"{PAWFLOW_SERVER_LABEL}={get_server_id()}"
                _r = _sp.run(  # nosec B603
                    docker_cmd() + ["ps", "-a", "-q", "--filter",
                                     f"label={_sel}"],
                    capture_output=True, text=True, timeout=5)
                _ids = [i for i in (_r.stdout or "").split() if i]
                if _ids:
                    _sp.run(docker_cmd() + ["rm", "-f"] + _ids,  # nosec B603
                            capture_output=True, timeout=20)
                    logger.info("Reaped %d container(s) spawned by this server",
                                len(_ids))
            except Exception:
                pass  # nosec B110
            # Pass 2 — legacy names, for containers started before the label
            # existed. Most of these prefixes carry no server id at all, so the
            # name match alone is host-wide: on a shared Docker daemon it also
            # names another PawFlow server's pools, relays and logins, which
            # survived pass 1 precisely because their label is not ours.
            # Reaping those would kill a running instance's agents.
            #
            # So the label decides here too, in the other direction: a
            # container that carries SOMEBODY ELSE's server id is left alone,
            # and only an unlabelled one -- which can only come from a build
            # older than the label -- is reaped by name.
            _own = get_server_id()[:12]
            _own_full = get_server_id()
            for _prefix in (
                "pf-cc-pool-",
                "pf-codex-pool-",
                "pf-gemini-pool-",
                f"pf-{_own}-cci-",
                f"pf-{_own}-agyobs-",
                "pawflow-relay-srv-",
                "pawflow-relay-min-",
                "pawflow-claude-login-",
                "pawflow-codex-login-",
                "pawflow-gemini-login-",
                "pawflow-agy-login-",
            ):
                try:
                    _r = _sp.run(  # nosec B603
                        docker_cmd() + ["ps", "-a",
                                         "--filter", f"name={_prefix}",
                                         "--format", LEGACY_REAP_FORMAT],
                        capture_output=True, text=True, timeout=5)
                    _ids = legacy_reap_ids(_r.stdout or "", _own_full)
                    if _ids:
                        _sp.run(docker_cmd() + ["rm", "-f"] + _ids,  # nosec B603
                                capture_output=True, timeout=10)
                        logger.info("Reaped %d orphan container(s) matching %s*",
                                    len(_ids), _prefix)
                except Exception:
                    pass  # nosec B110
        except Exception:
            pass  # nosec B110

    signal.signal(signal.SIGINT, _shutdown)
    signal.signal(signal.SIGTERM, _shutdown)
    # atexit belt-and-suspenders: reap spawned containers even on abnormal
    # exit (uncaught exception, sys.exit, etc.). Signal-driven _shutdown
    # already covers Ctrl-C; this covers everything else.
    import atexit as _atexit
    _atexit.register(_kill_spawned_docker_containers)

    # Guardian thread: force-exit if the shutdown handler hangs.
    # Budget must cover: executor stop (3s) + ConversationWriter drain
    # (up to 20s) + docker reap (~5s), with slack. Force-exit BEFORE the
    # drain finishes would re-introduce the message-loss bug we are
    # fixing here, so the guardian sits well past the drain budget.
    def _force_exit_guardian():
        import time as _t
        while not _shutting_down:
            _t.sleep(0.5)
        _t.sleep(45)
        logger.warning("Shutdown timeout - force exit")
        os._exit(1)
    import threading as _th_guard
    _guardian = _th_guard.Thread(target=_force_exit_guardian, daemon=True)
    _guardian.start()

    # On Windows: Ctrl-C is eaten by wsl.exe subprocess.
    # Use a dedicated thread that reads stdin for Ctrl-C detection.
    if os.name == "nt":
        def _ctrl_c_watcher():
            """Watch for Ctrl-C via msvcrt on Windows."""
            import msvcrt
            while not _shutting_down:
                if msvcrt.kbhit():
                    ch = msvcrt.getch()
                    if ch == b'\x03':  # Ctrl-C
                        logger.info("Ctrl-C detected via kbhit")
                        _shutdown(None, None)
                        return
                time.sleep(0.1)
        _watcher = _th_guard.Thread(target=_ctrl_c_watcher, daemon=True,
                                     name="ctrl-c-watcher")
        _watcher.start()

    try:
        while not _shutting_down:
            time.sleep(0.5)
    except KeyboardInterrupt:
        pass
    if not _shutting_down:
        _shutdown(None, None)


def cmd_import_flow(args):
    """Import a NiFi flow."""
    path = args.path
    from engine.nifi_converter import NiFiConverter
    with open(path) as f:
        content = f.read()
    converter = NiFiConverter()
    result = converter.convert(content)

    output_path = args.output or "imported_flow.json"
    Path(output_path).parent.mkdir(parents=True, exist_ok=True)
    with open(output_path, 'w') as f:
        json.dump(result.flow, f, indent=2)

    print(f"Imported flow: {output_path}")
    if result.warnings:
        print(f"Warnings ({len(result.warnings)}):")
        for w in result.warnings[:5]:
            print(f"  - {w}")
    if result.subflows:
        for sf in result.subflows:
            sf_path = f"{sf.get('id', 'subflow')}.json"
            with open(sf_path, 'w') as f_out:
                json.dump(sf, f_out, indent=2)
            print(f"  Subflow: {sf_path}")
    return 0


def main():
    from core import __version__ as _pf_version
    parser = argparse.ArgumentParser(
        prog='pawflow',
        description='PawFlow - Self-hosted AI agent orchestration platform'
    )
    parser.add_argument('--version', action='version', version=f'pawflow {_pf_version}')
    subparsers = parser.add_subparsers(dest='command', help='Command')

    # run
    run_parser = subparsers.add_parser('run', help='Run a flow')
    run_parser.add_argument('flow', help='Flow JSON file')
    run_parser.add_argument('--input', '-i', nargs='+', help='Input files')
    run_parser.add_argument('--output-dir', '-o', help='Output directory')
    run_parser.add_argument('--verbose', '-v', action='store_true', help='Verbose mode')
    run_parser.add_argument('--provenance', '-p', action='store_true', help='Enable provenance tracking')
    run_parser.add_argument('--workers', '-w', type=int, default=4, help='Parallel workers (default: 4)')
    run_parser.add_argument('--retries', '-r', type=int, default=3, help='Max retries (default: 3)')
    run_parser.add_argument('--timeout', '-t', type=int, default=300, help='Timeout in seconds (default: 300)')
    run_parser.add_argument('--param', action='append', metavar='KEY=VALUE',
                            help='Override a flow parameter (repeatable)')

    # validate
    val_parser = subparsers.add_parser('validate', help='Validate a flow')
    val_parser.add_argument('flow', help='Flow JSON file')

    # list-tasks
    subparsers.add_parser('list-tasks', help='List available tasks')

    # info
    info_parser = subparsers.add_parser('info', help='Show flow info')
    info_parser.add_argument('flow', help='Flow JSON file')

    # start
    start_parser = subparsers.add_parser('start', help='Start PawFlow server')
    start_parser.add_argument('--host', default='localhost', help='Host (default: localhost)')
    start_parser.add_argument('--port', type=int, required=True, help='Port selected for this runtime')
    start_parser.add_argument('--workspace-mount', choices=['off', 'ro', 'rw'], default=None,
                              help='Mount linked relay workspaces into CLI provider containers: off, ro, or rw. Overrides PAWFLOW_CLI_WORKSPACE_MOUNT.')



    # import
    import_parser = subparsers.add_parser('import', help='Import a NiFi flow or .pfp plugin')
    import_parser.add_argument('path', help='File to import (.pfp, .xml, or .json)')
    import_parser.add_argument('-o', '--output', help='Output flow JSON path')

    # triggers
    triggers_parser = subparsers.add_parser('triggers', help='Manage event triggers')
    triggers_parser.add_argument('action',
                                 choices=['list', 'create', 'start', 'stop', 'delete', 'history', 'run'],
                                 help='Trigger action')
    triggers_parser.add_argument('--trigger-id', dest='trigger_id', help='Trigger ID')
    triggers_parser.add_argument('--trigger-type', dest='trigger_type',
                                 choices=['file_watcher', 'webhook', 'event', 'polling'],
                                 help='Trigger type (for create)')
    triggers_parser.add_argument('--flow-path', dest='flow_path', help='Flow JSON path (for create)')
    triggers_parser.add_argument('--name', help='Trigger name (for create)')
    triggers_parser.add_argument('--config', help='Trigger config as JSON string (for create)')

    # cluster
    cluster_parser = subparsers.add_parser('cluster', help='Cluster management')
    cluster_parser.add_argument('action', choices=['status'], help='Cluster action')
    cluster_parser.add_argument('--api-url', dest='api_url', help='API URL (default: http://localhost:8000)')

    # re-embed-memories
    reembed_parser = subparsers.add_parser('re-embed-memories',
                                            help='Re-embed all memories with vector embeddings')
    reembed_parser.add_argument('--user-id', dest='user_id', required=True, help='User ID')
    reembed_parser.add_argument('--provider', choices=['openai', 'local', 'auto'],
                                default='auto', help='Embedding provider (default: auto)')
    reembed_parser.add_argument('--api-key', dest='api_key', help='OpenAI API key (optional, uses env var)')

    # admin-user
    admin_user_parser = subparsers.add_parser(
        'admin-user',
        help='Create or repair a local admin account for recovery'
    )
    admin_user_parser.add_argument('action', choices=['create', 'reset-password'],
                                   help='Create/repair an admin, or reset an existing user password')
    admin_user_parser.add_argument('--username', required=True, help='PawFlow username')
    admin_user_parser.add_argument('--password', help='Password value. Prefer --password-env or interactive prompt.')
    admin_user_parser.add_argument('--password-env', dest='password_env',
                                   help='Environment variable containing the password')
    admin_user_parser.add_argument('--email', default='', help='Optional user email for create/update')
    admin_user_parser.add_argument('--display-name', dest='display_name', default='',
                                   help='Optional display name for create/update')

    args = parser.parse_args()

    if args.command == 'run':
        sys.exit(cmd_run(args))
    elif args.command == 'validate':
        sys.exit(cmd_validate(args))
    elif args.command == 'list-tasks':
        cmd_list_tasks(args)
    elif args.command == 'info':
        sys.exit(cmd_info(args))
    elif args.command == 'start':
        sys.exit(cmd_start(args))
    elif args.command == 'import':
        sys.exit(cmd_import_flow(args))
    elif args.command == 'triggers':
        sys.exit(cmd_triggers(args))
    elif args.command == 'cluster':
        sys.exit(cmd_cluster(args))
    elif args.command == 're-embed-memories':
        sys.exit(cmd_re_embed(args))
    elif args.command == 'admin-user':
        sys.exit(cmd_admin_user(args))
    else:
        parser.print_help()


if __name__ == '__main__':
    # Windows + Python 3.14: Playwright/Patchright needs ProactorEventLoopPolicy
    # for subprocess support.  Without this, scrapling cleanup crashes on exit.
    import sys
    import warnings
    if sys.platform == "win32":
        import asyncio
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)
            asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
    main()