guanlan-src/scripts/java_contract_parity.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""P12-A：Java 后端迁移的**裁判**——契约端点逐一比对 Java 与 Python 的响应。

用途（与前端对拍门同思路：门禁当裁判，逐条收敛）：
  · 读冻契约 `app_backEnd/.../contract/http_api_v1.json`（42 条）；
  · 对每条端点，分别请求 Python 侧与 Java 侧，做**深度相等**比较（JSON 归一化后）；
  · 输出计分板：`已一致 / 有差异 / Java 未实现`，并把"必须一致"的端点清单交给 `--require` 用于门禁。

用法：
  python scripts/java_contract_parity.py [--gateway URL] [--python-base /detail] [--java-base /java]
                                         [--require kpi,fleet,...] [--limit N]
"""
from __future__ import annotations

import argparse
import json
import pathlib
import sys
import urllib.error
import urllib.request

ROOT = pathlib.Path(__file__).resolve().parents[1]
CONTRACT = ROOT / 'app_backEnd/app_backEnd_guanlan/contract/http_api_v1.json'


def get(url: str, timeout: float = 60.0):
    req = urllib.request.Request(url, headers={'Accept': 'application/json'})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            body = r.read().decode('utf-8', 'replace')
            try:
                return r.status, json.loads(body)
            except Exception:
                return r.status, body[:400]
    except urllib.error.HTTPError as e:
        return e.code, (e.read().decode('utf-8', 'replace')[:200])
    except Exception as e:                                   # noqa: BLE001
        return 0, f'{type(e).__name__}: {e}'


def norm(x):
    """比较用归一化：字典按键排序；列表保序（顺序也是行为）。"""
    if isinstance(x, dict):
        return {k: norm(v) for k, v in sorted(x.items())}
    if isinstance(x, list):
        return [norm(v) for v in x]
    return x


def first_diff(a, b, path='$', depth=0):
    if depth > 6:
        return None
    if type(a) is not type(b) and not (isinstance(a, (int, float)) and isinstance(b, (int, float))):
        return f'{path}: 类型 {type(a).__name__} ≠ {type(b).__name__}'
    if isinstance(a, dict):
        for k in sorted(set(a) | set(b)):
            if k not in a:
                return f'{path}.{k}: Java 多键'
            if k not in b:
                return f'{path}.{k}: Java 缺键'
            d = first_diff(a[k], b[k], f'{path}.{k}', depth + 1)
            if d:
                return d
        return None
    if isinstance(a, list):
        if len(a) != len(b):
            return f'{path}: 长度 {len(a)} ≠ {len(b)}'
        for i, (x, y) in enumerate(zip(a, b)):
            d = first_diff(x, y, f'{path}[{i}]', depth + 1)
            if d:
                return d
        return None
    if a != b:
        sa, sb = str(a), str(b)
        return f'{path}: {sa[:60]!r} ≠ {sb[:60]!r}'
    return None


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument('--gateway', default='http://127.0.0.1:28084')
    ap.add_argument('--python-base', default='/detail')
    ap.add_argument('--java-base', default='/java')
    ap.add_argument('--require', default='', help='逗号分隔：这些端点必须一致，否则退出码 1')
    ap.add_argument('--limit', type=int, default=0)
    a = ap.parse_args()

    contract = json.loads(CONTRACT.read_text(encoding='utf-8'))
    eps = contract['endpoints']
    rows = eps if isinstance(eps, list) else [dict(path=k, **(v if isinstance(v, dict) else {})) for k, v in eps.items()]
    if a.limit:
        rows = rows[:a.limit]

    same, diff, missing, skipped = [], [], [], []
    for e in rows:
        path = str(e.get('path') or e.get('url'))
        method = str(e.get('method') or 'GET').upper()
        if method != 'GET':
            skipped.append((path, 'POST 类端点本轮不自动比（需构造请求体）'))
            continue
        # ★ 一律真的问两侧：Java 走 /java 前缀、Python 走 /detail 前缀（灰度期两个并存入口）
        jurl = a.gateway.rstrip('/') + a.java_base + path
        purl = a.gateway.rstrip('/') + a.python_base + path
        js, jb = get(jurl)
        ps, pb = get(purl)
        if js == 0 or (isinstance(jb, str) and ('404' in str(jb) or 'Not Found' in str(jb))) or js == 404:
            missing.append((path, f'Java HTTP {js}'))
            continue
        if js != 200 or ps != 200:
            diff.append((path, f'HTTP Java {js} / Python {ps}'))
            continue
        d = first_diff(norm(jb), norm(pb))
        if d:
            diff.append((path, d))
        else:
            same.append(path)

    print('Java ↔ Python 契约端点比对（裁判：冻契约 %d 条）' % len(rows))
    print('  一致 %d · 有差异 %d · Java 未实现 %d · 跳过 %d\n' % (len(same), len(diff), len(missing), len(skipped)))
    if same:
        print('  [一致] ' + ', '.join(same[:12]) + (' …' if len(same) > 12 else ''))
    for p, why in diff[:10]:
        print('  [差异] %-28s %s' % (p, why))
    for p, why in missing[:10]:
        print('  [未实现] %-26s %s' % (p, why))
    for p, why in skipped[:6]:
        print('  [跳过] %-28s %s' % (p, why))

    need = [x.strip() for x in a.require.split(',') if x.strip()]
    bad = [p for p in need if p not in same]
    if need:
        print('\n门禁要求一致 %d 条 → %s' % (len(need), '通过' if not bad else '未通过: ' + ', '.join(bad)))
    print('\n结论: %s' % ('全部已迁端点一致' if not diff and not missing else
                          '迁移进行中（%d/%d 已一致）' % (len(same), len(rows))))
    return 0 if not bad else 1


if __name__ == '__main__':
    raise SystemExit(main())