markitect-main/gitea/package_registry.py

"""
Gitea Package Registry Client

This module provides functionality to publish Python packages to Gitea's package registry.
Gitea supports multiple package registries including PyPI-compatible registries.
"""

import os
import subprocess
import tempfile
from pathlib import Path
from typing import Optional, List, Dict
from .config import GiteaConfig
from .exceptions import GiteaError


class GiteaPackageRegistry:
    """Client for publishing packages to Gitea package registry."""

    def __init__(self, config: Optional[GiteaConfig] = None):
        """Initialize the package registry client.

        Args:
            config: Gitea configuration. If None, auto-detects from git repository.
        """
        self.config = config or GiteaConfig.from_git_repository()
        self.config.validate()

    @property
    def pypi_registry_url(self) -> str:
        """Get the PyPI-compatible registry URL for this repository."""
        return f"{self.config.gitea_url}/api/packages/{self.config.repo_owner}/pypi"

    @property
    def package_list_url(self) -> str:
        """Get the package listing URL for this repository."""
        return f"{self.config.gitea_url}/api/v1/packages/{self.config.repo_owner}"

    def check_auth(self) -> bool:
        """Check if authentication token is available and valid."""
        if not self.config.auth_token:
            return False

        try:
            # Test auth by trying to access packages API
            import requests
            headers = {"Authorization": f"token {self.config.auth_token}"}
            response = requests.get(self.package_list_url, headers=headers, timeout=10)
            return response.status_code in [200, 404]  # 404 is okay if no packages exist yet
        except Exception:
            return False

    def list_packages(self) -> List[Dict]:
        """List all packages for this repository owner.

        Returns:
            List of package information dictionaries
        """
        try:
            import requests
            headers = {}
            if self.config.auth_token:
                headers["Authorization"] = f"token {self.config.auth_token}"

            response = requests.get(self.package_list_url, headers=headers, timeout=10)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            raise GiteaError(f"Failed to list packages: {e}")

    def get_package_info(self, package_name: str) -> Optional[Dict]:
        """Get information about a specific package.

        Args:
            package_name: Name of the package

        Returns:
            Package information dictionary or None if not found
        """
        try:
            packages = self.list_packages()
            for package in packages:
                if package.get("name") == package_name:
                    return package
            return None
        except Exception:
            return None

    def upload_package(self,
                      wheel_path: Path,
                      sdist_path: Optional[Path] = None,
                      dry_run: bool = False) -> bool:
        """Upload package files to Gitea registry.

        Args:
            wheel_path: Path to wheel (.whl) file
            sdist_path: Optional path to source distribution (.tar.gz) file
            dry_run: If True, show what would be done without uploading

        Returns:
            True if upload successful, False otherwise
        """
        if not self.config.auth_token:
            raise GiteaError("Authentication token required for package upload. Set GITEA_API_TOKEN environment variable.")

        if not wheel_path.exists():
            raise GiteaError(f"Wheel file not found: {wheel_path}")

        if sdist_path and not sdist_path.exists():
            raise GiteaError(f"Source distribution file not found: {sdist_path}")

        files_to_upload = [wheel_path]
        if sdist_path:
            files_to_upload.append(sdist_path)

        if dry_run:
            print(f"[DRY RUN] Would upload to: {self.pypi_registry_url}")
            for file_path in files_to_upload:
                print(f"[DRY RUN] Would upload: {file_path}")
            return True

        success = True
        for file_path in files_to_upload:
            if not self._upload_file(file_path):
                success = False

        return success

    def _upload_file(self, file_path: Path) -> bool:
        """Upload a single file to the registry.

        Args:
            file_path: Path to file to upload

        Returns:
            True if upload successful, False otherwise
        """
        try:
            import requests

            # Upload using multipart form data (PyPI-compatible)
            with open(file_path, 'rb') as f:
                files = {
                    'content': (file_path.name, f, 'application/octet-stream')
                }
                headers = {
                    'Authorization': f'token {self.config.auth_token}'
                }

                upload_url = f"{self.pypi_registry_url}/simple/"
                response = requests.post(
                    upload_url,
                    headers=headers,
                    files=files,
                    timeout=60
                )

                if response.status_code in [200, 201, 409]:  # 409 = already exists
                    print(f"✅ Uploaded: {file_path.name}")
                    if response.status_code == 409:
                        print(f"   (already exists)")
                    return True
                else:
                    print(f"❌ Upload failed for {file_path.name}: {response.status_code} {response.text}")
                    return False

        except Exception as e:
            print(f"❌ Upload failed for {file_path.name}: {e}")
            return False

    def delete_package_version(self, package_name: str, version: str, dry_run: bool = False) -> bool:
        """Delete a specific version of a package.

        Args:
            package_name: Name of the package
            version: Version to delete
            dry_run: If True, show what would be done without deleting

        Returns:
            True if deletion successful, False otherwise
        """
        if not self.config.auth_token:
            raise GiteaError("Authentication token required for package deletion.")

        delete_url = f"{self.config.gitea_url}/api/v1/packages/{self.config.repo_owner}/pypi/{package_name}/{version}"

        if dry_run:
            print(f"[DRY RUN] Would delete: {package_name} v{version}")
            print(f"[DRY RUN] DELETE {delete_url}")
            return True

        try:
            import requests
            headers = {"Authorization": f"token {self.config.auth_token}"}
            response = requests.delete(delete_url, headers=headers, timeout=10)

            if response.status_code in [200, 204, 404]:  # 404 = already deleted
                print(f"✅ Deleted: {package_name} v{version}")
                return True
            else:
                print(f"❌ Delete failed: {response.status_code} {response.text}")
                return False

        except Exception as e:
            print(f"❌ Delete failed: {e}")
            return False

    def get_registry_info(self) -> Dict:
        """Get information about the package registry configuration.

        Returns:
            Dictionary with registry information
        """
        return {
            "gitea_url": self.config.gitea_url,
            "repo_owner": self.config.repo_owner,
            "repo_name": self.config.repo_name,
            "pypi_registry_url": self.pypi_registry_url,
            "package_list_url": self.package_list_url,
            "auth_configured": bool(self.config.auth_token),
            "auth_valid": self.check_auth() if self.config.auth_token else False
        }


