Spaces:
Running
Running
Update app/services/danger_zone_service.py
Browse files
app/services/danger_zone_service.py
CHANGED
|
@@ -2,10 +2,10 @@ import tempfile
|
|
| 2 |
import shutil
|
| 3 |
import logging
|
| 4 |
from typing import List, Dict, Any
|
| 5 |
-
from git import Repo
|
| 6 |
|
| 7 |
logger = logging.getLogger(__name__)
|
| 8 |
|
|
|
|
| 9 |
class DangerZoneService:
|
| 10 |
@staticmethod
|
| 11 |
def simulate_destructive_action(
|
|
@@ -19,21 +19,31 @@ class DangerZoneService:
|
|
| 19 |
Clones the repo into an ephemeral temp directory, performs the action,
|
| 20 |
and returns what would be lost. The temp dir is always cleaned up.
|
| 21 |
"""
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 22 |
temp_dir = tempfile.mkdtemp(prefix="gitgud_danger_")
|
| 23 |
repo_url = f"https://github.com/{owner}/{repo_name}.git"
|
| 24 |
|
| 25 |
try:
|
| 26 |
logger.info(f"Cloning {repo_url} into isolation dir for simulation...")
|
| 27 |
-
repo = Repo.clone_from(repo_url, temp_dir, depth=100)
|
| 28 |
|
| 29 |
-
# Record reachable commits before the action
|
| 30 |
before_shas = {c.hexsha for c in repo.iter_commits()}
|
| 31 |
|
| 32 |
-
# Apply the simulated operation
|
| 33 |
if action in ("hard_reset", "force_push"):
|
| 34 |
repo.git.reset("--hard", target_ref)
|
| 35 |
else:
|
| 36 |
-
# Future-proof: you can add rebase/squash later
|
| 37 |
repo.git.reset("--hard", target_ref)
|
| 38 |
|
| 39 |
after_shas = {c.hexsha for c in repo.iter_commits()}
|
|
@@ -52,10 +62,12 @@ class DangerZoneService:
|
|
| 52 |
except Exception:
|
| 53 |
continue
|
| 54 |
|
| 55 |
-
# Optional: filter only commits that the client currently has locally
|
| 56 |
if current_local_shas:
|
| 57 |
local_set = set(current_local_shas)
|
| 58 |
-
would_lose = [
|
|
|
|
|
|
|
|
|
|
| 59 |
|
| 60 |
summary = (
|
| 61 |
f"This {action.replace('_', ' ')} will eliminate "
|
|
@@ -64,7 +76,7 @@ class DangerZoneService:
|
|
| 64 |
|
| 65 |
return {
|
| 66 |
"wouldLose": would_lose,
|
| 67 |
-
"wouldRewrite": [],
|
| 68 |
"conflicts": [],
|
| 69 |
"summary": summary,
|
| 70 |
}
|
|
@@ -79,7 +91,6 @@ class DangerZoneService:
|
|
| 79 |
"error": str(e),
|
| 80 |
}
|
| 81 |
finally:
|
| 82 |
-
# Always clean the isolation directory
|
| 83 |
try:
|
| 84 |
shutil.rmtree(temp_dir, ignore_errors=True)
|
| 85 |
except Exception:
|
|
|
|
| 2 |
import shutil
|
| 3 |
import logging
|
| 4 |
from typing import List, Dict, Any
|
|
|
|
| 5 |
|
| 6 |
logger = logging.getLogger(__name__)
|
| 7 |
|
| 8 |
+
|
| 9 |
class DangerZoneService:
|
| 10 |
@staticmethod
|
| 11 |
def simulate_destructive_action(
|
|
|
|
| 19 |
Clones the repo into an ephemeral temp directory, performs the action,
|
| 20 |
and returns what would be lost. The temp dir is always cleaned up.
|
| 21 |
"""
|
| 22 |
+
# Lazy import so the whole app can still start even if git binary is missing
|
| 23 |
+
try:
|
| 24 |
+
from git import Repo
|
| 25 |
+
except ImportError as e:
|
| 26 |
+
logger.error(f"GitPython / git binary not available: {e}")
|
| 27 |
+
return {
|
| 28 |
+
"wouldLose": [],
|
| 29 |
+
"wouldRewrite": [],
|
| 30 |
+
"conflicts": [],
|
| 31 |
+
"summary": "Simulation unavailable: git binary is missing on the server.",
|
| 32 |
+
"error": "GIT_BINARY_MISSING"
|
| 33 |
+
}
|
| 34 |
+
|
| 35 |
temp_dir = tempfile.mkdtemp(prefix="gitgud_danger_")
|
| 36 |
repo_url = f"https://github.com/{owner}/{repo_name}.git"
|
| 37 |
|
| 38 |
try:
|
| 39 |
logger.info(f"Cloning {repo_url} into isolation dir for simulation...")
|
| 40 |
+
repo = Repo.clone_from(repo_url, temp_dir, depth=100)
|
| 41 |
|
|
|
|
| 42 |
before_shas = {c.hexsha for c in repo.iter_commits()}
|
| 43 |
|
|
|
|
| 44 |
if action in ("hard_reset", "force_push"):
|
| 45 |
repo.git.reset("--hard", target_ref)
|
| 46 |
else:
|
|
|
|
| 47 |
repo.git.reset("--hard", target_ref)
|
| 48 |
|
| 49 |
after_shas = {c.hexsha for c in repo.iter_commits()}
|
|
|
|
| 62 |
except Exception:
|
| 63 |
continue
|
| 64 |
|
|
|
|
| 65 |
if current_local_shas:
|
| 66 |
local_set = set(current_local_shas)
|
| 67 |
+
would_lose = [
|
| 68 |
+
c for c in would_lose
|
| 69 |
+
if any(c["id"] in s or s.startswith(c["id"]) for s in local_set)
|
| 70 |
+
]
|
| 71 |
|
| 72 |
summary = (
|
| 73 |
f"This {action.replace('_', ' ')} will eliminate "
|
|
|
|
| 76 |
|
| 77 |
return {
|
| 78 |
"wouldLose": would_lose,
|
| 79 |
+
"wouldRewrite": [],
|
| 80 |
"conflicts": [],
|
| 81 |
"summary": summary,
|
| 82 |
}
|
|
|
|
| 91 |
"error": str(e),
|
| 92 |
}
|
| 93 |
finally:
|
|
|
|
| 94 |
try:
|
| 95 |
shutil.rmtree(temp_dir, ignore_errors=True)
|
| 96 |
except Exception:
|