"""Phase 4: recover linked Git source for included sites into NEW directories. Read only toward remotes. Bare mirror clones run no repository hooks or code. No dependency installs, no checkouts of working trees.""" import json import os import subprocess import sys import zipfile sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from nlr_common import ROOT, checkpoint_update, log, now, safe_name, sha256_file, write_json ENV = dict(os.environ, GIT_TERMINAL_PROMPT="0", GIT_ASKPASS="/usr/bin/false", SSH_ASKPASS="/usr/bin/false", GIT_LFS_SKIP_SMUDGE="1") def git(*args, cwd=None, timeout=1800): r = subprocess.run(["git", "-c", "credential.helper=", "-c", "core.hooksPath=/dev/null", *args], cwd=cwd, env=ENV, capture_output=True, text=True, timeout=timeout) return r.returncode, r.stdout, r.stderr def main(): cls = json.load(open(os.path.join(ROOT, "checkpoints", "classification.json")))["sites"] for sid, v in sorted(cls.items(), key=lambda x: x[1]["name"]): if not v["include"]: continue m = json.load(open(os.path.join(v["dir"], "metadata", "site.json"))) repo = (m["build_settings"] or {}).get("repo_url") try: deps = json.load(open(os.path.join(v["dir"], "metadata", "deploys.json")))["deploys"] except FileNotFoundError: deps = [] commits = sorted({d["commit_ref"] for d in deps if d.get("commit_ref")}) rec = {"site_id": sid, "site_name": v["name"], "repo_url": repo, "checked_at": now(), "deploy_commits": {d["id"]: d.get("commit_ref") for d in deps}, "status": None} if not repo: rec["status"] = "NO_LINKED_REPOSITORY" rec["note"] = "Deployed by upload (CLI, API or drag and drop). Netlify holds no Git link for this project." write_json(os.path.join(v["dir"], "metadata", "source_recovery.json"), rec) checkpoint_update(lambda c: c["sites"].setdefault(sid, {}).update({"source_status": rec["status"]})) continue url = repo if repo.endswith(".git") else repo + ".git" sdir = os.path.join(v["dir"], "source") mirror = os.path.join(sdir, safe_name(repo.rstrip("/").split("/")[-1]) + ".git") if not os.path.isdir(mirror): os.makedirs(sdir, exist_ok=True) rc, out, err = git("clone", "--mirror", url, mirror) if rc != 0: rec["status"] = "BLOCKED_REPOSITORY_ACCESS" rec["error"] = err.strip().splitlines()[-1][:300] if err.strip() else f"git exit {rc}" rec["needed"] = ("Authorized GitHub access for this repository (for example run `gh auth login` " "or add an SSH key to GitHub), then rerun scripts/04_recover_source.py.") write_json(os.path.join(v["dir"], "metadata", "source_recovery.json"), rec) checkpoint_update(lambda c: c["sites"].setdefault(sid, {}).update({"source_status": rec["status"]})) log(f"source {v['name']}: BLOCKED ({rec['error']})") continue rc, refs, _ = git("for-each-ref", "--format=%(refname) %(objectname)", cwd=mirror) rec["refs"] = refs.strip().splitlines() rc, cnt, _ = git("rev-list", "--all", "--count", cwd=mirror) rec["commit_count_all_refs"] = int(cnt.strip() or 0) bundle = os.path.join(sdir, "repository_all_refs.bundle") if not os.path.exists(bundle): rc, out, err = git("bundle", "create", bundle, "--all", cwd=mirror) rec["bundle_error"] = err.strip()[-300:] if rc else None rc, out, err = git("bundle", "verify", bundle, cwd=mirror) rec["bundle_verified"] = rc == 0 hz = os.path.join(ROOT, "zips", f"{safe_name(v['name'])}__{sid}__REPOSITORY_HISTORY.zip") if not os.path.exists(hz): with zipfile.ZipFile(hz + ".part", "w", zipfile.ZIP_STORED, allowZip64=True) as z: z.write(bundle, "repository_all_refs.bundle") z.writestr("REFS.txt", "\n".join(rec["refs"]) + "\n") z.writestr("HOW_TO_RESTORE.txt", f"git clone repository_all_refs.bundle restored_repo\nSource: {repo}\n") os.replace(hz + ".part", hz) rec["history_zip"] = {"path": os.path.relpath(hz, ROOT), "bytes": os.path.getsize(hz), "sha256": sha256_file(hz)} rec["commits"] = {} for c in commits: info = {} rc, _, _ = git("cat-file", "-e", f"{c}^{{commit}}", cwd=mirror) if rc != 0: info["status"] = "COMMIT_NOT_IN_REPOSITORY" rec["commits"][c] = info continue rc, attrs, _ = git("show", f"{c}:.gitattributes", cwd=mirror) info["lfs_tracked_patterns"] = [l for l in attrs.splitlines() if "filter=lfs" in l] if rc == 0 else [] rc, gm, _ = git("show", f"{c}:.gitmodules", cwd=mirror) info["submodules_declared"] = gm.strip() if rc == 0 else None sz = os.path.join(ROOT, "zips", f"{safe_name(v['name'])}__{sid}__SOURCE__{c}.zip") if not os.path.exists(sz): rc, out, err = git("archive", "--format=zip", "-o", sz + ".part", c, cwd=mirror) if rc == 0: os.replace(sz + ".part", sz) else: info["archive_error"] = err.strip()[-300:] if os.path.exists(sz): info["source_zip"] = {"path": os.path.relpath(sz, ROOT), "bytes": os.path.getsize(sz), "sha256": sha256_file(sz)} info["status"] = "SOURCE_AT_DEPLOYED_COMMIT" if info["lfs_tracked_patterns"]: info["status"] += "_LFS_POINTERS_ONLY" info["lfs_note"] = "git-lfs is not installed and LFS payloads were not fetched; matching files are pointer files." if info["submodules_declared"]: info["submodule_note"] = "Submodule contents are not included; see submodules_declared." rec["commits"][c] = info if commits: ok = [c for c, i in rec["commits"].items() if i.get("status", "").startswith("SOURCE_AT_DEPLOYED_COMMIT")] rec["status"] = "SOURCE_RECOVERED" if len(ok) == len(commits) else ("SOURCE_PARTIAL" if ok else "HISTORY_ONLY") else: rec["status"] = "HISTORY_ONLY_NO_DEPLOY_COMMIT" rec["note"] = "Repository history recovered, but no deploy records a commit, so no snapshot is labeled as deployed source." write_json(os.path.join(v["dir"], "metadata", "source_recovery.json"), rec) checkpoint_update(lambda c: c["sites"].setdefault(sid, {}).update({"source_status": rec["status"]})) log(f"source {v['name']}: {rec['status']} ({rec['commit_count_all_refs']} commits, deploy commits {commits})") if __name__ == "__main__": main()