mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 10:10:43 +00:00
126 lines
4.2 KiB
Python
126 lines
4.2 KiB
Python
# -*- coding: utf-8 -*-
|
||
r"""验证「记住上次通的地址」是否让启动变快。
|
||
|
||
第一次启动：候选里没有记忆，要挨个探（默认地址现在是 192.168.2.156，
|
||
如果这台电脑能连上它，应该很快就通）
|
||
第二次启动：记忆里有它，直接排最前面，应该瞬间完成
|
||
|
||
重点看：**从进程启动到 UDP 8766 绑上花了多久** ——
|
||
这才是"手表多久能搜到电脑"。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import shutil
|
||
import socket
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote")
|
||
DISCOVERY = 8766
|
||
|
||
|
||
def clean() -> None:
|
||
subprocess.run(
|
||
["powershell", "-NoProfile", "-Command",
|
||
"Get-CimInstance Win32_Process -Filter \"Name='python.exe'\" | "
|
||
"Where-Object { $_.CommandLine -match 'run_agent' } | "
|
||
"ForEach-Object { Stop-Process -Id $_.ProcessId -Force "
|
||
"-ErrorAction SilentlyContinue }"],
|
||
capture_output=True, text=True, timeout=60)
|
||
time.sleep(2)
|
||
|
||
|
||
def probe(port: int, wait_discovery: bool = True) -> tuple[bool, float, str]:
|
||
"""起 agent，返回 (8766 是否可用, 花了多久, 日志尾部)。"""
|
||
t0 = time.time()
|
||
env = dict(os.environ)
|
||
env["PYTHONIOENCODING"] = "utf-8"
|
||
proc = subprocess.Popen(
|
||
[sys.executable, "run_agent.py", "--port", str(port),
|
||
"--new-code", "--add-device", "10"],
|
||
cwd=str(ROOT / "windows"), env=env,
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
try:
|
||
deadline = time.time() + 40
|
||
got = False
|
||
while time.time() < deadline:
|
||
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
s.settimeout(1.5)
|
||
try:
|
||
s.sendto(json.dumps({"lanremote": "LANREMOTE_SELFCHECK",
|
||
"proto": 1}).encode(), ("127.0.0.1",
|
||
DISCOVERY))
|
||
data, _ = s.recvfrom(2048)
|
||
if b"LANREMOTE_ACK" in data:
|
||
got = True
|
||
break
|
||
except (socket.timeout, ConnectionResetError, OSError):
|
||
pass
|
||
finally:
|
||
s.close()
|
||
time.sleep(0.3)
|
||
elapsed = time.time() - t0
|
||
finally:
|
||
proc.terminate()
|
||
try:
|
||
proc.wait(timeout=10)
|
||
except subprocess.TimeoutExpired:
|
||
proc.kill()
|
||
time.sleep(1.5)
|
||
|
||
log = ROOT / "windows" / "data" / "silent.log"
|
||
tail = ""
|
||
if log.is_file():
|
||
tail = log.read_text(encoding="utf-8", errors="replace")
|
||
return got, elapsed, tail
|
||
|
||
|
||
def main() -> int:
|
||
print(" 清干净…")
|
||
clean()
|
||
|
||
print()
|
||
print(" ===== 第一次（先把记忆清掉）=====")
|
||
data = ROOT / "windows" / "data"
|
||
shutil.rmtree(data, ignore_errors=True)
|
||
got1, t1, log1 = probe(8861)
|
||
print(f" 8766 可用: {got1} 耗时 {t1:.1f} 秒")
|
||
rel_lines = [ln for ln in log1.splitlines() if "选用" in ln or "中转已启用" in ln]
|
||
for ln in rel_lines:
|
||
print(f" {ln.strip()}")
|
||
|
||
print()
|
||
print(" ===== 第二次（有记忆了）=====")
|
||
# 保留 data 里的 settings.json（记忆就存在这里）
|
||
got2, t2, log2 = probe(8862)
|
||
print(f" 8766 可用: {got2} 耗时 {t2:.1f} 秒")
|
||
rel_lines = [ln for ln in log2.splitlines() if "选用" in ln or "中转已启用" in ln]
|
||
for ln in rel_lines:
|
||
print(f" {ln.strip()}")
|
||
|
||
print()
|
||
print(" ===== 结果 =====")
|
||
print(f" 第一次 {t1:.1f} 秒 -> 第二次 {t2:.1f} 秒")
|
||
if t2 < t1:
|
||
print(f" ✓ 快了 {t1 - t2:.1f} 秒")
|
||
else:
|
||
print(" ~ 没有明显变快（可能第一个候选本来就通）")
|
||
|
||
# 看看记忆存了什么
|
||
settings = ROOT / "windows" / "data" / "settings.json"
|
||
if settings.is_file():
|
||
try:
|
||
obj = json.loads(settings.read_text(encoding="utf-8"))
|
||
print(f" 记忆的中转地址: {obj.get('relay_last_good')}")
|
||
except Exception as exc: # noqa: BLE001
|
||
print(f" 读 settings 失败: {exc}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|