guanlan-src/app_qualityGate/app_qualityGate_guanlan/audits/check_transferable.py

# -*- coding: utf-8 -*-
r"""移植性大扫描 (2026-09-12) —— 换电脑/换盘/换系统之前跑它。

比 `scripts/check_portability.py`(只看 .py 的路径写法) 查得更宽, 因为"能不能独立运行"还取决于:

  ① **所有文件类型**里的机器相关绝对路径 (.py/.bat/.sh/.ps1/.json/.md/.txt/.html/.yaml/.cfg …)
     —— 不只是源码; 文档里的示例路径写死也会误导现场, 产物里烘进去的路径换机后就是死链。
  ② **虚拟环境的可移植性**: `.venv/pyvenv.cfg` 的 `home` 指向基础解释器 —— 换台没有同路径 Python 的
     机器, `.venv\Scripts\python.exe` 直接起不来(这是"拷包就能跑"最常见的误解)。
  ③ **离线依赖是否为本平台备齐**: `wheels/<平台>` 有没有; 缺了就只能联网装。
  ④ **便携运行时**是否在 (`vendor/python/…`), 以及 `vendor/MANIFEST.json` 有没有把它登记全。
  ⑤ **配置与端口**是否相对/可改, 有没有写死本机 IP。

用法:
    python scripts/check_transferable.py            # 全量检查 + 结论
    python scripts/check_transferable.py --brief    # 只报问题
    python scripts/check_transferable.py --simulate # **模拟移植**(推荐换机前跑): 把包拷成一份副本 →
                                                    # 在副本里跑安装 → 起服务核验页面 → 报结果 → 清理副本。
                                                    # 默认拷全量(除 data/.git/.venv/暂存区); --keep 保留副本。
"""
from __future__ import annotations

try:
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                     # 理论不可达；包结构异常时回退到按位置上跳
    from pathlib import Path as _P
    def _install_root(_f): return _P(_f).resolve().parents[4]

import argparse
import json
import os
import pathlib
import re
import shutil
import socket
import subprocess
import sys
import tempfile
import time
import urllib.request

ROOT = _install_root(__file__)
sys.path.insert(0, str(ROOT))
from src import paths as P                                            # 配置/路径唯一取用口 (2026-09-17 用户令 2)

# 机器相关绝对路径: 盘符路径 / mac 用户目录 / 家目录 / 本机用户名
RE_ABS = re.compile(r"""(?<![\w:/])(?:[A-Za-z]:[\\/](?![\\/])|/Users/[A-Za-z0-9_.\-]+|/home/[a-z][A-Za-z0-9_.\-]*|/opt/[A-Za-z0-9_.\-]+|/Volumes/[A-Za-z0-9_\- ]+)""")
# 扫描范围: 源码/脚本/配置/文档/外壳页面 (跳过 .venv、依赖、二进制、大体积产物里的二进制)
TEXT_EXT = {'.py', '.bat', '.sh', '.ps1', '.cmd', '.json', '.md', '.txt', '.html', '.htm', '.yaml', '.yml', '.cfg', '.ini', '.toml'}
SKIP_DIRS = {'.venv', '.git', '__pycache__', 'node_modules', '_products_off', 'logs', 'run'}
SKIP_PATH_HINT = ('_products_off_prev',)
# 扫描范围里要**排除现场原始件**: data/raw/** 是现场给的测量导出 (SCADA csv、Brande TCM JSON ——
# 2026-09-12 落位后单这一个目录就是 2.5 万个 .json / 150 GB), 且**不随包分发** (pack_dist.py 的
# EXCLUDE_GLOBS 含 'data')。文本闸门扫它: ①对"换机可用"零信息量 (那些文件里就算有绝对路径也不是我们的
# 接线) ②每次验收要多读几十 GB 文本 (8 MB 以下的一律会读进内存), 把闸门从秒级拖成小时级。
SKIP_RAW_PREFIX = ('data/raw/', 'data/_incoming/')
# 允许的例外: 文档里明确在讲"路径约定"或演示用的占位
ALLOW_LINE = ('portability-allow', '<安装目录>', '<现场包目录>', '<场站名称>', 'D:\\guanlan', '/opt/guanlan',
              'D:/guanlan', '示例', '例:', '例如')


