274 Zeilen
12 KiB
Python
274 Zeilen
12 KiB
Python
import logging
|
|
from typing import List, Dict, Any, Optional, Tuple
|
|
from pathlib import Path
|
|
from .gitea_client import GiteaClient, GiteaConfig
|
|
from .git_operations import GitOperationsManager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class RepositoryManager:
|
|
def __init__(self, gitea_client: Optional[GiteaClient] = None):
|
|
self.client = gitea_client or GiteaClient()
|
|
self._current_user = None
|
|
|
|
try:
|
|
# Get the actual username from Gitea
|
|
user_info = self.client.get_user_info()
|
|
actual_username = user_info.get('username', self.client.config.username)
|
|
self._current_user = user_info
|
|
except Exception as e:
|
|
logger.warning(f"Failed to get user info from Gitea: {e}")
|
|
# Fall back to config username
|
|
actual_username = self.client.config.username
|
|
|
|
self.git_ops = GitOperationsManager(
|
|
base_url=self.client.config.base_url,
|
|
token=self.client.config.api_token,
|
|
username=actual_username
|
|
)
|
|
|
|
@property
|
|
def current_user(self) -> Dict[str, Any]:
|
|
if self._current_user is None:
|
|
self._current_user = self.client.get_user_info()
|
|
return self._current_user
|
|
|
|
def list_all_repositories(self) -> List[Dict[str, Any]]:
|
|
all_repos = []
|
|
page = 1
|
|
|
|
while True:
|
|
repos = self.client.list_repositories(page=page)
|
|
if not repos:
|
|
break
|
|
all_repos.extend(repos)
|
|
page += 1
|
|
|
|
return all_repos
|
|
|
|
def list_organization_repositories(self, org_name: str, page: int = None, per_page: int = None) -> List[Dict[str, Any]]:
|
|
"""List all repositories for an organization"""
|
|
if page is not None and per_page is not None:
|
|
# Single page request
|
|
try:
|
|
repos = self.client._request("GET", f"orgs/{org_name}/repos", params={"page": page, "limit": per_page})
|
|
return repos if repos else []
|
|
except Exception as e:
|
|
logger.error(f"Failed to list org repositories: {e}")
|
|
return []
|
|
else:
|
|
# Get all pages
|
|
all_repos = []
|
|
current_page = 1
|
|
|
|
while True:
|
|
try:
|
|
repos = self.client._request("GET", f"orgs/{org_name}/repos", params={"page": current_page, "limit": 50})
|
|
if not repos:
|
|
break
|
|
all_repos.extend(repos)
|
|
current_page += 1
|
|
except Exception as e:
|
|
logger.error(f"Failed to list org repositories: {e}")
|
|
break
|
|
|
|
return all_repos
|
|
|
|
def create_repository(self, name: str, description: str = "", private: bool = False,
|
|
auto_init: bool = True, gitignore: str = "", license: str = "",
|
|
organization: str = None) -> Dict[str, Any]:
|
|
try:
|
|
# Try to create in organization first
|
|
if organization:
|
|
try:
|
|
data = {
|
|
"name": name,
|
|
"description": description,
|
|
"private": private,
|
|
"auto_init": auto_init,
|
|
"gitignores": gitignore,
|
|
"license": license,
|
|
"default_branch": "main" # Use main instead of master
|
|
}
|
|
repo = self.client._request("POST", f"orgs/{organization}/repos", json=data)
|
|
logger.info(f"Repository '{name}' created successfully in organization '{organization}'")
|
|
return repo
|
|
except Exception as org_error:
|
|
logger.error(f"Failed to create in organization {organization}: {org_error}")
|
|
# Don't fall back - raise the error so user knows what happened
|
|
raise Exception(f"Konnte Repository nicht in Organisation '{organization}' erstellen: {str(org_error)}")
|
|
|
|
# Create as user repository
|
|
repo = self.client.create_repository(
|
|
name=name,
|
|
description=description,
|
|
private=private,
|
|
auto_init=auto_init,
|
|
gitignores=gitignore,
|
|
license=license
|
|
)
|
|
logger.info(f"Repository '{name}' created successfully as user repository")
|
|
return repo
|
|
except Exception as e:
|
|
logger.error(f"Failed to create repository '{name}': {e}")
|
|
raise
|
|
|
|
def delete_repository(self, repo_name: str) -> bool:
|
|
try:
|
|
owner = self.current_user["username"]
|
|
self.client.delete_repository(owner, repo_name)
|
|
logger.info(f"Repository '{repo_name}' deleted successfully")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Failed to delete repository '{repo_name}': {e}")
|
|
return False
|
|
|
|
def clone_repository(self, repo_name: str, clone_dir: Optional[Path] = None) -> Tuple[bool, Path]:
|
|
owner = self.current_user["username"]
|
|
return self.git_ops.clone_repository(owner, repo_name, clone_dir)
|
|
|
|
def get_repository_info(self, repo_name: str) -> Optional[Dict[str, Any]]:
|
|
try:
|
|
owner = self.current_user["username"]
|
|
return self.client.get_repository(owner, repo_name)
|
|
except Exception as e:
|
|
logger.error(f"Failed to get repository info for '{repo_name}': {e}")
|
|
return None
|
|
|
|
def fork_repository(self, owner: str, repo_name: str) -> Optional[Dict[str, Any]]:
|
|
try:
|
|
fork = self.client.fork_repository(owner, repo_name)
|
|
logger.info(f"Repository '{owner}/{repo_name}' forked successfully")
|
|
return fork
|
|
except Exception as e:
|
|
logger.error(f"Failed to fork repository '{owner}/{repo_name}': {e}")
|
|
return None
|
|
|
|
def search_repositories(self, query: str) -> List[Dict[str, Any]]:
|
|
all_repos = self.list_all_repositories()
|
|
query_lower = query.lower()
|
|
|
|
return [
|
|
repo for repo in all_repos
|
|
if query_lower in repo["name"].lower() or
|
|
query_lower in repo.get("description", "").lower()
|
|
]
|
|
|
|
def sync_repository(self, repo_path: Path) -> Tuple[bool, str]:
|
|
success, fetch_result = self.git_ops.fetch(repo_path)
|
|
if not success:
|
|
return False, f"Fetch failed: {fetch_result}"
|
|
|
|
success, pull_result = self.git_ops.pull(repo_path)
|
|
if not success:
|
|
return False, f"Pull failed: {pull_result}"
|
|
|
|
return True, "Repository synchronized successfully"
|
|
|
|
def commit_and_push(self, repo_path: Path, message: str,
|
|
files: Optional[List[str]] = None) -> Tuple[bool, str]:
|
|
success, add_result = self.git_ops.add(repo_path, files)
|
|
if not success:
|
|
return False, f"Add failed: {add_result}"
|
|
|
|
success, commit_result = self.git_ops.commit(repo_path, message)
|
|
if not success:
|
|
return False, f"Commit failed: {commit_result}"
|
|
|
|
success, push_result = self.git_ops.push(repo_path)
|
|
if not success:
|
|
return False, f"Push failed: {push_result}"
|
|
|
|
return True, "Changes committed and pushed successfully"
|
|
|
|
def get_repository_status(self, repo_path: Path) -> Dict[str, Any]:
|
|
success, status = self.git_ops.status(repo_path)
|
|
success_branch, branches = self.git_ops.branch(repo_path)
|
|
success_remote, remotes = self.git_ops.remote_list(repo_path)
|
|
|
|
current_branch = None
|
|
if success_branch:
|
|
for line in branches.split('\n'):
|
|
if line.startswith('*'):
|
|
current_branch = line[2:].strip()
|
|
break
|
|
|
|
return {
|
|
"has_changes": bool(status.strip()) if success else None,
|
|
"status": status if success else "Unable to get status",
|
|
"current_branch": current_branch,
|
|
"remotes": remotes if success_remote else "Unable to get remotes"
|
|
}
|
|
|
|
def create_branch(self, repo_path: Path, branch_name: str) -> Tuple[bool, str]:
|
|
return self.git_ops.checkout(repo_path, branch_name, create=True)
|
|
|
|
def switch_branch(self, repo_path: Path, branch_name: str) -> Tuple[bool, str]:
|
|
return self.git_ops.checkout(repo_path, branch_name)
|
|
|
|
def list_branches(self, repo_name: str) -> List[Dict[str, Any]]:
|
|
try:
|
|
owner = self.current_user["username"]
|
|
return self.client.list_branches(owner, repo_name)
|
|
except Exception as e:
|
|
logger.error(f"Failed to list branches for '{repo_name}': {e}")
|
|
return []
|
|
|
|
def create_remote_branch(self, repo_name: str, branch_name: str,
|
|
base_branch: str = "main") -> Optional[Dict[str, Any]]:
|
|
try:
|
|
owner = self.current_user["username"]
|
|
return self.client.create_branch(owner, repo_name, branch_name, base_branch)
|
|
except Exception as e:
|
|
logger.error(f"Failed to create branch '{branch_name}' in '{repo_name}': {e}")
|
|
return None
|
|
|
|
def delete_remote_branch(self, repo_name: str, branch_name: str) -> bool:
|
|
try:
|
|
owner = self.current_user["username"]
|
|
self.client.delete_branch(owner, repo_name, branch_name)
|
|
logger.info(f"Branch '{branch_name}' deleted from '{repo_name}'")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Failed to delete branch '{branch_name}' from '{repo_name}': {e}")
|
|
return False
|
|
|
|
def push_local_repo_to_gitea(self, local_repo_path: Path, repo_name: str,
|
|
description: str = "", private: bool = False,
|
|
branch: str = "main", organization: str = None) -> Tuple[bool, str]:
|
|
"""Create a new repo on Gitea and push an existing local repository to it"""
|
|
try:
|
|
# First create the repository on Gitea
|
|
repo = self.create_repository(
|
|
name=repo_name,
|
|
description=description,
|
|
private=private,
|
|
auto_init=False, # Important: don't initialize since we're pushing existing code
|
|
organization=organization
|
|
)
|
|
|
|
# Determine the correct owner
|
|
if organization:
|
|
owner = organization
|
|
elif 'owner' in repo and repo['owner']:
|
|
owner = repo['owner']['username'] if 'username' in repo['owner'] else repo['owner']['login']
|
|
else:
|
|
owner = self.current_user["username"]
|
|
|
|
logger.info(f"Repository created, owner determined as: {owner}")
|
|
|
|
# Then push the local repository
|
|
success, result = self.git_ops.push_existing_repo_to_gitea(
|
|
local_repo_path, owner, repo_name, branch
|
|
)
|
|
|
|
if success:
|
|
logger.info(f"Successfully pushed local repository to '{repo_name}'")
|
|
return True, f"Repository '{repo_name}' created and pushed successfully"
|
|
else:
|
|
logger.error(f"Failed to push to '{repo_name}': {result}")
|
|
return False, f"Repository created but push failed: {result}"
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to push local repository: {e}")
|
|
return False, str(e) |