guanlan-src/scripts/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

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 = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

# 机器相关绝对路径: 盘符路径 / 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',)
# 允许的例外: 文档里明确在讲"路径约定"或演示用的占位
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 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 = ROOT / 'configs' / '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} 里跑安装 (离线) …')
    cmd = (['powershell', '-ExecutionPolicy', 'Bypass', '-File', str(dst / 'install.ps1')]
           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


def run_and_probe(dst: pathlib.Path, py: pathlib.Path, base_port=28084) -> int:
    """在副本里起服务并核验页面 —— **不动正在跑的实例**: 网关改用空闲端口, 组件端口若被占则跳过。

    原先网关端口被占就整段跳过(实测踩到: 主实例在跑 → 模拟移植只做了拷贝+安装)。现在:
      · 组件端口(18033 等)空闲 → 用它们; 网关端口(28084)被占 → 换成空闲端口(如 28700);
      · 网关 ROUTES 里上游端口是写死的(文档 §三 的已知限制), 所以组件端口**必须**用默认值, 不能随网关一起改。
    """
    comp_ports = (18033, 18020, 18791, 18792, 64292)
    busy_comp = [p for p in comp_ports if _busy(p)]
    if busy_comp:
        print(f'   [!] 组件端口被占用 {busy_comp} —— 跳过启动核验(先停掉正在跑的实例再试)')
        return 0
    gw = base_port if not _busy(base_port) else _free_port(28700)
    if gw is None:
        print('   [!] 找不到空闲网关端口 —— 跳过启动核验'); return 0
    if gw != base_port:
        cfg_p = dst / 'configs' / 'serve.json'
        cfg = json.loads(cfg_p.read_text(encoding='utf-8-sig'))
        cfg['gateway'] = gw
        cfg_p.write_text(json.dumps(cfg, ensure_ascii=False, indent=2), encoding='utf-8')
        print(f'   网关端口 {base_port} 已被占用 → 副本改用 {gw} (不打扰正在跑的实例)')
    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:
        urls = [('/', 20225828), ('/detail/', 229643), ('/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})')
                    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:
        subprocess.run([str(py), 'guanlan.py', 'stop'], cwd=str(dst), capture_output=True)
    return ok


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())