From a73465cf4b881af81d90a86cda99dde112a02e62 Mon Sep 17 00:00:00 2001 From: retoor Date: Thu, 5 Jun 2025 23:32:45 +0200 Subject: [PATCH] Initial commit. --- .gitignore | 53 +++++++++++++ gitea_api.py | 205 +++++++++++++++++++++++++++++++++++++++++++++++++++ main.py | 92 +++++++++++++++++++++++ read_env.py | 65 ++++++++++++++++ 4 files changed, 415 insertions(+) create mode 100644 .gitignore create mode 100644 gitea_api.py create mode 100644 main.py create mode 100644 read_env.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2857fe7 --- /dev/null +++ b/.gitignore @@ -0,0 +1,53 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# Virtual environment +venv/ +.env +.envrc +.venv/ + +# Distribution / packaging +build/ +dist/ +*.egg-info/ +.eggs/ + +# PyInstaller +# Usually these files are generated in the build process +*.spec + +# Jupyter Notebook +.ipynb_checkpoints + +# Pytest cache +.cache/ + +# MyPy cache +.mypy_cache/ +.dmypy.json +.dmypy.json + +# Coverage reports +htmlcov/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover + +# IDEs and editors +.vscode/ +.idea/ +*.sublime-project +*.sublime-workspace + +# Logs +logs/ +*.log + +# Environment variables +.env diff --git a/gitea_api.py b/gitea_api.py new file mode 100644 index 0000000..4aefe02 --- /dev/null +++ b/gitea_api.py @@ -0,0 +1,205 @@ +""" +MIT License + +Author: retoor + +This module provides a `GiteaRepoManager` class for managing Gitea repositories. +It supports authentication via username/password, token generation, listing repositories, +checking for commits, and deleting repositories with safety and logging. + +Usage: + Instantiate the class with your Gitea API URL, username, and password. + Use methods like `list_repositories()`, `manage_repositories()`, etc., to perform operations. + +Dependencies: + - requests + +""" + +import requests +import logging +from typing import Optional, List, Dict, Any + +# Configure logger for the module +logger = logging.getLogger("GiteaRepoManager") +logger.setLevel(logging.INFO) # Default level; override as needed externally + +handler = logging.StreamHandler() +formatter = logging.Formatter('[%(asctime)s] %(levelname)s %(name)s: %(message)s') +handler.setFormatter(formatter) +if not logger.hasHandlers(): + logger.addHandler(handler) + +class GiteaRepoManager: + """ + Manage Gitea repositories using username/password authentication. + Generates a personal access token internally for API calls. + """ + + def __init__(self, api_url: str, username: str, password: str, token: Optional[str] = None, otp: Optional[str] = None): + """ + Initialize the GiteaRepoManager with API URL and credentials. + + Args: + api_url (str): Base URL of the Gitea API (e.g., 'https://gitea.example.com'). + username (str): Gitea username. + password (str): Gitea password. + token (Optional[str]): Optional pre-existing token. If None, a new token is generated. + otp (Optional[str]): Optional one-time password for 2FA. + + Raises: + ValueError: If token generation fails. + """ + self.api_url = api_url.rstrip('/') + self.username = username + self.password = password + self.token = token or self.generate_token(otp) + if self.token: + self.headers = { + "Authorization": f"token {self.token}", + "Content-Type": "application/json" + } + logger.info("Created API token 'script_token'. Delete from Gitea settings when done.") + else: + logger.error("Failed to generate token during initialization.") + raise ValueError("Failed to generate token") + + def generate_token(self, otp: Optional[str] = None) -> Optional[str]: + """ + Generate a personal access token for API authentication. + + Args: + otp (Optional[str]): One-time password for 2FA, if enabled. + + Returns: + Optional[str]: The generated token SHA1 string if successful, None otherwise. + """ + url = f"{self.api_url}/api/v1/users/{self.username}/tokens" + auth = (self.username, self.password) + headers = {"X-Gitea-OTP": otp} if otp else {} + data = {"name": "script_token"} + try: + response = requests.post(url, auth=auth, headers=headers, json=data) + logger.debug(f"Token generation request URL: {url}") + logger.debug(f"Request headers: {headers}") + logger.debug(f"Request data: {data}") + logger.debug(f"Response status code: {response.status_code}") + logger.debug(f"Response text: {response.text}") + if response.status_code == 201: + return response.json()["sha1"] + logger.error(f"Error generating token: {response.status_code} - {response.text}") + except requests.RequestException as e: + logger.error(f"Exception during token generation: {e}", exc_info=True) + return None + + def list_repositories(self) -> List[Dict[str, Any]]: + """ + List repositories accessible with the current token. + + Returns: + List[Dict[str, Any]]: List of repository data dictionaries. + """ + url = f"{self.api_url}/repos/search" + try: + response = requests.get(url, headers=self.headers) + if response.status_code == 200: + logger.debug("Successfully listed repositories.") + return response.json().get('data', []) + logger.warning(f"Error listing repositories: {response.status_code} - {response.text}") + except requests.RequestException as e: + logger.error(f"Exception during listing repositories: {e}", exc_info=True) + return [] + + def delete_repository(self, owner: str, repo_name: str, dry_run: bool = False) -> None: + """ + Delete a repository by owner and name. + + Args: + owner (str): Username of the repository owner. + repo_name (str): Name of the repository. + dry_run (bool): If True, only logs the intended deletion without executing. + + Returns: + None + """ + url = f"{self.api_url}/api/v1/repos/{owner}/{repo_name}" + if dry_run: + logger.info(f"Would delete repository: {owner}/{repo_name}") + return + try: + response = requests.delete(url, headers=self.headers) + if response.status_code == 204: + logger.info(f"Deleted repository: {owner}/{repo_name}") + else: + logger.warning(f"Error deleting repository {owner}/{repo_name}: {response.status_code} - {response.text}") + except requests.RequestException as e: + logger.error(f"Exception during deleting repository {owner}/{repo_name}: {e}", exc_info=True) + + def has_commits(self, owner: str, repo_name: str) -> bool: + """ + Check if a repository has commits. + + Args: + owner (str): Username of the repository owner. + repo_name (str): Repository name. + + Returns: + bool: True if commits exist, False otherwise. + """ + url = f"{self.api_url}/repos/{owner}/{repo_name}/commits" + try: + response = requests.get(url, headers=self.headers) + if response.status_code == 409: + # 409 indicates no commits (empty repository) + logger.debug(f"No commits found for {owner}/{repo_name}.") + return False + elif response.status_code == 200: + logger.debug(f"Commits found for {owner}/{repo_name}.") + return True + else: + logger.warning(f"Unexpected status code checking commits for {owner}/{repo_name}: {response.status_code}") + except requests.RequestException as e: + logger.error(f"Exception during checking commits for {owner}/{repo_name}: {e}", exc_info=True) + return False + + def should_delete(self, repo: Dict[str, Any]) -> bool: + """ + Determine if a repository should be deleted based on criteria. + + Args: + repo (Dict[str, Any]): Repository data dictionary. + + Returns: + bool: True if the repository should be deleted, False otherwise. + """ + try: + owner = repo['owner']['username'] + repo_name = repo['name'] + # Placeholder logic: delete if no commits + return not self.has_commits(owner, repo_name) + except Exception as e: + logger.error(f"Exception in should_delete for repo {repo}: {e}", exc_info=True) + return False + + def manage_repositories(self, dry_run: bool = False) -> List[Dict[str, Any]]: + """ + List repositories and delete those that meet deletion criteria. + + Args: + dry_run (bool): If True, only logs actions without executing deletions. + + Returns: + List[Dict[str, Any]]: List of repositories with updated 'pending_delete' flags. + """ + repos = self.list_repositories() + if not repos: + logger.warning("No repositories found or error occurred.") + return [] + for repo in repos: + try: + repo['pending_delete'] = self.should_delete(repo) + if repo['pending_delete']: + self.delete_repository(repo['owner']['username'], repo['name'], dry_run=dry_run) + except Exception as e: + logger.error(f"Exception managing repository {repo.get('name', 'unknown')}: {e}", exc_info=True) + return repos diff --git a/main.py b/main.py new file mode 100644 index 0000000..7b726cc --- /dev/null +++ b/main.py @@ -0,0 +1,92 @@ +import read_env +import os +import argparse +from gitea_api import GiteaRepoManager + +# Reads by default .env +read_env.load_env() + +def parse_arguments() -> argparse.Namespace: + """ + Parse command-line arguments with defaults from environment variables. + + Returns: + argparse.Namespace: Parsed arguments with attributes: + - gitea_username (str) + - gitea_password (str) + - gitea_api_url (str) + - gitea_api_token (str) + - dry_run (bool) + """ + parser = argparse.ArgumentParser( + description=( + "Manage Gitea repositories. " + "Configure via command-line arguments or environment variables. " + "Contact: retoor for support." + ) + ) + + # Add arguments with defaults from environment variables + parser.add_argument( + '--gitea-username', + type=str, + default=os.environ.get('GITEA_USERNAME'), + help='Gitea username (default: from GITEA_USERNAME env variable)' + ) + parser.add_argument( + '--gitea-password', + type=str, + default=os.environ.get('GITEA_PASSWORD'), + help='Gitea password (default: from GITEA_PASSWORD env variable)' + ) + parser.add_argument( + '--gitea-api-url', + type=str, + default=os.environ.get('GITEA_API_URL'), + help='Gitea API URL (default: from GITEA_API_URL env variable)' + ) + parser.add_argument( + '--gitea-api-token', + type=str, + default=os.environ.get('GITEA_API_TOKEN'), + help='Gitea API token (default: from GITEA_API_TOKEN env variable)' + ) + + # Mutually exclusive dry-run flags + dry_run_group = parser.add_mutually_exclusive_group() + dry_run_group.add_argument( + '--dry-run', + dest='dry_run', + action='store_true', + help='Perform a dry run without making changes (default: enabled)' + ) + dry_run_group.add_argument( + '--no-dry-run', + dest='dry_run', + action='store_false', + help='Disable dry run and make real changes' + ) + parser.set_defaults(dry_run=True) + + return parser.parse_args() + +def main() -> None: + """ + Main function to initialize the GiteaRepoManager with parsed arguments + and manage repositories. + """ + args = parse_arguments() + + # Initialize the manager with arguments + manager = GiteaRepoManager( + api_url=args.gitea_api_url, + token=args.gitea_api_token, + username=args.gitea_username, + password=args.gitea_password, + ) + + # Manage repositories with dry_run as specified + manager.manage_repositories(dry_run=args.dry_run) + +if __name__ == '__main__': + main() diff --git a/read_env.py b/read_env.py new file mode 100644 index 0000000..3fbb660 --- /dev/null +++ b/read_env.py @@ -0,0 +1,65 @@ +""" +MIT License + +Copyright (c) 2024 retoor + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. + +Author: retoor +Description: Utility to load environment variables from a .env file. +""" + +import os +from typing import Optional + +def load_env(file_path: str = '.env') -> None: + """ + Loads environment variables from a specified file into the current process environment. + + Args: + file_path (str): Path to the environment file. Defaults to '.env'. + + Returns: + None + + Raises: + FileNotFoundError: If the specified file does not exist. + IOError: If an I/O error occurs during file reading. + """ + try: + with open(file_path, 'r') as env_file: + for line in env_file: + line = line.strip() + # Skip empty lines and comments + if not line or line.startswith('#'): + continue + # Skip lines without '=' + if '=' not in line: + continue + # Split into key and value at the first '=' + key, value = line.split('=', 1) + # Set environment variable + os.environ[key.strip()] = value.strip() + except FileNotFoundError: + raise FileNotFoundError(f"Environment file '{file_path}' not found.") + except IOError as e: + raise IOError(f"Error reading environment file '{file_path}': {e}") + + +