diff --git a/Makefile b/Makefile index 0073c29..e9a6a4b 100644 --- a/Makefile +++ b/Makefile @@ -44,7 +44,7 @@ install: # ------------------------------------------------------------ # Default: keep current auto-detection behavior -setup: setup-nix setup-venv +setup: setup-venv # Explicit: developer setup (Python venv + shell RC + install) setup-venv: setup-nix diff --git a/src/pkgmgr/cli/commands/version.py b/src/pkgmgr/cli/commands/version.py index cdf05c0..e739202 100644 --- a/src/pkgmgr/cli/commands/version.py +++ b/src/pkgmgr/cli/commands/version.py @@ -9,8 +9,13 @@ from pkgmgr.core.repository.dir import get_repo_dir from pkgmgr.core.repository.identifier import get_repo_identifier from pkgmgr.core.git import get_tags from pkgmgr.core.version.semver import SemVer, find_latest_version +from pkgmgr.core.version.installed import ( + get_installed_python_version, + get_installed_nix_profile_version, +) from pkgmgr.core.version.source import ( read_pyproject_version, + read_pyproject_project_name, read_flake_version, read_pkgbuild_version, read_debian_changelog_version, @@ -18,10 +23,54 @@ from pkgmgr.core.version.source import ( read_ansible_galaxy_version, ) - Repository = Dict[str, Any] +def _print_pkgmgr_self_version() -> None: + """ + Print version information for pkgmgr itself (installed env + nix profile), + used when no repository is selected (e.g. user is not inside a repo). + """ + print("pkgmgr version info") + print("====================") + print("\nRepository: ") + print("----------------------------------------") + + # Common distribution/module naming variants. + python_candidates = [ + "package-manager", # PyPI dist name in your project + "package_manager", # module-ish variant + "pkgmgr", # console/alias-ish + ] + nix_candidates = [ + "pkgmgr", + "package-manager", + ] + + installed_python = get_installed_python_version(*python_candidates) + installed_nix = get_installed_nix_profile_version(*nix_candidates) + + if installed_python: + print( + f"Installed (Python env): {installed_python.version} " + f"(dist: {installed_python.name})" + ) + else: + print("Installed (Python env): ") + + if installed_nix: + print( + f"Installed (Nix profile): {installed_nix.version} " + f"(match: {installed_nix.name})" + ) + else: + print("Installed (Nix profile): ") + + # Helpful context for debugging "why do versions differ?" + print(f"Python executable: {sys.executable}") + print(f"Python prefix: {sys.prefix}") + + def handle_version( args, ctx: CLIContext, @@ -30,20 +79,39 @@ def handle_version( """ Handle the 'version' command. - Shows version information from various sources (git tags, pyproject, - flake.nix, PKGBUILD, debian, spec, Ansible Galaxy). - """ + Shows version information from: + - Git tags + - packaging metadata + - installed Python environment + - installed Nix profile - repo_list = selected - if not repo_list: - print("No repositories selected for version.") - sys.exit(1) + Special case: + - If no repositories are selected (e.g. not in a repo and no identifiers), + print pkgmgr's own installed versions instead of exiting with an error. + """ + if not selected: + _print_pkgmgr_self_version() + return print("pkgmgr version info") print("====================") - for repo in repo_list: - # Resolve repository directory + for repo in selected: + identifier = get_repo_identifier(repo, ctx.all_repositories) + + python_candidates: list[str] = [] + nix_candidates: list[str] = [identifier] + + for key in ("pypi", "pip", "python_package", "distribution", "package"): + val = repo.get(key) + if isinstance(val, str) and val.strip(): + python_candidates.append(val.strip()) + + python_candidates.append(identifier) + + installed_python = get_installed_python_version(*python_candidates) + installed_nix = get_installed_nix_profile_version(*nix_candidates) + repo_dir = repo.get("directory") if not repo_dir: try: @@ -51,51 +119,79 @@ def handle_version( except Exception: repo_dir = None - # If no local clone exists, skip gracefully with info message if not repo_dir or not os.path.isdir(repo_dir): - identifier = get_repo_identifier(repo, ctx.all_repositories) print(f"\nRepository: {identifier}") print("----------------------------------------") print( - "[INFO] Skipped: repository directory does not exist " - "locally, version detection is not possible." + "[INFO] Skipped: repository directory does not exist locally, " + "version detection is not possible." ) + + if installed_python: + print( + f"Installed (Python env): {installed_python.version} " + f"(dist: {installed_python.name})" + ) + else: + print("Installed (Python env): ") + + if installed_nix: + print( + f"Installed (Nix profile): {installed_nix.version} " + f"(match: {installed_nix.name})" + ) + else: + print("Installed (Nix profile): ") + continue print(f"\nRepository: {repo_dir}") print("----------------------------------------") - # 1) Git tags (SemVer) try: tags = get_tags(cwd=repo_dir) except Exception as exc: print(f"[ERROR] Could not read git tags: {exc}") tags = [] - latest_tag_info: Optional[Tuple[str, SemVer]] - latest_tag_info = find_latest_version(tags) if tags else None + latest_tag_info: Optional[Tuple[str, SemVer]] = ( + find_latest_version(tags) if tags else None + ) - if latest_tag_info is None: - latest_tag_str = None - latest_ver = None + if latest_tag_info: + tag, ver = latest_tag_info + print(f"Git (latest SemVer tag): {tag} (parsed: {ver})") else: - latest_tag_str, latest_ver = latest_tag_info + print("Git (latest SemVer tag): ") - # 2) Packaging / metadata sources pyproject_version = read_pyproject_version(repo_dir) + pyproject_name = read_pyproject_project_name(repo_dir) flake_version = read_flake_version(repo_dir) pkgbuild_version = read_pkgbuild_version(repo_dir) debian_version = read_debian_changelog_version(repo_dir) spec_version = read_spec_version(repo_dir) ansible_version = read_ansible_galaxy_version(repo_dir) - # 3) Print version summary - if latest_ver is not None: + if pyproject_name: + installed_python = get_installed_python_version( + pyproject_name, *python_candidates + ) + + if installed_python: print( - f"Git (latest SemVer tag): {latest_tag_str} (parsed: {latest_ver})" + f"Installed (Python env): {installed_python.version} " + f"(dist: {installed_python.name})" ) else: - print("Git (latest SemVer tag): ") + print("Installed (Python env): ") + + if installed_nix: + print( + f"Installed (Nix profile): {installed_nix.version} " + f"(match: {installed_nix.name})" + ) + else: + print("Installed (Nix profile): ") print(f"pyproject.toml: {pyproject_version or ''}") print(f"flake.nix: {flake_version or ''}") @@ -104,15 +200,16 @@ def handle_version( print(f"package-manager.spec: {spec_version or ''}") print(f"Ansible Galaxy meta: {ansible_version or ''}") - # 4) Consistency hint (Git tag vs. pyproject) - if latest_ver is not None and pyproject_version is not None: + if latest_tag_info and pyproject_version: try: file_ver = SemVer.parse(pyproject_version) - if file_ver != latest_ver: + if file_ver != latest_tag_info[1]: print( - f"[WARN] Version mismatch: Git={latest_ver}, pyproject={file_ver}" + f"[WARN] Version mismatch: " + f"Git={latest_tag_info[1]}, pyproject={file_ver}" ) except ValueError: print( - f"[WARN] pyproject version {pyproject_version!r} is not valid SemVer." + f"[WARN] pyproject version {pyproject_version!r} " + f"is not valid SemVer." ) diff --git a/src/pkgmgr/core/version/installed.py b/src/pkgmgr/core/version/installed.py new file mode 100644 index 0000000..72b3ecd --- /dev/null +++ b/src/pkgmgr/core/version/installed.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +import json +import re +import shutil +import subprocess +from dataclasses import dataclass +from typing import Iterable, Optional, Tuple + + +@dataclass(frozen=True) +class InstalledVersion: + """ + Represents a resolved installed version and the matched name. + """ + name: str + version: str + + +def _normalize(name: str) -> str: + return re.sub(r"[-_.]+", "-", (name or "").strip()).lower() + + +def _unique_candidates(names: Iterable[str]) -> list[str]: + seen: set[str] = set() + out: list[str] = [] + for n in names: + if not n: + continue + key = _normalize(n) + if key in seen: + continue + seen.add(key) + out.append(n) + return out + + +def get_installed_python_version(*candidates: str) -> Optional[InstalledVersion]: + """ + Detect installed Python package version in the CURRENT Python environment. + + Strategy: + 1) Exact normalized match using importlib.metadata.version() + 2) Substring fallback by scanning installed distributions + """ + try: + from importlib import metadata as importlib_metadata + except Exception: + return None + + candidates = _unique_candidates(candidates) + + expanded: list[str] = [] + for c in candidates: + n = _normalize(c) + expanded.extend([c, n, n.replace("-", "_"), n.replace("-", ".")]) + expanded = _unique_candidates(expanded) + + # 1) Direct queries first (fast path) + for name in expanded: + try: + version = importlib_metadata.version(name) + return InstalledVersion(name=name, version=version) + except Exception: + continue + + # 2) Fallback: scan distributions (last resort) + try: + dists = importlib_metadata.distributions() + except Exception: + return None + + norm_candidates = {_normalize(c) for c in candidates} + + for dist in dists: + dist_name = dist.metadata.get("Name", "") or "" + norm_dist = _normalize(dist_name) + for c in norm_candidates: + if c and (c in norm_dist or norm_dist in c): + ver = getattr(dist, "version", None) + if ver: + return InstalledVersion(name=dist_name, version=ver) + + return None + + +def _run_nix(args: list[str]) -> Tuple[int, str, str]: + p = subprocess.run( + args, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=False, + ) + return p.returncode, p.stdout or "", p.stderr or "" + + +def _extract_version_from_store_path(path: str) -> Optional[str]: + if not path: + return None + base = path.rstrip("/").split("/")[-1] + if "-" not in base: + return None + tail = base.split("-")[-1] + if re.match(r"\d+(\.\d+){0,3}([a-z0-9+._-]*)?$", tail, re.I): + return tail + return None + + +def get_installed_nix_profile_version(*candidates: str) -> Optional[InstalledVersion]: + """ + Detect installed version from the current Nix profile. + + Strategy: + 1) JSON output (exact normalized match) + 2) Text fallback (substring) + """ + if shutil.which("nix") is None: + return None + + candidates = _unique_candidates(candidates) + if not candidates: + return None + + norm_candidates = {_normalize(c) for c in candidates} + + # Preferred: JSON output + rc, out, _ = _run_nix(["nix", "profile", "list", "--json"]) + if rc == 0 and out.strip(): + try: + data = json.loads(out) + elements = data.get("elements") or data.get("items") or {} + if isinstance(elements, dict): + for elem in elements.values(): + if not isinstance(elem, dict): + continue + name = (elem.get("name") or elem.get("pname") or "").strip() + version = (elem.get("version") or "").strip() + norm_name = _normalize(name) + + if norm_name in norm_candidates: + if version: + return InstalledVersion(name=name, version=version) + for sp in elem.get("storePaths", []) or []: + guess = _extract_version_from_store_path(sp) + if guess: + return InstalledVersion(name=name, version=guess) + except Exception: + pass + + # Fallback: text mode + rc, out, _ = _run_nix(["nix", "profile", "list"]) + if rc != 0: + return None + + for line in out.splitlines(): + norm_line = _normalize(line) + for c in norm_candidates: + if c in norm_line: + m = re.search(r"\b\d+(\.\d+){0,3}[a-z0-9+._-]*\b", line, re.I) + if m: + return InstalledVersion(name=c, version=m.group(0)) + if "/nix/store/" in line: + guess = _extract_version_from_store_path(line.split()[-1]) + if guess: + return InstalledVersion(name=c, version=guess) + + return None diff --git a/src/pkgmgr/core/version/source.py b/src/pkgmgr/core/version/source.py index 97823a0..194b497 100644 --- a/src/pkgmgr/core/version/source.py +++ b/src/pkgmgr/core/version/source.py @@ -1,21 +1,3 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- - -""" -Helpers to extract version information from various packaging files. - -All functions take a repository directory and return either a version -string or None if the corresponding file or version field is missing. - -Supported sources: -- pyproject.toml (PEP 621, [project].version) -- flake.nix (version = "X.Y.Z";) -- PKGBUILD (pkgver / pkgrel) -- debian/changelog (first entry line: package (version) ...) -- RPM spec file (package-manager.spec: Version / Release) -- Ansible Galaxy (galaxy.yml or meta/main.yml) -""" - from __future__ import annotations import os @@ -30,46 +12,61 @@ def read_pyproject_version(repo_dir: str) -> Optional[str]: Read the version from pyproject.toml in repo_dir, if present. Expects a PEP 621-style [project] table with a 'version' field. - Returns the version string or None. """ path = os.path.join(repo_dir, "pyproject.toml") - if not os.path.exists(path): + if not os.path.isfile(path): return None try: - try: - import tomllib # Python 3.11+ - except ModuleNotFoundError: # pragma: no cover - tomllib = None - - if tomllib is None: - return None + import tomllib # Python 3.11+ + except Exception: + import tomli as tomllib # type: ignore + try: with open(path, "rb") as f: data = tomllib.load(f) - - project = data.get("project", {}) - if isinstance(project, dict): - version = project.get("version") - if isinstance(version, str): - return version.strip() or None + project = data.get("project") or {} + version = project.get("version") + return str(version).strip() if version else None except Exception: - # Intentionally swallow errors and fall back to None. return None - return None + +def read_pyproject_project_name(repo_dir: str) -> Optional[str]: + """ + Read distribution name from pyproject.toml ([project].name). + + This is required to correctly resolve installed Python package + versions via importlib.metadata. + """ + path = os.path.join(repo_dir, "pyproject.toml") + if not os.path.isfile(path): + return None + + try: + import tomllib # Python 3.11+ + except Exception: + import tomli as tomllib # type: ignore + + try: + with open(path, "rb") as f: + data = tomllib.load(f) + project = data.get("project") or {} + name = project.get("name") + return str(name).strip() if name else None + except Exception: + return None def read_flake_version(repo_dir: str) -> Optional[str]: """ Read the version from flake.nix in repo_dir, if present. - Looks for a line like: - version = "1.2.3"; - and returns the string inside the quotes. + Looks for: + version = "X.Y.Z"; """ path = os.path.join(repo_dir, "flake.nix") - if not os.path.exists(path): + if not os.path.isfile(path): return None try: @@ -81,22 +78,21 @@ def read_flake_version(repo_dir: str) -> Optional[str]: match = re.search(r'version\s*=\s*"([^"]+)"', text) if not match: return None - version = match.group(1).strip() - return version or None + + return match.group(1).strip() or None def read_pkgbuild_version(repo_dir: str) -> Optional[str]: """ - Read the version from PKGBUILD in repo_dir, if present. + Read the version from PKGBUILD in repo_dir. - Expects: - pkgver=1.2.3 - pkgrel=1 - - Returns either "1.2.3-1" (if both are present) or just "1.2.3". + Combines pkgver and pkgrel if both exist: + pkgver=1.2.3 + pkgrel=1 + -> 1.2.3-1 """ path = os.path.join(repo_dir, "PKGBUILD") - if not os.path.exists(path): + if not os.path.isfile(path): return None try: @@ -121,15 +117,13 @@ def read_pkgbuild_version(repo_dir: str) -> Optional[str]: def read_debian_changelog_version(repo_dir: str) -> Optional[str]: """ - Read the latest Debian version from debian/changelog in repo_dir, if present. + Read the latest version from debian/changelog. - The first non-empty line typically looks like: - package-name (1.2.3-1) unstable; urgency=medium - - We extract the text inside the first parentheses. + Expected format: + package (1.2.3-1) unstable; urgency=medium """ path = os.path.join(repo_dir, "debian", "changelog") - if not os.path.exists(path): + if not os.path.isfile(path): return None try: @@ -140,8 +134,7 @@ def read_debian_changelog_version(repo_dir: str) -> Optional[str]: continue match = re.search(r"\(([^)]+)\)", line) if match: - version = match.group(1).strip() - return version or None + return match.group(1).strip() or None break except Exception: return None @@ -151,81 +144,72 @@ def read_debian_changelog_version(repo_dir: str) -> Optional[str]: def read_spec_version(repo_dir: str) -> Optional[str]: """ - Read the version from a RPM spec file. + Read the version from an RPM spec file. - For now, we assume a fixed file name 'package-manager.spec' - in repo_dir with lines like: - - Version: 1.2.3 - Release: 1%{?dist} - - Returns either "1.2.3-1" (if Release is present) or "1.2.3". - Any RPM macro suffix like '%{?dist}' is stripped from the release. + Combines: + Version: 1.2.3 + Release: 1%{?dist} + -> 1.2.3-1 """ - path = os.path.join(repo_dir, "package-manager.spec") - if not os.path.exists(path): - return None + for fn in os.listdir(repo_dir): + if not fn.endswith(".spec"): + continue - try: - with open(path, "r", encoding="utf-8") as f: - text = f.read() - except Exception: - return None + path = os.path.join(repo_dir, fn) + try: + with open(path, "r", encoding="utf-8") as f: + text = f.read() + except Exception: + return None - ver_match = re.search(r"^Version:\s*(.+)$", text, re.MULTILINE) - if not ver_match: - return None - version = ver_match.group(1).strip() + ver_match = re.search(r"^Version:\s*(.+)$", text, re.MULTILINE) + if not ver_match: + return None + version = ver_match.group(1).strip() - rel_match = re.search(r"^Release:\s*(.+)$", text, re.MULTILINE) - if rel_match: - release_raw = rel_match.group(1).strip() - # Strip common RPM macro suffix like %... (e.g. 1%{?dist}) - release = release_raw.split("%", 1)[0].strip() - # Also strip anything after first whitespace, just in case - release = release.split(" ", 1)[0].strip() - if release: - return f"{version}-{release}" + rel_match = re.search(r"^Release:\s*(.+)$", text, re.MULTILINE) + if rel_match: + release_raw = rel_match.group(1).strip() + release = release_raw.split("%", 1)[0].split(" ", 1)[0].strip() + if release: + return f"{version}-{release}" - return version or None + return version or None + + return None def read_ansible_galaxy_version(repo_dir: str) -> Optional[str]: """ - Read the version from Ansible Galaxy metadata, if present. + Read the version from Ansible Galaxy metadata. - Supported locations: - - galaxy.yml (preferred for modern roles/collections) - - meta/main.yml (legacy style roles; uses galaxy_info.version or version) + Supported: + - galaxy.yml + - meta/main.yml (galaxy_info.version or version) """ - # 1) galaxy.yml in repo root - galaxy_path = os.path.join(repo_dir, "galaxy.yml") - if os.path.exists(galaxy_path): + galaxy_yml = os.path.join(repo_dir, "galaxy.yml") + if os.path.isfile(galaxy_yml): try: - with open(galaxy_path, "r", encoding="utf-8") as f: + with open(galaxy_yml, "r", encoding="utf-8") as f: data = yaml.safe_load(f) or {} version = data.get("version") if isinstance(version, str) and version.strip(): return version.strip() except Exception: - # Ignore parse errors and fall through to meta/main.yml pass - # 2) meta/main.yml (classic Ansible role) - meta_path = os.path.join(repo_dir, "meta", "main.yml") - if os.path.exists(meta_path): + meta_yml = os.path.join(repo_dir, "meta", "main.yml") + if os.path.isfile(meta_yml): try: - with open(meta_path, "r", encoding="utf-8") as f: + with open(meta_yml, "r", encoding="utf-8") as f: data = yaml.safe_load(f) or {} - # Preferred: galaxy_info.version galaxy_info = data.get("galaxy_info") or {} if isinstance(galaxy_info, dict): version = galaxy_info.get("version") if isinstance(version, str) and version.strip(): return version.strip() - # Fallback: top-level 'version' version = data.get("version") if isinstance(version, str) and version.strip(): return version.strip()