resolve_requirements.py

#!/usr/bin/env python3
# Copyright (C) 2016, 2018-2020 Bioinformatics Research Group (BioRG)
#                    Florida International University
# SPDX-License-Identifier: MIT
#
# Scans plugins/*/requirements.txt, detects version conflicts between
# plugins, and installs the merged set into a shared .venv at the
# project root.

import os
import re
import sys
import venv
import subprocess
from glob import glob
from collections import defaultdict
from packaging.requirements import Requirement, InvalidRequirement
from packaging.specifiers import SpecifierSet
from packaging.version import Version


VENV_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".venv")
MERGED_REQUIREMENTS = os.path.join(
    os.path.dirname(os.path.abspath(__file__)), "requirements-plugins.txt"
)


def discover_requirements(plugin_root="plugins"):
    """Return {plugin_name: [Requirement, ...]} for every plugin that ships a requirements.txt."""
    results = {}
    for req_file in sorted(glob(os.path.join(plugin_root, "*", "requirements.txt"))):
        plugin_name = os.path.basename(os.path.dirname(req_file))
        reqs = []
        with open(req_file) as fh:
            for lineno, raw in enumerate(fh, 1):
                line = raw.split("#")[0].strip()
                if not line or line.startswith("-"):
                    continue
                try:
                    reqs.append(Requirement(line))
                except InvalidRequirement as exc:
                    print(
                        f"WARNING: {req_file}:{lineno}: skipping unparseable "
                        f"requirement '{line}': {exc}",
                        file=sys.stderr,
                    )
        if reqs:
            results[plugin_name] = reqs
    return results


def _normalize(name):
    return re.sub(r"[-_.]+", "-", name).lower()


def check_conflicts(requirements_by_plugin):
    """Detect irreconcilable version conflicts. Returns a list of human-readable error strings."""
    pkg_sources = defaultdict(list)
    for plugin, reqs in requirements_by_plugin.items():
        for req in reqs:
            pkg_sources[_normalize(req.name)].append((plugin, req))

    errors = []
    for pkg, entries in sorted(pkg_sources.items()):
        if len(entries) < 2:
            continue

        pinned = {}
        for plugin, req in entries:
            for spec in req.specifier:
                if spec.operator == "==":
                    pinned.setdefault(spec.version, []).append(plugin)

        if len(pinned) > 1:
            lines = [f"  Conflicting pinned versions for '{pkg}':"]
            for ver, plugins in sorted(pinned.items()):
                lines.append(f"    =={ver}  (required by: {', '.join(plugins)})")
            errors.append("\n".join(lines))
            continue

        combined = SpecifierSet()
        contributing = []
        for plugin, req in entries:
            contributing.append((plugin, str(req.specifier) or "(any)"))
            combined &= req.specifier

        if not _specifier_set_satisfiable(combined):
            lines = [f"  Unsatisfiable combined specifiers for '{pkg}':"]
            for plugin, spec_str in contributing:
                lines.append(f"    {spec_str}  (required by: {plugin})")
            errors.append("\n".join(lines))

    return errors


def _specifier_set_satisfiable(spec_set):
    """Heuristic check: extract explicit bounds and verify lower <= upper."""
    if not str(spec_set):
        return True

    lower = None
    upper = None
    for spec in spec_set:
        try:
            ver = Version(spec.version)
        except Exception:
            continue
        if spec.operator in (">=", ">", "~=", "=="):
            if lower is None or ver > lower:
                lower = ver
        if spec.operator in ("<=", "<"):
            if upper is None or ver < upper:
                upper = ver

    if lower is not None and upper is not None:
        return lower <= upper
    return True


def merge_requirements(requirements_by_plugin):
    """Merge per-plugin requirements into a single {normalized_name: combined_specifier_string}."""
    merged = {}
    for plugin, reqs in requirements_by_plugin.items():
        for req in reqs:
            key = _normalize(req.name)
            if key not in merged:
                merged[key] = (req.name, SpecifierSet(str(req.specifier)))
            else:
                orig_name, existing = merged[key]
                merged[key] = (orig_name, existing & req.specifier)
    return merged


def write_merged(merged, path):
    """Write the merged requirements file."""
    with open(path, "w") as fh:
        fh.write("# Auto-generated by resolve_requirements.py — do not edit.\n")
        fh.write("# Merged from all plugins/*/requirements.txt files.\n\n")
        for key in sorted(merged):
            name, specifiers = merged[key]
            line = f"{name}{specifiers}\n" if str(specifiers) else f"{name}\n"
            fh.write(line)


def ensure_venv(venv_dir):
    """Create the shared venv if it does not exist."""
    if not os.path.isdir(venv_dir):
        print(f"[PluMA] Creating shared plugin venv at {venv_dir}")
        venv.create(venv_dir, with_pip=True, clear=False)
    else:
        print(f"[PluMA] Using existing plugin venv at {venv_dir}")


def pip_install(venv_dir, requirements_file):
    """Install the merged requirements into the venv."""
    if sys.platform == "win32":
        pip = os.path.join(venv_dir, "Scripts", "pip")
    else:
        pip = os.path.join(venv_dir, "bin", "pip")

    cmd = [pip, "install", "-r", requirements_file]
    print(f"[PluMA] Installing plugin requirements: {' '.join(cmd)}")
    result = subprocess.run(cmd, check=False)
    if result.returncode != 0:
        print(
            "ERROR: pip install failed. Check output above for details.",
            file=sys.stderr,
        )
        sys.exit(1)


def resolve_and_install(plugin_root="plugins"):
    """Top-level entry point used by SConstruct."""
    reqs = discover_requirements(plugin_root)
    if not reqs:
        print("[PluMA] No plugin requirements.txt files found — skipping venv setup.")
        return

    print(f"[PluMA] Found requirements.txt in {len(reqs)} plugin(s): {', '.join(sorted(reqs))}")

    conflicts = check_conflicts(reqs)
    if conflicts:
        print("\n" + "=" * 72, file=sys.stderr)
        print("FATAL: Plugin dependency conflicts detected!\n", file=sys.stderr)
        for msg in conflicts:
            print(msg, file=sys.stderr)
        print(
            "\nResolve these conflicts before building. Each plugin's "
            "requirements.txt must be mutually satisfiable.",
            file=sys.stderr,
        )
        print("=" * 72 + "\n", file=sys.stderr)
        sys.exit(1)

    merged = merge_requirements(reqs)
    write_merged(merged, MERGED_REQUIREMENTS)
    print(f"[PluMA] Wrote merged requirements to {MERGED_REQUIREMENTS}")

    ensure_venv(VENV_DIR)
    pip_install(VENV_DIR, MERGED_REQUIREMENTS)
    print("[PluMA] Plugin venv is ready.")


if __name__ == "__main__":
    plugin_root = sys.argv[1] if len(sys.argv) > 1 else "plugins"
    resolve_and_install(plugin_root)