chendelian
84157758e6
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 27s
Co-authored-by: Cursor <cursoragent@cursor.com>
281 lines
9.4 KiB
Python
281 lines
9.4 KiB
Python
"""软唤醒策略梯 + 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,
|
||
find_same_pid_hwnd_by_class,
|
||
get_window_pid,
|
||
require_windows,
|
||
)
|
||
|
||
_logger = logging.getLogger("jiangchang_desktop_wake.wake")
|
||
|
||
# strategy -> (detail, status) status: ok | fail | skipped
|
||
StrategyFn = Callable[[int, Profile], Tuple[str, str]]
|
||
|
||
|
||
def resolve_native_hwnd(probe_hwnd: int, profile: Profile) -> int:
|
||
"""native_helper 实际注入 hwnd。
|
||
|
||
若 profile.native_hwnd_class 已设：优先同 PID 下该 class 窗（如浮层 probe、主窗注入）。
|
||
找不到则回退 probe_hwnd，并打 warning。
|
||
"""
|
||
cls = (getattr(profile, "native_hwnd_class", None) or "").strip()
|
||
if not cls:
|
||
return int(probe_hwnd)
|
||
try:
|
||
pid = get_window_pid(int(probe_hwnd))
|
||
hit = find_same_pid_hwnd_by_class(pid=pid, class_name=cls)
|
||
except Exception as exc:
|
||
_logger.warning(
|
||
"resolve_native_hwnd failed profile=%s probe=%s err=%s; fallback probe",
|
||
profile.id,
|
||
probe_hwnd,
|
||
exc,
|
||
)
|
||
return int(probe_hwnd)
|
||
if not hit:
|
||
_logger.warning(
|
||
"resolve_native_hwnd miss profile=%s native_hwnd_class=%s pid=%s "
|
||
"probe=%s; fallback probe (may crash popup UI)",
|
||
profile.id,
|
||
cls,
|
||
pid,
|
||
probe_hwnd,
|
||
)
|
||
return int(probe_hwnd)
|
||
if int(hit) != int(probe_hwnd):
|
||
_logger.info(
|
||
"native_hwnd redirect profile=%s probe=%s -> inject=%s class=%s pid=%s",
|
||
profile.id,
|
||
probe_hwnd,
|
||
hit,
|
||
cls,
|
||
pid,
|
||
)
|
||
return int(hit)
|
||
|
||
|
||
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
|
||
# probe / activate 目标（如 SearchResultWindow2）
|
||
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:
|
||
# native_helper：可重定向到同 PID 主窗注入；其余策略仍作用在 probe 窗
|
||
target = resolve_native_hwnd(h, prof) if name == "native_helper" else h
|
||
detail, status = fn(target, prof)
|
||
if name == "native_helper" and int(target) != int(h):
|
||
detail = f"inject_hwnd={target} probe_hwnd={h} | {detail}"
|
||
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:
|
||
# 始终对 profile 目标窗（浮层）判暖，不是 inject 窗
|
||
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
|