AI Update: فاز ۱ و ۲ نقشه راه را روی ربات تلگرام پیاده‌سازی کن

2026-08-30 12:45:15 +03:30
parent f477963a56
commit 03d3e92760
4 changed files with 275 additions and 10 deletions
@@ -486,6 +486,112 @@ class GitManager:
            logger.error(f"Error resetting to {commit_hash} in {ws}: {e}")
            return False, f"خطا در بازگردانی به کامیت {commit_hash}: {e}"

+    async def git_revert(
+        self,
+        workspace_path: str,
+        commit_target: str = "HEAD",
+        strategy: Optional[str] = None,
+    ) -> Tuple[bool, str, Dict[str, Any]]:
+        """
+        Reverts the specified commit (or HEAD) cleanly.
+        Attempts deterministic Git revert first without using AI tokens.
+        If conflicts occur, detects them and safely aborts to keep working tree clean.
+        """
+        ws = Path(workspace_path).expanduser().resolve()
+        if not (ws / ".git").exists():
+            return False, "مسیر مشخص‌شده یک مخزن گیت نیست.", {}
+
+        try:
+            # 1. Check dirty files, stash if necessary
+            porcelain = await self._run_cmd(["git", "status", "--porcelain"], cwd=str(ws))
+            has_stash = False
+            if porcelain.strip():
+                try:
+                    await self._run_cmd(["git", "stash", "save", "Auto-stash before revert"], cwd=str(ws))
+                    has_stash = True
+                except Exception:
+                    pass
+
+            # 2. Get target commit info before revert
+            target_info = (await self._run_cmd(["git", "log", "-1", "--format=%h - %s", commit_target], cwd=str(ws))).strip()
+
+            # 3. Attempt revert
+            cmd = ["git", "revert", "--no-edit", commit_target]
+            if strategy:
+                cmd.extend(["--strategy", strategy])
+
+            try:
+                out = await self._run_cmd(cmd, cwd=str(ws))
+            except Exception as e:
+                # Check for conflicts
+                status_out = await self._run_cmd(["git", "status", "--porcelain"], cwd=str(ws))
+                conflicted_files = []
+                for line in status_out.splitlines():
+                    line_s = line.strip()
+                    if line_s.startswith(("UU ", "AA ", "UD ", "DU ", "U ", "DD ")):
+                        parts = line_s.split(maxsplit=1)
+                        if len(parts) > 1:
+                            conflicted_files.append(parts[1])
+
+                if conflicted_files:
+                    # Abort revert to leave working directory clean
+                    try:
+                        await self._run_cmd(["git", "revert", "--abort"], cwd=str(ws))
+                    except Exception:
+                        pass
+                    if has_stash:
+                        try:
+                            await self._run_cmd(["git", "stash", "pop"], cwd=str(ws))
+                        except Exception:
+                            pass
+                    file_list_str = "\n".join(f"• <code>{f}</code>" for f in conflicted_files)
+                    return False, f"⚠️ تداخل (Conflict) در لغو کامیت رخ داد:\n{file_list_str}\n\n<i>فرآیند لغو متوقف شد تا کدها بدون دستکاری باقی بمانند.</i>", {
+                        "conflict": True,
+                        "conflicted_files": conflicted_files,
+                        "target": commit_target,
+                    }
+
+                # If other revert error, abort
+                try:
+                    await self._run_cmd(["git", "revert", "--abort"], cwd=str(ws))
+                except Exception:
+                    pass
+                if has_stash:
+                    try:
+                        await self._run_cmd(["git", "stash", "pop"], cwd=str(ws))
+                    except Exception:
+                        pass
+                return False, f"خطا در اجرای revert: {e}", {}
+
+            # 4. Push to remote Gitea
+            try:
+                await self._run_cmd(["git", "push", "origin", "main"], cwd=str(ws))
+            except Exception as pe:
+                logger.warning(f"Failed to push after revert in {ws}: {pe}")
+
+            # 5. Pop stash if we stashed earlier
+            if has_stash:
+                try:
+                    await self._run_cmd(["git", "stash", "pop"], cwd=str(ws))
+                except Exception:
+                    pass
+
+            # 6. New commit log
+            new_log = (await self._run_cmd(["git", "log", "-1", "--format=%h - %s"], cwd=str(ws))).strip()
+            return True, f"✅ تغییرات کامیت <code>{target_info}</code> با موفقیت لغو شد (Reverted).\nکامیت جدید ثبت و پوش شد:\n<code>{new_log}</code>", {
+                "success": True,
+                "revert_log": new_log,
+                "target_info": target_info,
+            }
+
+        except Exception as e:
+            logger.error(f"Error reverting {commit_target} in {ws}: {e}")
+            return False, f"❌ خطا در لغو کامیت: {e}", {}
+
+    async def git_undo_last_action(self, workspace_path: str) -> Tuple[bool, str, Dict[str, Any]]:
+        """Undoes the last commit by reverting HEAD."""
+        return await self.git_revert(workspace_path, commit_target="HEAD")
+
    async def _run_cmd(self, cmd: List[str], cwd: str) -> str:
        """Helper to run a shell command asynchronously."""
        proc = await asyncio.create_subprocess_exec(