From 259fe538d40c3d86e5bec1aee19c3051570c3a7c Mon Sep 17 00:00:00 2001 From: xcaliber Date: Tue, 13 Jan 2026 18:22:43 +0000 Subject: [PATCH] Add gitea/admin.py --- gitea/admin.py | 2052 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 2052 insertions(+) create mode 100644 gitea/admin.py diff --git a/gitea/admin.py b/gitea/admin.py new file mode 100644 index 0000000..6a6e052 --- /dev/null +++ b/gitea/admin.py @@ -0,0 +1,2052 @@ +""" +title: Gitea Admin +author: Jeff Smith + minimax + Claude +version: 1.2.0 +license: MIT +description: Gitea organization management - teams, members, repository settings, and admin operations +requirements: pydantic, httpx +""" + +from typing import Optional, Callable, Any, List, Dict +from pydantic import BaseModel, Field +import httpx + + +class Tools: + class Valves(BaseModel): + """Shared configuration (set by admin)""" + + GITEA_URL: str = Field( + default="https://gitea.example.com", + description="Gitea URL (requires admin token for most operations)", + ) + DEFAULT_ORG: str = Field( + default="", + description="Default organization for team operations", + ) + ALLOW_USER_OVERRIDES: bool = Field( + default=True, + description="Allow users to override defaults via UserValves", + ) + + class UserValves(BaseModel): + """Per-user configuration""" + + GITEA_TOKEN: str = Field( + default="", + description="Your Gitea API token with admin scope", + ) + USER_DEFAULT_ORG: str = Field( + default="", + description="Override default organization", + ) + + def __init__(self): + self.valves = self.Valves() + self.user_valves = self.UserValves() + self.citation = False + self._team_cache: Dict[str, Dict[str, int]] = {} + + def _api_url(self, endpoint: str) -> str: + base = self.valves.GITEA_URL.rstrip("/") + return f"{base}/api/v1{endpoint}" + + def _get_token( + self, + __user__: dict = None, + ) -> str: + """Get token - always user-specific for security.""" + if __user__ and "valves" in __user__: + return __user__["valves"].GITEA_TOKEN + return "" + + def _headers(self, token: str) -> dict: + return { + "Authorization": f"token {token}", + "Content-Type": "application/json", + } + + def _get_org( + self, + org: Optional[str], + __user__: dict = None, + ) -> str: + """Get effective org with priority.""" + if org: + return org + if self.valves.ALLOW_USER_OVERRIDES: + if __user__ and "valves" in __user__: + user_org = __user__["valves"].USER_DEFAULT_ORG + if user_org: + return user_org + return self.valves.DEFAULT_ORG + + def _resolve_repo(self, repo: Optional[str]) -> tuple[str, str]: + if not repo: + raise ValueError("No repository specified") + if "/" not in repo: + raise ValueError(f"Repository must be in owner/repo format, got: {repo}") + return repo.split("/", 1) + + async def _get_team_id( + self, + team_name: str, + org: str, + __user__: dict = None, + ) -> Optional[int]: + """Resolve team name to numeric ID.""" + cache_key = org.lower() + if cache_key in self._team_cache: + if team_name in self._team_cache[cache_key]: + return self._team_cache[cache_key][team_name] + + token = self._get_token(__user__) + if not token: + return None + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.get( + self._api_url(f"/orgs/{org}/teams"), + headers=self._headers(token), + ) + response.raise_for_status() + teams = response.json() + + if not isinstance(teams, list): + return None + + if cache_key not in self._team_cache: + self._team_cache[cache_key] = {} + + for team in teams: + name = team.get("name", "") + team_id = team.get("id") + if name and team_id: + self._team_cache[cache_key][name] = team_id + + return self._team_cache[cache_key].get(team_name) + except Exception: + return None + + async def _invalidate_team_cache(self, org: str): + """Clear team cache for an org.""" + cache_key = org.lower() + if cache_key in self._team_cache: + del self._team_cache[cache_key] + + # ========================================================================= + # USER MANAGEMENT (ADMIN) + # ========================================================================= + + async def create_user( + self, + username: str, + email: str, + password: Optional[str] = None, + full_name: Optional[str] = None, + login_name: Optional[str] = None, + source_id: int = 0, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Create a new user (admin only). + + :param username: Unique username + :param email: User email address + :param password: Password (optional, may be generated) + :param full_name: Full name + :param login_name: Login name + :param source_id: Authentication source ID + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured. Admin token required." + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Creating user {username}...", + "done": False, + }, + } + ) + + payload = { + "username": username, + "email": email, + } + if password: + payload["password"] = password + if full_name: + payload["full_name"] = full_name + if login_name: + payload["login_name"] = login_name + if source_id: + payload["source_id"] = source_id + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.post( + self._api_url("/admin/users"), + headers=self._headers(token), + json=payload, + ) + response.raise_for_status() + user = response.json() + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return f"Created user: {username} (ID: {user.get('id', 'unknown')})\nEmail: {email}\nFull Name: {full_name or 'N/A'}" + except httpx.HTTPStatusError as e: + if e.response.status_code == 422: + return f"User '{username}' or email '{email}' already exists" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def delete_user( + self, + username: str, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + __event_call__: Callable[[dict], Any] = None, + ) -> str: + """Delete a user (admin only). + + :param username: Username to delete + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured. Admin token required." + + if __event_call__: + result = await __event_call__( + { + "type": "confirmation", + "data": { + "title": "Delete User?", + "message": f"Permanently delete user '{username}'? This action cannot be undone.", + }, + } + ) + if result is None or result is False: + return "delete_user CANCELLED" + if isinstance(result, dict) and not result.get("confirmed"): + return "delete_user CANCELLED" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Deleting user {username}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.delete( + self._api_url(f"/admin/users/{username}"), + headers=self._headers(token), + ) + if response.status_code == 204: + if __event_emitter__: + await __event_emitter__( + {"type": "status", "data": {"done": True, "hidden": True}} + ) + return f"Deleted user: {username}" + + response.raise_for_status() + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + return f"User '{username}' not found" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def edit_user( + self, + username: str, + email: Optional[str] = None, + password: Optional[str] = None, + full_name: Optional[str] = None, + website: Optional[str] = None, + location: Optional[str] = None, + active: Optional[bool] = None, + admin: Optional[bool] = None, + allow_create_org: Optional[bool] = None, + allow_git_hook: Optional[bool] = None, + allow_import_local: Optional[bool] = None, + max_repo_creation: Optional[int] = None, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Edit a user (admin only). + + :param username: Username to edit + :param email: New email + :param password: New password + :param full_name: New full name + :param website: Website URL + :param location: Location + :param active: Set active status + :param admin: Set admin status + :param allow_create_org: Allow organization creation + :param allow_git_hook: Allow git hooks + :param allow_import_local: Allow local imports + :param max_repo_creation: Max repository creation limit (-1 for unlimited) + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured. Admin token required." + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Editing user {username}...", + "done": False, + }, + } + ) + + payload = {} + if email is not None: + payload["email"] = email + if password is not None: + payload["password"] = password + if full_name is not None: + payload["full_name"] = full_name + if website is not None: + payload["website"] = website + if location is not None: + payload["location"] = location + if active is not None: + payload["active"] = active + if admin is not None: + payload["admin"] = admin + if allow_create_org is not None: + payload["allow_create_organization"] = allow_create_org + if allow_git_hook is not None: + payload["allow_git_hook"] = allow_git_hook + if allow_import_local is not None: + payload["allow_import_local"] = allow_import_local + if max_repo_creation is not None: + payload["max_repo_creation"] = max_repo_creation + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.patch( + self._api_url(f"/admin/users/{username}"), + headers=self._headers(token), + json=payload, + ) + response.raise_for_status() + user = response.json() + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return f"Updated user: {username}\nEmail: {user.get('email', email or 'N/A')}\nFull Name: {user.get('full_name', full_name or 'N/A')}\nAdmin: {user.get('admin', admin)}" + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + return f"User '{username}' not found" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def list_users( + self, + page: int = 1, + limit: int = 50, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """List all users (admin only). + + :param page: Page number + :param limit: Items per page + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured. Admin token required." + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Fetching users...", "done": False}, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.get( + self._api_url(f"/admin/users?page={page}&limit={limit}"), + headers=self._headers(token), + ) + response.raise_for_status() + users = response.json() + + output = f"Users (Page {page}, Limit {limit}):\n\n" + for user in users: + login = user.get("login", "") + email = user.get("email", "") + full_name = user.get("full_name", "") + admin = user.get("admin", False) + active = user.get("active", False) + created = user.get("created_at", "")[:10] + admin_badge = " [ADMIN]" if admin else "" + active_badge = " ✓" if active else " ✗" + output += f"- {login}{admin_badge}{active_badge}\n" + if email: + output += f" Email: {email}\n" + if full_name: + output += f" Name: {full_name}\n" + output += f" Created: {created}\n" + + output += f"\nTotal: {len(users)} users" + if len(users) == limit: + output += f" (use page={page+1} for more)" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return output + except httpx.HTTPStatusError as e: + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def get_user( + self, + username: str, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Get user details (admin only). + + :param username: Username to get + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured. Admin token required." + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Fetching user {username}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.get( + self._api_url(f"/users/{username}"), + headers=self._headers(token), + ) + response.raise_for_status() + user = response.json() + + output = f"# User: {username}\n\n" + output += f"**ID:** {user.get('id', 'N/A')}\n" + output += f"**Login:** {user.get('login', 'N/A')}\n" + output += f"**Full Name:** {user.get('full_name', 'N/A')}\n" + output += f"**Email:** {user.get('email', 'N/A')}\n" + output += f"**Avatar URL:** {user.get('avatar_url', 'N/A')}\n" + output += f"**Website:** {user.get('website', 'N/A')}\n" + output += f"**Location:** {user.get('location', 'N/A')}\n" + output += f"**Admin:** {'Yes' if user.get('admin') else 'No'}\n" + output += f"**Active:** {'Yes' if user.get('active') else 'No'}\n" + output += f"**Repos:** {user.get('repo_count', 0)}\n" + output += f"**Followers:** {user.get('followers_count', 0)}\n" + output += f"**Following:** {user.get('following_count', 0)}\n" + output += f"**Created:** {user.get('created_at', 'N/A')[:10]}\n" + output += f"**Updated:** {user.get('updated_at', 'N/A')[:10]}\n" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return output + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + return f"User '{username}' not found" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + # ========================================================================= + # ORGANIZATION MANAGEMENT + # ========================================================================= + + async def create_org( + self, + org_name: str, + full_name: Optional[str] = None, + description: Optional[str] = None, + website: Optional[str] = None, + location: Optional[str] = None, + visibility: str = "public", + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Create a new organization. + + :param org_name: Unique organization name + :param full_name: Full display name + :param description: Organization description + :param website: Website URL + :param location: Location + :param visibility: public, limited, or private + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + valid_visibility = ["public", "limited", "private"] + if visibility not in valid_visibility: + return f"Error: Invalid visibility '{visibility}'. Use: {', '.join(valid_visibility)}" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Creating organization {org_name}...", + "done": False, + }, + } + ) + + payload = { + "username": org_name, + } + if full_name: + payload["full_name"] = full_name + if description: + payload["description"] = description + if website: + payload["website"] = website + if location: + payload["location"] = location + payload["visibility"] = visibility + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.post( + self._api_url("/admin/users"), + headers=self._headers(token), + json=payload, + ) + response.raise_for_status() + org = response.json() + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return f"Created organization: {org_name}\nFull Name: {full_name or org_name}\nVisibility: {visibility}\nURL: {org.get('html_url', 'N/A')}" + except httpx.HTTPStatusError as e: + if e.response.status_code == 422: + return f"Organization '{org_name}' already exists" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def delete_org( + self, + org: str, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + __event_call__: Callable[[dict], Any] = None, + ) -> str: + """Delete an organization (admin only). + + :param org: Organization name + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured. Admin token required." + + if __event_call__: + result = await __event_call__( + { + "type": "confirmation", + "data": { + "title": "Delete Organization?", + "message": f"Permanently delete organization '{org}' and all its repositories? This action cannot be undone.", + }, + } + ) + if result is None or result is False: + return "delete_org CANCELLED" + if isinstance(result, dict) and not result.get("confirmed"): + return "delete_org CANCELLED" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Deleting organization {org}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.delete( + self._api_url(f"/orgs/{org}"), + headers=self._headers(token), + ) + if response.status_code == 204: + if __event_emitter__: + await __event_emitter__( + {"type": "status", "data": {"done": True, "hidden": True}} + ) + return f"Deleted organization: {org}" + + response.raise_for_status() + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + return f"Organization '{org}' not found" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def edit_org( + self, + org: str, + full_name: Optional[str] = None, + description: Optional[str] = None, + website: Optional[str] = None, + location: Optional[str] = None, + visibility: Optional[str] = None, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Edit an organization. + + :param org: Organization name + :param full_name: New full name + :param description: New description + :param website: New website + :param location: New location + :param visibility: public, limited, or private + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + if visibility and visibility not in ["public", "limited", "private"]: + return f"Error: Invalid visibility '{visibility}'. Use: public, limited, private" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Editing organization {org}...", + "done": False, + }, + } + ) + + payload = {} + if full_name is not None: + payload["full_name"] = full_name + if description is not None: + payload["description"] = description + if website is not None: + payload["website"] = website + if location is not None: + payload["location"] = location + if visibility is not None: + payload["visibility"] = visibility + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.patch( + self._api_url(f"/orgs/{org}"), + headers=self._headers(token), + json=payload, + ) + response.raise_for_status() + org_info = response.json() + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return f"Updated organization: {org}\nFull Name: {org_info.get('full_name', full_name or org)}\nVisibility: {org_info.get('visibility', visibility or 'N/A')}" + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + return f"Organization '{org}' not found" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def list_orgs( + self, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """List organizations the authenticated user belongs to.""" + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured. Add it in User Settings > Gitea Admin." + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Fetching organizations...", "done": False}, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.get( + self._api_url("/user/orgs"), + headers=self._headers(token), + ) + response.raise_for_status() + orgs = response.json() + + output = "Your Organizations:\n\n" + for org in orgs: + name = org.get("username", "") + desc = org.get("description", "")[:60] or "No description" + output += f"- {name}: {desc}\n" + + effective_org = self._get_org(None, __user__) + if effective_org: + output += f"\nDefault org: {effective_org}" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return output + except httpx.HTTPStatusError as e: + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def get_org( + self, + org: Optional[str] = None, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Get organization details.""" + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + effective_org = self._get_org(org, __user__) + if not effective_org: + return "Error: Organization not specified and no default set" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Fetching {effective_org}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.get( + self._api_url(f"/orgs/{effective_org}"), + headers=self._headers(token), + ) + response.raise_for_status() + org_info = response.json() + + output = f"# Organization: {effective_org}\n\n" + output += f"**Full Name:** {org_info.get('full_name', effective_org)}\n" + output += ( + f"**Description:** {org_info.get('description', 'No description')}\n" + ) + output += f"**Website:** {org_info.get('website', 'N/A')}\n" + output += f"**Location:** {org_info.get('location', 'N/A')}\n" + output += f"**Visibility:** {org_info.get('visibility', 'unknown')}\n" + output += f"**Repos:** {org_info.get('repo_count', 0)}\n" + output += f"**Members:** {org_info.get('members', 0)}\n" + output += f"**Teams:** {org_info.get('teams', 0)}\n" + output += f"**Created:** {org_info.get('created_at', 'N/A')[:10]}\n" + output += f"**Updated:** {org_info.get('updated_at', 'N/A')[:10]}\n" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return output + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + return f"Organization not found: {effective_org}" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def list_org_members( + self, + org: Optional[str] = None, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """List members of an organization. + + :param org: Organization name + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + effective_org = self._get_org(org, __user__) + if not effective_org: + return "Error: Organization not specified and no default set" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Fetching members of {effective_org}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.get( + self._api_url(f"/orgs/{effective_org}/members"), + headers=self._headers(token), + ) + response.raise_for_status() + members = response.json() + + output = f"Members of {effective_org}:\n\n" + for member in members: + login = member.get("login", "") + full_name = member.get("full_name", "") or member.get("username", "") + output += f"- {login} ({full_name})\n" + + output += f"\nTotal: {len(members)} members" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return output + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + return f"Organization '{effective_org}' not found or no access" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def add_org_member( + self, + username: str, + org: Optional[str] = None, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Add a member to an organization. + + :param username: Username to add + :param org: Organization name + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + effective_org = self._get_org(org, __user__) + if not effective_org: + return "Error: Organization not specified and no default set" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Adding {username} to {effective_org}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.put( + self._api_url(f"/orgs/{effective_org}/members/{username}"), + headers=self._headers(token), + ) + if response.status_code == 204: + if __event_emitter__: + await __event_emitter__( + {"type": "status", "data": {"done": True, "hidden": True}} + ) + return f"Added {username} to organization '{effective_org}'" + + response.raise_for_status() + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + return f"User '{username}' or organization '{effective_org}' not found" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def remove_org_member( + self, + username: str, + org: Optional[str] = None, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Remove a member from an organization. + + :param username: Username to remove + :param org: Organization name + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + effective_org = self._get_org(org, __user__) + if not effective_org: + return "Error: Organization not specified and no default set" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Removing {username} from {effective_org}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.delete( + self._api_url(f"/orgs/{effective_org}/members/{username}"), + headers=self._headers(token), + ) + if response.status_code == 204: + if __event_emitter__: + await __event_emitter__( + {"type": "status", "data": {"done": True, "hidden": True}} + ) + return f"Removed {username} from organization '{effective_org}'" + + response.raise_for_status() + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + return f"User '{username}' not in organization '{effective_org}'" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def list_org_repos( + self, + org: Optional[str] = None, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """List repositories in an organization. + + :param org: Organization name + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + effective_org = self._get_org(org, __user__) + if not effective_org: + return "Error: Organization not specified and no default set" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Fetching repositories for {effective_org}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.get( + self._api_url(f"/orgs/{effective_org}/repos"), + headers=self._headers(token), + ) + response.raise_for_status() + repos = response.json() + + output = f"Repositories in {effective_org}:\n\n" + for repo in repos: + full_name = repo.get("full_name", "") + private = "🔒" if repo.get("private") else "🌐" + desc = repo.get("description", "")[:50] or "No description" + stars = repo.get("stars_count", 0) + forks = repo.get("forks_count", 0) + output += f"- {private} {full_name}\n" + output += f" {desc} | ⭐ {stars} | 🍴 {forks}\n" + + output += f"\nTotal: {len(repos)} repositories" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return output + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + return f"Organization '{effective_org}' not found" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def create_org_repo( + self, + repo_name: str, + org: Optional[str] = None, + description: str = "", + private: bool = False, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Create a repository in an organization. + + :param repo_name: Repository name + :param org: Organization name + :param description: Repository description + :param private: Make repository private + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + effective_org = self._get_org(org, __user__) + if not effective_org: + return "Error: Organization not specified and no default set" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Creating repository {effective_org}/{repo_name}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.post( + self._api_url(f"/orgs/{effective_org}/repos"), + headers=self._headers(token), + json={ + "name": repo_name, + "description": description, + "private": private, + "auto_init": True, + }, + ) + response.raise_for_status() + repo = response.json() + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + visibility = "Private" if private else "Public" + return f"Created repository: {effective_org}/{repo_name}\nVisibility: {visibility}\nDescription: {description or 'N/A'}\nURL: {repo.get('html_url', 'N/A')}" + except httpx.HTTPStatusError as e: + if e.response.status_code == 422: + return f"Repository '{repo_name}' already exists in {effective_org}" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + # ========================================================================= + # TEAMS (UPDATED - now uses team ID resolution) + # ========================================================================= + + async def list_teams( + self, + org: Optional[str] = None, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """List teams in an organization.""" + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + effective_org = self._get_org(org, __user__) + if not effective_org: + return "Error: Organization not specified and no default set" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Fetching teams in {effective_org}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.get( + self._api_url(f"/orgs/{effective_org}/teams"), + headers=self._headers(token), + ) + response.raise_for_status() + teams = response.json() + + output = f"Teams in {effective_org}:\n\n" + for team in teams: + name = team.get("name", "") + team_id = team.get("id", "?") + desc = team.get("description", "")[:50] or "No description" + members = team.get("members_count", 0) + repos = team.get("repos_count", 0) + permission = team.get("permission", "unknown") + output += f"- {name} (ID: {team_id}): {desc}\n" + output += ( + f" Permission: {permission} | {members} members | {repos} repos\n" + ) + + output += f"\nTotal: {len(teams)} teams" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return output + except httpx.HTTPStatusError as e: + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def create_team( + self, + name: str, + org: Optional[str] = None, + description: str = "", + permission: str = "push", + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Create a new team in an organization.""" + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + effective_org = self._get_org(org, __user__) + if not effective_org: + return "Error: Organization not specified and no default set" + + valid_permissions = ["read", "write", "admin"] + if permission not in valid_permissions: + return f"Error: Invalid permission '{permission}'. Use: {', '.join(valid_permissions)}" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": f"Creating team {name}...", "done": False}, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.post( + self._api_url(f"/orgs/{effective_org}/teams"), + headers=self._headers(token), + json={ + "name": name, + "description": description, + "permission": permission, + "units": [ + "repo.code", + "repo.issues", + "repo.pulls", + "repo.releases", + "repo.wiki", + ], + }, + ) + response.raise_for_status() + team = response.json() + team_id = team.get("id", "unknown") + + await self._invalidate_team_cache(effective_org) + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return f"Created team: {name} (ID: {team_id}) in {effective_org}\nPermission: {permission}" + except httpx.HTTPStatusError as e: + if e.response.status_code == 422: + return f"Team '{name}' already exists in {effective_org}" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def delete_team( + self, + team_name: str, + org: Optional[str] = None, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + __event_call__: Callable[[dict], Any] = None, + ) -> str: + """Delete a team from an organization.""" + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + effective_org = self._get_org(org, __user__) + if not effective_org: + return "Error: Organization not specified and no default set" + + team_id = await self._get_team_id(team_name, effective_org, __user__) + if not team_id: + return f"Error: Team '{team_name}' not found in {effective_org}" + + if __event_call__: + result = await __event_call__( + { + "type": "confirmation", + "data": { + "title": "Delete Team?", + "message": f"Permanently delete team '{team_name}' (ID: {team_id}) from {effective_org}?", + }, + } + ) + if result is None or result is False: + return "delete_team CANCELLED" + if isinstance(result, dict) and not result.get("confirmed"): + return "delete_team CANCELLED" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Deleting team {team_name}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.delete( + self._api_url(f"/teams/{team_id}"), + headers=self._headers(token), + ) + if response.status_code == 204: + await self._invalidate_team_cache(effective_org) + if __event_emitter__: + await __event_emitter__( + {"type": "status", "data": {"done": True, "hidden": True}} + ) + return f"Deleted team '{team_name}' (ID: {team_id}) from {effective_org}" + + response.raise_for_status() + except httpx.HTTPStatusError as e: + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def add_team_member( + self, + username: str, + team_name: str, + org: Optional[str] = None, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Add a user to a team.""" + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + effective_org = self._get_org(org, __user__) + if not effective_org: + return "Error: Organization not specified and no default set" + + team_id = await self._get_team_id(team_name, effective_org, __user__) + if not team_id: + return f"Error: Team '{team_name}' not found in {effective_org}. Use list_teams() to see available teams." + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Adding {username} to {team_name}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.put( + self._api_url(f"/teams/{team_id}/members/{username}"), + headers=self._headers(token), + ) + if response.status_code == 204: + if __event_emitter__: + await __event_emitter__( + {"type": "status", "data": {"done": True, "hidden": True}} + ) + return f"Added {username} to team '{team_name}' (ID: {team_id}) in {effective_org}" + + response.raise_for_status() + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + return f"User '{username}' not found" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def remove_team_member( + self, + username: str, + team_name: str, + org: Optional[str] = None, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Remove a user from a team.""" + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + effective_org = self._get_org(org, __user__) + if not effective_org: + return "Error: Organization not specified and no default set" + + team_id = await self._get_team_id(team_name, effective_org, __user__) + if not team_id: + return f"Error: Team '{team_name}' not found in {effective_org}" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Removing {username} from {team_name}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.delete( + self._api_url(f"/teams/{team_id}/members/{username}"), + headers=self._headers(token), + ) + if response.status_code == 204: + if __event_emitter__: + await __event_emitter__( + {"type": "status", "data": {"done": True, "hidden": True}} + ) + return f"Removed {username} from team '{team_name}' (ID: {team_id}) in {effective_org}" + + response.raise_for_status() + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + return f"User '{username}' not in team '{team_name}'" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def list_team_members( + self, + team_name: str, + org: Optional[str] = None, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """List members of a team.""" + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + effective_org = self._get_org(org, __user__) + if not effective_org: + return "Error: Organization not specified and no default set" + + team_id = await self._get_team_id(team_name, effective_org, __user__) + if not team_id: + return f"Error: Team '{team_name}' not found in {effective_org}. Use list_teams() to see available teams." + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Fetching members of {team_name}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.get( + self._api_url(f"/teams/{team_id}/members"), + headers=self._headers(token), + ) + response.raise_for_status() + members = response.json() + + output = f"Members of '{team_name}' (ID: {team_id}) in {effective_org}:\n\n" + for member in members: + login = member.get("login", "") + full_name = member.get("full_name", "") or member.get("username", "") + is_admin = member.get("is_admin", False) + admin_badge = " [ADMIN]" if is_admin else "" + output += f"- {login} ({full_name}){admin_badge}\n" + + output += f"\nTotal: {len(members)} members" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return output + except httpx.HTTPStatusError as e: + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def add_team_repo( + self, + team_name: str, + repo: str, + org: Optional[str] = None, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Add a repository to a team's access list.""" + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + effective_org = self._get_org(org, __user__) + if not effective_org: + return "Error: Organization not specified" + + team_id = await self._get_team_id(team_name, effective_org, __user__) + if not team_id: + return f"Error: Team '{team_name}' not found in {effective_org}" + + try: + owner, repo_name = self._resolve_repo(repo) + except ValueError as e: + return f"Error: {e}" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": f"Adding {repo} to team...", "done": False}, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.put( + self._api_url(f"/teams/{team_id}/repos/{owner}/{repo_name}"), + headers=self._headers(token), + ) + if response.status_code == 204: + if __event_emitter__: + await __event_emitter__( + {"type": "status", "data": {"done": True, "hidden": True}} + ) + return f"Added {repo} to team '{team_name}'" + + response.raise_for_status() + except httpx.HTTPStatusError as e: + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def remove_team_repo( + self, + team_name: str, + repo: str, + org: Optional[str] = None, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Remove a repository from a team's access list.""" + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + effective_org = self._get_org(org, __user__) + if not effective_org: + return "Error: Organization not specified" + + team_id = await self._get_team_id(team_name, effective_org, __user__) + if not team_id: + return f"Error: Team '{team_name}' not found in {effective_org}" + + try: + owner, repo_name = self._resolve_repo(repo) + except ValueError as e: + return f"Error: {e}" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Removing {repo} from team...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.delete( + self._api_url(f"/teams/{team_id}/repos/{owner}/{repo_name}"), + headers=self._headers(token), + ) + if response.status_code == 204: + if __event_emitter__: + await __event_emitter__( + {"type": "status", "data": {"done": True, "hidden": True}} + ) + return f"Removed {repo} from team '{team_name}'" + + response.raise_for_status() + except httpx.HTTPStatusError as e: + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def list_team_repos( + self, + team_name: str, + org: Optional[str] = None, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """List repositories accessible to a team.""" + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + effective_org = self._get_org(org, __user__) + if not effective_org: + return "Error: Organization not specified" + + team_id = await self._get_team_id(team_name, effective_org, __user__) + if not team_id: + return f"Error: Team '{team_name}' not found in {effective_org}" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Fetching repos for {team_name}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.get( + self._api_url(f"/teams/{team_id}/repos"), + headers=self._headers(token), + ) + response.raise_for_status() + repos = response.json() + + output = f"Repositories for team '{team_name}' (ID: {team_id}):\n\n" + for repo in repos: + full_name = repo.get("full_name", "") + private = "🔒" if repo.get("private") else "🌐" + desc = repo.get("description", "")[:40] or "No description" + output += f"- {private} {full_name}: {desc}\n" + + output += f"\nTotal: {len(repos)} repositories" + + if __event_emitter__: + await __event_emitter__( + {"type": "status", "data": {"done": True, "hidden": True}} + ) + return output + except httpx.HTTPStatusError as e: + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + # ========================================================================= + # COLLABORATORS + # ========================================================================= + + async def add_collaborator( + self, + username: str, + repo: str, + permission: str = "push", + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Add a collaborator to a repository.""" + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + try: + owner, repo_name = self._resolve_repo(repo) + except ValueError as e: + return f"Error: {e}" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Adding {username} to {owner}/{repo_name}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.put( + self._api_url( + f"/repos/{owner}/{repo_name}/collaborators/{username}" + ), + headers=self._headers(token), + json={"permission": permission}, + ) + response.raise_for_status() + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return ( + f"Added {username} to {owner}/{repo_name} with {permission} permission" + ) + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + return f"Repository {owner}/{repo_name} or user '{username}' not found" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def remove_collaborator( + self, + username: str, + repo: str, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Remove a collaborator from a repository.""" + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + try: + owner, repo_name = self._resolve_repo(repo) + except ValueError as e: + return f"Error: {e}" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Removing {username} from {owner}/{repo_name}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.delete( + self._api_url( + f"/repos/{owner}/{repo_name}/collaborators/{username}" + ), + headers=self._headers(token), + ) + if response.status_code == 204: + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": "Done", + "done": True, + "hidden": True, + }, + } + ) + return f"Removed {username} from {owner}/{repo_name}" + + response.raise_for_status() + except httpx.HTTPStatusError as e: + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def list_collaborators( + self, + repo: str, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """List collaborators on a repository.""" + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + try: + owner, repo_name = self._resolve_repo(repo) + except ValueError as e: + return f"Error: {e}" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Fetching collaborators for {owner}/{repo_name}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.get( + self._api_url(f"/repos/{owner}/{repo_name}/collaborators"), + headers=self._headers(token), + ) + response.raise_for_status() + collabs = response.json() + + output = f"Collaborators on {owner}/{repo_name}:\n\n" + for col in collabs: + login = col.get("login", "") + perm = col.get("permissions", {}) + role = ( + "admin" + if perm.get("admin") + else "push" if perm.get("push") else "pull" + ) + output += f"- {login} ({role})\n" + + output += f"\nTotal: {len(collabs)} collaborators" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return output + except httpx.HTTPStatusError as e: + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + # ========================================================================= + # REPOSITORY MANAGEMENT + # ========================================================================= + + async def create_repo( + self, + repo_name: str, + description: str = "", + private: bool = False, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Create a personal repository. + + :param repo_name: Repository name + :param description: Repository description + :param private: Make repository private + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Creating repository {repo_name}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.post( + self._api_url("/user/repos"), + headers=self._headers(token), + json={ + "name": repo_name, + "description": description, + "private": private, + "auto_init": True, + }, + ) + response.raise_for_status() + repo = response.json() + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + visibility = "Private" if private else "Public" + return f"Created repository: {repo_name}\nVisibility: {visibility}\nDescription: {description or 'N/A'}\nURL: {repo.get('html_url', 'N/A')}" + except httpx.HTTPStatusError as e: + if e.response.status_code == 422: + return f"Repository '{repo_name}' already exists" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def delete_repo( + self, + repo: str, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + __event_call__: Callable[[dict], Any] = None, + ) -> str: + """Delete a repository. + + :param repo: Repository in owner/repo format + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + try: + owner, repo_name = self._resolve_repo(repo) + except ValueError as e: + return f"Error: {e}" + + if __event_call__: + result = await __event_call__( + { + "type": "confirmation", + "data": { + "title": "Delete Repository?", + "message": f"Permanently delete repository '{owner}/{repo_name}'? This action cannot be undone.", + }, + } + ) + if result is None or result is False: + return "delete_repo CANCELLED" + if isinstance(result, dict) and not result.get("confirmed"): + return "delete_repo CANCELLED" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Deleting repository {owner}/{repo_name}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.delete( + self._api_url(f"/repos/{owner}/{repo_name}"), + headers=self._headers(token), + ) + if response.status_code == 204: + if __event_emitter__: + await __event_emitter__( + {"type": "status", "data": {"done": True, "hidden": True}} + ) + return f"Deleted repository: {owner}/{repo_name}" + + response.raise_for_status() + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + return f"Repository '{owner}/{repo_name}' not found" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" + + async def get_repo( + self, + repo: str, + __user__: dict = None, + __event_emitter__: Callable[[dict], Any] = None, + ) -> str: + """Get repository details. + + :param repo: Repository in owner/repo format + """ + token = self._get_token(__user__) + if not token: + return "Error: GITEA_TOKEN not configured" + + try: + owner, repo_name = self._resolve_repo(repo) + except ValueError as e: + return f"Error: {e}" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": { + "description": f"Fetching repository {owner}/{repo_name}...", + "done": False, + }, + } + ) + + try: + async with httpx.AsyncClient(timeout=30.0, verify=False) as client: + response = await client.get( + self._api_url(f"/repos/{owner}/{repo_name}"), + headers=self._headers(token), + ) + response.raise_for_status() + repo_info = response.json() + + output = f"# Repository: {owner}/{repo_name}\n\n" + output += f"**Full Name:** {repo_info.get('full_name', 'N/A')}\n" + output += ( + f"**Description:** {repo_info.get('description', 'No description')}\n" + ) + output += f"**Private:** {'Yes' if repo_info.get('private') else 'No'}\n" + output += f"**Fork:** {'Yes' if repo_info.get('fork') else 'No'}\n" + output += f"**Stars:** {repo_info.get('stars_count', 0)}\n" + output += f"**Forks:** {repo_info.get('forks_count', 0)}\n" + output += f"**Watchers:** {repo_info.get('watchers_count', 0)}\n" + output += f"**Open Issues:** {repo_info.get('open_issues_count', 0)}\n" + output += f"**Size:** {repo_info.get('size', 0)} KB\n" + output += f"**Default Branch:** {repo_info.get('default_branch', 'main')}\n" + output += f"**Created:** {repo_info.get('created_at', 'N/A')[:10]}\n" + output += f"**Updated:** {repo_info.get('updated_at', 'N/A')[:10]}\n" + output += f"**HTML URL:** {repo_info.get('html_url', 'N/A')}\n" + + if __event_emitter__: + await __event_emitter__( + { + "type": "status", + "data": {"description": "Done", "done": True, "hidden": True}, + } + ) + return output + except httpx.HTTPStatusError as e: + if e.response.status_code == 404: + return f"Repository '{owner}/{repo_name}' not found" + return f"HTTP Error {e.response.status_code}: {e.response.text[:200]}" + except Exception as e: + return f"Error: {type(e).__name__}: {e}" \ No newline at end of file