def walk_files():
    for p in ROOT.rglob('*'):
        if not p.is_file() or p.suffix.lower() not in TEXT_EXT:
            continue
        rel = p.relative_to(ROOT).as_posix()
        parts = set(p.relative_to(ROOT).parts)
        if parts & SKIP_DIRS or any(h in rel for h in SKIP_PATH_HINT):
            continue
        if rel.startswith(SKIP_RAW_PREFIX):
            continue
        if rel.startswith(SKIP_RAW_PREFIX):
            continue
        if p.stat().st_size > 8 * 1024 * 1024:      # 超大文本(如 20MB 门户)另算, 见下
            continue
        yield p


def scan_paths(verbose=True):
    hits = []
    for p in walk_files():
        try:
            text = p.read_text(encoding='utf-8', errors='replace')
        except Exception:
            continue
        for i, line in enumerate(text.splitlines(), 1):
            if any(a in line for a in ALLOW_LINE):
                continue
            if line.lstrip().startswith(('#', 'rem ', '::')) and 'src' not in line and '--' in line:
                continue
            m = RE_ABS.search(line)
            if m:
                hits.append((p.relative_to(ROOT).as_posix(), i, m.group(0), line.strip()[:110]))
    print(f'① 文本文件里的机器相关绝对路径: {len(hits)} 处')
    if hits and verbose:
        for rel, i, found, line in hits[:25]:
            print(f'     {rel}:{i}  ← {found}\n        {line}')
        if len(hits) > 25:
            print(f'     … 另有 {len(hits) - 25} 处')
    return hits


def scan_big_text():
    """大文件(门户 20MB 等)单独扫: 里面烘进的绝对路径换机后就是死链。"""
    out = []
    for p in ROOT.rglob('*.html'):
        rel = p.relative_to(ROOT).as_posix()
        if any(x in rel for x in SKIP_PATH_HINT) or '.venv' in rel or p.stat().st_size < 1024:
            continue
        try:
            t = p.read_text(encoding='utf-8', errors='replace')
        except Exception:
            continue
        n = len(RE_ABS.findall(t))
        if n:
            out.append((rel, n, p.stat().st_size))
    if out:
        print(f'② HTML 里烘进的绝对路径: {len(out)} 个文件')
        for rel, n, sz in out[:10]:
            print(f'     {rel}  {n} 处 ({sz/1e6:.1f} MB)')
    else:
        print('② HTML 里没有烘进的绝对路径')
    return out


def check_venv():
    cfg = ROOT / '.venv' / 'pyvenv.cfg'
    print('③ 虚拟环境')
    if not cfg.is_file():
        print('     [--] 没有 .venv/pyvenv.cfg (还没装? 目标机上跑 install.bat / sh install.sh)')
        return 'absent'
    kv = {}
    for l in cfg.read_text(encoding='utf-8', errors='replace').splitlines():
        if '=' in l:
            k, v = l.split('=', 1)
            kv[k.strip()] = v.strip()          # ★键要去空格: pyvenv.cfg 写的是 "home = D:\..." (等号前有空格)
    home = (kv.get('home') or '').strip()
    print(f'     home(基础解释器) = {home}')
    print(f'     版本 {kv.get("version", "?").strip()}')
    here = pathlib.Path(home).is_dir() if home else False
    print(f'     本机存在该目录: {here}')
    print('     ★ 换台机器若没有同路径的 Python, .venv 里的解释器起不来 —— 移植时**必须在目标机重跑一次安装**'
          '(install.bat / sh install.sh; 离线, 用包内轮子或便携运行时)。')
    return kv


