jiangchang-platform-kit/src/jiangchang_desktop_wake/wake.py

"""软唤醒策略梯 + native helper（N1+）。"""

from __future__ import annotations

import json
import logging
import os
import subprocess
import time
from pathlib import Path
from typing import Callable, Optional, Tuple, Union

from .errors import DesktopWakeError, ERR_WAKE_FAILED
from .probe import probe
from .profiles import load_profile
from .types import Profile, WakeStepResult
from . import uia_backend
from .window import activate_hwnd, find_hwnd_for_profile, require_windows

_logger = logging.getLogger("jiangchang_desktop_wake.wake")

# strategy -> (detail, status)  status: ok | fail | skipped
StrategyFn = Callable[[int, Profile], Tuple[str, str]]


def _strategy_activate(hwnd: int, profile: Profile) -> Tuple[str, str]:
    maximize = bool(getattr(profile, "activate_maximize", True))
    activate_hwnd(hwnd, maximize=maximize)
    return ("foreground+maximize" if maximize else "foreground"), "ok"


def _strategy_wm_getobject(hwnd: int, _profile: Profile) -> Tuple[str, str]:
    return uia_backend.send_wm_getobject(hwnd), "ok"


def _strategy_msaa(hwnd: int, _profile: Profile) -> Tuple[str, str]:
    return uia_backend.msaa_from_window(hwnd), "ok"


def _strategy_uia_walk(hwnd: int, _profile: Profile) -> Tuple[str, str]:
    return uia_backend.deep_uia_walk(hwnd), "ok"


def native_helper_path(name: str) -> Optional[Path]:
    here = Path(__file__).resolve().parent / "native" / name
    if here.is_file():
        return here
    env = os.environ.get("JIANGCHANG_DESKTOP_WAKE_NATIVE")
    if env:
        p = Path(env)
        if p.is_file():
            return p
        cand = p / name
        if cand.is_file():
            return cand
    app_root = os.environ.get("JIANGCHANG_APP_ROOT") or os.environ.get("JIANGCHANG_SKILLS_ROOT")
    if app_root:
        for sub in (
            Path(app_root) / "resources" / "desktop-wake" / name,
            Path(app_root) / "desktop-wake" / name,
        ):
            if sub.is_file():
                return sub
    return None


def _strategy_native(hwnd: int, profile: Profile) -> Tuple[str, str]:
    helper_name = profile.native_helper or "desktop-wake.exe"
    path = native_helper_path(helper_name)
    if path is None:
        return "skipped:native_helper_missing", "skipped"
    args = []
    for a in profile.native_args or ["0x{hwnd_hex}"]:
        args.append(
            a.replace("{hwnd}", str(int(hwnd)))
            # 必须带 0x：纯 hex 如 542144 会被 C# 当成十进制，导致 window_invalid
            .replace("{hwnd_hex}", format(int(hwnd), "x"))
            .replace("{profile}", profile.id)
        )
    cmd = [str(path), *args]
    mode_n2_wake = "n2-wake" in args
    # n2-wake：注入 + DuiLib 子类化，需更长超时；暖树由后续 probe 判定
    timeout_s = 60 if mode_n2_wake else 25
    _logger.info("native_helper cmd=%s timeout=%s", cmd, timeout_s)
    try:
        proc = subprocess.run(
            cmd, capture_output=True, text=True, timeout=timeout_s, check=False
        )
    except subprocess.TimeoutExpired:
        return "timeout", "fail"
    except OSError as exc:
        return f"spawn_error:{exc}", "fail"

    if mode_n2_wake:
        time.sleep(1.0)

    out = (proc.stdout or "").strip()
    err = (proc.stderr or "").strip()
    detail = f"rc={proc.returncode} out={out[:300]!r} err={err[:120]!r}"

    native_ok = False
    if out:
        # 取最后一行 JSON
        line = out.splitlines()[-1].strip()
        try:
            payload = json.loads(line)
            if mode_n2_wake:
                # n2-wake：ok=注入成功且 UI 仍响应；是否暖树看外层 probe
                native_ok = bool(payload.get("ok")) and bool(
                    payload.get("inject_ok", payload.get("ok"))
                )
                if payload.get("ui_responsive") is False:
                    native_ok = False
            else:
                # n1/auto：ok 表示已暖
                native_ok = bool(payload.get("ok"))
            detail = (
                f"rc={proc.returncode} ok={native_ok} mode={'n2-wake' if mode_n2_wake else 'other'} "
                f"inject_ok={payload.get('inject_ok')} ui_responsive={payload.get('ui_responsive')} "
                f"walk={payload.get('walk_count')} findall={payload.get('findall_count')} "
                f"markers={payload.get('markers_found')} detail={payload.get('detail')}"
            )
        except json.JSONDecodeError:
            native_ok = proc.returncode == 0
    else:
        native_ok = proc.returncode == 0

    return detail, ("ok" if native_ok else "fail")