def configure_pip_for_gitea(config: Optional[GiteaConfig] = None,
                           pip_conf_path: Optional[Path] = None) -> Path:
    """Configure pip to use Gitea package registry as additional index.

    Args:
        config: Gitea configuration
        pip_conf_path: Custom path for pip.conf file

    Returns:
        Path to created/updated pip.conf file
    """
    config = config or GiteaConfig.from_git_repository()

    if pip_conf_path is None:
        # Default pip config location
        pip_conf_path = Path.home() / ".pip" / "pip.conf"

    pip_conf_path.parent.mkdir(parents=True, exist_ok=True)

    registry = GiteaPackageRegistry(config)
    gitea_index = f"{registry.pypi_registry_url}/simple/"

    # Read existing config or create new
    config_content = ""
    if pip_conf_path.exists():
        config_content = pip_conf_path.read_text()

    # Add Gitea index if not already present
    if "extra-index-url" not in config_content:
        if "[global]" not in config_content:
            config_content = "[global]\n" + config_content

        lines = config_content.split('\n')
        global_section_idx = next(i for i, line in enumerate(lines) if line.strip() == "[global]")
        lines.insert(global_section_idx + 1, f"extra-index-url = {gitea_index}")
        config_content = '\n'.join(lines)
    elif gitea_index not in config_content:
        # Add to existing extra-index-url
        lines = config_content.split('\n')
        for i, line in enumerate(lines):
            if line.startswith("extra-index-url"):
                lines[i] = f"{line} {gitea_index}"
                break
        config_content = '\n'.join(lines)

    pip_conf_path.write_text(config_content)
    return pip_conf_path