def check_wheels():
    w = ROOT / 'wheels'
    print('④ 离线依赖轮子')
    if not w.is_dir():
        print('     [X] 没有 wheels/ 目录 —— 装依赖需要联网')
        return []
    dirs = sorted(d.name for d in w.iterdir() if d.is_dir())
    print(f'     平台目录: {dirs or "(无)"}')
    req = [l.split('#')[0].strip() for l in (ROOT / 'requirements.txt').read_text(encoding='utf-8').splitlines()]
    req = [r for r in req if r]
    names = {r.split('==')[0].lower().replace('-', '_') for r in req}
    for d in dirs:
        have = {p.name.split('-')[0].lower().replace('-', '_') for p in (w / d).glob('*.whl')}
        miss = sorted(n for n in names if n not in have)
        tag = 'OK' if not miss else f'缺 {len(miss)}: {miss[:6]}'
        print(f'     {d}: {len(list((w / d).glob("*.whl")))} 个轮子  {tag}')
    plat = 'win_amd64' if os.name == 'nt' else ('linux_x86_64' if sys.platform.startswith('linux') else 'macos')
    if plat not in dirs:
        print(f'     [!] 本机平台 {plat} 没有对应轮子目录: Linux/macOS 上 install.sh 会自动**改成联网安装**')
    return dirs


def check_vendor():
    v = ROOT / 'vendor'
    print('⑤ 便携运行时 / 离线件')
    if not v.is_dir():
        print('     [--] 没有 vendor/')
        return
    for p in sorted(v.rglob('*')):
        if p.is_file():
            print(f'     {p.relative_to(ROOT).as_posix()}  {p.stat().st_size/1e6:.1f} MB')
    man = v / 'MANIFEST.json'
    if man.is_file():
        m = json.loads(man.read_text(encoding='utf-8'))
        files = {x.relative_to(v).as_posix() for x in v.rglob('*') if x.is_file() and x.name != 'MANIFEST.json'}
        undoc = sorted(files - set(m))
        if undoc:
            print(f'     [!] MANIFEST.json 没登记: {undoc} (只影响校验, 不影响运行)')


def check_configs():
    print('⑥ 配置与端口')
    p = P.SERVE_JSON                      # 配置唯一取用口
    if p.is_file():
        c = json.loads(p.read_text(encoding='utf-8-sig'))
        print(f'     serve.json: host={c.get("host")} gateway={c.get("gateway")} python={c.get("python")!r} '
              f'raw_dir={c.get("raw_dir")!r}')
        for k in ('python', 'raw_dir', 'release_dir', 'viewer_dir', 'sim_dir'):
            v = str(c.get(k) or '')
            if re.match(r'^[A-Za-z]:[\\/]|^/', v):
                print(f'     [!] {k} 是绝对路径: {v}')
    # 写死的本机端口/入口(允许出现, 但要知道)
    n_port = sum(1 for p in walk_files() if '28084' in p.read_text(encoding='utf-8', errors='replace'))
    print(f'     提到 28084 的文本文件: {n_port} 个 (端口真源是 configs/serve.json, 其余多为文档)')


def _free_port(start=28700, tries=40):
    for p in range(start, start + tries):
        with socket.socket() as s:
            try:
                s.bind(('127.0.0.1', p)); return p
            except OSError:
                continue
    return None


def _busy(port, host='127.0.0.1'):
    with socket.socket() as s:
        s.settimeout(0.3)
        return s.connect_ex((host, port)) == 0


IGNORE = shutil.ignore_patterns('data', '.git', '.venv', '__pycache__', 'logs', 'run',
                                '_products_off', '_products_off_prev_*', '*.pyc')


def install_in(dst: pathlib.Path) -> pathlib.Path | None:
    """在 dst 里跑一次离线安装 (Windows: install.ps1; POSIX: install.sh), 返回该副本的解释器路径。"""
    print(f'   在 {dst} 里跑安装 (离线) …')
    # -NoDesktopShortcut: install.ps1 默认会在**桌面**建「观澜·如东样板 v2」快捷方式, 而这里是临时副本,
    # 建出来的快捷方式指向马上要被删掉的目录 (2026-09-16 加)。副本里的 `启动观澜.lnk` 照建 —— 那正是要核验的东西。
    cmd = (['powershell', '-ExecutionPolicy', 'Bypass', '-File', str(dst / 'install.ps1'), '-NoDesktopShortcut']
           if os.name == 'nt' else ['sh', str(dst / 'install.sh')])
    t0 = time.time()
    r = subprocess.run(cmd, cwd=str(dst))
    print(f'   install 退出码 {r.returncode}, 耗时 {time.time() - t0:.0f}s')
    py = dst / ('.venv/Scripts/python.exe' if os.name == 'nt' else '.venv/bin/python')
    if not py.is_file():
        print(f'   [X] 副本里没有 {py} —— 安装没成')
        return None
    v = subprocess.run([str(py), '-c', 'import sys,pandas,pyarrow,polars,yaml;print(sys.version.split()[0])'],
                       capture_output=True, text=True, errors='replace')
    print(f'   副本解释器: {v.stdout.strip() or v.stderr.strip()[:120]}')
    return py


