Gradle-mirror/gradle_mirror/mirror_manager.py

"""Core business logic for activating/deactivating Gradle mirrors."""

from __future__ import annotations

import logging
import re
from dataclasses import dataclass
from pathlib import Path

from .constants import (
    BACKUP_SUFFIX,
    END_MARKER,
    INIT_FILE_NAME,
    START_MARKER,
)
from .utils import read_text_if_exists, write_text


@dataclass(frozen=True)
class OperationResult:
    """Result object returned by mirror operations."""

    changed: bool
    message: str
    init_file: Path
    backup_file: Path


class MirrorManager:
    """Manage insertion/removal of mirror repositories in init.gradle.kts."""

    def __init__(self, gradle_home: Path, logger: logging.Logger) -> None:
        self.gradle_home = gradle_home
        self.logger = logger
        self.init_file = self.gradle_home / INIT_FILE_NAME
        self.backup_file = self.gradle_home / f"{INIT_FILE_NAME}{BACKUP_SUFFIX}"

    def activate(self) -> OperationResult:
        """Activate global repository mirrors via init.gradle.kts."""

        self.gradle_home.mkdir(parents=True, exist_ok=True)
        existing = read_text_if_exists(self.init_file)

        if self.init_file.exists():
            # Keep the original state so deactivation can restore it exactly.
            write_text(self.backup_file, existing)
            self.logger.info("Created backup at %s", self.backup_file)

        block = self._build_marked_block()
        updated = self._upsert_marked_block(existing, block)

        if updated == existing:
            self.logger.info("Mirror block already up-to-date: %s", self.init_file)
            return OperationResult(
                changed=False,
                message="Gradle mirror is already active",
                init_file=self.init_file,
                backup_file=self.backup_file,
            )

        write_text(self.init_file, updated)
        self.logger.info("Activated mirrors in %s", self.init_file)
        return OperationResult(
            changed=True,
            message="Gradle mirror activated",
            init_file=self.init_file,
            backup_file=self.backup_file,
        )

    def deactivate(self) -> OperationResult:
        """Deactivate mirror configuration and restore original behavior."""

        if self.backup_file.exists():
            backup_content = read_text_if_exists(self.backup_file)
            if backup_content.strip():
                write_text(self.init_file, backup_content)
                self.logger.info("Restored init script from backup %s", self.backup_file)
            else:
                if self.init_file.exists():
                    self.init_file.unlink()
                    self.logger.info("Deleted empty init script %s", self.init_file)
            self.backup_file.unlink(missing_ok=True)
            self.logger.info("Removed backup file %s", self.backup_file)
            return OperationResult(
                changed=True,
                message="Gradle mirror deactivated (restored from backup)",
                init_file=self.init_file,
                backup_file=self.backup_file,
            )

        if not self.init_file.exists():
            self.logger.info("Nothing to deactivate; %s does not exist", self.init_file)
            return OperationResult(
                changed=False,
                message="Nothing to deactivate",
                init_file=self.init_file,
                backup_file=self.backup_file,
            )

        current = read_text_if_exists(self.init_file)
        updated = self._remove_marked_block(current)

        if updated == current:
            self.logger.info("No mirror markers found in %s", self.init_file)
            return OperationResult(
                changed=False,
                message="Mirror markers were not found",
                init_file=self.init_file,
                backup_file=self.backup_file,
            )

        if updated.strip():
            write_text(self.init_file, updated)
            self.logger.info("Removed mirror block from %s", self.init_file)
        else:
            self.init_file.unlink(missing_ok=True)
            self.logger.info("Removed empty init script %s", self.init_file)

        return OperationResult(
            changed=True,
            message="Gradle mirror deactivated",
            init_file=self.init_file,
            backup_file=self.backup_file,
        )

    def _upsert_marked_block(self, content: str, marked_block: str) -> str:
        pattern = self._marker_pattern()

        if pattern.search(content):
            return pattern.sub(marked_block, content)

        base = content.rstrip()
        if base:
            return f"{base}\n\n{marked_block}\n"
        return f"{marked_block}\n"

    def _remove_marked_block(self, content: str) -> str:
        pattern = self._marker_pattern()
        removed = pattern.sub("", content)

        # Normalize repeated blank lines left by block removal.
        removed = re.sub(r"\n{3,}", "\n\n", removed)
        return removed.strip() + ("\n" if removed.strip() else "")

    def _marker_pattern(self) -> re.Pattern[str]:
        escaped_start = re.escape(START_MARKER)
        escaped_end = re.escape(END_MARKER)
        return re.compile(rf"{escaped_start}[\s\S]*?{escaped_end}", re.MULTILINE)

    def _build_marked_block(self) -> str:
        script_body = """allprojects {
    repositories {
        clear()

        maven { url = uri(\"https://maven.myket.ir/\") }
        maven { url = uri(\"https://mirror-maven.runflare.com/maven2/\") }
        maven { url = uri(\"https://mirror-maven.runflare.com/android/maven2/\") }
        maven { url = uri(\"https://mirror-maven.runflare.com/gradle-plugins/\") }

        mavenCentral()
        google()
    }
}

gradle.settingsEvaluated {
    settings.pluginManagement.repositories.apply {
        clear()

        maven { url = uri(\"https://maven.myket.ir/\") }
        maven { url = uri(\"https://mirror-maven.runflare.com/gradle-plugins/\") }

        mavenCentral()
        google()
    }
}"""
        return f"{START_MARKER}\n{script_body}\n{END_MARKER}"