import requests import json from typing import Dict, List, Optional, Any from dataclasses import dataclass from datetime import datetime import logging from pathlib import Path logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class GiteaConfig: base_url: str = "https://gitea-undso.intelsight.de" api_token: str = "3b4a6ba1ade3f34640f3c85d2333b4a3a0627471" api_version: str = "v1" username: str = "StuXn3t" def __post_init__(self): """Load settings from config file if available""" self.load_from_settings() def load_from_settings(self): """Load Gitea settings from UI settings file""" try: settings_file = Path.home() / ".claude_project_manager" / "ui_settings.json" if settings_file.exists(): with open(settings_file, 'r') as f: settings = json.load(f) # Override with saved settings if available if "gitea_server_url" in settings and settings["gitea_server_url"]: self.base_url = settings["gitea_server_url"] if "gitea_api_token" in settings and settings["gitea_api_token"]: self.api_token = settings["gitea_api_token"] if "gitea_username" in settings and settings["gitea_username"]: self.username = settings["gitea_username"] logger.info(f"Loaded Gitea settings from config file") except Exception as e: logger.warning(f"Could not load Gitea settings from config: {e}") @property def api_url(self) -> str: return f"{self.base_url}/api/{self.api_version}" class GiteaClient: def __init__(self, config: Optional[GiteaConfig] = None): self.config = config or GiteaConfig() self.session = requests.Session() self.session.headers.update({ "Authorization": f"token {self.config.api_token}", "Content-Type": "application/json" }) def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]: url = f"{self.config.api_url}/{endpoint}" try: response = self.session.request(method, url, **kwargs) response.raise_for_status() return response.json() if response.content else {} except requests.exceptions.RequestException as e: logger.error(f"API request failed: {e}") raise def get_user_info(self) -> Dict[str, Any]: return self._request("GET", "user") def list_user_organizations(self) -> List[Dict[str, Any]]: """List all organizations the user is a member of""" return self._request("GET", "user/orgs") def get_user_teams_in_org(self, org_name: str) -> List[Dict[str, Any]]: """Get user's teams in a specific organization""" try: return self._request("GET", f"user/teams", params={"org": org_name}) except Exception as e: logger.error(f"Failed to get teams for org {org_name}: {e}") return [] def list_repositories(self, page: int = 1, limit: int = 50) -> List[Dict[str, Any]]: params = {"page": page, "limit": limit} return self._request("GET", "user/repos", params=params) def create_repository(self, name: str, description: str = "", private: bool = False, auto_init: bool = True, gitignores: str = "", license: str = "") -> Dict[str, Any]: data = { "name": name, "description": description, "private": private, "auto_init": auto_init, "gitignores": gitignores, "license": license, "default_branch": "main" # Use main instead of master } return self._request("POST", "user/repos", json=data) def delete_repository(self, owner: str, repo: str) -> None: self._request("DELETE", f"repos/{owner}/{repo}") def get_repository(self, owner: str, repo: str) -> Dict[str, Any]: return self._request("GET", f"repos/{owner}/{repo}") def fork_repository(self, owner: str, repo: str, organization: Optional[str] = None) -> Dict[str, Any]: data = {"organization": organization} if organization else {} return self._request("POST", f"repos/{owner}/{repo}/forks", json=data) def list_issues(self, owner: str, repo: str, state: str = "open", labels: Optional[List[str]] = None, page: int = 1, limit: int = 50) -> List[Dict[str, Any]]: params = { "state": state, "page": page, "limit": limit } if labels: params["labels"] = ",".join(labels) return self._request("GET", f"repos/{owner}/{repo}/issues", params=params) def create_issue(self, owner: str, repo: str, title: str, body: str = "", assignees: Optional[List[str]] = None, labels: Optional[List[int]] = None) -> Dict[str, Any]: data = { "title": title, "body": body, "assignees": assignees or [], "labels": labels or [] } return self._request("POST", f"repos/{owner}/{repo}/issues", json=data) def update_issue(self, owner: str, repo: str, index: int, **kwargs) -> Dict[str, Any]: return self._request("PATCH", f"repos/{owner}/{repo}/issues/{index}", json=kwargs) def close_issue(self, owner: str, repo: str, index: int) -> Dict[str, Any]: return self.update_issue(owner, repo, index, state="closed") def list_pull_requests(self, owner: str, repo: str, state: str = "open", page: int = 1, limit: int = 50) -> List[Dict[str, Any]]: params = { "state": state, "page": page, "limit": limit } return self._request("GET", f"repos/{owner}/{repo}/pulls", params=params) def create_pull_request(self, owner: str, repo: str, title: str, head: str, base: str, body: str = "", assignees: Optional[List[str]] = None) -> Dict[str, Any]: data = { "title": title, "head": head, "base": base, "body": body, "assignees": assignees or [] } return self._request("POST", f"repos/{owner}/{repo}/pulls", json=data) def merge_pull_request(self, owner: str, repo: str, index: int, merge_style: str = "merge") -> Dict[str, Any]: data = {"Do": merge_style} return self._request("POST", f"repos/{owner}/{repo}/pulls/{index}/merge", json=data) def list_branches(self, owner: str, repo: str) -> List[Dict[str, Any]]: return self._request("GET", f"repos/{owner}/{repo}/branches") def create_branch(self, owner: str, repo: str, branch_name: str, old_branch_name: str = "main") -> Dict[str, Any]: data = { "new_branch_name": branch_name, "old_branch_name": old_branch_name } return self._request("POST", f"repos/{owner}/{repo}/branches", json=data) def delete_branch(self, owner: str, repo: str, branch_name: str) -> None: self._request("DELETE", f"repos/{owner}/{repo}/branches/{branch_name}") def list_releases(self, owner: str, repo: str, page: int = 1, limit: int = 50) -> List[Dict[str, Any]]: params = {"page": page, "limit": limit} return self._request("GET", f"repos/{owner}/{repo}/releases", params=params) def create_release(self, owner: str, repo: str, tag_name: str, name: str, body: str = "", target: str = "main", draft: bool = False, prerelease: bool = False) -> Dict[str, Any]: data = { "tag_name": tag_name, "name": name, "body": body, "target_commitish": target, "draft": draft, "prerelease": prerelease } return self._request("POST", f"repos/{owner}/{repo}/releases", json=data) def list_webhooks(self, owner: str, repo: str) -> List[Dict[str, Any]]: return self._request("GET", f"repos/{owner}/{repo}/hooks") def create_webhook(self, owner: str, repo: str, url: str, events: List[str], active: bool = True) -> Dict[str, Any]: data = { "type": "gitea", "config": { "url": url, "content_type": "json" }, "events": events, "active": active } return self._request("POST", f"repos/{owner}/{repo}/hooks", json=data) def get_repository_contents(self, owner: str, repo: str, filepath: str = "", ref: str = "main") -> Dict[str, Any]: params = {"ref": ref} return self._request("GET", f"repos/{owner}/{repo}/contents/{filepath}", params=params) def create_or_update_file(self, owner: str, repo: str, filepath: str, content: str, message: str, branch: str = "main", sha: Optional[str] = None) -> Dict[str, Any]: import base64 data = { "content": base64.b64encode(content.encode()).decode(), "message": message, "branch": branch } if sha: data["sha"] = sha return self._request("PUT", f"repos/{owner}/{repo}/contents/{filepath}", json=data) def delete_file(self, owner: str, repo: str, filepath: str, message: str, sha: str, branch: str = "main") -> Dict[str, Any]: data = { "message": message, "sha": sha, "branch": branch } return self._request("DELETE", f"repos/{owner}/{repo}/contents/{filepath}", json=data) # Global config instance gitea_config = GiteaConfig() # Default client instance client = GiteaClient(gitea_config)