dffdbf54f5
Hardcoded ALLOWED_OWNER="euphon" blocked non-euphon repos from being processed. Move it to config.yaml as a list so each deployment can specify its own allowed owners. Also make git user.name/email configurable for multi-instance deployments (e.g. ms vs yoyo).
61 lines
2.1 KiB
Python
61 lines
2.1 KiB
Python
import asyncio
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
from config import GITEA_URL, GITEA_TOKEN, WORKSPACE_DIR, BOT_USERNAME, BOT_GIT_NAME, BOT_GIT_EMAIL
|
|
|
|
log = logging.getLogger("gitea-bot")
|
|
|
|
|
|
async def _run(cmd: str, cwd: Path | None = None) -> tuple[int, str, str]:
|
|
proc = await asyncio.create_subprocess_shell(
|
|
cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
cwd=cwd,
|
|
)
|
|
stdout, stderr = await proc.communicate()
|
|
return proc.returncode, stdout.decode(), stderr.decode()
|
|
|
|
|
|
def workspace_path(owner: str, repo: str, issue_number: int) -> Path:
|
|
return WORKSPACE_DIR / f"{owner}-{repo}-{issue_number}"
|
|
|
|
|
|
def repo_path(owner: str, repo: str, issue_number: int) -> Path:
|
|
return workspace_path(owner, repo, issue_number) / "repo"
|
|
|
|
|
|
def attachments_path(owner: str, repo: str, issue_number: int) -> Path:
|
|
return workspace_path(owner, repo, issue_number) / "attachments"
|
|
|
|
|
|
async def ensure_repo(owner: str, repo: str, issue_number: int) -> Path:
|
|
"""Clone or update the repo for this issue's workspace. Returns repo dir."""
|
|
ws = workspace_path(owner, repo, issue_number)
|
|
rp = repo_path(owner, repo, issue_number)
|
|
att = attachments_path(owner, repo, issue_number)
|
|
|
|
ws.mkdir(parents=True, exist_ok=True)
|
|
att.mkdir(parents=True, exist_ok=True)
|
|
|
|
clone_url = f"https://{BOT_USERNAME}:{GITEA_TOKEN}@{GITEA_URL.split('://', 1)[1]}/{owner}/{repo}.git"
|
|
|
|
if (rp / ".git").exists():
|
|
# Reset and pull latest
|
|
log.info(f"Updating existing repo at {rp}")
|
|
rc, out, err = await _run("git checkout main 2>/dev/null || git checkout master", cwd=rp)
|
|
await _run("git fetch origin", cwd=rp)
|
|
await _run("git reset --hard FETCH_HEAD", cwd=rp)
|
|
else:
|
|
log.info(f"Cloning {owner}/{repo} to {rp}")
|
|
rc, out, err = await _run(f"git clone {clone_url} {rp}")
|
|
if rc != 0:
|
|
raise RuntimeError(f"git clone failed: {err}")
|
|
|
|
# Configure git user for bot commits
|
|
await _run(f'git config user.name "{BOT_GIT_NAME}"', cwd=rp)
|
|
await _run(f'git config user.email "{BOT_GIT_EMAIL}"', cwd=rp)
|
|
|
|
return rp
|