#!/usr/bin/env python3 """Compare WCC src/ at an exact Git commit with the official ApexCharts 4.7.0 npm source tarball. This is a source-file lineage inventory, not a semantic equivalence, ownership, or code quality audit. Requires Python 3.9+ and git. The tarball is verified against npm's published SRI before use. """ from __future__ import annotations import argparse import base64 import difflib import hashlib import json import subprocess import tarfile import tempfile from collections import defaultdict from datetime import datetime, timezone from pathlib import PurePosixPath from pathlib import Path NPM_SRI = "sha512-iZSrrBGvVlL+nt2B1NpqfDuBZ9jX61X9I2+XV0hlYXHtTwhwLTHDKGXjNXAgFBDLuvSYCB/rq2nPWVPRv2DrGA==" NPM_SHA1 = "27e578a2fd66847d8ea425aeea1de56d6be42f21" NPM_TARBALL_URL = "https://registry.npmjs.org/apexcharts/-/apexcharts-4.7.0.tgz" NPM_GIT_HEAD = "2c72f8299dba81dddbb50d2b8178163efbde2305" def run(*args: str, cwd: Path | None = None) -> str: return subprocess.check_output(args, cwd=cwd, text=True).strip() def sha256(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def group(path: str) -> str: parts = PurePosixPath(path).parts if len(parts) < 2: return "src/__root__" if parts[1] == "modules": return "src/modules/" + (parts[2] if len(parts) > 2 else "__root__") return "src/" + parts[1] def classify(path: str) -> str: p = PurePosixPath(path) name = p.name.lower() if p.parts[:2] == ("src", "premium"): return "premium-commercial-source" if p.parts[:2] == ("src", "agent"): return "agent-source" if p.parts[:2] == ("src", "svgjs") or p.parts[:2] == ("src", "libs"): return "vendored-third-party-or-adapter" if p.parts[:2] == ("src", "assets") or p.suffix.lower() in {".css", ".scss", ".svg", ".png", ".jpg", ".woff", ".woff2"}: return "style-or-asset" if p.parts[:2] == ("src", "locales") or p.suffix.lower() == ".json": return "locale-or-data" if "license" in name or "copyright" in name or "notice" in name: return "license-or-notice" return "source-code" def diff_lines(a: bytes, b: bytes) -> tuple[int, int, int, int]: # Text line deltas are descriptive only: line similarity is not authorship or semantic equivalence. try: al = a.decode("utf-8").splitlines(keepends=True) bl = b.decode("utf-8").splitlines(keepends=True) except UnicodeDecodeError: return 0, 0, 0, 0 sm = difflib.SequenceMatcher(a=al, b=bl, autojunk=True) same = added = removed = 0 for tag, i1, i2, j1, j2 in sm.get_opcodes(): if tag == "equal": same += i2 - i1 elif tag == "insert": added += j2 - j1 elif tag == "delete": removed += i2 - i1 else: removed += i2 - i1 added += j2 - j1 return len(al), len(bl), same, added + removed def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--repo", type=Path, required=True) ap.add_argument("--commit", required=True) ap.add_argument("--apex-tgz", type=Path, required=True) ap.add_argument("--upstream-git-dir", type=Path, help="optional local ApexCharts clone containing the official tag and npm GitHead commits") ap.add_argument("--tag-commit", default="1e93a0d47b834111cf616fb6c78a263bbae7c1d8") ap.add_argument("--npm-git-head", default=NPM_GIT_HEAD) ap.add_argument("--public-safe", action="store_true", default=True, help="omit per-file detail for non-Apex target-only files (including premium); emit a module CSV") ap.add_argument("--out", type=Path, required=True) args = ap.parse_args() repo = args.repo.resolve() tgz = args.apex_tgz.resolve() out = args.out.resolve() out.mkdir(parents=True, exist_ok=True) verified = subprocess.check_output(["git", "-C", str(repo), "rev-parse", f"{args.commit}^{{commit}}"], text=True).strip() tree = subprocess.check_output(["git", "-C", str(repo), "rev-parse", f"{verified}^{{tree}}"], text=True).strip() tar_bytes = tgz.read_bytes() actual_sri = "sha512-" + base64.b64encode(hashlib.sha512(tar_bytes).digest()).decode("ascii") actual_sha1 = hashlib.sha1(tar_bytes).hexdigest() if actual_sri != NPM_SRI or actual_sha1 != NPM_SHA1: raise SystemExit(f"Baseline integrity mismatch: SRI={actual_sri} SHA1={actual_sha1}") upstream: dict[str, bytes] = {} with tarfile.open(tgz, "r:gz") as tf: for member in tf.getmembers(): if not member.isfile() or not member.name.startswith("package/src/"): continue rel = member.name[len("package/"):] stream = tf.extractfile(member) if stream is not None: upstream[rel] = stream.read() package_meta = json.loads(tf.extractfile("package/package.json").read().decode("utf-8")) if package_meta.get("name") != "apexcharts" or package_meta.get("version") != "4.7.0": raise SystemExit("Tarball package identity is not apexcharts@4.7.0") names = run("git", "-C", str(repo), "ls-tree", "-r", "--name-only", "-z", verified, "--", "src").split("\0") target: dict[str, bytes] = {} for name in filter(None, names): target[name] = subprocess.check_output(["git", "-C", str(repo), "show", f"{verified}:{name}"]) allpaths = sorted(set(upstream) | set(target)) per = defaultdict(lambda: {"group": "", "classification": "", "baselineFiles": 0, "targetFiles": 0, "identicalFiles": 0, "modifiedMatchedFiles": 0, "addedTargetFiles": 0, "removedBaselineFiles": 0, "baselineBytes": 0, "targetBytes": 0, "baselineTotalLines": 0, "targetTotalLines": 0, "addedTargetFileLines": 0, "removedBaselineFileLines": 0, "identicalFileLinesBaseline": 0, "identicalFileLinesTarget": 0, "modifiedBaselineLines": 0, "modifiedTargetLines": 0, "modifiedFilesUnchangedLinesEstimate": 0, "modifiedFilesChangedLinesEstimate": 0, "gitDiffLineAdds": 0, "gitDiffLineDeletes": 0}) files = [] summary = {k: 0 for k in ("baselineFiles", "targetFiles", "matchedPathFiles", "identicalFiles", "modifiedMatchedFiles", "addedTargetFiles", "removedBaselineFiles", "baselineBytes", "targetBytes", "baselineTotalLines", "targetTotalLines", "addedTargetFileLines", "removedBaselineFileLines", "identicalFileLinesBaseline", "identicalFileLinesTarget", "modifiedBaselineLines", "modifiedTargetLines", "modifiedFilesUnchangedLinesEstimate", "modifiedFilesChangedLinesEstimate")} summary["gitDiffLineAdds"] = 0 summary["gitDiffLineDeletes"] = 0 git_changes: dict[str, tuple[int, int]] = {} with tempfile.TemporaryDirectory(prefix="wcc-apex470-lineage-") as tmp_name: tmp = Path(tmp_name) for path in allpaths: a, b = upstream.get(path), target.get(path) if a is None or b is None or a == b: continue left, right = tmp / "baseline", tmp / "target" left.write_bytes(a); right.write_bytes(b) proc = subprocess.run(["git", "diff", "--no-index", "--numstat", "--no-ext-diff", "--diff-algorithm=myers", "--", str(left), str(right)], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if proc.returncode not in (0, 1): raise SystemExit(f"git diff numstat failed for {path}: {proc.stderr}") cols = proc.stdout.strip().split("\t") if len(cols) >= 2 and cols[0] != "-" and cols[1] != "-": git_changes[path] = (int(cols[0]), int(cols[1])) summary["gitDiffLineAdds"] += int(cols[0]) summary["gitDiffLineDeletes"] += int(cols[1]) for path in allpaths: a, b = upstream.get(path), target.get(path) g, c = group(path), classify(path) row = per[g] row["group"], row["classification"] = g, c if path in git_changes: row["gitDiffLineAdds"] += git_changes[path][0] row["gitDiffLineDeletes"] += git_changes[path][1] f = {"path": path, "group": g, "classification": c, "baselinePresent": a is not None, "targetPresent": b is not None, "baselineBytes": len(a) if a is not None else 0, "targetBytes": len(b) if b is not None else 0, "baselineSha256": sha256(a) if a is not None else None, "targetSha256": sha256(b) if b is not None else None} if a is not None: row["baselineFiles"] += 1; row["baselineBytes"] += len(a) summary["baselineFiles"] += 1; summary["baselineBytes"] += len(a) an = len(a.decode("utf-8", errors="replace").splitlines()) row["baselineTotalLines"] += an; summary["baselineTotalLines"] += an if b is not None: row["targetFiles"] += 1; row["targetBytes"] += len(b) summary["targetFiles"] += 1; summary["targetBytes"] += len(b) bn = len(b.decode("utf-8", errors="replace").splitlines()) row["targetTotalLines"] += bn; summary["targetTotalLines"] += bn if a is not None and b is not None: summary["matchedPathFiles"] += 1 if a == b: row["identicalFiles"] += 1; summary["identicalFiles"] += 1 try: n = len(a.decode("utf-8").splitlines(keepends=True)) except UnicodeDecodeError: n = 0 row["identicalFileLinesBaseline"] += n; row["identicalFileLinesTarget"] += n summary["identicalFileLinesBaseline"] += n; summary["identicalFileLinesTarget"] += n f["status"] = "byte-identical" f["identicalLines"] = n else: row["modifiedMatchedFiles"] += 1; summary["modifiedMatchedFiles"] += 1 al, bl, unchanged, changed = diff_lines(a, b) row["modifiedBaselineLines"] += al; row["modifiedTargetLines"] += bl row["modifiedFilesUnchangedLinesEstimate"] += unchanged; row["modifiedFilesChangedLinesEstimate"] += changed summary["modifiedBaselineLines"] += al; summary["modifiedTargetLines"] += bl summary["modifiedFilesUnchangedLinesEstimate"] += unchanged; summary["modifiedFilesChangedLinesEstimate"] += changed f["status"] = "modified-at-same-path" f["lineComparison"] = {"baselineLines": al, "targetLines": bl, "unchangedLineEstimate": unchanged, "addedPlusRemovedLineEstimate": changed} elif b is not None: row["addedTargetFiles"] += 1; summary["addedTargetFiles"] += 1; f["status"] = "new-in-target" f["targetLines"] = bn row["addedTargetFileLines"] += bn; summary["addedTargetFileLines"] += bn else: row["removedBaselineFiles"] += 1; summary["removedBaselineFiles"] += 1; f["status"] = "absent-from-target" f["baselineLines"] = an row["removedBaselineFileLines"] += an; summary["removedBaselineFileLines"] += an files.append(f) groups = [] for g, row in sorted(per.items()): groups.append(row) scope_keys = ("baselineFiles", "targetFiles", "identicalFiles", "modifiedMatchedFiles", "addedTargetFiles", "removedBaselineFiles", "baselineTotalLines", "targetTotalLines", "identicalFileLinesBaseline", "modifiedFilesUnchangedLinesEstimate", "modifiedFilesChangedLinesEstimate", "addedTargetFileLines", "removedBaselineFileLines", "gitDiffLineAdds", "gitDiffLineDeletes") def aggregate(selected: list[dict]) -> dict: result = {key: sum(row.get(key, 0) for row in selected) for key in scope_keys} result["retainedLineMatchEstimate"] = result["identicalFileLinesBaseline"] + result["modifiedFilesUnchangedLinesEstimate"] result["retainedAsPercentOfBaselineLines"] = round(100 * result["retainedLineMatchEstimate"] / result["baselineTotalLines"], 2) if result["baselineTotalLines"] else None result["retainedAsPercentOfTargetLines"] = round(100 * result["retainedLineMatchEstimate"] / result["targetTotalLines"], 2) if result["targetTotalLines"] else None result["reconciledTargetLines"] = result["baselineTotalLines"] - result["removedBaselineFileLines"] + result["addedTargetFileLines"] + result["gitDiffLineAdds"] - result["gitDiffLineDeletes"] result["lineTotalReconciles"] = result["reconciledTargetLines"] == result["targetTotalLines"] return result derived_scopes = { "allTrackedSrc": aggregate(groups), "coreRelatedSourcePaths": aggregate([g for g in groups if g["group"] == "src/apexcharts.js" or g["group"] == "src/charts" or g["group"].startswith("src/modules/") or g["group"] == "src/utils"]), "nonPremiumTrackedSrc": aggregate([g for g in groups if g["group"] != "src/premium"]), "vendoredThirdPartyOrAdapter": aggregate([g for g in groups if g["classification"] == "vendored-third-party-or-adapter"]), "styleOrAsset": aggregate([g for g in groups if g["classification"] == "style-or-asset"]), "localeOrData": aggregate([g for g in groups if g["classification"] == "locale-or-data"]), } github_compare = None if args.upstream_git_dir: upstream_repo = args.upstream_git_dir.resolve() tag_commit = run("git", "-C", str(upstream_repo), "rev-parse", f"{args.tag_commit}^{{commit}}") npm_commit = run("git", "-C", str(upstream_repo), "rev-parse", f"{args.npm_git_head}^{{commit}}") def git_src(commit: str) -> dict[str, bytes]: paths = run("git", "-C", str(upstream_repo), "ls-tree", "-r", "--name-only", "-z", commit, "--", "src").split("\0") return {p: subprocess.check_output(["git", "-C", str(upstream_repo), "show", f"{commit}:{p}"]) for p in filter(None, paths)} tag_src, npm_src = git_src(tag_commit), git_src(npm_commit) def source_pair_stats(source: dict[str, bytes]) -> dict: common = set(source) & set(upstream) exact = sorted(p for p in common if source[p] == upstream[p]) differing = sorted(p for p in common if source[p] != upstream[p]) return {"gitSourceFileCount": len(source), "npmTarballSourceFileCount": len(upstream), "samePathCount": len(common), "byteIdenticalFileCount": len(exact), "differentSamePathFileCount": len(differing), "gitOnlyFileCount": len(set(source) - set(upstream)), "npmOnlyFileCount": len(set(upstream) - set(source)), "differentSamePathFiles": differing} github_compare = {"repository": "https://github.com/apexcharts/apexcharts.js.git", "tagRef": "refs/tags/v4.7.0", "tagCommit": tag_commit, "registryReportedNpmGitHead": npm_commit, "npmGitHeadSourceVsTarball": source_pair_stats(npm_src), "tagSourceVsTarball": source_pair_stats(tag_src), "note": "Tag and registry GitHead are distinct commits. Counts compare their tracked src/ trees with the integrity-verified npm tarball source; the tag is not substituted for the registry artifact."} result = { "reportType": "source-file-lineage-inventory", "generatedAtUtc": datetime.now(timezone.utc).isoformat(), "interpretation": { "notAClaim": ["semantic equivalence", "authorship of modified lines", "bug-free code", "independent implementation", "license determination", "bundle size reduction"], "lineMethod": "UTF-8 line-level difflib.SequenceMatcher; for changed same-path files, unchanged line counts are estimates sensitive to formatting/reordering; gitDiffLineAdds/Deletes are raw git diff --no-index --numstat with the Myers algorithm and reconcile net line totals. Exact byte identity is the strongest unchanged-file result.", "scope": "tracked src/** files only, as they exist in the selected WCC Git commit; generated dist, tests, docs, package metadata and dependencies are excluded.", "vendorPolicy": "src/svgjs and src/libs are labeled vendored-third-party-or-adapter by path. Source headers/package licenses must be read before attributing ownership; source similarity alone does not establish authorship. CSS/assets and locale/data are reported in separate groups." }, "baseline": {"name": "apexcharts", "version": "4.7.0", "source": "npm registry package tarball", "tarballUrl": NPM_TARBALL_URL, "registryGitHead": NPM_GIT_HEAD, "tarballSha1": actual_sha1, "tarballSha512Integrity": actual_sri, "packageJson": {k: package_meta.get(k) for k in ("name", "version", "repository", "license", "gitHead")}}, "officialGitHubSourceComparison": github_compare, "target": {"repositoryPath": str(repo), "commit": verified, "tree": tree}, "summary": summary, "derivedLineageScopes": derived_scopes, "groups": groups, "files": files } output_result = result if args.public_safe: output_result = dict(result) output_result["interpretation"] = dict(result["interpretation"]) output_result["interpretation"]["publicSafeDetail"] = "Per-file target-only records are omitted. Per-file rows cover only paths present in the public ApexCharts baseline; commercial/premium source paths and hashes are never exported. Aggregate full-checkout counts are retained only for denominator reconciliation." output_result["target"] = {"commit": verified, "tree": tree} output_result["groups"] = [g for g in groups if g["group"] != "src/premium"] output_result["files"] = [f for f in files if f["baselinePresent"]] output_result["omittedCommercialDetail"] = {"pathDetailsOmitted": True, "fileCount": next((g["targetFiles"] for g in groups if g["group"] == "src/premium"), 0), "lineCount": next((g["targetTotalLines"] for g in groups if g["group"] == "src/premium"), 0)} import csv with (out / "module-summary.csv").open("w", encoding="utf-8", newline="") as stream: writer = csv.DictWriter(stream, fieldnames=["group", "classification", "baselineFiles", "targetFiles", "identicalFiles", "modifiedMatchedFiles", "addedTargetFiles", "removedBaselineFiles", "baselineTotalLines", "targetTotalLines", "identicalFileLinesBaseline", "modifiedFilesUnchangedLinesEstimate", "modifiedFilesChangedLinesEstimate", "gitDiffLineAdds", "gitDiffLineDeletes"]) writer.writeheader() for row in output_result["groups"]: writer.writerow({key: row.get(key, "") for key in writer.fieldnames}) report_path = out / "lineage-report.json" report_path.write_text(json.dumps(output_result, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") print(json.dumps({"out": str(report_path), "baselineIntegrity": actual_sri, "targetCommit": verified, "summary": summary, "publicSafe": args.public_safe}, indent=2)) if __name__ == "__main__": main()