506 lines
20 KiB
Python
506 lines
20 KiB
Python
import os
|
|
import sys
|
|
import json
|
|
import logging
|
|
import asyncio
|
|
import urllib.request
|
|
import urllib.error
|
|
from pathlib import Path
|
|
from typing import Optional, Dict, Any, List, Tuple
|
|
|
|
from config import settings
|
|
|
|
logger = logging.getLogger("AGYGitManager")
|
|
|
|
DEFAULT_GITIGNORE = """# Byte-compiled / optimized / DLL files
|
|
__pycache__/
|
|
*.py[cod]
|
|
*$py.class
|
|
|
|
# C extensions
|
|
*.so
|
|
|
|
# Distribution / packaging
|
|
.Python
|
|
build/
|
|
develop-eggs/
|
|
dist/
|
|
downloads/
|
|
eggs/
|
|
.eggs/
|
|
lib/
|
|
lib64/
|
|
parts/
|
|
sdist/
|
|
var/
|
|
wheels/
|
|
share/python-wheels/
|
|
*.egg-info/
|
|
.installed.cfg
|
|
*.egg
|
|
MANIFEST
|
|
|
|
# Virtual environments
|
|
.env
|
|
.venv
|
|
env/
|
|
venv/
|
|
ENV/
|
|
env.bak/
|
|
venv.bak/
|
|
|
|
# Node.js
|
|
node_modules/
|
|
npm-debug.log*
|
|
yarn-debug.log*
|
|
yarn-error.log*
|
|
.pnpm-debug.log*
|
|
|
|
# Bot internal & session data
|
|
sessions_data/
|
|
*.db-shm
|
|
*.db-wal
|
|
.agents/
|
|
.gemini/
|
|
.tmp/
|
|
tmp/
|
|
*.log
|
|
|
|
# IDE & OS
|
|
.DS_Store
|
|
.idea/
|
|
.vscode/
|
|
*.swp
|
|
*.swo
|
|
"""
|
|
|
|
class GitManager:
|
|
def __init__(self):
|
|
self.gitea_url = settings.gitea_url
|
|
self.internal_url = settings.gitea_internal_url
|
|
self.admin_user = settings.gitea_admin_user
|
|
self.admin_token = settings.gitea_admin_token
|
|
|
|
def _api_request(self, endpoint: str, method: str = "GET", data: Optional[Dict[str, Any]] = None) -> Tuple[int, Optional[Dict[str, Any]]]:
|
|
"""Makes a synchronous HTTP request to Gitea REST API."""
|
|
url = f"{self.internal_url}/api/v1{endpoint}"
|
|
headers = {
|
|
"Authorization": f"token {self.admin_token}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
}
|
|
body = json.dumps(data).encode("utf-8") if data is not None else None
|
|
req = urllib.request.Request(url, data=body, headers=headers, method=method)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
status = resp.status
|
|
resp_text = resp.read().decode("utf-8", errors="replace")
|
|
res_json = json.loads(resp_text) if resp_text else {}
|
|
return status, res_json
|
|
except urllib.error.HTTPError as e:
|
|
resp_text = e.read().decode("utf-8", errors="replace")
|
|
try:
|
|
res_json = json.loads(resp_text)
|
|
except Exception:
|
|
res_json = {"error": resp_text}
|
|
return e.code, res_json
|
|
except Exception as e:
|
|
logger.error(f"Gitea API connection error ({method} {url}): {e}")
|
|
return 500, {"error": str(e)}
|
|
|
|
async def ensure_gitea_repo(
|
|
self,
|
|
repo_name: str,
|
|
owner: str = "root",
|
|
description: str = "",
|
|
private: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Ensures a repository exists on Gitea. Creates it if not present."""
|
|
clean_name = repo_name.strip().lower()
|
|
clean_name = clean_name.replace(" ", "-")
|
|
|
|
# 1. Check if repo exists
|
|
status, res = await asyncio.to_thread(self._api_request, f"/repos/{owner}/{clean_name}")
|
|
if status == 200 and isinstance(res, dict) and "name" in res:
|
|
return res
|
|
|
|
# 2. Create repo under owner
|
|
create_payload = {
|
|
"name": clean_name,
|
|
"description": description or f"Project {clean_name} managed by Antigravity",
|
|
"private": private,
|
|
"auto_init": False,
|
|
"default_branch": "main",
|
|
}
|
|
create_status, create_res = await asyncio.to_thread(
|
|
self._api_request,
|
|
"/user/repos",
|
|
method="POST",
|
|
data=create_payload,
|
|
)
|
|
|
|
if create_status in (200, 201) and isinstance(create_res, dict):
|
|
logger.info(f"Created new Gitea repository: {owner}/{clean_name}")
|
|
return create_res
|
|
|
|
logger.warning(f"Could not create Gitea repo {owner}/{clean_name}: {create_res}")
|
|
return create_res or {}
|
|
|
|
def get_repo_urls(self, repo_name: str, owner: str = "root") -> Dict[str, str]:
|
|
"""Returns standard URLs for a Gitea repository."""
|
|
clean_name = repo_name.strip().lower().replace(" ", "-")
|
|
return {
|
|
"web_url": f"{self.gitea_url}/{owner}/{clean_name}",
|
|
"clone_url": f"{self.gitea_url}/{owner}/{clean_name}.git",
|
|
"ssh_url": f"git@git.msa.artacloud.ir:{owner}/{clean_name}.git",
|
|
"internal_auth_url": f"http://{self.admin_user}:{self.admin_token}@127.0.0.1:3000/{owner}/{clean_name}.git",
|
|
}
|
|
|
|
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.
|
|
"""
|
|
ws = Path(workspace_path).expanduser().resolve()
|
|
os.makedirs(ws, exist_ok=True)
|
|
clean_name = repo_name.strip().lower().replace(" ", "-")
|
|
|
|
# 1. Ensure repo in Gitea
|
|
repo_data = await self.ensure_gitea_repo(clean_name, owner=owner)
|
|
urls = self.get_repo_urls(clean_name, owner=owner)
|
|
remote_url = urls["internal_auth_url"]
|
|
|
|
# 2. Ensure .gitignore
|
|
gitignore_path = ws / ".gitignore"
|
|
if not gitignore_path.exists():
|
|
try:
|
|
with open(gitignore_path, "w", encoding="utf-8") as f:
|
|
f.write(DEFAULT_GITIGNORE)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to write .gitignore in {ws}: {e}")
|
|
|
|
# 3. Setup git repo
|
|
git_dir = ws / ".git"
|
|
is_new = not git_dir.exists()
|
|
|
|
try:
|
|
if is_new:
|
|
await self._run_cmd(["git", "init", "-b", "main"], cwd=str(ws))
|
|
|
|
# Configure author
|
|
await self._run_cmd(["git", "config", "user.name", "Antigravity Bot"], cwd=str(ws))
|
|
await self._run_cmd(["git", "config", "user.email", "bot@msa.artacloud.ir"], cwd=str(ws))
|
|
await self._run_cmd(["git", "config", "pull.rebase", "false"], cwd=str(ws))
|
|
|
|
# Remote origin
|
|
remotes_out = await self._run_cmd(["git", "remote"], cwd=str(ws))
|
|
if "origin" in remotes_out:
|
|
await self._run_cmd(["git", "remote", "set-url", "origin", remote_url], cwd=str(ws))
|
|
else:
|
|
await self._run_cmd(["git", "remote", "add", "origin", remote_url], cwd=str(ws))
|
|
|
|
# Initial add, commit and push
|
|
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))
|
|
|
|
logger.info(f"Initialized Git repo for {clean_name} at {ws}")
|
|
return {
|
|
"success": True,
|
|
"name": clean_name,
|
|
"web_url": urls["web_url"],
|
|
"clone_url": urls["clone_url"],
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Error initializing git repo for {clean_name} in {ws}: {e}")
|
|
return {
|
|
"success": False,
|
|
"error": str(e),
|
|
"web_url": urls["web_url"],
|
|
"clone_url": urls["clone_url"],
|
|
}
|
|
|
|
async def git_status(self, workspace_path: str) -> Dict[str, Any]:
|
|
"""Gets the git status, active branch, and last commit info."""
|
|
ws = Path(workspace_path).expanduser().resolve()
|
|
if not (ws / ".git").exists():
|
|
return {"is_git": False, "dirty": False, "files": [], "last_commit": None, "branch": None}
|
|
|
|
try:
|
|
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"
|
|
|
|
# Last commit
|
|
log_out = await self._run_cmd(["git", "log", "-1", "--format=%h|%an|%ar|%s"], cwd=str(ws))
|
|
last_commit = None
|
|
if log_out and "|" in log_out:
|
|
parts = log_out.strip().split("|", 3)
|
|
if len(parts) == 4:
|
|
last_commit = {
|
|
"hash": parts[0],
|
|
"author": parts[1],
|
|
"time": parts[2],
|
|
"message": parts[3],
|
|
}
|
|
|
|
return {
|
|
"is_git": True,
|
|
"dirty": len(dirty_files) > 0,
|
|
"files_count": len(dirty_files),
|
|
"files": dirty_files[:10],
|
|
"branch": branch,
|
|
"last_commit": last_commit,
|
|
}
|
|
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"}
|
|
|
|
async def git_commit_and_push(
|
|
self,
|
|
workspace_path: str,
|
|
message: str = "Auto-commit from Antigravity",
|
|
repo_name: Optional[str] = None,
|
|
owner: str = "root",
|
|
) -> Tuple[bool, str]:
|
|
"""Stages all changes, commits with message, and pushes to Gitea."""
|
|
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"Failed to initialize repo: {init_res.get('error')}"
|
|
|
|
try:
|
|
# Check for uncommitted changes
|
|
porcelain = await self._run_cmd(["git", "status", "--porcelain"], cwd=str(ws))
|
|
if not porcelain.strip():
|
|
return True, "درخت کاری تمیز است (تغییری برای کامیت وجود نداشت)."
|
|
|
|
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"
|
|
|
|
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))
|
|
|
|
# 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>"
|
|
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."""
|
|
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>"
|
|
except Exception as e:
|
|
logger.warning(f"Git pull warning/error in {ws}: {e}")
|
|
return False, f"⚠️ خطا در دریافت تغییرات (Pull): {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."""
|
|
ws = Path(workspace_path).expanduser().resolve()
|
|
if not (ws / ".git").exists():
|
|
name = repo_name or ws.name
|
|
await self.init_project_repo(str(ws), name)
|
|
|
|
results = []
|
|
# 1. Pull
|
|
pull_ok, pull_msg = await self.git_pull(str(ws))
|
|
results.append(pull_msg)
|
|
|
|
# 2. Commit & Push if dirty
|
|
status = await self.git_status(str(ws))
|
|
if status.get("dirty"):
|
|
push_ok, push_msg = await self.git_commit_and_push(str(ws), message=message, repo_name=repo_name)
|
|
results.append(push_msg)
|
|
else:
|
|
results.append("ℹ️ تغییرات محلی جدیدی برای کامیت وجود ندارد.")
|
|
|
|
return True, "\n".join(results)
|
|
|
|
async def delete_gitea_repo(self, repo_name: str, owner: str = "root") -> Tuple[bool, str]:
|
|
"""Deletes a repository from Gitea."""
|
|
clean_name = repo_name.strip().lower().replace(" ", "-")
|
|
status, res = await asyncio.to_thread(
|
|
self._api_request,
|
|
f"/repos/{owner}/{clean_name}",
|
|
method="DELETE",
|
|
)
|
|
if status in (200, 204):
|
|
logger.info(f"Deleted Gitea repo: {owner}/{clean_name}")
|
|
return True, f"مخزن {clean_name} با موفقیت از Gitea حذف شد."
|
|
elif status == 404:
|
|
return True, f"مخزن {clean_name} در Gitea وجود نداشت."
|
|
else:
|
|
err = (res.get("error") or res.get("message") or f"HTTP {status}") if isinstance(res, dict) else str(res)
|
|
logger.warning(f"Failed to delete Gitea repo {owner}/{clean_name}: {err}")
|
|
return False, f"خطا در حذف مخزن Gitea: {err}"
|
|
|
|
async def sync_all_projects(self) -> List[Dict[str, Any]]:
|
|
"""Scans ONLY active registered projects in sessions.json, ensuring Gitea repos exist and are synced."""
|
|
from agy_engine import session_manager
|
|
synced = []
|
|
|
|
# Scan active registered projects only
|
|
for uid, sess in session_manager.sessions.items():
|
|
for p_name, proj in sess.projects.items():
|
|
if p_name == "default" and proj.workspace == "/root":
|
|
continue
|
|
ws = proj.workspace
|
|
if os.path.isdir(ws):
|
|
try:
|
|
res = await self.init_project_repo(ws, p_name)
|
|
synced.append({"project": p_name, "workspace": ws, "result": res})
|
|
except Exception as e:
|
|
logger.error(f"Failed to sync project {p_name}: {e}")
|
|
|
|
logger.info(f"Synced {len(synced)} active projects with Gitea.")
|
|
return synced
|
|
|
|
async def get_commit_history(self, workspace_path: str, limit: int = 20) -> List[Dict[str, Any]]:
|
|
"""Returns the list of the last `limit` commits with details and active HEAD indicator."""
|
|
ws = Path(workspace_path).expanduser().resolve()
|
|
if not (ws / ".git").exists():
|
|
return []
|
|
|
|
try:
|
|
# Check current HEAD commit hash
|
|
head_hash = ""
|
|
try:
|
|
head_hash = (await self._run_cmd(["git", "rev-parse", "HEAD"], cwd=str(ws))).strip()
|
|
except Exception:
|
|
pass
|
|
|
|
cmd = ["git", "log", f"-n", str(limit), "--format=%H|%h|%an|%ar|%cd|%s", "--date=short"]
|
|
out = await self._run_cmd(cmd, cwd=str(ws))
|
|
|
|
commits = []
|
|
for line in out.splitlines():
|
|
line = line.strip()
|
|
if not line or "|" not in line:
|
|
continue
|
|
parts = line.split("|", 5)
|
|
if len(parts) == 6:
|
|
full_h, short_h, author, rel_time, date_str, subject = parts
|
|
is_current = (full_h == head_hash) or (short_h == head_hash[:len(short_h)])
|
|
commits.append({
|
|
"hash": full_h,
|
|
"short_hash": short_h,
|
|
"author": author,
|
|
"relative_time": rel_time,
|
|
"date": date_str,
|
|
"subject": subject,
|
|
"is_current": is_current,
|
|
})
|
|
return commits
|
|
except Exception as e:
|
|
logger.error(f"Error fetching commit history in {ws}: {e}")
|
|
return []
|
|
|
|
async def get_commit_detail(self, workspace_path: str, commit_hash: str) -> Dict[str, Any]:
|
|
"""Gets full commit details including diff stat and full commit message."""
|
|
ws = Path(workspace_path).expanduser().resolve()
|
|
if not (ws / ".git").exists():
|
|
return {"error": "Not a git repo"}
|
|
|
|
try:
|
|
info_out = await self._run_cmd(
|
|
["git", "log", "-1", "--format=%H|%h|%an|%ae|%ar|%cd|%s", "--date=iso", commit_hash],
|
|
cwd=str(ws)
|
|
)
|
|
parts = info_out.strip().split("|", 6)
|
|
if len(parts) < 7:
|
|
return {"error": "Invalid commit hash"}
|
|
|
|
stat_out = await self._run_cmd(["git", "show", "--stat", "--oneline", commit_hash], cwd=str(ws))
|
|
stat_lines = stat_out.strip().splitlines()
|
|
file_stats = "\n".join(stat_lines[1:]) if len(stat_lines) > 1 else ""
|
|
|
|
head_hash = ""
|
|
try:
|
|
head_hash = (await self._run_cmd(["git", "rev-parse", "HEAD"], cwd=str(ws))).strip()
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"hash": parts[0],
|
|
"short_hash": parts[1],
|
|
"author": parts[2],
|
|
"email": parts[3],
|
|
"relative_time": parts[4],
|
|
"date": parts[5],
|
|
"subject": parts[6],
|
|
"stat": file_stats.strip(),
|
|
"is_current": (parts[0] == head_hash or parts[1] == head_hash[:len(parts[1])]),
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Error getting commit detail {commit_hash} in {ws}: {e}")
|
|
return {"error": str(e)}
|
|
|
|
async def git_checkout(self, workspace_path: str, target: str) -> Tuple[bool, str]:
|
|
"""Checks out a commit hash or branch (e.g. main)."""
|
|
ws = Path(workspace_path).expanduser().resolve()
|
|
if not (ws / ".git").exists():
|
|
return False, "مسیر مشخصشده یک مخزن گیت نیست."
|
|
|
|
try:
|
|
porcelain = await self._run_cmd(["git", "status", "--porcelain"], cwd=str(ws))
|
|
if porcelain.strip():
|
|
try:
|
|
await self._run_cmd(["git", "stash", "save", "Auto-stash before checkout"], cwd=str(ws))
|
|
except Exception:
|
|
pass
|
|
|
|
out = await self._run_cmd(["git", "checkout", target], cwd=str(ws))
|
|
current_head = (await self._run_cmd(["git", "rev-parse", "--short", "HEAD"], cwd=str(ws))).strip()
|
|
return True, f"جابهجایی به {target} انجام شد (HEAD اکنون: {current_head})."
|
|
except Exception as e:
|
|
logger.error(f"Error checkout {target} in {ws}: {e}")
|
|
return False, f"خطا در جابهجایی به کامیت/شاخه {target}: {e}"
|
|
|
|
async def git_reset_hard(self, workspace_path: str, commit_hash: str) -> Tuple[bool, str]:
|
|
"""Resets the repository hard to the specified commit."""
|
|
ws = Path(workspace_path).expanduser().resolve()
|
|
if not (ws / ".git").exists():
|
|
return False, "مسیر مشخصشده یک مخزن گیت نیست."
|
|
|
|
try:
|
|
out = await self._run_cmd(["git", "reset", "--hard", commit_hash], cwd=str(ws))
|
|
log_out = await self._run_cmd(["git", "log", "-1", "--format=%h - %s"], cwd=str(ws))
|
|
return True, f"✅ پروژه با موفقیت به کامیت بازگردانی شد:\n<code>{log_out.strip()}</code>"
|
|
except Exception as e:
|
|
logger.error(f"Error resetting to {commit_hash} in {ws}: {e}")
|
|
return False, f"خطا در بازگردانی به کامیت {commit_hash}: {e}"
|
|
|
|
async def _run_cmd(self, cmd: List[str], cwd: str) -> str:
|
|
"""Helper to run a shell command asynchronously."""
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
cwd=cwd,
|
|
)
|
|
stdout, stderr = await proc.communicate()
|
|
if proc.returncode != 0:
|
|
err_msg = stderr.decode("utf-8", errors="replace").strip()
|
|
# If stdout has info, include it
|
|
out_msg = stdout.decode("utf-8", errors="replace").strip()
|
|
raise RuntimeError(err_msg or out_msg or f"Command failed with exit code {proc.returncode}")
|
|
return stdout.decode("utf-8", errors="replace")
|
|
|
|
git_manager = GitManager()
|