Add Debian BTS source for bugs filed and owned

New `bts` subcommand backed by python-debianbts (bugs.debian.org SOAP):

- Identify by email address: get_bugs(submitter=...) for bugs filed,
  get_bugs(owner=...) for bugs owned
- Filed filtered on creation date, owned on log_modified; status
  lookups capped at 500 bugs per collect
- Per-query failures degrade to warnings like other sources
This commit is contained in:
2026-08-25 19:44:56 +02:00
parent 5b89313c4a
commit 98eb6ab542
5 changed files with 135 additions and 11 deletions
+6
View File
@@ -9,6 +9,7 @@ from datetime import UTC, datetime
from weekly_activity.model import last_week_window
from weekly_activity.report import format_report
from weekly_activity.sources.debian_bts import DebianBtsSource
from weekly_activity.sources.github import GitHubSource
from weekly_activity.sources.gitlab import GitLabSource
from weekly_activity.sources.launchpad import LaunchpadSource
@@ -68,6 +69,9 @@ def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
help="Env var holding the API token (see README for discovery order).",
)
bts = subparsers.add_parser("bts", parents=[base], help="Debian Bug Tracking System activity.")
bts.add_argument("username", help="Email address used on Debian bugs (submitter/owner).")
return parser.parse_args(argv)
@@ -106,6 +110,8 @@ def main() -> None:
source = GitHubSource(token=args.token)
elif args.source == "gitlab":
source = GitLabSource(url=args.url, token=args.token, token_env=args.token_env)
elif args.source == "bts":
source = DebianBtsSource()
else:
raise SystemExit(f"error: unknown source {args.source!r}")
+95
View File
@@ -0,0 +1,95 @@
"""Debian Bug Tracking System activity source — bugs filed and owned.
Queries bugs.debian.org over SOAP via python-debianbts. Identification is
by email address (the one used on Debian bugs), not a username.
"""
from __future__ import annotations
from datetime import datetime
from typing import Any
import debianbts
from weekly_activity.model import ActivityReport, Section
def _to_naive_utc(value: object) -> datetime | None:
"""Normalize a datetime (naive or aware) to naive UTC."""
if not isinstance(value, datetime):
return None
if value.tzinfo is not None:
return value.astimezone(value.tzinfo).replace(tzinfo=None)
return value
def _fmt_date(dt: datetime | None) -> str:
return dt.strftime("%Y-%m-%d") if dt is not None else ""
class DebianBtsSource:
"""Debian BTS activity source implementing ActivitySource."""
name = "debian-bts"
MAX_BUGS = 500 # safety cap on per-collect status lookups
def __init__(self, *, server: str = "https://bugs.debian.org") -> None:
self._server = server.rstrip("/")
def collect(self, username: str, since: datetime, until: datetime) -> ActivityReport:
warnings: list[str] = []
def safe(label: str, fn: Any) -> Any:
try:
return fn()
except Exception as exc: # noqa: BLE001 - degrade section, keep report
warnings.append(f"{label}: {type(exc).__name__}: {exc}")
return None
# --- Bugs filed / owned ---
filed_nums = safe("bugs filed", lambda: debianbts.get_bugs(submitter=username)) or []
owned_nums = safe("bugs owned", lambda: debianbts.get_bugs(owner=username)) or []
filed_set = set(filed_nums)
owned_set = set(owned_nums)
all_nums = sorted(filed_set | owned_set)[: self.MAX_BUGS]
reports: list[Any] = []
if all_nums:
result = safe("bug status", lambda: debianbts.get_status(all_nums))
if result is not None:
reports = list(result)
filed_entries: list[tuple[datetime | None, str]] = []
owned_entries: list[tuple[datetime | None, str]] = []
for report in reports:
num = getattr(report, "bug_num", None)
subject = getattr(report, "subject", None) or "(untitled)"
package = getattr(report, "package", None) or "(no package)"
severity = getattr(report, "severity", None) or "unknown"
pending = getattr(report, "pending", None) or "unknown"
created = _to_naive_utc(getattr(report, "date", None))
modified = _to_naive_utc(getattr(report, "log_modified", None))
base = f" [#{num}] {subject}\n {package} · {pending} · severity {severity}"
url = f"{self._server}/{num}"
if num in filed_set and created is not None and since <= created < until:
filed_entries.append((created, f"{base} · created {_fmt_date(created)}\n {url}"))
if num in owned_set and modified is not None and since <= modified < until:
owned_entries.append(
(modified, f"{base} · last modified {_fmt_date(modified)}\n {url}")
)
filed_entries.sort(key=lambda x: x[0] or datetime.min, reverse=True)
owned_entries.sort(key=lambda x: x[0] or datetime.min, reverse=True)
return ActivityReport(
source=self.name,
username=username,
since=since,
until=until,
sections=[
Section("Bugs filed", [entry for _, entry in filed_entries]),
Section("Bugs owned, updated", [entry for _, entry in owned_entries]),
],
warnings=warnings,
)