Add new update-pkg command.

debian_todo/__init__.py

@@ -9,6 +9,8 @@ import json
 import re
 import glob
 import os
+import subprocess
+import sys
 from pathlib import Path
 from typing import Any, Optional, cast
 from urllib.request import urlopen
@@ -33,6 +35,7 @@ CACHE_PATH = Path(".vcs_git_cache.json")
 CACHE_VERSION = 4
 SourceInfo = dict[str, str]
 HIDE_UPLOADER = "Edward Betts <edward@4angle.com>"
+DEBIAN_SRC_BASE = Path.home() / "src" / "debian"
 
 
 def parse_details(details: str) -> tuple[str, Optional[str]]:
@@ -396,6 +399,138 @@ def update_todos() -> None:
     print_changes(filter_todo_list(old_list), filter_todo_list(todo_list))
 
 
+def team_slug_to_display_name(team_slug: str) -> str:
+    """Convert a team slug to a display name for directory structure.
+
+    homeassistant-team -> HA
+    python-team -> python
+    openstack-team -> openstack
+    """
+    if team_slug == "homeassistant-team":
+        return "HA"
+    if team_slug.endswith("-team"):
+        return team_slug[:-5]
+    return team_slug
+
+
+def has_uncommitted_changes(pkg_dir: Path) -> bool:
+    """Check if the git repository has uncommitted changes."""
+    result = subprocess.run(
+        ["git", "status", "--porcelain"],
+        cwd=pkg_dir,
+        capture_output=True,
+        text=True,
+    )
+    return bool(result.stdout.strip())
+
+
+def extract_upstream_version_from_git_log(pkg_dir: Path) -> Optional[str]:
+    """Extract upstream version from the latest git commit message."""
+    result = subprocess.run(
+        ["git", "log", "-1"],
+        cwd=pkg_dir,
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode != 0:
+        return None
+    match = re.search(
+        r"Update upstream source from tag 'upstream/(.+?)'", result.stdout
+    )
+    if not match:
+        return None
+    return match.group(1)
+
+
+def add_changelog_entry(pkg_dir: Path) -> bool:
+    """Add a new debian/changelog entry for the upstream version."""
+    version = extract_upstream_version_from_git_log(pkg_dir)
+    if not version:
+        print("Could not find upstream version in git log.", file=sys.stderr)
+        return False
+
+    new_version = f"{version}-1"
+    result = subprocess.run(
+        [
+            "dch",
+            "--release-heuristic",
+            "log",
+            "--newversion",
+            new_version,
+            "New upstream release.",
+        ],
+        cwd=pkg_dir,
+    )
+    return result.returncode == 0
+
+
+def update_package(package: str) -> None:
+    """Update a Debian package to a new upstream version.
+
+    Checks out the package if not already present, imports the new upstream
+    version using gbp, and creates a new changelog entry.
+    """
+    source_info_map = load_source_info_map()
+    source_info = source_info_map.get(package)
+
+    if not source_info or not source_info.get("vcs_git"):
+        print(
+            f"Error: Could not find team info for package '{package}'", file=sys.stderr
+        )
+        sys.exit(1)
+
+    team_slug = source_info["vcs_git"]
+    display_team = team_slug_to_display_name(team_slug).lower()
+
+    team_dir = DEBIAN_SRC_BASE / display_team
+    pkg_dir = team_dir / package
+    repo_dir = pkg_dir / package
+
+    # Create package directory if it doesn't exist
+    pkg_dir.mkdir(parents=True, exist_ok=True)
+
+    if not repo_dir.exists():
+        # Package not checked out, do salsa checkout
+        print(f"Checking out {package} into {pkg_dir}...")
+        salsa_path = f"{team_slug}/deps/{package}"
+        result = subprocess.run(
+            ["salsa", "checkout", salsa_path],
+            cwd=pkg_dir,
+        )
+        if result.returncode != 0:
+            print(f"Error: salsa checkout failed for {salsa_path}", file=sys.stderr)
+            sys.exit(1)
+    else:
+        # Package already checked out, check for uncommitted changes
+        if has_uncommitted_changes(repo_dir):
+            print(
+                f"Error: {repo_dir} has uncommitted changes. "
+                "Please commit or stash them first.",
+                file=sys.stderr,
+            )
+            sys.exit(1)
+        print(f"Package {package} already checked out at {pkg_dir}")
+
+    # Import new upstream version
+    print("Importing new upstream version...")
+    result = subprocess.run(
+        ["gbp", "import-orig", "--uscan", "--pristine-tar", "--no-interactive"],
+        cwd=repo_dir,
+    )
+    if result.returncode != 0:
+        print("Error: gbp import-orig failed", file=sys.stderr)
+        sys.exit(1)
+
+    # Add changelog entry
+    print("Adding changelog entry...")
+    if not add_changelog_entry(repo_dir):
+        print("Error: Failed to add changelog entry", file=sys.stderr)
+        sys.exit(1)
+
+    print(f"Successfully updated {package}")
+    print(repo_dir)
+
+
 @click.group(invoke_without_command=True)
 @click.pass_context
 def cli(context: click.Context) -> None:
@@ -419,6 +554,12 @@ def update_command() -> None:
     update_todos()
 
 
+@cli.command("update-pkg", help="Update a package to a new upstream version.")
+@click.argument("package")
+def update_pkg_command(package: str) -> None:
+    update_package(package)
+
+
 def main() -> None:
     """Entry point for the CLI."""
     cli()