1from __future__ import annotations
  2
  3import shlex
  4import subprocess
  5import sys
  6import tomllib
  7from pathlib import Path
  8
  9import click
 10
 11LOCK_FILE = Path("uv.lock")
 12
 13
 14@click.command()
 15@click.argument("packages", nargs=-1)
 16@click.option(
 17    "--diff", is_flag=True, help="Read versions from unstaged uv.lock changes"
 18)
 19@click.option(
 20    "--agent-command",
 21    envvar="PLAIN_UPGRADE_AGENT_COMMAND",
 22    help="Run command with generated prompt",
 23)
 24def cli(
 25    packages: tuple[str, ...], diff: bool, agent_command: str | None = None
 26) -> None:
 27    """Generate an upgrade prompt for plain packages."""
 28    require_git_repo()
 29
 30    if not packages:
 31        click.secho("Getting installed packages...", bold=True, err=True)
 32        packages = tuple(sorted(get_installed_plain_packages()))
 33        for pkg in packages:
 34            click.secho(f"- {click.style(pkg, fg='yellow')}", err=True)
 35        click.echo(err=True)
 36
 37    if not packages:
 38        raise click.UsageError("No plain packages found or specified.")
 39
 40    if diff:
 41        before_after = versions_from_diff(packages)
 42    else:
 43        before_after = upgrade_packages(packages)
 44
 45    # Remove all packages that were not upgraded
 46    before_after = {
 47        pkg: versions
 48        for pkg, versions in before_after.items()
 49        if versions[0] != versions[1]
 50    }
 51
 52    if not before_after:
 53        click.secho(
 54            "No packages were upgraded. If uv.lock has already been updated, use --diff instead.",
 55            fg="green",
 56            err=True,
 57        )
 58        return
 59
 60    prompt = build_prompt(before_after)
 61
 62    if agent_command:
 63        cmd = shlex.split(agent_command)
 64        cmd.append(prompt)
 65        result = subprocess.run(cmd, check=False)
 66        if result.returncode != 0:
 67            click.secho(
 68                f"Agent command failed with exit code {result.returncode}",
 69                fg="red",
 70                err=True,
 71            )
 72    else:
 73        click.secho(
 74            "\nCopy this prompt to a coding agent. To run an agent automatically, use --agent-command or set the PLAIN_UPGRADE_AGENT_COMMAND environment variable.\n",
 75            dim=True,
 76            italic=True,
 77            err=True,
 78        )
 79        click.echo(prompt)
 80
 81
 82def require_git_repo() -> None:
 83    result = subprocess.run(
 84        ["git", "rev-parse", "--is-inside-work-tree"], capture_output=True, text=True
 85    )
 86    if result.returncode != 0 or result.stdout.strip() != "true":
 87        raise click.UsageError("This command must be run inside a git repository")
 88
 89
 90def get_installed_plain_packages() -> list[str]:
 91    lock_text = LOCK_FILE.read_text()
 92    data = tomllib.loads(lock_text)
 93    names: list[str] = []
 94    for pkg in data.get("package", []):
 95        name = pkg.get("name", "")
 96        if name.startswith("plain") and name != "plain-upgrade":
 97            names.append(name)
 98    return names
 99
100
101def parse_lock_versions(lock_text: str, packages: set[str]) -> dict[str, str]:
102    data = tomllib.loads(lock_text)
103    versions: dict[str, str] = {}
104    for pkg in data.get("package", []):
105        name = pkg.get("name")
106        if name in packages:
107            versions[name] = pkg.get("version")
108    return versions
109
110
111def versions_from_diff(
112    packages: tuple[str, ...],
113) -> dict[str, tuple[str | None, str | None]]:
114    result = subprocess.run(
115        ["git", "status", "--porcelain", str(LOCK_FILE)], capture_output=True, text=True
116    )
117    if not result.stdout.strip():
118        raise click.UsageError(
119            "--diff specified but uv.lock has no uncommitted changes"
120        )
121
122    prev_text = subprocess.run(
123        ["git", "show", f"HEAD:{LOCK_FILE}"], capture_output=True, text=True, check=True
124    ).stdout
125    current_text = LOCK_FILE.read_text()
126
127    packages_set = set(packages)
128    before = parse_lock_versions(prev_text, packages_set)
129    after = parse_lock_versions(current_text, packages_set)
130
131    return {pkg: (before.get(pkg), after.get(pkg)) for pkg in packages}
132
133
134def upgrade_packages(
135    packages: tuple[str, ...],
136) -> dict[str, tuple[str | None, str | None]]:
137    before = parse_lock_versions(LOCK_FILE.read_text(), set(packages))
138
139    upgrade_args = ["uv", "sync"]
140    for pkg in packages:
141        upgrade_args.extend(["--upgrade-package", pkg])
142
143    click.secho("Upgrading with uv sync...", bold=True, err=True)
144    subprocess.run(upgrade_args, check=True, stdout=sys.stderr)
145    click.echo(err=True)
146
147    after = parse_lock_versions(LOCK_FILE.read_text(), set(packages))
148    return {pkg: (before.get(pkg), after.get(pkg)) for pkg in packages}
149
150
151def build_prompt(before_after: dict[str, tuple[str | None, str | None]]) -> str:
152    lines = [
153        "These packages have been updated and may require additional changes to the code:",
154        "",
155    ]
156    for pkg, (before, after) in before_after.items():
157        lines.append(f"- {pkg}: {before} -> {after}")
158    lines.extend(
159        [
160            "",
161            'For each package run `plain-changelog {package} --from {before} --to {after}` and read the "Upgrade instructions" for what to do.',
162            "Work through each package in order, making all necessary code changes (if applicable).",
163            "When ALL of the package upgrade steps have been completed, you can check the results with `plain preflight` and `plain test`.",
164            "Do not commit any changes.",
165        ]
166    )
167    return "\n".join(lines)