refactor(*): 代码质量 - 优化前后端实现并完善协议与画布验证

scripts/audit-source-overlap.py

@@ -0,0 +1,123 @@
+#!/usr/bin/env python3
+"""Triage source overlap in tracked and non-ignored untracked working-tree files.
+
+Exact resources and normalized line blocks need separate provenance review.
+This tool does not establish ownership or independent implementation.
+"""
+import argparse
+from collections import defaultdict
+import difflib
+import hashlib
+import io
+import json
+from pathlib import Path
+import subprocess
+import tarfile
+
+SOURCE_SUFFIXES = {".ts", ".tsx", ".js", ".mjs", ".css", ".mdx"}
+MIN_LINES = 5
+MIN_CHARACTERS = 160
+
+
+def normalized_lines(name, content):
+    if Path(name).suffix not in SOURCE_SUFFIXES:
+        return None
+    try:
+        return [line.strip() for line in content.decode().splitlines()]
+    except UnicodeDecodeError:
+        return None
+
+
+def seeds(lines):
+    # Every retained block has at least one five-line seed. Do not filter short
+    # seeds: a longer qualifying block can consist entirely of short lines.
+    return {tuple(lines[i:i + MIN_LINES]) for i in range(len(lines) - MIN_LINES + 1)}
+
+
+def matching_report(name, baseline_name, old, current):
+    blocks = [
+        block for block in difflib.SequenceMatcher(None, old, current, autojunk=False).get_matching_blocks()
+        if block.size >= MIN_LINES and sum(map(len, old[block.a:block.a + block.size])) >= MIN_CHARACTERS
+    ]
+    if not blocks:
+        return None
+    return {
+        "file": name, "baseline_file": baseline_name,
+        "baseline_lines": len(old), "current_lines": len(current),
+        "matched_lines": sum(block.size for block in blocks),
+        "blocks": [{"baseline_start": block.a + 1, "current_start": block.b + 1, "lines": block.size} for block in blocks],
+    }
+
+
+def audit(baseline, current):
+    by_hash = defaultdict(list)
+    index = defaultdict(set)
+    sources = {}
+    for name, content in baseline.items():
+        by_hash[hashlib.sha256(content).hexdigest()].append(name)
+        lines = normalized_lines(name, content)
+        if lines is not None:
+            sources[name] = lines
+            for seed in seeds(lines):
+                index[seed].add(name)
+
+    exact, same_path, cross_path = [], [], []
+    for name, content in sorted(current.items()):
+        digest = hashlib.sha256(content).hexdigest()
+        if digest in by_hash:
+            exact.append({"file": name, "baseline_files": sorted(by_hash[digest]), "bytes": len(content)})
+        lines = normalized_lines(name, content)
+        if lines is None:
+            continue
+        candidates = {name} if name in sources else set()
+        for seed in seeds(lines):
+            candidates.update(index.get(seed, ()))
+        for baseline_name in sorted(candidates):
+            report = matching_report(name, baseline_name, sources[baseline_name], lines)
+            if report:
+                (same_path if name == baseline_name else cross_path).append(report)
+
+    order = lambda item: (-item["matched_lines"], item["file"], item["baseline_file"])
+    return {
+        "baseline_files": len(baseline), "working_tree_files": len(current),
+        "exact": exact, "same_path_blocks": sorted(same_path, key=order),
+        "cross_path_blocks": sorted(cross_path, key=order),
+        "limitations": "Exact hashes and whitespace-normalized line blocks (at least 5 lines and 160 characters). Includes tracked and non-ignored untracked files; checks cross-path blocks using five-line seeds. Transformed code, shorter blocks and excluded files may be missed. Cross-path totals count file pairs and can double-count lines. Matches are candidates, not ownership conclusions.",
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("archive_url")
+    parser.add_argument("--output", required=True)
+    args = parser.parse_args()
+    archive = subprocess.check_output(["curl", "--fail", "--location", "--silent", "--show-error", "--max-time", "60", args.archive_url])
+    baseline = {}
+    with tarfile.open(fileobj=io.BytesIO(archive), mode="r:gz") as tar:
+        for member in tar.getmembers():
+            if member.isfile():
+                name = member.name.partition("/")[2]
+                if name:
+                    baseline[name] = tar.extractfile(member).read()
+    paths = subprocess.check_output(["git", "ls-files", "--cached", "--others", "--exclude-standard", "-z"]).decode().split("\0")
+    current = {name: Path(name).read_bytes() for name in sorted(set(paths)) if name and Path(name).is_file()}
+    report = {
+        "head": subprocess.check_output(["git", "rev-parse", "HEAD"]).decode().strip(),
+        "archive_url": args.archive_url,
+        "archive_sha256": hashlib.sha256(archive).hexdigest(),
+        **audit(baseline, current),
+    }
+    output = Path(args.output)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n")
+    print(json.dumps({
+        "report": args.output, "baseline_files": len(baseline), "working_tree_files": len(current),
+        "exact_files": len(report["exact"]), "same_path_candidate_files": len(report["same_path_blocks"]),
+        "same_path_matched_lines": sum(item["matched_lines"] for item in report["same_path_blocks"]),
+        "cross_path_candidate_pairs": len(report["cross_path_blocks"]),
+        "largest_cross_path_candidates": [{key: item[key] for key in ("file", "baseline_file", "matched_lines")} for item in report["cross_path_blocks"][:15]],
+    }, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()

scripts/test_audit_source_overlap.py

@@ -0,0 +1,39 @@
+import importlib.util
+from pathlib import Path
+import unittest
+
+spec = importlib.util.spec_from_file_location("overlap", Path(__file__).with_name("audit-source-overlap.py"))
+overlap = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(overlap)
+
+
+class SourceOverlapTests(unittest.TestCase):
+    def test_renamed_source_is_found_after_small_edits(self):
+        source = "\n".join(f"export const resource_{i} = 'long-enough-contract-value-{i}';" for i in range(12))
+        result = overlap.audit({"old.ts": source.encode()}, {"new.ts": ("// new header\n" + source + "\n// footer").encode()})
+        self.assertEqual(result["same_path_blocks"], [])
+        self.assertEqual(len(result["cross_path_blocks"]), 1)
+        self.assertEqual(result["cross_path_blocks"][0]["matched_lines"], 12)
+        self.assertEqual(result["cross_path_blocks"][0]["baseline_file"], "old.ts")
+
+    def test_short_lines_can_form_a_long_qualifying_block(self):
+        source = "\n".join(f"value{i}," for i in range(40))
+        result = overlap.audit({"old.ts": source.encode()}, {"new.ts": ("// extra\n" + source).encode()})
+        self.assertEqual(result["cross_path_blocks"][0]["matched_lines"], 40)
+
+    def test_identical_renamed_binary_resources_are_reported(self):
+        result = overlap.audit({"old.svg": b"\x00\xffsame-resource"}, {"renamed.svg": b"\x00\xffsame-resource"})
+        self.assertEqual(result["exact"][0]["baseline_files"], ["old.svg"])
+        self.assertEqual(result["cross_path_blocks"], [])
+
+    def test_whitespace_normalization_preserves_same_path_thresholds(self):
+        source = "\n".join(f"const contract_{i} = 'serialized-value-{i}';" for i in range(8))
+        result = overlap.audit({"same.ts": source.encode()}, {"same.ts": ("\n".join("    " + line for line in source.splitlines())).encode()})
+        self.assertEqual(result["same_path_blocks"][0]["matched_lines"], 8)
+        self.assertEqual(result["exact"], [])
+        short = overlap.audit({"old.ts": b"a\nb\nc\nd\ne"}, {"new.ts": b"a\nb\nc\nd\ne"})
+        self.assertEqual(short["cross_path_blocks"], [])
+
+
+if __name__ == "__main__":
+    unittest.main()