root/default
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

AI Update: تسک ها را آماده کن این ساختار عالیه

Browse Source
This commit is contained in:
Antigravity Bot
2026-08-30 18:36:58 +03:30
parent d5c225ebf6
commit bf4df87e86
7 changed files with 930 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
telegram-agy-bot/git_manager.py
+140 -19
View File
@@ -159,8 +159,11 @@ class GitManager:
async def init_project_repo(self, workspace_path: str, repo_name: str, owner: str = "root") -> Dict[str, Any]:
"""
Initializes git in workspace_path, links with Gitea remote, creates .gitignore,
and pushes the initial commit.
Initializes git in workspace_path with dual branches:
- `production`: base/production branch
- `dev`: active development branch
Links with Gitea remote, creates .gitignore, pushes initial commits on both branches,
and leaves `dev` as the active branch.
"""
ws = Path(workspace_path).expanduser().resolve()
os.makedirs(ws, exist_ok=True)
@@ -186,7 +189,7 @@ class GitManager:
try:
if is_new:
await self._run_cmd(["git", "init", "-b", "main"], cwd=str(ws))
await self._run_cmd(["git", "init", "-b", "production"], cwd=str(ws))
# Configure author
await self._run_cmd(["git", "config", "user.name", "Antigravity Bot"], cwd=str(ws))
@@ -200,19 +203,36 @@ class GitManager:
else:
await self._run_cmd(["git", "remote", "add", "origin", remote_url], cwd=str(ws))
# Initial add, commit and push
# Check if production branch exists, if on main rename to production
curr_branch = (await self._run_cmd(["git", "branch", "--show-current"], cwd=str(ws))).strip()
if curr_branch in ("main", "master"):
try:
await self._run_cmd(["git", "branch", "-M", "production"], cwd=str(ws))
except Exception:
pass
# Initial add, commit and push to production
status_out = await self._run_cmd(["git", "status", "--porcelain"], cwd=str(ws))
if status_out.strip() or is_new:
await self._run_cmd(["git", "add", "-A"], cwd=str(ws))
await self._run_cmd(["git", "commit", "-m", f"Initial commit for {clean_name}"], cwd=str(ws))
await self._run_cmd(["git", "push", "-u", "origin", "main", "--force"], cwd=str(ws))
await self._run_cmd(["git", "push", "-u", "origin", "production", "--force"], cwd=str(ws))
logger.info(f"Initialized Git repo for {clean_name} at {ws}")
# Ensure dev branch exists and is pushed
branches_out = await self._run_cmd(["git", "branch"], cwd=str(ws))
if "dev" not in branches_out:
await self._run_cmd(["git", "checkout", "-b", "dev"], cwd=str(ws))
await self._run_cmd(["git", "push", "-u", "origin", "dev", "--force"], cwd=str(ws))
else:
await self._run_cmd(["git", "checkout", "dev"], cwd=str(ws))
logger.info(f"Initialized Git repo for {clean_name} at {ws} with dev & production branches.")
return {
"success": True,
"name": clean_name,
"web_url": urls["web_url"],
"clone_url": urls["clone_url"],
"active_branch": "dev",
}
except Exception as e:
logger.error(f"Error initializing git repo for {clean_name} in {ws}: {e}")
@@ -233,7 +253,7 @@ class GitManager:
porcelain = await self._run_cmd(["git", "status", "--porcelain"], cwd=str(ws))
dirty_files = [line.strip() for line in porcelain.splitlines() if line.strip()]
branch = (await self._run_cmd(["git", "branch", "--show-current"], cwd=str(ws))).strip() or "main"
branch = (await self._run_cmd(["git", "branch", "--show-current"], cwd=str(ws))).strip() or "dev"
# Last commit
log_out = await self._run_cmd(["git", "log", "-1", "--format=%h|%an|%ar|%s"], cwd=str(ws))
@@ -258,7 +278,7 @@ class GitManager:
}
except Exception as e:
logger.debug(f"Git status error in {ws}: {e}")
return {"is_git": True, "dirty": False, "files": [], "error": str(e), "branch": "main"}
return {"is_git": True, "dirty": False, "files": [], "error": str(e), "branch": "dev"}
async def git_commit_and_push(
self,
@@ -266,8 +286,9 @@ class GitManager:
message: str = "Auto-commit from Antigravity",
repo_name: Optional[str] = None,
owner: str = "root",
branch: Optional[str] = None,
) -> Tuple[bool, str]:
"""Stages all changes, commits with message, and pushes to Gitea."""
"""Stages all changes, commits with message, and pushes to Gitea on the active branch (default dev)."""
ws = Path(workspace_path).expanduser().resolve()
if not (ws / ".git").exists():
name = repo_name or ws.name
@@ -276,42 +297,142 @@ class GitManager:
return False, f"Failed to initialize repo: {init_res.get('error')}"
try:
curr_branch = branch or (await self._run_cmd(["git", "branch", "--show-current"], cwd=str(ws))).strip() or "dev"
# Check for uncommitted changes
porcelain = await self._run_cmd(["git", "status", "--porcelain"], cwd=str(ws))
if not porcelain.strip():
return True, "درخت کاری تمیز است (تغییری برای کامیت وجود نداشت)."
return True, f"درخت کاری شاخه <code>{curr_branch}</code> تمیز است (تغییری برای کامیت وجود نداشت)."
await self._run_cmd(["git", "add", "-A"], cwd=str(ws))
clean_msg = message.replace('"', '\\"').replace("\n", " ")
if not clean_msg.strip():
clean_msg = "Updates from Antigravity Bot"
clean_msg = f"Updates on {curr_branch} from Antigravity Bot"
commit_out = await self._run_cmd(["git", "commit", "-m", clean_msg], cwd=str(ws))
push_out = await self._run_cmd(["git", "push", "origin", "main"], cwd=str(ws))
push_out = await self._run_cmd(["git", "push", "origin", curr_branch], cwd=str(ws))
# Extract hash
log_out = await self._run_cmd(["git", "log", "-1", "--format=%h - %s"], cwd=str(ws))
return True, f"✅ تغییرات با موفقیت کامیت و پوش شد:\n<code>{log_out.strip()}</code>"
return True, f"✅ تغییرات با موفقیت در شاخه <code>{curr_branch}</code> کامیت و پوش شد:\n<code>{log_out.strip()}</code>"
except Exception as e:
logger.error(f"Git commit/push error in {ws}: {e}")
return False, f"❌ خطا در کامیت یا پوش: {e}"
async def git_pull(self, workspace_path: str) -> Tuple[bool, str]:
"""Pulls latest changes from remote Gitea repository."""
async def git_pull(self, workspace_path: str, branch: Optional[str] = None) -> Tuple[bool, str]:
"""Pulls latest changes from remote Gitea repository on current or specified branch."""
ws = Path(workspace_path).expanduser().resolve()
if not (ws / ".git").exists():
return False, "مخزن گیت برای این مسیر مقداردهی نشده است."
try:
# Fetch and check
out = await self._run_cmd(["git", "pull", "origin", "main"], cwd=str(ws))
return True, f"✅ وضعیت دریافت تغییرات:\n<code>{out.strip()}</code>"
curr_branch = branch or (await self._run_cmd(["git", "branch", "--show-current"], cwd=str(ws))).strip() or "dev"
out = await self._run_cmd(["git", "pull", "origin", curr_branch], cwd=str(ws))
return True, f"✅ وضعیت دریافت تغییرات شاخه <code>{curr_branch}</code>:\n<code>{out.strip()}</code>"
except Exception as e:
logger.warning(f"Git pull warning/error in {ws}: {e}")
return False, f"⚠️ خطا در دریافت تغییرات (Pull): {e}"
async def git_publish(
self,
workspace_path: str,
message: str = "Publish dev to production",
repo_name: Optional[str] = None,
owner: str = "root",
) -> Tuple[bool, str, Dict[str, Any]]:
"""
Publishes latest changes from `dev` to `production`:
1. Ensures changes on `dev` are committed and pushed.
2. Checks out `production` branch.
3. Merges `dev` into `production`.
4. Pushes `production` to Gitea origin.
5. Switches back to `dev`.
"""
ws = Path(workspace_path).expanduser().resolve()
if not (ws / ".git").exists():
name = repo_name or ws.name
init_res = await self.init_project_repo(str(ws), name, owner=owner)
if not init_res.get("success"):
return False, f"خطا در مقداردهی اولیه مخزن: {init_res.get('error')}", {}
try:
# 1. Commit any pending changes on dev
status = await self.git_status(str(ws))
if status.get("dirty"):
await self.git_commit_and_push(str(ws), message=f"Pre-publish commit: {message}", repo_name=repo_name, owner=owner)
# Ensure we are on dev branch
curr_branch = (await self._run_cmd(["git", "branch", "--show-current"], cwd=str(ws))).strip()
if curr_branch != "dev":
# Ensure dev exists
branches = await self._run_cmd(["git", "branch"], cwd=str(ws))
if "dev" in branches:
await self._run_cmd(["git", "checkout", "dev"], cwd=str(ws))
else:
await self._run_cmd(["git", "checkout", "-b", "dev"], cwd=str(ws))
# Push dev
try:
await self._run_cmd(["git", "push", "origin", "dev"], cwd=str(ws))
except Exception as pe:
logger.warning(f"Push dev warning before publish in {ws}: {pe}")
# Ensure production branch exists
branches = await self._run_cmd(["git", "branch"], cwd=str(ws))
if "production" not in branches:
await self._run_cmd(["git", "branch", "production"], cwd=str(ws))
# 2. Checkout production
await self._run_cmd(["git", "checkout", "production"], cwd=str(ws))
# Pull production if remote exists
try:
await self._run_cmd(["git", "pull", "origin", "production"], cwd=str(ws))
except Exception:
pass
# 3. Merge dev into production
merge_msg = f"Merge branch 'dev' into production: {message}".replace('"', '\\"')
merge_out = await self._run_cmd(["git", "merge", "dev", "--no-edit", "-m", merge_msg], cwd=str(ws))
# 4. Push production to Gitea
push_out = await self._run_cmd(["git", "push", "origin", "production"], cwd=str(ws))
# Get production latest commit
prod_commit = (await self._run_cmd(["git", "log", "-1", "--format=%h|%an|%ar|%s"], cwd=str(ws))).strip()
# 5. Switch back to dev branch
await self._run_cmd(["git", "checkout", "dev"], cwd=str(ws))
commit_parts = prod_commit.split("|", 3) if "|" in prod_commit else [prod_commit, "", "", ""]
commit_hash = commit_parts[0]
commit_text = commit_parts[3] if len(commit_parts) > 3 else prod_commit
res_msg = (
f"🚀 <b>پروژه با موفقیت روی شاخه <code>production</code> پابلیش شد!</b>\n\n"
f"• 🌿 <b>شاخه مبدأ:</b> <code>dev</code>\n"
f"• 🚀 <b>شاخه مقصد:</b> <code>production</code>\n"
f"• 🔖 <b>آخرین کامیت:</b> <code>{commit_hash}</code> - {escape_html(commit_text)}\n"
f"• 🔄 <b>وضعیت کاری:</b> مجدداً به شاخه <code>dev</code> بازگشت داده شد تا توسعه ادامه یابد."
)
return True, res_msg, {
"commit_hash": commit_hash,
"commit_text": commit_text,
"source_branch": "dev",
"target_branch": "production",
}
except Exception as e:
logger.error(f"Error publishing dev to production in {ws}: {e}")
# Ensure we return to dev
try:
await self._run_cmd(["git", "checkout", "dev"], cwd=str(ws))
except Exception:
pass
return False, f"❌ خطا در عملیات پابلیش به شاخه پروداکشن: {e}", {}
async def git_sync(self, workspace_path: str, message: str = "Sync with Gitea", repo_name: Optional[str] = None) -> Tuple[bool, str]:
"""Performs full sync: Pulls remote changes, then commits and pushes any local changes."""
"""Performs full sync: Pulls remote changes, then commits and pushes any local changes on active branch."""
ws = Path(workspace_path).expanduser().resolve()
if not (ws / ".git").exists():
name = repo_name or ws.name