"""Recipe version parsing, compose.yml reading, and metadata. Handles the Co-op Cloud version format: +v 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: ..+v 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