# -*- 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": {}}