309 lines
12 KiB
Python
309 lines
12 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""
|
|||
|
|
sync-diff: 上游三向同步差异分析工具
|
|||
|
|
|
|||
|
|
功能
|
|||
|
|
----
|
|||
|
|
给定「上游 git 仓库 + 基线 ref(base) + 上游新版本 ref(head) + 本地仓库(local)」，
|
|||
|
|
对指定子路径做三向文件级 diff，输出可执行分类清单，供同步合并使用：
|
|||
|
|
|
|||
|
|
same base == head == local 无需动作
|
|||
|
|
upstream_only base == local != head 仅上游改 → 直接采用上游
|
|||
|
|
local_only base == head != local 仅本地改 → 保留本地
|
|||
|
|
both 三方互异 双改 → 人工裁决
|
|||
|
|
only_head 仅 head 有 上游新增 → 引入
|
|||
|
|
only_local 仅 local 有 本地新增 → 保留
|
|||
|
|
removed_upstream base 有、head/local 都无 双方都删 → 忽略
|
|||
|
|
deleted_upstream base 有、head 无、local 有 上游删了本地还留 → 决策是否跟随
|
|||
|
|
deleted_local base 有、head 有、local 无 本地删了上游还在 → 决策是否恢复
|
|||
|
|
|
|||
|
|
路径映射
|
|||
|
|
--------
|
|||
|
|
本地若对上游目录做过重命名（如 src/qwenpaw -> src/pineagents），通过 --map 指定，
|
|||
|
|
导出后会先把上游树按映射改名再比对，避免把「重命名」误判为「删除+新增」。
|
|||
|
|
|
|||
|
|
测试文件
|
|||
|
|
--------
|
|||
|
|
默认将 *test*/__tests__/*.test.*/*.spec.* 归入独立的 tests_* 清单输出，
|
|||
|
|
不混入功能源码统计（--include-tests 可合并）。
|
|||
|
|
|
|||
|
|
用法示例
|
|||
|
|
--------
|
|||
|
|
# 前端 console/src 比对（本次 QwenPaw 2.1.0b1 -> 2.2.0）
|
|||
|
|
python scripts/sync-diff.py \
|
|||
|
|
--upstream "/path/to/QwenPaw" --base v2.1.0-beta.1 --head v2.2.0 \
|
|||
|
|
--local . --subpath console/src --out sync_out
|
|||
|
|
|
|||
|
|
# 后端（上游 src/qwenpaw -> 本地 src/pineagents）
|
|||
|
|
python scripts/sync-diff.py --upstream /path/to/QwenPaw \
|
|||
|
|
--base v2.1.0-beta.1 --head v2.2.0 --local . \
|
|||
|
|
--subpath src --map src/qwenpaw=src/pineagents --out sync_out_backend
|
|||
|
|
"""
|
|||
|
|
import argparse
|
|||
|
|
import hashlib
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
import shutil
|
|||
|
|
import subprocess
|
|||
|
|
import sys
|
|||
|
|
import tempfile
|
|||
|
|
|
|||
|
|
IGNORE_DIRS = {"node_modules", "dist", "build", ".venv", "venv", "__pycache__", ".git", "target", "out", "logs"}
|
|||
|
|
TEST_MARKERS = ("test", "__tests__", "e2e", "tests")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def log(msg):
|
|||
|
|
print(msg, file=sys.stderr)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def run_git(args, cwd):
|
|||
|
|
r = subprocess.run(["git"] + args, cwd=cwd, capture_output=True, text=True)
|
|||
|
|
if r.returncode != 0:
|
|||
|
|
raise RuntimeError("git %s failed: %s" % (" ".join(args), r.stderr.strip()))
|
|||
|
|
return r.stdout
|
|||
|
|
|
|||
|
|
|
|||
|
|
def resolve_commit(repo, ref):
|
|||
|
|
"""ref 支持 tag / 分支 / commit，返回完整 commit hash。"""
|
|||
|
|
return run_git(["rev-parse", "%s^{commit}" % ref], repo).strip()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def export_tree(repo, ref, subpath, dest):
|
|||
|
|
"""git archive 导出 ref 的 subpath 到 dest（去掉 subpath 前缀）。"""
|
|||
|
|
os.makedirs(dest, exist_ok=True)
|
|||
|
|
strip = len(subpath.split("/")) if subpath else 0
|
|||
|
|
with open(os.devnull, "wb") as devnull:
|
|||
|
|
p = subprocess.Popen(
|
|||
|
|
["git", "archive", ref, subpath], cwd=repo,
|
|||
|
|
stdout=subprocess.PIPE, stderr=devnull,
|
|||
|
|
)
|
|||
|
|
t = subprocess.Popen(
|
|||
|
|
["tar", "-x", "-C", dest, "--strip-components", str(strip)],
|
|||
|
|
stdin=p.stdout, stdout=devnull, stderr=devnull,
|
|||
|
|
)
|
|||
|
|
if p.stdout is not None:
|
|||
|
|
p.stdout.close()
|
|||
|
|
t.wait()
|
|||
|
|
p.wait()
|
|||
|
|
if p.returncode != 0 or t.returncode != 0:
|
|||
|
|
raise RuntimeError("git archive %s:%s failed" % (ref, subpath))
|
|||
|
|
|
|||
|
|
|
|||
|
|
def export_local(repo, subpath, dest, path_map):
|
|||
|
|
"""用 git ls-files 取本地 tracked 文件，导出到 dest（应用路径映射）。"""
|
|||
|
|
os.makedirs(dest, exist_ok=True)
|
|||
|
|
files = [f for f in run_git(["ls-files", "-z", subpath], repo).split("\0") if f]
|
|||
|
|
for f in files:
|
|||
|
|
rel = f[len(subpath):].lstrip("/") if subpath else f
|
|||
|
|
if not rel:
|
|||
|
|
continue
|
|||
|
|
# 应用路径映射（把上游名替换为本地名）
|
|||
|
|
out_rel = rel
|
|||
|
|
for old, new in path_map:
|
|||
|
|
if rel == old or rel.startswith(old + "/"):
|
|||
|
|
out_rel = new + rel[len(old):]
|
|||
|
|
break
|
|||
|
|
target = os.path.join(dest, out_rel)
|
|||
|
|
os.makedirs(os.path.dirname(target), exist_ok=True)
|
|||
|
|
blob = run_git(["show", "HEAD:" + f], repo)
|
|||
|
|
with open(target, "w", encoding="utf-8", newline="") as fh:
|
|||
|
|
fh.write(blob)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def apply_map(tree_root, path_map):
|
|||
|
|
"""把 tree_root 下符合映射的文件/目录改名（处理 export 后未映射的上游树）。"""
|
|||
|
|
for old, new in path_map:
|
|||
|
|
src = os.path.join(tree_root, old)
|
|||
|
|
if os.path.exists(src):
|
|||
|
|
dst = os.path.join(tree_root, new)
|
|||
|
|
os.makedirs(os.path.dirname(dst), exist_ok=True)
|
|||
|
|
if os.path.exists(dst):
|
|||
|
|
shutil.rmtree(dst)
|
|||
|
|
os.rename(src, dst)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def md5_file(path):
|
|||
|
|
h = hashlib.md5()
|
|||
|
|
with open(path, "rb") as f:
|
|||
|
|
for chunk in iter(lambda: f.read(65536), b""):
|
|||
|
|
h.update(chunk)
|
|||
|
|
return h.hexdigest()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def walk_files(root, ignore_dirs, include_tests):
|
|||
|
|
"""返回 {rel_path: md5hex}。"""
|
|||
|
|
result = {}
|
|||
|
|
for dirpath, dirnames, filenames in os.walk(root):
|
|||
|
|
dirnames[:] = [d for d in dirnames if d not in ignore_dirs]
|
|||
|
|
for fn in filenames:
|
|||
|
|
full = os.path.join(dirpath, fn)
|
|||
|
|
rel = os.path.relpath(full, root)
|
|||
|
|
is_test = is_test_file(rel)
|
|||
|
|
if is_test and not include_tests:
|
|||
|
|
continue
|
|||
|
|
result[rel] = md5_file(full)
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
|
|||
|
|
def is_test_file(rel):
|
|||
|
|
parts = rel.split(os.sep)
|
|||
|
|
if any(m in p for p in parts for m in TEST_MARKERS):
|
|||
|
|
return True
|
|||
|
|
if rel.endswith((".test.ts", ".test.tsx", ".test.js", ".spec.ts", ".spec.tsx", ".spec.js", ".test.py")):
|
|||
|
|
return True
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
|
|||
|
|
def classify(base, head, local):
|
|||
|
|
cats = {k: [] for k in (
|
|||
|
|
"same", "upstream_only", "local_only", "both", "only_head", "only_local",
|
|||
|
|
"removed_upstream", "deleted_upstream", "deleted_local",
|
|||
|
|
)}
|
|||
|
|
tests = {k: [] for k in cats}
|
|||
|
|
base_set = set(base); head_set = set(head); local_set = set(local)
|
|||
|
|
|
|||
|
|
for p in sorted(base_set | head_set | local_set):
|
|||
|
|
b, h, l = p in base_set, p in head_set, p in local_set
|
|||
|
|
if b and h and l:
|
|||
|
|
if base[p] == head[p] == local[p]:
|
|||
|
|
key = "same"
|
|||
|
|
elif base[p] == local[p]:
|
|||
|
|
key = "upstream_only"
|
|||
|
|
elif base[p] == head[p]:
|
|||
|
|
key = "local_only"
|
|||
|
|
else:
|
|||
|
|
key = "both"
|
|||
|
|
elif h and not b and not l:
|
|||
|
|
key = "only_head"
|
|||
|
|
elif l and not b and not h:
|
|||
|
|
key = "only_local"
|
|||
|
|
elif b and not h and not l:
|
|||
|
|
key = "removed_upstream"
|
|||
|
|
elif b and not h and l:
|
|||
|
|
key = "deleted_upstream"
|
|||
|
|
else: # b and h and not l
|
|||
|
|
key = "deleted_local"
|
|||
|
|
(tests if is_test_file(p) else cats)[key].append(p)
|
|||
|
|
return cats, tests
|
|||
|
|
|
|||
|
|
|
|||
|
|
def write_list(path, items):
|
|||
|
|
with open(path, "w", encoding="utf-8") as f:
|
|||
|
|
for p in items:
|
|||
|
|
f.write(p + "\n")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main():
|
|||
|
|
ap = argparse.ArgumentParser(description="上游三向同步差异分析")
|
|||
|
|
ap.add_argument("--upstream", required=True, help="上游 git 仓库路径")
|
|||
|
|
ap.add_argument("--base", required=True, help="基线 ref（tag/commit）")
|
|||
|
|
ap.add_argument("--head", required=True, help="上游新版本 ref")
|
|||
|
|
ap.add_argument("--local", default=".", help="本地仓库路径（默认当前目录）")
|
|||
|
|
ap.add_argument("--subpath", default="console/src", help="比对的子路径（默认 console/src）")
|
|||
|
|
ap.add_argument("--map", action="append", default=[], metavar="OLD=NEW",
|
|||
|
|
help="上游→本地路径映射，可多次指定，如 src/qwenpaw=src/pineagents")
|
|||
|
|
ap.add_argument("--out", default="sync_out", help="输出目录")
|
|||
|
|
ap.add_argument("--include-tests", action="store_true", help="将测试文件并入主分类（默认单独输出 tests_*）")
|
|||
|
|
ap.add_argument("--keep-tmp", action="store_true", help="保留临时导出目录（调试用）")
|
|||
|
|
args = ap.parse_args()
|
|||
|
|
|
|||
|
|
path_map = []
|
|||
|
|
for m in args.map:
|
|||
|
|
if "=" not in m:
|
|||
|
|
ap.error("--map 需为 OLD=NEW 格式: %s" % m)
|
|||
|
|
old, new = m.split("=", 1)
|
|||
|
|
path_map.append((old.strip("/"), new.strip("/")))
|
|||
|
|
|
|||
|
|
local = os.path.abspath(args.local)
|
|||
|
|
upstream = os.path.abspath(args.upstream)
|
|||
|
|
if not os.path.isdir(os.path.join(local, ".git")):
|
|||
|
|
ap.error("本地不是 git 仓库: %s" % local)
|
|||
|
|
if not os.path.isdir(os.path.join(upstream, ".git")):
|
|||
|
|
ap.error("上游不是 git 仓库: %s" % upstream)
|
|||
|
|
|
|||
|
|
base_commit = resolve_commit(upstream, args.base)
|
|||
|
|
head_commit = resolve_commit(upstream, args.head)
|
|||
|
|
local_commit = resolve_commit(local, "HEAD")
|
|||
|
|
log("base = %s (%s)" % (args.base, base_commit[:12]))
|
|||
|
|
log("head = %s (%s)" % (args.head, head_commit[:12]))
|
|||
|
|
log("local = HEAD (%s)" % local_commit[:12])
|
|||
|
|
|
|||
|
|
tmp = tempfile.mkdtemp(prefix="sync-diff-")
|
|||
|
|
try:
|
|||
|
|
d_base = os.path.join(tmp, "base")
|
|||
|
|
d_head = os.path.join(tmp, "head")
|
|||
|
|
d_local = os.path.join(tmp, "local")
|
|||
|
|
export_tree(upstream, args.base, args.subpath, d_base)
|
|||
|
|
export_tree(upstream, args.head, args.subpath, d_head)
|
|||
|
|
apply_map(d_base, path_map)
|
|||
|
|
apply_map(d_head, path_map)
|
|||
|
|
export_local(local, args.subpath, d_local, path_map)
|
|||
|
|
|
|||
|
|
ignore = IGNORE_DIRS
|
|||
|
|
base_map = walk_files(d_base, ignore, args.include_tests)
|
|||
|
|
head_map = walk_files(d_head, ignore, args.include_tests)
|
|||
|
|
local_map = walk_files(d_local, ignore, args.include_tests)
|
|||
|
|
|
|||
|
|
cats, tests = classify(base_map, head_map, local_map)
|
|||
|
|
finally:
|
|||
|
|
if not args.keep_tmp:
|
|||
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|||
|
|
|
|||
|
|
os.makedirs(args.out, exist_ok=True)
|
|||
|
|
meta = {
|
|||
|
|
"upstream": upstream,
|
|||
|
|
"base_ref": args.base, "base_commit": base_commit,
|
|||
|
|
"head_ref": args.head, "head_commit": head_commit,
|
|||
|
|
"local": local, "local_commit": local_commit,
|
|||
|
|
"subpath": args.subpath, "path_map": args.map,
|
|||
|
|
"include_tests": args.include_tests,
|
|||
|
|
}
|
|||
|
|
with open(os.path.join(args.out, "meta.json"), "w", encoding="utf-8") as f:
|
|||
|
|
json.dump(meta, f, ensure_ascii=False, indent=2)
|
|||
|
|
|
|||
|
|
order = ["same", "upstream_only", "local_only", "both",
|
|||
|
|
"only_head", "only_local", "removed_upstream",
|
|||
|
|
"deleted_upstream", "deleted_local"]
|
|||
|
|
label = {
|
|||
|
|
"same": "一致(无需动作)", "upstream_only": "仅上游改(直接采用上游)",
|
|||
|
|
"local_only": "仅本地改(保留本地)", "both": "双改(人工裁决)",
|
|||
|
|
"only_head": "上游新增(引入)", "only_local": "本地新增(保留)",
|
|||
|
|
"removed_upstream": "双方删除(忽略)",
|
|||
|
|
"deleted_upstream": "上游已删但本地保留(决策是否跟随删除)",
|
|||
|
|
"deleted_local": "本地已删但上游保留(决策是否恢复)",
|
|||
|
|
}
|
|||
|
|
summary = []
|
|||
|
|
for k in order:
|
|||
|
|
n = len(cats[k]) + len(tests[k])
|
|||
|
|
summary.append((k, label[k], n))
|
|||
|
|
write_list(os.path.join(args.out, k + ".txt"),
|
|||
|
|
cats[k] + tests[k])
|
|||
|
|
if tests[k]:
|
|||
|
|
write_list(os.path.join(args.out, "tests_" + k + ".txt"), tests[k])
|
|||
|
|
# 仅测试文件专属清单
|
|||
|
|
all_test_files = sorted(set(sum(tests.values(), [])))
|
|||
|
|
if all_test_files:
|
|||
|
|
write_list(os.path.join(args.out, "tests_all.txt"), all_test_files)
|
|||
|
|
|
|||
|
|
with open(os.path.join(args.out, "REPORT.md"), "w", encoding="utf-8") as f:
|
|||
|
|
f.write("# 上游同步差异报告（sync-diff）\n\n")
|
|||
|
|
f.write("- 上游: `%s`\n- base: `%s` (%s)\n- head: `%s` (%s)\n- local: `%s` (%s)\n- 子路径: `%s`\n- 路径映射: %s\n\n" % (
|
|||
|
|
upstream, args.base, base_commit[:12], args.head, head_commit[:12],
|
|||
|
|
local, local_commit[:12], args.subpath, args.map or "无"))
|
|||
|
|
f.write("| 分类 | 说明 | 数量 |\n|---|---|---|\n")
|
|||
|
|
for k, lab, n in summary:
|
|||
|
|
f.write("| %s | %s | %d |\n" % (k, lab, n))
|
|||
|
|
f.write("\n> 双改文件清单见 `both.txt`；each 分类清单见同目录 `*.txt`。\n")
|
|||
|
|
|
|||
|
|
width = max(len(lab) for _, lab, _ in summary)
|
|||
|
|
log("\n=== 同步差异汇总 ===")
|
|||
|
|
for k, lab, n in summary:
|
|||
|
|
log(" %-*s %4d" % (width, lab, n))
|
|||
|
|
log("\n输出目录: %s" % os.path.abspath(args.out))
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
sys.exit(main())
|