feat(factory): serve dashboard at root and create project repos refs NOISSUE
This commit is contained in:
@@ -1,12 +1,11 @@
|
||||
"""Gitea API integration for commits and PRs."""
|
||||
"""Gitea API integration for repository and pull request operations."""
|
||||
|
||||
import os
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class GiteaAPI:
|
||||
"""Gitea API client for repository operations."""
|
||||
|
||||
|
||||
def __init__(self, token: str, base_url: str, owner: str | None = None, repo: str | None = None):
|
||||
self.token = token
|
||||
self.base_url = base_url.rstrip("/")
|
||||
@@ -14,69 +13,100 @@ class GiteaAPI:
|
||||
self.repo = repo
|
||||
self.headers = {
|
||||
"Authorization": f"token {token}",
|
||||
"Content-Type": "application/json"
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
|
||||
def get_config(self) -> dict:
|
||||
"""Load configuration from environment."""
|
||||
base_url = os.getenv("GITEA_URL", "https://gitea.local")
|
||||
token = os.getenv("GITEA_TOKEN", "")
|
||||
owner = os.getenv("GITEA_OWNER", "ai-test")
|
||||
repo = os.getenv("GITEA_REPO", "")
|
||||
|
||||
# Allow empty repo for any repo mode (org/repo pattern)
|
||||
if not repo:
|
||||
repo = "any-repo" # Use this as a placeholder for org/repo operations
|
||||
|
||||
# Check for repo suffix pattern (e.g., repo-* for multiple repos)
|
||||
repo_suffix = os.getenv("GITEA_REPO_SUFFIX", "")
|
||||
|
||||
return {
|
||||
"base_url": base_url.rstrip("/"),
|
||||
"token": token,
|
||||
"owner": owner,
|
||||
"repo": repo,
|
||||
"repo_suffix": repo_suffix,
|
||||
"supports_any_repo": not repo or repo_suffix
|
||||
"supports_project_repos": not bool(repo),
|
||||
}
|
||||
|
||||
|
||||
def get_auth_headers(self) -> dict:
|
||||
"""Get authentication headers."""
|
||||
return {
|
||||
"Authorization": f"token {self.token}",
|
||||
"Content-Type": "application/json"
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
async def create_branch(self, branch: str, base: str = "main", owner: str | None = None, repo: str | None = None):
|
||||
"""Create a new branch.
|
||||
|
||||
Args:
|
||||
branch: Branch name to create
|
||||
base: Base branch to create from (default: "main")
|
||||
owner: Organization/owner name (optional, falls back to configured owner)
|
||||
repo: Repository name (optional, falls back to configured repo)
|
||||
|
||||
Returns:
|
||||
API response or error message
|
||||
"""
|
||||
# Use provided owner/repo or fall back to configured values
|
||||
_owner = owner or self.owner
|
||||
_repo = repo or self.repo
|
||||
|
||||
url = f"{self.base_url}/repos/{_owner}/{_repo}/branches/{branch}"
|
||||
payload = {"base": base}
|
||||
|
||||
|
||||
def _api_url(self, path: str) -> str:
|
||||
"""Build a Gitea API URL from a relative path."""
|
||||
return f"{self.base_url}/api/v1/{path.lstrip('/')}"
|
||||
|
||||
async def _request(self, method: str, path: str, payload: dict | None = None) -> dict:
|
||||
"""Perform a Gitea API request and normalize the response."""
|
||||
try:
|
||||
import aiohttp
|
||||
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, headers=self.get_auth_headers(), json=payload) as resp:
|
||||
if resp.status == 201:
|
||||
async with session.request(
|
||||
method,
|
||||
self._api_url(path),
|
||||
headers=self.get_auth_headers(),
|
||||
json=payload,
|
||||
) as resp:
|
||||
if resp.status in (200, 201):
|
||||
return await resp.json()
|
||||
else:
|
||||
return {"error": await resp.text()}
|
||||
return {"error": await resp.text(), "status_code": resp.status}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
|
||||
def build_project_repo_name(self, project_id: str, project_name: str | None = None) -> str:
|
||||
"""Build a repository name for a generated project."""
|
||||
preferred = (project_name or project_id or "project").strip().lower().replace(" ", "-")
|
||||
sanitized = "".join(ch if ch.isalnum() or ch in {"-", "_"} else "-" for ch in preferred)
|
||||
while "--" in sanitized:
|
||||
sanitized = sanitized.replace("--", "-")
|
||||
return sanitized.strip("-") or project_id
|
||||
|
||||
async def create_repo(
|
||||
self,
|
||||
repo_name: str,
|
||||
owner: str | None = None,
|
||||
description: str | None = None,
|
||||
private: bool = False,
|
||||
auto_init: bool = True,
|
||||
) -> dict:
|
||||
"""Create a repository inside the configured organization."""
|
||||
_owner = owner or self.owner
|
||||
if not _owner:
|
||||
return {"error": "Owner or organization is required"}
|
||||
|
||||
payload = {
|
||||
"name": repo_name,
|
||||
"description": description or f"AI-generated project repository for {repo_name}",
|
||||
"private": private,
|
||||
"auto_init": auto_init,
|
||||
"default_branch": "main",
|
||||
}
|
||||
result = await self._request("POST", f"orgs/{_owner}/repos", payload)
|
||||
if result.get("status_code") == 409:
|
||||
existing = await self.get_repo_info(owner=_owner, repo=repo_name)
|
||||
if not existing.get("error"):
|
||||
existing["status"] = "exists"
|
||||
return existing
|
||||
if not result.get("error"):
|
||||
result.setdefault("status", "created")
|
||||
return result
|
||||
|
||||
async def create_branch(self, branch: str, base: str = "main", owner: str | None = None, repo: str | None = None):
|
||||
"""Create a new branch."""
|
||||
_owner = owner or self.owner
|
||||
_repo = repo or self.repo
|
||||
return await self._request(
|
||||
"POST",
|
||||
f"repos/{_owner}/{_repo}/branches",
|
||||
{"new_branch_name": branch, "old_ref_name": base},
|
||||
)
|
||||
|
||||
async def create_pull_request(
|
||||
self,
|
||||
title: str,
|
||||
@@ -84,107 +114,50 @@ class GiteaAPI:
|
||||
owner: str,
|
||||
repo: str,
|
||||
base: str = "main",
|
||||
head: str | None = None
|
||||
head: str | None = None,
|
||||
) -> dict:
|
||||
"""Create a pull request.
|
||||
|
||||
Args:
|
||||
title: PR title
|
||||
body: PR description
|
||||
owner: Organization/owner name
|
||||
repo: Repository name
|
||||
base: Base branch (default: "main")
|
||||
head: Head branch (optional, auto-generated if not provided)
|
||||
|
||||
Returns:
|
||||
API response or error message
|
||||
"""
|
||||
"""Create a pull request."""
|
||||
_owner = owner or self.owner
|
||||
_repo = repo or self.repo
|
||||
|
||||
url = f"{self.base_url}/repos/{_owner}/{_repo}/pulls"
|
||||
|
||||
payload = {
|
||||
"title": title,
|
||||
"body": body,
|
||||
"base": {"branch": base},
|
||||
"head": head or f"{_owner}-{_repo}-ai-gen-{hash(title) % 10000}"
|
||||
"base": base,
|
||||
"head": head or f"{_owner}-{_repo}-ai-gen-{hash(title) % 10000}",
|
||||
}
|
||||
|
||||
try:
|
||||
import aiohttp
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(url, headers=self.get_auth_headers(), json=payload) as resp:
|
||||
if resp.status == 201:
|
||||
return await resp.json()
|
||||
else:
|
||||
return {"error": await resp.text()}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
return await self._request("POST", f"repos/{_owner}/{_repo}/pulls", payload)
|
||||
|
||||
async def push_commit(
|
||||
self,
|
||||
branch: str,
|
||||
files: list[dict],
|
||||
message: str,
|
||||
owner: str | None = None,
|
||||
repo: str | None = None
|
||||
repo: str | None = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Push files to a branch.
|
||||
|
||||
"""Push files to a branch.
|
||||
|
||||
In production, this would use gitea's API or git push.
|
||||
For now, we'll simulate the operation.
|
||||
|
||||
Args:
|
||||
branch: Branch name
|
||||
files: List of files to push
|
||||
message: Commit message
|
||||
owner: Organization/owner name (optional, falls back to configured owner)
|
||||
repo: Repository name (optional, falls back to configured repo)
|
||||
|
||||
Returns:
|
||||
Status response
|
||||
For now, this remains simulated.
|
||||
"""
|
||||
# Use provided owner/repo or fall back to configured values
|
||||
_owner = owner or self.owner
|
||||
_repo = repo or self.repo
|
||||
|
||||
|
||||
return {
|
||||
"status": "simulated",
|
||||
"branch": branch,
|
||||
"message": message,
|
||||
"files": files,
|
||||
"owner": _owner,
|
||||
"repo": _repo
|
||||
"repo": _repo,
|
||||
}
|
||||
|
||||
|
||||
async def get_repo_info(self, owner: str | None = None, repo: str | None = None) -> dict:
|
||||
"""Get repository information.
|
||||
|
||||
Args:
|
||||
owner: Organization/owner name (optional, falls back to configured owner)
|
||||
repo: Repository name (optional, falls back to configured repo)
|
||||
|
||||
Returns:
|
||||
Repository info or error message
|
||||
"""
|
||||
# Use provided owner/repo or fall back to configured values
|
||||
"""Get repository information."""
|
||||
_owner = owner or self.owner
|
||||
_repo = repo or self.repo
|
||||
|
||||
|
||||
if not _repo:
|
||||
return {"error": "Repository name required for org operations"}
|
||||
|
||||
url = f"{self.base_url}/repos/{_owner}/{_repo}"
|
||||
|
||||
try:
|
||||
import aiohttp
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url, headers=self.get_auth_headers()) as resp:
|
||||
if resp.status == 200:
|
||||
return await resp.json()
|
||||
else:
|
||||
return {"error": await resp.text()}
|
||||
except Exception as e:
|
||||
return {"error": str(e)}
|
||||
|
||||
return await self._request("GET", f"repos/{_owner}/{_repo}")
|
||||
Reference in New Issue
Block a user