STRATEGIES: dict[str, StrategyFn] = {
    "activate": _strategy_activate,
    "wm_getobject": _strategy_wm_getobject,
    "msaa_from_window": _strategy_msaa,
    "uia_attach_walk": _strategy_uia_walk,
    "native_helper": _strategy_native,
}


def wake(
    profile: Union[str, Profile] = "wecom",
    *,
    hwnd: Optional[int] = None,
    stop_when_warm: bool = True,
    pause_s: float = 0.4,
) -> list[WakeStepResult]:
    """按 profile.wake_strategies 依次尝试唤醒。

    - ``skipped``：策略未真正执行（如缺 exe），不算成功
    - ``fail``：执行了但 helper/异常失败
    - 每步后若 ``stop_when_warm`` 且 probe 已暖，则提前结束
    """
    require_windows()
    prof = load_profile(profile) if isinstance(profile, str) else profile
    h = find_hwnd_for_profile(prof, hwnd=hwnd)
    steps: list[WakeStepResult] = []

    for name in prof.wake_strategies:
        fn = STRATEGIES.get(name)
        if fn is None:
            steps.append(
                WakeStepResult(strategy=name, ok=False, status="fail", detail="unknown_strategy")
            )
            continue
        try:
            detail, status = fn(h, prof)
            steps.append(
                WakeStepResult(
                    strategy=name,
                    ok=(status == "ok"),
                    status=status,
                    detail=detail,
                )
            )
            if status == "skipped":
                continue
        except Exception as exc:
            _logger.debug("strategy %s failed: %s", name, exc)
            steps.append(
                WakeStepResult(strategy=name, ok=False, status="fail", detail=str(exc))
            )
            continue

        time.sleep(pause_s)
        if stop_when_warm:
            try:
                pr = probe(prof, hwnd=h)
                if pr.warm:
                    steps.append(
                        WakeStepResult(
                            strategy="probe_check",
                            ok=True,
                            status="ok",
                            detail=f"warm after {name} findall={pr.findall_count}",
                        )
                    )
                    break
            except Exception as exc:
                steps.append(
                    WakeStepResult(
                        strategy="probe_check",
                        ok=False,
                        status="fail",
                        detail=str(exc),
                    )
                )

    return steps


def wake_or_raise(
    profile: Union[str, Profile] = "wecom",
    *,
    hwnd: Optional[int] = None,
) -> list[WakeStepResult]:
    steps = wake(profile, hwnd=hwnd)
    prof = load_profile(profile) if isinstance(profile, str) else profile
    h = find_hwnd_for_profile(prof, hwnd=hwnd)
    pr = probe(prof, hwnd=h)
    if not pr.warm:
        raise DesktopWakeError(
            ERR_WAKE_FAILED,
            f"唤醒后仍冷 findall={pr.findall_count} missing={pr.markers_missing} "
            f"steps={[s.strategy + ':' + s.status for s in steps]}",
        )
    return steps