#!/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())