devx/src/devx/tools/create_pr.py

#!/usr/bin/env python3
"""Create a pull request with the correct title from the Vikunja task.

This tool is run **after** pushing a feature branch.  It:

1. Extracts the task ID from the branch name (e.g. ``DEVX-31-fix-foo`` → ``DEVX-31``).
2. Fetches the Vikunja task title for that task ID.
3. Creates a PR with title ``{TASK_PREFIX}-N: <vikunja task title>``.

This eliminates manual PR title entry and ensures the title always
matches the Vikunja task — which is what the auto-merge workflow
validates.

If a PR already exists for the branch, the tool prints its URL and
exits successfully (idempotent).

Usage::

    python -m devx.tools.create_pr --branch DEVX-31-fix-foo

The repository is auto-detected from ``DEVX_REPO_OWNER`` /
``DEVX_REPO_NAME`` or ``GITHUB_REPOSITORY`` environment variables.
"""

from __future__ import annotations

import os
import subprocess  # nosec B404

import click
from dotenv import load_dotenv

from devx.api_clients import GiteaClient, VikunjaClient
from devx.config import (
    DEFAULT_PER_PAGE,
    GITEA_API_URL,
    REPO_NAME,
    REPO_OWNER,
    TASK_ID_RE,
    TASK_PREFIX,
    VIKUNJA_API_URL,
    VIKUNJA_PROJECT_ID,
)
from devx.i18n import _
from devx.tokens import get_developer_token, get_vikunja_token

load_dotenv()


def get_repo_name() -> str:
    """Auto-detect repository name from env vars, pyproject.toml, or git remote."""
    name = os.environ.get("DEVX_REPO_NAME", "")
    if name:
        return name
    github_repo = os.environ.get("GITHUB_REPOSITORY", "")
    if github_repo and "/" in github_repo:
        return github_repo.split("/", 1)[1]
    if REPO_NAME:
        return REPO_NAME
    raise click.ClickException(
        _("Repository name not set. Use DEVX_REPO_NAME, [tool.devx] repo_name, or GITHUB_REPOSITORY env var."),
    )


def extract_task_id(branch: str) -> str:
    """Extract the task ID (e.g. ``DEVX-31``) from a branch name."""
    match = TASK_ID_RE.search(branch)
    return match.group(0) if match else ""


def get_vikunja_task_title(task_id: str) -> str:
    """Fetch the Vikunja task title for the given task identifier.

    Raises ClickException if VIKUNJA_TOKEN is not set or the task is not found.
    """
    try:
        token = get_vikunja_token()
    except click.ClickException:
        raise click.ClickException(_("VIKUNJA_TOKEN is not set. Required to derive PR title.")) from None
    client = VikunjaClient(VIKUNJA_API_URL, token)
    task = client.find_task_by_identifier(VIKUNJA_PROJECT_ID, task_id, per_page=DEFAULT_PER_PAGE)
    if not task:
        raise click.ClickException(
            _(
                "Could not find Vikunja task {task_id} in project {project_id}.",
                task_id=task_id,
                project_id=VIKUNJA_PROJECT_ID,
            ),
        )
    return str(task.get("title", ""))


def find_existing_pr(client: GiteaClient, branch: str) -> dict | None:
    """Return an existing open PR for the branch, or None."""
    prs = client.list_prs(state="open")
    for pr in prs:
        if pr.get("head", {}).get("ref") == branch:
            return pr
    return None


def create_pr(
    branch: str,
    base: str,
    body: str,
    repo_owner: str,
    repo_name: str,
) -> dict:
    """Create a PR with the title derived from the Vikunja task.

    Returns the PR dict from the Gitea API.
    """
    task_id = extract_task_id(branch)
    if not task_id:
        raise click.ClickException(
            _(
                "Branch '{branch}' does not contain a task ID.\n  Expected format: {prefix}-N-short-description",
                branch=branch,
                prefix=TASK_PREFIX,
            ),
        )

    try:
        token = get_developer_token()
    except click.ClickException:
        raise click.ClickException(_("CI_GITEA_TOKEN is not set. Required to create a PR.")) from None

    vikunja_title = get_vikunja_task_title(task_id)
    pr_title = f"{task_id}: {vikunja_title}"

    client = GiteaClient(GITEA_API_URL, token, repo_owner, repo_name)

    existing = find_existing_pr(client, branch)
    if existing:
        click.echo(
            _(
                "PR already exists: #{index} — {url}",
                index=existing.get("number", "?"),
                url=existing.get("html_url", ""),
            ),
        )
        return existing

    pr = client.create_pr(title=pr_title, head=branch, base=base, body=body)
    click.echo(
        _(
            "Created PR #{index}: {title}\n  {url}",
            index=pr.get("number", "?"),
            title=pr_title,
            url=pr.get("html_url", ""),
        ),
    )
    return pr


@click.command()
@click.option("--branch", default=None, help="Head branch (default: auto-detect from git).")
@click.option("--base", default="master", show_default=True, help="Base branch.")
@click.option("--body", default="", help="PR body (markdown). Read from stdin if '-' is passed.")
@click.option("--owner", default=None, help="Repository owner (default: DEVX_REPO_OWNER).")
@click.option("--repo", default=None, help="Repository name (default: DEVX_REPO_NAME or GITHUB_REPOSITORY).")
def cli(branch: str | None, base: str, body: str, owner: str | None, repo: str | None) -> None:
    """Create a PR with the correct title from the Vikunja task."""
    if branch is None:
        result = subprocess.run(  # nosec
            ["git", "rev-parse", "--abbrev-ref", "HEAD"],
            capture_output=True,
            text=True,
            check=False,
        )
        if result.returncode != 0:
            raise click.ClickException(
                _("Could not detect current branch: {error}", error=result.stderr.strip()),
            )
        branch = result.stdout.strip()

    if body == "-":
        body = click.get_text_stream("stdin").read().strip()

    repo_owner = owner or REPO_OWNER
    if not repo_owner:
        raise click.ClickException(_("Repository owner not set. Use --owner or DEVX_REPO_OWNER env var."))
    repo_name = repo or get_repo_name()

    create_pr(branch, base, body, repo_owner, repo_name)


if __name__ == "__main__":  # pragma: no cover
    cli()  # pragma: no cover