LAST_GATEWAY_PORT: int | None = None      # run_and_probe 实际用的网关端口（卸载核验要拿它查"释放了没"）


def run_and_probe(dst: pathlib.Path, py: pathlib.Path, base_port=28084) -> int:
    """在副本里起服务并核验页面 —— **完全不动正在跑的实例**: 副本整套换端口跑。

    ★ 2026-09-17 改法 (现场事故驱动): 以前网关 ROUTES 的上游端口是写死的, 所以副本**必须**占用
    18033/18020/18791/18792/64292 —— 于是"开箱验证"就得先把用户正在跑的服务停掉; 实测后果:
    用户在验证窗口里点门户「登录」, `/detail/` 报"模块未启动 (127.0.0.1:18033)"。
    现在 `guanlan_gateway.ROUTES` 的上游端口可被 `configs/serve.json` 覆盖, 于是副本可以**整套**
    换到空闲端口 (网关 + 5 个组件), 主实例照常跑, 谁也不打扰谁。
    端口占用仍会兜底: 找不到空闲端口就跳过启动核验 (不计失败)。
    """
    global LAST_GATEWAY_PORT
    comp_keys = ('detail', 'cms', 'sim', 'sim_sys', 'viewer')
    cfg_p = dst / 'configs' / P.SERVE_JSON.name          # config-path-allow: 改的是**副本**里的那份配置
    try:
        cfg = json.loads(cfg_p.read_text(encoding='utf-8-sig'))
    except Exception as e:
        print(f'   [!] 副本配置读不出来 ({type(e).__name__}) —— 跳过启动核验'); return 0

    # ① 主实例在不在: 决定要不要给副本换端口, 以及能否做"门户同字节"比对
    main_up = not _busy(base_port)
    gw = base_port if not _busy(base_port) else _free_port(28700)
    if gw is None:
        print('   [!] 找不到空闲网关端口 —— 跳过启动核验'); return 0
    LAST_GATEWAY_PORT = gw
    changed = {}
    if gw != base_port:
        changed['gateway'] = gw
    for k in comp_keys:
        want = int(cfg.get(k) or 0)
        if want and _busy(want):
            free = _free_port(28800 + 100 * (len(changed) % 5))
            if free is None:
                print(f'   [!] 找不到空闲端口给组件 {k} —— 跳过启动核验'); return 0
            changed[k] = free
    if changed:
        cfg.update(changed)
        cfg_p.write_text(json.dumps(cfg, ensure_ascii=False, indent=2), encoding='utf-8')
        print('   副本改用独立端口 (不打扰正在跑的实例): ' +
              ', '.join(f'{k}={v}' for k, v in changed.items()))
    else:
        print('   主实例未在跑 → 副本直接用默认端口')

    t0 = time.time()
    rs = subprocess.run([str(py), 'guanlan.py', 'serve'], cwd=str(dst), capture_output=True, text=True, errors='replace')
    print(f'   serve 退出码 {rs.returncode} (1 = 有模块未就绪, 属正常), 耗时 {time.time() - t0:.0f}s')
    ok = 0
    try:
        # ★门户期望字节数**不写死** (2026-09-16): 原来常量 20225828 是某一版门户"服务出来"的字节数,
        #   门户外壳一改 (例如按用户令在菜单里加「数据重算」) 它就过期, 于是核验会打印一条永远不成立的
        #   "★与主包不同", 看起来像移植出了问题。改为**从正在跑的主实例现量**: 主实例没起就不比这一项。
        exp_portal = None
        try:
            with urllib.request.urlopen(f'http://127.0.0.1:{base_port}/', timeout=10) as rr:
                exp_portal = len(rr.read())
                print(f'   参照: 主实例门户 {exp_portal:,} B (端口 {base_port})')
        except Exception as e:
            print(f'   参照: 主实例 ({base_port}) 未响应 → 门户字节数不做同字节比对 ({type(e).__name__})')
        urls = [('/', exp_portal), ('/detail/', None), ('/cms/', None), ('/ops', None), ('/healthz', None)]
        for path, exp_size in urls:
            try:
                with urllib.request.urlopen(f'http://127.0.0.1:{gw}{path}', timeout=30) as rr:
                    body = rr.read()
                    tag = '' if exp_size is None else (' 与主包同字节' if len(body) == exp_size else f' ★与主包不同(主包 {exp_size})')
                    if exp_size is not None and len(body) != exp_size:
                        # 副本换了网关端口时, 门户里的绝对地址(127.0.0.1:<端口>)会被网关改写成副本自己的
                        # 端口 ⇒ 字节数**必然**不同。这是预期, 不是移植问题 (2026-09-17 加这句, 免得看错)。
                        tag += f' ← 副本网关端口 {gw} ≠ 主实例 {base_port}, 门户内的绝对地址被改写, 字节不同属预期'
                    print(f'   {path:10s} HTTP {rr.status}  {len(body):,} B{tag}')
                    ok += 1
            except Exception as e:
                print(f'   {path:10s} 失败: {type(e).__name__} {e}')
        print(f'   页面可用 {ok}/{len(urls)}')
    finally:
        # 只停**副本**自己起的那些 (用副本里的 pids.json), 绝不碰主实例
        subprocess.run([str(py), 'guanlan.py', 'stop'], cwd=str(dst), capture_output=True)
    return ok


