edward/debian-todo
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Various improvements

Browse source
This commit is contained in:
Edward Betts 2026-05-05 09:28:44 +01:00
parent 256fdef129
commit c15b57ec87
2 changed files with 110 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

139
debian_todo/__init__.py
View file

@ -39,25 +39,30 @@ CACHE_VERSION = 4
SourceInfo = dict[str, str]
HIDE_UPLOADER = "Edward Betts <edward@4angle.com>"
DEBIAN_SRC_BASE = Path.home() / "src" / "debian"
UDD_HIGH_LOAD_MARKER = "Current system load"
class PackageUpdateError(Exception):
"""Base exception for package update operations."""
pass
class PackageNotFoundError(PackageUpdateError):
"""Package not found in source info."""
pass
class RepositoryStateError(PackageUpdateError):
"""Repository in invalid state (uncommitted changes, etc)."""
pass
class ExternalCommandError(PackageUpdateError):
"""External command failed (salsa, gbp, dch)."""
def __init__(self, command: str, returncode: int, message: str):
self.command = command
self.returncode = returncode
@ -66,6 +71,7 @@ class ExternalCommandError(PackageUpdateError):
class MissingToolError(PackageUpdateError):
"""Required external tool not available."""
pass
@ -104,7 +110,10 @@ def run_command(
raise ExternalCommandError(
command=cmd_str,
returncode=result.returncode,
message=f"{description} failed: {cmd_str} (exit code {result.returncode})",
message=(
f"{description} failed: {cmd_str} "
f"(exit code {result.returncode})"
),
)
return result
except FileNotFoundError as e:
@ -128,7 +137,8 @@ def check_required_tools() -> None:
if missing_tools:
tools_str = ", ".join(missing_tools)
raise MissingToolError(
f"Required tools not found: {tools_str}. Please install them to continue."
f"Required tools not found: {tools_str}. "
"Please install them to continue."
)
@ -217,15 +227,16 @@ def load_cache(source_paths: list[str]) -> Optional[dict[str, SourceInfo]]:
if not isinstance(key, str):
return None
if isinstance(value, str):
normalized[key] = {"vcs_git": value, "uploaders": ""}
normalized[key] = {"team": value, "uploaders": ""}
continue
if not isinstance(value, dict):
return None
vcs_git = value.get("vcs_git")
team = value.get("team")
uploaders = value.get("uploaders")
if not isinstance(vcs_git, str) or not isinstance(uploaders, str):
vcs_git = value.get("vcs_git")
if not isinstance(team, str) or not isinstance(uploaders, str):
return None
normalized[key] = {"vcs_git": vcs_git, "uploaders": uploaders}
normalized[key] = {"team": team, "uploaders": uploaders, "vcs_git": vcs_git}
return normalized
@ -282,8 +293,9 @@ def load_source_info_map() -> dict[str, SourceInfo]:
)
if team or uploaders_text:
vcs_by_source[source] = {
"vcs_git": team or "",
"team": team or "",
"uploaders": uploaders_text,
"vcs_git": vcs_git,
}
save_cache(source_paths, vcs_by_source)
return vcs_by_source
@ -299,29 +311,41 @@ def fetch_todo_list(timeout: int = 30) -> TodoList:
List of TODO items
Raises:
PackageUpdateError: If network request fails or JSON is invalid
PackageUpdateError: If network request fails or JSON is
invalid
"""
try:
with urlopen(TODO_URL, timeout=timeout) as response:
payload = response.read().decode("utf-8")
payload = response.read()
except urllib.error.HTTPError as e:
raise PackageUpdateError(
f"HTTP error {e.code} while fetching TODO list from {TODO_URL}: {e.reason}"
f"HTTP error {e.code} while fetching TODO list from "
f"{TODO_URL}: {e.reason}"
) from e
except urllib.error.URLError as e:
raise PackageUpdateError(
f"Network error while fetching TODO list from {TODO_URL}: {e.reason}"
f"Network error while fetching TODO list from " f"{TODO_URL}: {e.reason}"
) from e
except TimeoutError as e:
raise PackageUpdateError(
f"Timeout after {timeout}s while fetching TODO list from {TODO_URL}"
f"Timeout after {timeout}s while fetching TODO list " f"from {TODO_URL}"
) from e
try:
return cast(TodoList, json.loads(payload))
except json.JSONDecodeError as e:
response_text = payload.decode("utf-8", errors="replace").strip()
if UDD_HIGH_LOAD_MARKER in response_text:
raise PackageUpdateError(
"UDD returned a temporary overload page instead of JSON. "
"Please retry later."
) from e
preview = response_text[:120]
if len(response_text) > 120:
preview += "..."
raise PackageUpdateError(
f"Invalid JSON in TODO list response: {e}"
f"Invalid JSON in TODO list response: {preview}"
) from e
@ -456,7 +480,10 @@ def list_todos(include_prerelease: bool) -> None:
save_todo_list(todo_list)
except PackageUpdateError as e:
print(f"Error: {e}", file=sys.stderr)
print("Please try again later or check your network connection.", file=sys.stderr)
print(
"Please try again later or check your network connection.",
file=sys.stderr,
)
sys.exit(1)
else:
with TODO_PATH.open("r", encoding="utf-8") as handle:
@ -480,14 +507,14 @@ def list_todos(include_prerelease: bool) -> None:
for todo in filtered:
new_version, current_version = parse_details(todo[":details"])
source_info = source_info_map.get(todo[":source"], {})
vcs_git = source_info.get("vcs_git")
team = source_info.get("team")
uploaders = source_info.get("uploaders", "")
source = todo[":source"]
note = notes_by_source.get(source, "")
display_new = new_version
if is_prerelease_version(todo[":details"]):
display_new = f"[yellow]{new_version} (pre)[/yellow]"
display_team = vcs_git or "-"
display_team = team or "-"
if display_team == "homeassistant-team":
display_team = "HA"
elif display_team.endswith("-team"):
@ -516,7 +543,7 @@ def list_todos(include_prerelease: bool) -> None:
console.print(f"Packages: {len(filtered)}")
def update_todos() -> None:
def refresh_todos() -> None:
"""Fetch latest TODO list from UDD and show changes."""
old_list: TodoList = []
if TODO_PATH.exists():
@ -609,20 +636,21 @@ def validate_package_info(package: str) -> SourceInfo:
Source info dictionary containing vcs_git and uploaders
Raises:
PackageNotFoundError: If package not found or has no vcs_git
PackageNotFoundError: If package not found or has no
vcs_git
"""
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"):
raise PackageNotFoundError(
f"Could not find team info for package '{package}'"
)
if not source_info or not source_info.get("team"):
raise PackageNotFoundError(f"Could not find team info for package '{package}'")
return source_info
def resolve_package_directories(package: str, source_info: SourceInfo) -> tuple[Path, Path, Path]:
def resolve_package_directories(
package: str, source_info: SourceInfo
) -> tuple[Path, Path, Path]:
"""Resolve team_dir, pkg_dir, and repo_dir paths.
Args:
@ -632,7 +660,7 @@ def resolve_package_directories(package: str, source_info: SourceInfo) -> tuple[
Returns:
Tuple of (team_dir, pkg_dir, repo_dir)
"""
team_slug = source_info["vcs_git"]
team_slug = source_info["team"]
display_team = team_slug_to_display_name(team_slug).lower()
team_dir = DEBIAN_SRC_BASE / display_team
@ -642,7 +670,9 @@ def resolve_package_directories(package: str, source_info: SourceInfo) -> tuple[
return team_dir, pkg_dir, repo_dir
def ensure_package_checkout(package: str, source_info: SourceInfo, pkg_dir: Path) -> None:
def ensure_package_checkout(
package: str, source_info: SourceInfo, pkg_dir: Path
) -> None:
"""Checkout package from salsa.
Args:
@ -657,13 +687,12 @@ def ensure_package_checkout(package: str, source_info: SourceInfo, pkg_dir: Path
try:
pkg_dir.mkdir(parents=True, exist_ok=True)
except OSError as e:
raise RepositoryStateError(
f"Failed to create directory {pkg_dir}: {e}"
) from e
raise RepositoryStateError(f"Failed to create directory {pkg_dir}: {e}") from e
print(f"Checking out {package} into {pkg_dir}...")
team_slug = source_info["vcs_git"]
salsa_path = f"{team_slug}/deps/{package}"
assert source_info["vcs_git"].startswith("https://salsa.debian.org/")
salsa_path = source_info["vcs_git"][len("https://salsa.debian.org/") : -4]
print(" ".join(["salsa", "checkout", salsa_path]))
run_command(
["salsa", "checkout", salsa_path],
pkg_dir,
@ -682,8 +711,7 @@ def validate_repository_state(repo_dir: Path) -> None:
"""
if has_uncommitted_changes(repo_dir):
raise RepositoryStateError(
f"{repo_dir} has uncommitted changes. "
"Please commit or stash them first."
f"{repo_dir} has uncommitted changes. " "Please commit or stash them first."
)
@ -696,6 +724,12 @@ def run_gbp_pq_workflow(repo_dir: Path) -> None:
Raises:
ExternalCommandError: If gbp pq commands fail
"""
run_command(
["gbp", "pq", "drop"],
repo_dir,
"Import patch queue",
)
run_command(
["gbp", "pq", "import"],
repo_dir,
@ -731,7 +765,7 @@ def update_debian_control(repo_dir: Path) -> None:
- Removes obsolete 'Priority: optional' (now default)
- Removes obsolete 'Rules-Requires-Root: no' (now default)
- Updates Standards-Version to current (4.7.3)
- Updates Standards-Version to current (4.7.4)
Args:
repo_dir: Repository directory
@ -744,7 +778,7 @@ def update_debian_control(repo_dir: Path) -> None:
return
lines = control_path.read_text().splitlines()
current_standards_version = "4.7.3"
current_standards_version = "4.7.4"
new_lines = []
for line in lines:
@ -763,7 +797,7 @@ def update_debian_control(repo_dir: Path) -> None:
)
continue
if line.startswith("Standards-Version: "):
standards_version = line[len("Standards-Version: "):]
standards_version = line[len("Standards-Version: ") :]
if standards_version != current_standards_version:
line = "Standards-Version: " + current_standards_version
run_command(
@ -781,8 +815,10 @@ def update_debian_copyright_year(repo_dir: Path) -> None:
"""Update Edward Betts' copyright years in debian/copyright.
Transforms:
- "Copyright: 2025 Edward Betts <...>" -> "Copyright: 2025-2026 Edward Betts <...>"
- "Copyright: 2022-2024 Edward Betts <...>" -> "Copyright: 2022-2026 Edward Betts <...>"
- "Copyright: 2025 Edward Betts <...>" ->
"Copyright: 2025-2026 Edward Betts <...>"
- "Copyright: 2022-2024 Edward Betts <...>" ->
"Copyright: 2022-2026 Edward Betts <...>"
Args:
repo_dir: Repository directory
@ -831,7 +867,9 @@ def update_debian_copyright_year(repo_dir: Path) -> None:
for line in lines:
stripped = line.rstrip("\n")
if field_start_re.match(stripped) and not copyright_field_start_re.match(stripped):
if field_start_re.match(stripped) and not copyright_field_start_re.match(
stripped
):
in_copyright_field = False
if copyright_field_start_re.match(stripped):
@ -1024,10 +1062,12 @@ def add_salsa_ci(repo_dir: Path) -> None:
if not salsa_repo:
return
content = """---
include:
- https://salsa.debian.org/salsa-ci-team/pipeline/raw/master/recipes/debian.yml
"""
content = (
"---\n"
"include:\n"
" - https://salsa.debian.org/salsa-ci-team/pipeline"
"/raw/master/recipes/debian.yml\n"
)
salsa_ci_path.write_text(content)
@ -1142,8 +1182,11 @@ def update_package(package: str) -> None:
# Run package updates
run_package_updates(repo_dir)
run_command(["debuild", "-S"], repo_dir, "debuild -S")
os.remove(repo_dir / "debian/files")
print(f"Successfully updated {package}")
print(repo_dir)
print(pkg_dir)
@click.group(invoke_without_command=True)
@ -1164,14 +1207,14 @@ def list_command(show_prerelease: bool) -> None:
list_todos(include_prerelease=show_prerelease)
@cli.command("update", help="Fetch the latest todo list and show changes.")
def update_command() -> None:
update_todos()
@cli.command("refresh", help="Fetch the latest todo list and show changes.")
def refresh_command() -> None:
refresh_todos()
@cli.command("update-pkg", help="Update a package to a new upstream version.")
@cli.command("update", help="Update a package to a new upstream version.")
@click.argument("package")
def update_pkg_command(package: str) -> None:
def update_command(package: str) -> None:
try:
update_package(package)
except PackageUpdateError as e: