ykxmz/_build/agent_keepalive.py

# -*- coding: utf-8 -*-
"""被控端改用**复用连接**做长轮询 —— 这是稳定性问题的真正解法。

实测证据（真机 192.168.10.105）：
    A) 每个请求新建连接（现状）  : 约 70% 可用
    C) 一条连接连打 10 次        : 10/10 = 100%
    D) 直连本地 relay（对照）    : 20/20 = 100%
    G) 到 1.1.1.1 的丢包率       : 40%
    E) 经隧道单次请求耗时        : 1.2 ~ 2.5 秒

结论：
    链路丢包 40%，**每次新建 TLS 连接有约 30% 概率失败**；
    但只要连接建起来，之后就能一直用（TCP 自己会重传）。
    而 RelayLink._post() 用的是 urllib.request.urlopen —— 每次都新建，
    等于白白扔掉 30% 的成功率。

改法：
    用 http.client.HTTP(S)Connection 复用一条连接。
    请求失败（连接断了）就关掉重开一次再试；
    重试仍失败才当真的失败 —— 这样短抖动也能自己扛过去。
"""
from __future__ import annotations

import sys
from pathlib import Path

RELAY = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\windows"
             r"\lanremote\relay.py")

# ---------- 1. __init__ 里加连接缓存字段 ----------

INIT_ANCHOR = '''        self._socket = None'''

INIT_NEW = '''        self._socket = None
        # 复用的 HTTP 连接（见 _post 的说明：每次新建在丢包链路上会丢 30%）
        self._http = None
        self._http_lock = threading.RLock()'''

# ---------- 2. _post 改成复用连接 ----------

POST_OLD = '''    def _post(self, path: str, body: bytes, headers: dict,
              timeout: float = POLL_TIMEOUT, want_headers: bool = False):
        headers = dict(headers or {})
        headers.setdefault("User-Agent", USER_AGENT)
        request = urllib.request.Request(self.url + path, data=body,
                                         method="POST")
        request.add_header("Content-Type", "application/octet-stream")
        request.add_header("User-Agent", "LanRemote-Agent/" + _version())
        for name, value in headers.items():
            request.add_header(name, value)

        context = _ssl_context() if self.url.startswith("https://") else None
        with urllib.request.urlopen(request, timeout=timeout,
                                    context=context) as response:
            payload = response.read(MAX_BODY)
            if want_headers:
                extra = {name: value for name, value in response.headers.items()}
                return response.status, payload, extra
            return response.status, payload'''

POST_NEW = '''    def _open_http(self):
        """按 self.url 建一条 HTTP(S) 连接。"""
        import http.client
        parts = urllib.parse.urlsplit(self.url)
        host = parts.hostname or ""
        if parts.scheme == "https":
            return http.client.HTTPSConnection(
                host, parts.port or 443, timeout=POLL_TIMEOUT,
                context=_ssl_context())
        return http.client.HTTPConnection(host, parts.port or 80,
                                          timeout=POLL_TIMEOUT)

    def _drop_http(self) -> None:
        """把缓存的连接关掉（出错后调用，下次会重开）。"""
        with self._http_lock:
            connection, self._http = self._http, None
        if connection is not None:
            try:
                connection.close()
            except Exception:                             # noqa: BLE001
                pass

    def _post(self, path: str, body: bytes, headers: dict,
              timeout: float = POLL_TIMEOUT, want_headers: bool = False):
        """POST 一个请求，**复用同一条连接**。

        为什么要复用（真机实测）：
            这条链路丢包约 40%，每次新建 TLS 连接约 30% 概率握手失败。
            但连接建好之后，后续请求 100% 成功（TCP 会自己重传）。
            实测：每次新建 -> 70% 可用；复用 -> 100%。

        所以这里：
            * 优先用缓存的那条连接
            * 断了就重开一次再试（短抖动自己扛过去）
            * 两次都失败才算真失败，往上抛
        """
        import http.client

        send = dict(headers or {})
        send["Content-Type"] = "application/octet-stream"
        send["User-Agent"] = "LanRemote-Agent/" + _version()
        send["Connection"] = "keep-alive"

        last_error = None
        for attempt in (1, 2):
            with self._http_lock:
                connection = self._http
                if connection is None:
                    connection = self._open_http()
                    self._http = connection
            try:
                connection.timeout = timeout
                connection.request("POST", path, body=body, headers=send)
                response = connection.getresponse()
                payload = response.read(MAX_BODY)
                status = response.status
                # 4xx/5xx 也走这里：HTTPError 不带 body，自己判断更清楚
                if status >= 400:
                    self._drop_http()
                    if status == 401:
                        raise urllib.error.HTTPError(
                            self.url + path, status, "unauthorized",
                            response.headers, None)
                    raise urllib.error.HTTPError(
                        self.url + path, status, "http error",
                        response.headers, None)
                if want_headers:
                    extra = {name: value
                             for name, value in response.headers.items()}
                    return status, payload, extra
                return status, payload
            except urllib.error.HTTPError:
                raise                     # HTTP 状态码是业务层的，不算连接问题
            except Exception as exc:                      # noqa: BLE001
                # 连接层面的问题：关掉，下一轮重开
                last_error = exc
                self._drop_http()
                if attempt == 2:
                    break
                self._log_quiet(f"连接断了，重开一次再试（{exc}）")

        raise RelayError(f"请求失败: {last_error}")

    def _log_quiet(self, message: str) -> None:
        """记一条不打扰用户的日志（重连这种事不用弹通知）。"""
        try:
            self._log(message)
        except Exception:                                 # noqa: BLE001
            pass'''


def main() -> int:
    text = RELAY.read_text(encoding="utf-8")
    ok = True

    if "_open_http" in text:
        print("  已经改过")
        return 0

    for old, new, label in ((INIT_ANCHOR, INIT_NEW, "加连接缓存字段"),
                            (POST_OLD, POST_NEW, "_post 改成复用连接")):
        if old in text:
            text = text.replace(old, new, 1)
            print(f"  {label}")
        else:
            print(f"  [失败] {label} —— 锚点没找到")
            ok = False

    # 确保 threading 已导入（_http_lock 要用）
    if "import threading" not in text:
        text = text.replace("import ssl", "import ssl\nimport threading", 1)
        print("  补了 import threading")

    RELAY.write_text(text, encoding="utf-8", newline="\n")

    import ast
    try:
        ast.parse(text)
        print("  relay.py 语法 OK")
    except SyntaxError as exc:
        print(f"  relay.py 语法错误: 行 {exc.lineno}: {exc.msg}")
        ok = False
    return 0 if ok else 1


if __name__ == "__main__":
    sys.exit(main())