def _files_of(p: pathlib.Path) -> int:
    """目录/文件的件数（卸载核验里"数据原样在位"的比对面）。"""
    if not p.exists():
        return -1
    return 1 if p.is_file() else sum(1 for _ in p.rglob('*') if _.is_file())


def uninstall_probe(dst: pathlib.Path, py: pathlib.Path, gw: int | None = None) -> int:
    """卸载核验（用户令 2026-09-17：增加卸载脚本）——在**副本**上真跑一遍卸载器。

    为什么放在开箱验证里：卸载脚本这类东西"在开发机上跑一次看着没事"不算数，它要证明的是
    **在刚装好的那份副本上**：① 计划齐全且 --dry-run 一个字节都不动；② 真卸载会把服务/运行态/
    安装记录/快捷方式拆掉；③ **数据与产物一件不少**（默认口径不碰用户数据）。

    → 返回失败条数（0 = 全过）
    """
    bad = 0
    say = lambda s: print('   ' + s)                                        # noqa: E731
    # ★ 子进程输出的编码必须点名 utf-8：卸载器写中文, Windows 上默认按 cp936 解会得到
    #   "�ƻ���ȫ" 这种乱码（2026-09-17 实测: 核验打印成了天书, 证据就没法看了）。
    sub_env = dict(os.environ, PYTHONUTF8='1')
    r = subprocess.run([str(py), 'scripts/guanlan_uninstall.py', '--selftest'], cwd=str(dst),
                       capture_output=True, text=True, encoding='utf-8', errors='replace',
                       env=sub_env, timeout=600)
    for ln in (r.stdout or '').strip().splitlines():
        if '[FAIL]' in ln or '结论' in ln or '[OK]' in ln:
            say(ln.strip())
    if r.returncode != 0:
        bad += 1
        say('[X] 卸载器 --selftest 未通过')
    keep = [d for d in ('data', 'outputs') if (dst / d).exists()]
    before = {d: _files_of(dst / d) for d in keep}
    r2 = subprocess.run([str(py), 'scripts/guanlan_uninstall.py', '--yes'], cwd=str(dst),
                        capture_output=True, text=True, encoding='utf-8', errors='replace',
                        env=sub_env, timeout=900)
    tail = (r2.stdout or '').strip().splitlines()
    for ln in tail[-8:]:
        say(ln.rstrip())
    if r2.returncode != 0:
        bad += 1
        say(f'[X] 卸载器退出码 {r2.returncode}')
    checks = [('安装记录 install-info.json 已删', not (dst / 'install-info.json').exists()),
              ('运行态 run/ 已清', not (dst / 'run').exists()),
              ('快捷方式已撤', not list(dst.glob('*.lnk')))]
    for d in keep:
        checks.append((f'数据 {d}/ 原样在位（{before[d]} 件）', _files_of(dst / d) == before[d]))
    if gw:
        checks.append((f'卸载后端口 {gw} 已释放', not _busy(gw)))
    for name, good in checks:
        say(('[OK] ' if good else '[X] ') + name)
        bad += 0 if good else 1
    return bad


