ykxmz/_build/probe_admin_width.py

# -*- coding: utf-8 -*-
"""量出管理页在窄屏下是哪个元素把页面撑宽了。

做法：抓一份管理页 HTML，注入一段脚本把"最宽的元素"写进 title，
然后用 Edge 无头 + --dump-dom 读回来。
"""
from __future__ import annotations

import http.cookiejar
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path


# ⚠ 不再把密码写死在脚本里（这个仓库是公开的）。
# 从环境变量或 _build/.admin_password（已 gitignore）读，读不到就报错。
def _admin_password() -> str:
    import os
    env = os.environ.get("YKXMZ_ADMIN_PASSWORD", "").strip()
    if env:
        return env
    f = Path(__file__).resolve().parent / ".admin_password"
    if f.is_file():
        return f.read_text(encoding="utf-8").strip()
    raise SystemExit(
        "没拿到管理密码：设 YKXMZ_ADMIN_PASSWORD，或写 _build/.admin_password")


ROOT = Path(__file__).resolve().parent.parent
UA = {"User-Agent": "LanRemote/1.4.0"}
PORT = 8111
BASE = f"http://127.0.0.1:{PORT}"

EDGE = next((p for p in (
    Path(r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe"),
    Path(r"C:\Program Files\Microsoft\Edge\Application\msedge.exe"))
    if p.is_file()), None)

PROBE = """
<script>
setTimeout(() => {
  const vw = document.documentElement.clientWidth;
  const rows = [];
  document.querySelectorAll('*').forEach(el => {
    const r = el.getBoundingClientRect();
    if (r.right > vw + 1 || r.width > vw + 1) {
      rows.push(Math.round(r.width) + 'x' + Math.round(r.height)
        + ' right=' + Math.round(r.right)
        + ' <' + el.tagName.toLowerCase()
        + (el.className ? '.' + String(el.className).split(' ').join('.') : '')
        + '>');
    }
  });
  document.title = 'VW=' + vw + ' BODY=' + document.body.scrollWidth
    + ' || ' + rows.slice(0, 12).join(' ;; ');
});
</script>
"""


def main() -> int:
    work = Path(tempfile.mkdtemp(prefix="admin_probe_"))
    relay = None
    try:
        data = work / "d"
        data.mkdir(parents=True, exist_ok=True)
        env = dict(os.environ)
        env.update({"RELAY_PORT": str(PORT), "RELAY_BIND": "127.0.0.1",
                    "RELAY_DATA": str(data),
                    "PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1"})
        relay = subprocess.Popen(
            [sys.executable, "-u", str(ROOT / "relay" / "relay.py")], env=env,
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        for _ in range(40):
            time.sleep(0.5)
            try:
                urllib.request.urlopen(urllib.request.Request(
                    BASE + "/v1/health", headers=UA), timeout=2)
                break
            except Exception:                             # noqa: BLE001
                continue

        for index, (name, code) in enumerate(
                [("书房台式机", "482913"), ("笔记本", "771204")], start=1):
            body = json.dumps({"deviceId": f"pc-{index}", "code": code,
                               "name": name, "version": "1.4.0"}).encode()
            urllib.request.urlopen(urllib.request.Request(
                BASE + "/v1/agent/register", data=body, method="POST",
                headers={**UA, "Content-Type": "application/json"}),
                timeout=10).read()

        jar = http.cookiejar.CookieJar()
        opener = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(jar))
        form = urllib.parse.urlencode({"password": _admin_password()}).encode()
        req = urllib.request.Request(BASE + "/admin/login", data=form,
                                     method="POST", headers=UA)
        req.add_header("Content-Type", "application/x-www-form-urlencoded")
        try:
            opener.open(req, timeout=15).read()
        except urllib.error.HTTPError:
            pass

        html = opener.open(urllib.request.Request(BASE + "/", headers=UA),
                           timeout=15).read().decode("utf-8")
        html = html.replace("</body>", PROBE + "</body>")
        page = work / "probe.html"
        page.write_text(html, encoding="utf-8")

        if not EDGE:
            print("  没有 Edge")
            return 1

        proc = subprocess.run(
            [str(EDGE), "--headless=new", "--disable-gpu",
             "--window-size=420,900", "--virtual-time-budget=3000",
             "--dump-dom", page.as_uri()],
            capture_output=True, text=True, encoding="utf-8",
            errors="replace", timeout=90)
        match = re.search(r"<title>(.*?)</title>", proc.stdout or "", re.S)
        if not match:
            print("  没拿到 title")
            print((proc.stdout or "")[:300])
            return 1
        for part in match.group(1).split(";;"):
            print("  " + part.strip()[:150])
        return 0
    finally:
        if relay is not None:
            try:
                relay.terminate()
                relay.wait(timeout=8)
            except Exception:                             # noqa: BLE001
                try:
                    relay.kill()
                except Exception:                     # noqa: BLE001
                    pass
        time.sleep(0.5)
        shutil.rmtree(work, ignore_errors=True)


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