381 lines
18 KiB
Python
381 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""手机远程控制台的桌面端通道。
|
||
|
||
客服电脑在内网、没有入站端口，所以永远是电脑去找服务端：一个后台线程长轮询
|
||
`/api/v2/desktop/commands/poll`（最长挂起 15 秒），每次顺带上报运行状态；领到
|
||
指令就交给界面线程执行（启动 / 停止监听），执行完回执结果。
|
||
|
||
黑名单以服务端为准：长轮询会带回服务端黑名单版本，和本机缓存的版本不一致就
|
||
补传本机新增、拉回全量，合并进 `conversation_blacklist` 的本地库。协议引擎和
|
||
视觉流程只读本地库，离线时照旧生效。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import concurrent.futures
|
||
import threading
|
||
import time
|
||
from collections import OrderedDict
|
||
from typing import Any, Callable
|
||
|
||
import backend_client
|
||
import conversation_blacklist
|
||
|
||
CAPABILITIES = ["commands_v1", "blacklist_sync_v1"]
|
||
POLL_WAIT_SECONDS = 15
|
||
# 界面线程执行一条指令最多等这么久；启动监听要等企业微信连接，停止要等后台任务退出。
|
||
EXECUTE_TIMEOUT_SECONDS = 45
|
||
# 同步失败后的冷却期。版本不一致时服务端会立即结束长轮询，同步又一直失败的话，
|
||
# 不冷却就会变成空转请求。
|
||
SYNC_RETRY_SECONDS = 60
|
||
# 任何原因导致长轮询没等就返回时，两轮之间至少间隔这么久。
|
||
MIN_POLL_INTERVAL_SECONDS = 1.0
|
||
# 出错后的首次退避，之后翻倍，最长 60 秒。
|
||
ERROR_BACKOFF_SECONDS = 2.0
|
||
|
||
|
||
def _current_tenant() -> str:
|
||
try:
|
||
account = backend_client.load_settings().get("desktop_account") or {}
|
||
except Exception:
|
||
return ""
|
||
return str(account.get("tenant_id") or "") if isinstance(account, dict) else ""
|
||
|
||
|
||
class RemoteControlChannel:
|
||
def __init__(
|
||
self,
|
||
*,
|
||
status_provider: Callable[[], dict[str, Any]],
|
||
command_executor: Callable[[dict[str, Any]], dict[str, Any]],
|
||
on_blacklist_synced: Callable[[dict[str, Any]], None] | None = None,
|
||
on_log: Callable[[str, str], None] | None = None,
|
||
root: Any = None,
|
||
poll_wait: int = POLL_WAIT_SECONDS,
|
||
tenant_provider: Callable[[], str] = _current_tenant,
|
||
):
|
||
self._status_provider = status_provider
|
||
self._execute = command_executor
|
||
self._on_blacklist_synced = on_blacklist_synced
|
||
self._log = on_log or (lambda _text, _level: None)
|
||
self._root = root
|
||
self._poll_wait = poll_wait
|
||
self._tenant_provider = tenant_provider
|
||
self._stop = threading.Event()
|
||
self._thread: threading.Thread | None = None
|
||
self._sync_lock = threading.Lock()
|
||
self._sync_again = False
|
||
self._sync_retry_at = 0.0
|
||
self._report_lock = threading.Lock()
|
||
self._report_pending = False
|
||
# 服务端时间 - 本机时间。指令有效期按服务端时钟判断，本机时钟不准也不会误拒。
|
||
self._clock_offset = 0.0
|
||
self._last_error = ""
|
||
# 已执行过的指令 id → 回执内容。长轮询的响应可能在路上丢，服务端会再给一次，
|
||
# 这里按 id 去重，重复的只补回执、绝不重复执行。
|
||
self._done: OrderedDict[str, dict[str, Any]] = OrderedDict()
|
||
self.unsupported = False
|
||
|
||
# ── 生命周期 ─────────────────────────────────────────────────────────
|
||
def start(self) -> None:
|
||
if self._thread is not None and self._thread.is_alive():
|
||
return
|
||
self._stop.clear()
|
||
self._thread = threading.Thread(target=self._run, name="remote-control-channel", daemon=True)
|
||
self._thread.start()
|
||
|
||
def stop(self) -> None:
|
||
self._stop.set()
|
||
|
||
@property
|
||
def running(self) -> bool:
|
||
return self._thread is not None and self._thread.is_alive() and not self._stop.is_set()
|
||
|
||
def status(self) -> dict[str, Any]:
|
||
try:
|
||
status = dict(self._status_provider() or {})
|
||
except Exception:
|
||
status = {}
|
||
status["capabilities"] = list(CAPABILITIES)
|
||
return status
|
||
|
||
def _tenant(self) -> str:
|
||
try:
|
||
return str(self._tenant_provider() or "")
|
||
except Exception:
|
||
return ""
|
||
|
||
# ── 主循环 ───────────────────────────────────────────────────────────
|
||
def _run(self) -> None:
|
||
backoff = ERROR_BACKOFF_SECONDS
|
||
self._tenant_seen: str | None = None
|
||
self._startup_synced = False
|
||
while not self._stop.is_set():
|
||
started = time.monotonic()
|
||
try:
|
||
delay = self._iterate()
|
||
backoff = ERROR_BACKOFF_SECONDS
|
||
self._last_error = ""
|
||
except Exception as exc:
|
||
# 线程死了手机端就永远看到"离线"，所以任何意外都只退避、不退出。
|
||
message = f"{exc.__class__.__name__}: {exc}"
|
||
if message != self._last_error:
|
||
self._last_error = message
|
||
self._log(f"远程控制通道暂时异常，稍后自动重试：{message}", "warn")
|
||
delay = backoff
|
||
backoff = min(backoff * 2, 60.0)
|
||
elapsed = time.monotonic() - started
|
||
if delay is None:
|
||
delay = max(0.0, MIN_POLL_INTERVAL_SECONDS - elapsed)
|
||
if delay:
|
||
self._stop.wait(delay)
|
||
|
||
def _iterate(self) -> float | None:
|
||
"""跑一轮；返回下一轮前要额外等待的秒数（None = 按最小间隔）。"""
|
||
if not backend_client.desktop_access_token():
|
||
return 10.0
|
||
tenant = self._tenant()
|
||
first = not self._startup_synced
|
||
if first or (tenant and tenant != self._tenant_seen):
|
||
previous = self._tenant_seen
|
||
if tenant:
|
||
if conversation_blacklist.adopt_tenant(tenant, root=self._root) and previous:
|
||
self._log("登录账号已变更，本机忽略改为按新账号同步。", "notify")
|
||
self._tenant_seen = tenant
|
||
self._startup_synced = True
|
||
self.sync_blacklist(reason="startup" if first else "account_changed", wait=True)
|
||
sync_allowed = time.monotonic() >= self._sync_retry_at
|
||
try:
|
||
response = backend_client.poll_remote_commands(
|
||
self.status(),
|
||
wait=self._poll_wait,
|
||
blacklist_version=(
|
||
conversation_blacklist.synced_version(self._root, tenant or None) if sync_allowed else None
|
||
),
|
||
)
|
||
self.unsupported = False
|
||
except backend_client.AuthenticationError:
|
||
# 登录失效由界面自己的心跳处理（强制下线弹窗），这里安静等待重新登录。
|
||
return 15.0
|
||
except backend_client.BackendError as exc:
|
||
if getattr(exc, "status", None) == 404:
|
||
# 管理端还没升级到带命令通道的版本：隔一段时间再看，不刷日志。
|
||
if not self.unsupported:
|
||
self._log("管理端暂不支持手机远程控制，升级后自动启用。", "warn")
|
||
self.unsupported = True
|
||
return 300.0
|
||
raise
|
||
if self._stop.is_set():
|
||
return 0.0
|
||
server_time = response.get("server_time")
|
||
if isinstance(server_time, (int, float)):
|
||
self._clock_offset = float(server_time) - time.time()
|
||
version = response.get("blacklist_version")
|
||
|
||
def needs_sync() -> bool:
|
||
return (
|
||
isinstance(version, int)
|
||
and version != conversation_blacklist.synced_version(self._root, tenant or None)
|
||
) or conversation_blacklist.pending_count(self._root) > 0
|
||
|
||
if time.monotonic() >= self._sync_retry_at and needs_sync():
|
||
# 等正在进行的同步（例如本机刚拉黑触发的那一次）做完再判断，不空转。
|
||
self.sync_blacklist(reason="server_changed", wait=True, needed=needs_sync)
|
||
for command in response.get("commands") or []:
|
||
if isinstance(command, dict) and command.get("id"):
|
||
self._handle(command)
|
||
return None
|
||
|
||
def _handle(self, command: dict[str, Any]) -> None:
|
||
command_id = str(command["id"])
|
||
if command_id in self._done:
|
||
self._ack(command_id, self._done[command_id])
|
||
return
|
||
expires_at = command.get("expires_at")
|
||
if isinstance(expires_at, (int, float)) and expires_at < time.time() + self._clock_offset:
|
||
outcome = {"ok": False, "error": "指令已超过有效期，电脑未执行", "result": {}}
|
||
else:
|
||
label = {"start_listening": "启动监听", "stop_listening": "停止监听"}.get(
|
||
str(command.get("type")), str(command.get("type")))
|
||
by = str(command.get("created_by") or "手机控制台")
|
||
self._log(f"收到远程指令：{label}（来自 {by}）", "notify")
|
||
try:
|
||
outcome = self._execute(command) or {}
|
||
except Exception as exc:
|
||
outcome = {"ok": False, "error": f"电脑端执行出错：{exc}", "result": {}}
|
||
outcome = {
|
||
"ok": bool(outcome.get("ok")),
|
||
"error": str(outcome.get("error") or ""),
|
||
"result": dict(outcome.get("result") or {}),
|
||
}
|
||
self._done[command_id] = outcome
|
||
while len(self._done) > 200:
|
||
self._done.popitem(last=False)
|
||
self._ack(command_id, outcome)
|
||
|
||
def _ack(self, command_id: str, outcome: dict[str, Any]) -> None:
|
||
for attempt in range(3):
|
||
try:
|
||
backend_client.ack_remote_command(
|
||
command_id, ok=outcome["ok"], result=outcome["result"],
|
||
error=outcome["error"], status=self.status(),
|
||
)
|
||
return
|
||
except backend_client.BackendError as exc:
|
||
if getattr(exc, "status", None) == 404:
|
||
return
|
||
except Exception:
|
||
pass
|
||
# 回执没送到也不要紧：服务端会再下发同一条，按 id 去重后只补回执。
|
||
if self._stop.wait(1.5 * (attempt + 1)):
|
||
return
|
||
|
||
# ── 状态上报 ─────────────────────────────────────────────────────────
|
||
def report_status_soon(self) -> None:
|
||
"""监听状态一变就立即上报一次，不必等下一轮长轮询。"""
|
||
with self._report_lock:
|
||
if self._report_pending:
|
||
return
|
||
self._report_pending = True
|
||
|
||
def send() -> None:
|
||
time.sleep(0.8)
|
||
with self._report_lock:
|
||
self._report_pending = False
|
||
try:
|
||
if backend_client.desktop_access_token():
|
||
backend_client.report_remote_status(self.status())
|
||
except Exception:
|
||
pass
|
||
|
||
threading.Thread(target=send, name="remote-status-report", daemon=True).start()
|
||
|
||
# ── 黑名单 ───────────────────────────────────────────────────────────
|
||
def sync_blacklist(
|
||
self, *, reason: str = "", wait: bool = False, needed: Callable[[], bool] | None = None,
|
||
) -> dict[str, Any] | None:
|
||
"""补传本机新增 + 拉回全量。
|
||
|
||
同一时间只跑一个。`wait=True`（命令通道线程）会等正在进行的那一轮结束，
|
||
拿到锁后再用 `needed` 复核一次——刚结束的那一轮往往已经把版本对齐了；
|
||
`wait=False` 时立即返回，并让正在进行的那一轮结束后再补一轮。
|
||
"""
|
||
if not self._sync_lock.acquire(blocking=wait, timeout=30 if wait else -1):
|
||
self._sync_again = True
|
||
return None
|
||
try:
|
||
if needed is not None and not needed():
|
||
return None
|
||
summary = None
|
||
for _round in range(3):
|
||
self._sync_again = False
|
||
summary = self._sync_once(reason)
|
||
if not self._sync_again or summary is None:
|
||
break
|
||
return summary
|
||
finally:
|
||
self._sync_lock.release()
|
||
|
||
def _sync_once(self, reason: str) -> dict[str, Any] | None:
|
||
tenant = self._tenant()
|
||
try:
|
||
if tenant:
|
||
conversation_blacklist.adopt_tenant(tenant, root=self._root)
|
||
adds = conversation_blacklist.pending_entries(self._root)
|
||
response = backend_client.sync_remote_blacklist(adds)
|
||
items = response.get("items")
|
||
version = response.get("version")
|
||
if not isinstance(items, list) or not isinstance(version, int):
|
||
raise backend_client.BackendError("后台返回的忽略格式不正确")
|
||
summary = conversation_blacklist.apply_server_snapshot(
|
||
items, version, root=self._root, tenant=tenant or None)
|
||
rejected = [row for row in response.get("rejected") or [] if isinstance(row, dict)]
|
||
if rejected:
|
||
# 服务端拒收的条目（标识不合法）本机继续拦截，但不再每轮重传。
|
||
conversation_blacklist.mark_rejected(rejected, root=self._root)
|
||
except Exception as exc:
|
||
self._sync_retry_at = time.monotonic() + SYNC_RETRY_SECONDS
|
||
if getattr(exc, "status", None) != 404:
|
||
self._log(f"忽略同步失败，本机仍按缓存拦截：{exc}", "warn")
|
||
return None
|
||
self._sync_retry_at = 0.0
|
||
summary.update(version=version, uploaded=len(adds), reason=reason,
|
||
rejected=len(response.get("rejected") or []))
|
||
if summary.get("added") or summary.get("removed"):
|
||
self._log(
|
||
f"忽略已与服务端同步：新增 {summary['added']} 个、移出 {summary['removed']} 个。",
|
||
"notify",
|
||
)
|
||
if self._on_blacklist_synced is not None:
|
||
try:
|
||
self._on_blacklist_synced(summary)
|
||
except Exception:
|
||
pass
|
||
return summary
|
||
|
||
def sync_blacklist_soon(self) -> None:
|
||
threading.Thread(
|
||
target=lambda: self.sync_blacklist(reason="local_change"),
|
||
name="remote-blacklist-sync", daemon=True,
|
||
).start()
|
||
|
||
|
||
def remove_blacklist_entries(items: list[dict[str, Any]], *, root: Any = None) -> dict[str, Any]:
|
||
"""移出黑名单：只在本机、尚未上传的条目直接删；其余必须先让服务端同意。
|
||
|
||
返回 {removed, denied: [...], failed: [...], message, ok}。服务端按 ZYT 权限
|
||
wecom_rpa.blacklist/remove 决定能不能移出；没有权限的条目原样保留。第一次
|
||
连不上服务端就停止，不在界面线程上对每一条都等超时。
|
||
"""
|
||
pairs = conversation_blacklist._validated_pairs(items)
|
||
states = conversation_blacklist.local_states(root)
|
||
local_only, remote = [], []
|
||
for pair in sorted(pairs):
|
||
(remote if states.get(pair) == "synced" else local_only).append(pair)
|
||
approved: list[tuple[str, str]] = []
|
||
denied: list[str] = []
|
||
failed: list[str] = []
|
||
for index, (account, conversation) in enumerate(remote):
|
||
try:
|
||
backend_client.remove_remote_blacklist(account, conversation, timeout=4.0)
|
||
approved.append((account, conversation))
|
||
except backend_client.RemotePermissionDenied as exc:
|
||
denied.append(str(exc))
|
||
except backend_client.BackendError as exc:
|
||
if getattr(exc, "status", None) == 404 and getattr(exc, "code", "") == "not_blocked":
|
||
# 服务端已经没有这一条（例如刚在手机上移出），本机跟着删。
|
||
approved.append((account, conversation))
|
||
else:
|
||
# 路由不存在（旧版或回滚后的管理端）、网络不通：一律保留拦截。
|
||
failed.extend([str(exc)] * (len(remote) - index))
|
||
break
|
||
removed = 0
|
||
if approved:
|
||
removed += conversation_blacklist.remove_entries(
|
||
[{"account": a, "conversation": c} for a, c in approved], root=root)
|
||
if local_only:
|
||
# 读状态之后，并行的同步可能刚把某条上传成"已同步"，那条就只能由服务端移出。
|
||
removed += conversation_blacklist.remove_entries(
|
||
[{"account": a, "conversation": c} for a, c in local_only], root=root, only_pending=True)
|
||
if denied:
|
||
message = denied[0] if not removed else f"已移出 {removed} 个；其余 {len(denied)} 个需要主管权限：{denied[0]}"
|
||
elif failed:
|
||
message = (f"已移出 {removed} 个；{len(failed)} 个暂时无法连接服务器，请稍后重试"
|
||
if removed else f"暂时无法移出：{failed[0]}")
|
||
else:
|
||
message = f"已移出 {removed} 个忽略会话" if removed else "所选会话已不在忽略中"
|
||
return {"removed": removed, "denied": denied, "failed": failed, "message": message,
|
||
"ok": bool(removed) or not (denied or failed)}
|
||
|
||
|
||
def future_result(future: concurrent.futures.Future, timeout: float = EXECUTE_TIMEOUT_SECONDS) -> dict[str, Any]:
|
||
try:
|
||
return future.result(timeout=timeout)
|
||
except concurrent.futures.TimeoutError:
|
||
# 界面线程还没来得及处理就作废，免得回执说"超时"之后电脑又悄悄执行了。
|
||
if not future.cancel() and future.done() and not future.cancelled():
|
||
return future.result()
|
||
return {"ok": False, "error": "电脑端执行超时，请到电脑上确认状态", "result": {}}
|
||
except concurrent.futures.CancelledError:
|
||
return {"ok": False, "error": "电脑端未执行该指令", "result": {}}
|