def simulate(into=None, keep=False) -> int:
    """模拟移植: 拷副本 → 副本内安装 → 起服务核验 → 清理。"""
    dst = pathlib.Path(into) if into else pathlib.Path(tempfile.gettempdir()) / 'guanlan_portsim'
    if dst.exists():
        print(f'  清理旧副本 {dst}'); shutil.rmtree(dst, ignore_errors=True)
    print(f'① 拷副本 → {dst}   (排除 data/.git/.venv/暂存区/logs/run)')
    t0 = time.time()
    shutil.copytree(ROOT, dst, ignore=IGNORE, symlinks=False)
    n = sum(1 for _ in dst.rglob('*') if _.is_file())
    mb = sum(_.stat().st_size for _ in dst.rglob('*') if _.is_file()) / 1e6
    print(f'   完成: {n} 件 / {mb:.0f} MB, 耗时 {time.time() - t0:.0f}s')

    # 产物: 主包当前可能被"清除产物"挪走了 → 从暂存区补进副本, 好让页面有数可验
    prod_dst = dst / 'outputs' / 'rudong'
    if not (prod_dst / 'windscada' / 'alarms.parquet').is_file():
        for src in sorted(ROOT.glob('_products_off*/rudong')):
            if (src / 'windscada' / 'alarms.parquet').is_file():
                print(f'② 主包产物不在位 → 从 {src.relative_to(ROOT)} 拷进副本')
                shutil.copytree(src, prod_dst, dirs_exist_ok=True)
                break
        else:
            print('② 找不到可用产物(主包与暂存区都没有) —— 页面会是"无产物", 属预期')
    else:
        print('② 副本已带产物 (从主包拷来)')

    print('③ 在副本里跑安装 (离线)')
    py = install_in(dst)
    if py is None:
        return 1

    print('④ 起服务核验 (副本独占端口; 组件端口被占则跳过这一步)')
    ok = run_and_probe(dst, py)

    print('⑤ 收尾')
    if keep:
        print(f'   副本保留在 {dst} (--keep); 用完手工删掉即可')
    else:
        shutil.rmtree(dst, ignore_errors=True)
        print('   副本已删除')
    print('\n模拟移植结论: 见上面各步 —— 安装成 + 页面与主包同字节 = 换机可行(记得在目标机也跑一次 install)')
    return 0


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument('--brief', action='store_true')
    ap.add_argument('--simulate', action='store_true', help='拷副本→副本内安装→起服务核验→清理')
    ap.add_argument('--into', default=None, help='--simulate 的副本目录 (默认 %%TEMP%%/guanlan_portsim)')
    ap.add_argument('--keep', action='store_true', help='--simulate 后保留副本')
    a = ap.parse_args()
    print(f'移植性大扫描 @ {ROOT}\n')
    if a.simulate:
        rc = simulate(a.into, a.keep)
        print()
    hits = scan_paths(verbose=not a.brief)
    big = scan_big_text()
    kv = check_venv()
    dirs = check_wheels()
    check_vendor()
    check_configs()
    bad = len(hits) + len(big)
    print(f'\n结论: {"路径层面干净" if not bad else f"{bad} 处机器相关路径要处理"};'
          f' 虚拟环境/轮子/便携运行时见上面各节 —— '
          f'"拷包就能跑"只在**同路径同基础 Python**时成立, 换机请按 README 先跑一次 install。')
    return 0 if not bad else 1


if __name__ == '__main__':
    for _s in (sys.stdout, sys.stderr):
        try: _s.reconfigure(errors='replace')
        except Exception: pass
    sys.exit(main())