Spaces:
Sleeping
Sleeping
| import tempfile | |
| import shutil | |
| import logging | |
| from typing import List, Dict, Any | |
| logger = logging.getLogger(__name__) | |
| class DangerZoneService: | |
| def simulate_destructive_action( | |
| owner: str, | |
| repo_name: str, | |
| action: str, | |
| target_ref: str, | |
| current_local_shas: List[str] = None, | |
| ) -> Dict[str, Any]: | |
| """ | |
| Clones the repo into an ephemeral temp directory, performs the action, | |
| and returns what would be lost. The temp dir is always cleaned up. | |
| """ | |
| # Lazy import so the whole app can still start even if git binary is missing | |
| try: | |
| from git import Repo | |
| except ImportError as e: | |
| logger.error(f"GitPython / git binary not available: {e}") | |
| return { | |
| "wouldLose": [], | |
| "wouldRewrite": [], | |
| "conflicts": [], | |
| "summary": "Simulation unavailable: git binary is missing on the server.", | |
| "error": "GIT_BINARY_MISSING" | |
| } | |
| temp_dir = tempfile.mkdtemp(prefix="gitgud_danger_") | |
| repo_url = f"https://github.com/{owner}/{repo_name}.git" | |
| try: | |
| logger.info(f"Cloning {repo_url} into isolation dir for simulation...") | |
| repo = Repo.clone_from(repo_url, temp_dir, depth=100) | |
| before_shas = {c.hexsha for c in repo.iter_commits()} | |
| if action in ("hard_reset", "force_push"): | |
| repo.git.reset("--hard", target_ref) | |
| else: | |
| repo.git.reset("--hard", target_ref) | |
| after_shas = {c.hexsha for c in repo.iter_commits()} | |
| lost_shas = before_shas - after_shas | |
| would_lose = [] | |
| for sha in lost_shas: | |
| try: | |
| commit = repo.commit(sha) | |
| would_lose.append({ | |
| "id": sha[:7], | |
| "author": commit.author.name if commit.author else "unknown", | |
| "message": commit.summary, | |
| "timestamp": commit.authored_date, | |
| }) | |
| except Exception: | |
| continue | |
| if current_local_shas: | |
| local_set = set(current_local_shas) | |
| would_lose = [ | |
| c for c in would_lose | |
| if any(c["id"] in s or s.startswith(c["id"]) for s in local_set) | |
| ] | |
| summary = ( | |
| f"This {action.replace('_', ' ')} will eliminate " | |
| f"{len(would_lose)} commits from the remote history." | |
| ) | |
| return { | |
| "wouldLose": would_lose, | |
| "wouldRewrite": [], | |
| "conflicts": [], | |
| "summary": summary, | |
| } | |
| except Exception as e: | |
| logger.error(f"Danger zone simulation failed: {e}", exc_info=True) | |
| return { | |
| "wouldLose": [], | |
| "wouldRewrite": [], | |
| "conflicts": [], | |
| "summary": f"Simulation failed: {str(e)}", | |
| "error": str(e), | |
| } | |
| finally: | |
| try: | |
| shutil.rmtree(temp_dir, ignore_errors=True) | |
| except Exception: | |
| pass |