recipe-maintainer/lib/recipe.py

"""Recipe version parsing, compose.yml reading, and metadata.

Handles the Co-op Cloud version format: <recipe-semver>+v<upstream-version>
Example: 0.2.6+v4.5.0
"""

import os
import re
import subprocess
from dataclasses import dataclass
from pathlib import Path

from lib.config import paths


# ---------------------------------------------------------------------------
# Version parsing
# ---------------------------------------------------------------------------

@dataclass
class RecipeVersion:
    """Parses and manipulates Co-op Cloud version strings.

    Format: <major>.<minor>.<patch>+v<upstream_version>
    Example: 0.2.6+v4.5.0
    """
    recipe_major: int
    recipe_minor: int
    recipe_patch: int
    upstream_version: str

    @classmethod
    def parse(cls, version_string: str) -> "RecipeVersion":
        """Parse a version string like '0.2.6+v4.5.0'."""
        # Split on '+v' to separate recipe semver from upstream
        if "+v" in version_string:
            recipe_part, upstream = version_string.split("+v", 1)
        elif "+" in version_string:
            recipe_part, upstream = version_string.split("+", 1)
            if upstream.startswith("v"):
                upstream = upstream[1:]
        else:
            recipe_part = version_string
            upstream = ""

        parts = recipe_part.split(".")
        if len(parts) != 3:
            raise ValueError(
                f"Invalid recipe version format: {version_string}"
            )
        return cls(
            recipe_major=int(parts[0]),
            recipe_minor=int(parts[1]),
            recipe_patch=int(parts[2]),
            upstream_version=upstream,
        )

    def bump(self, level: str) -> "RecipeVersion":
        """Return a new version with the specified level bumped.

        level: "patch", "minor", or "major"
        """
        if level == "patch":
            return RecipeVersion(
                self.recipe_major, self.recipe_minor,
                self.recipe_patch + 1, self.upstream_version,
            )
        elif level == "minor":
            return RecipeVersion(
                self.recipe_major, self.recipe_minor + 1,
                0, self.upstream_version,
            )
        elif level == "major":
            return RecipeVersion(
                self.recipe_major + 1, 0, 0,
                self.upstream_version,
            )
        else:
            raise ValueError(f"Invalid bump level: {level}")

    def with_upstream(self, upstream: str) -> "RecipeVersion":
        """Return a new version with a different upstream version."""
        return RecipeVersion(
            self.recipe_major, self.recipe_minor,
            self.recipe_patch, upstream,
        )

    def __str__(self) -> str:
        base = f"{self.recipe_major}.{self.recipe_minor}.{self.recipe_patch}"
        if self.upstream_version:
            return f"{base}+v{self.upstream_version}"
        return base


# ---------------------------------------------------------------------------
# Recipe directory operations
# ---------------------------------------------------------------------------

def recipe_dir(recipe: str) -> Path:
    """Return the abra recipe checkout directory."""
    return paths.ABRA_RECIPES_DIR / recipe


def compose_path(recipe: str) -> Path:
    """Return the path to a recipe's compose.yml."""
    return recipe_dir(recipe) / "compose.yml"


def has_local_changes(recipe: str) -> bool:
    """Check if a recipe checkout has uncommitted changes."""
    d = recipe_dir(recipe)
    if not d.exists():
        return False
    result = subprocess.run(
        ["git", "-C", str(d), "status", "--short"],
        capture_output=True, text=True, timeout=10,
    )
    return bool(result.stdout.strip())


def read_compose_version(recipe: str) -> RecipeVersion | None:
    """Read the version label from a recipe's compose.yml.

    Looks for: coop-cloud.${STACK_NAME}.version=X.Y.Z+vU.P.S
    """
    p = compose_path(recipe)
    if not p.exists():
        return None
    with open(p) as f:
        content = f.read()
    # Match the version label pattern
    m = re.search(r'coop-cloud\.\$\{STACK_NAME\}\.version=([^\s"]+)', content)
    if m:
        return RecipeVersion.parse(m.group(1))
    return None


def write_compose_version(recipe: str, version: RecipeVersion) -> None:
    """Update the version label in a recipe's compose.yml."""
    p = compose_path(recipe)
    with open(p) as f:
        content = f.read()
    content = re.sub(
        r'(coop-cloud\.\$\{STACK_NAME\}\.version=)[^\s"]+',
        rf'\g<1>{version}',
        content,
    )
    with open(p, "w") as f:
        f.write(content)


def read_compose_images(recipe: str) -> dict[str, str]:
    """Read service→image:tag mapping from compose.yml.

    Returns a dict like {"app": "cryptpad/cryptpad:version-2025.9.0"}.
    Simple regex-based parsing (not full YAML).
    """
    p = compose_path(recipe)
    if not p.exists():
        return {}

    images = {}
    current_service = None
    with open(p) as f:
        in_services = False
        indent_level = 0
        for line in f:
            stripped = line.strip()
            # Detect services: block
            if stripped == "services:":
                in_services = True
                continue
            if in_services:
                # Service name: indented once, ends with ":"
                if line.startswith("  ") and not line.startswith("    ") and stripped.endswith(":"):
                    current_service = stripped.rstrip(":")
                # Image line: indented under a service
                elif current_service and "image:" in stripped:
                    image = stripped.split("image:", 1)[1].strip()
                    images[current_service] = image
                # New top-level block ends services
                elif not line.startswith(" ") and stripped and stripped != "---":
                    in_services = False
                    current_service = None

    return images