Restructure to multi-source architecture; fix GitHub repo timeline filter

- Rename package launchpad_weekly_activity -> weekly_activity
- Split into model.py (ActivityReport, Section), report.py (source-agnostic
  formatter), cli.py (subcommand dispatch), sources/ (per-source modules)
- Move Launchpad code to sources/launchpad.py implementing ActivitySource
- Add GitHub source (sources/github.py) using httpx: PRs, issues, repos, commits
- Fix GitHub 'Repositories modified' filtering on pushed_at not updated_at
  (metadata changes were showing stale repos with no recent code pushes)
- Add --credentials-file for Launchpad OAuth (bypass keyring for headless)
- Add keyring dep (launchpadlib optional dep; bare import crashes without it)
- Graceful per-query error handling: degraded sections + concise warnings
- Rolling 7-day window (was previous calendar week)
This commit is contained in:
2026-08-25 15:24:10 +02:00
parent bc451f9830
commit 3625229056
11 changed files with 662 additions and 275 deletions
+6 -6
View File
@@ -1,12 +1,12 @@
[project]
name = "launchpad-weekly-activity"
version = "0.1.0"
description = "Weekly Launchpad activity report (merge proposals, branches, bugs) for a user."
name = "weekly-activity"
version = "0.2.0"
description = "Weekly activity report (merge proposals, pull requests, bugs, issues) for Launchpad and GitHub."
requires-python = ">=3.11"
dependencies = ["launchpadlib>=2.1.0", "keyring>=25"]
dependencies = ["launchpadlib>=2.1.0", "keyring>=25", "httpx>=0.27"]
[project.scripts]
launchpad-weekly-activity = "launchpad_weekly_activity.cli:main"
weekly-activity = "weekly_activity.cli:main"
[build-system]
requires = ["hatchling"]
@@ -16,7 +16,7 @@ build-backend = "hatchling.build"
dev = ["ruff>=0.6", "ty>=0.0.1"]
[tool.hatch.build.targets.wheel]
packages = ["src/launchpad_weekly_activity"]
packages = ["src/weekly_activity"]
[tool.ruff]
target-version = "py311"
@@ -1,3 +0,0 @@
"""Weekly Launchpad activity report generator."""
__version__ = "0.1.0"
-180
View File
@@ -1,180 +0,0 @@
"""Command-line entry point: parse args, fetch activity, print the report."""
from __future__ import annotations
import argparse
import sys
from collections.abc import Sequence
from datetime import UTC, datetime, timedelta
from launchpad_weekly_activity.api import (
LaunchpadClient,
MergeProposal,
WeeklyActivity,
last_week_window,
)
def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
prog="launchpad-weekly-activity",
description="Summarize a Launchpad user's activity for last week (or a custom range).",
)
parser.add_argument("username", help="Launchpad username (the ~name).")
parser.add_argument(
"--since",
help="Inclusive start as an ISO date/datetime (UTC). Default: last Monday 00:00 UTC.",
)
parser.add_argument(
"--until",
help="Exclusive end as an ISO date/datetime (UTC). Default: this Monday 00:00 UTC.",
)
parser.add_argument(
"--service",
choices=["production", "staging"],
default="production",
help="Launchpad service root (default: production).",
)
parser.add_argument(
"--anonymous",
action="store_true",
help="Anonymous access: public data only, no OAuth. Default uses OAuth.",
)
parser.add_argument(
"--credentials-file",
default=None,
help="OAuth token file path (bypasses keyring; useful headless). Default: system keyring.",
)
return parser.parse_args(argv)
def _parse_dt(value: str | None) -> datetime | None:
if value is None:
return None
dt = datetime.fromisoformat(value)
if dt.tzinfo is not None:
dt = dt.astimezone(UTC).replace(tzinfo=None)
return dt
def resolve_period(args: argparse.Namespace) -> tuple[datetime, datetime]:
if args.since or args.until:
since = _parse_dt(args.since)
until = _parse_dt(args.until)
if since is None or until is None:
raise SystemExit("error: --since and --until must be provided together as ISO dates.")
if since >= until:
raise SystemExit("error: --since must be earlier than --until.")
return since, until
return last_week_window()
def _fmt_date(dt: datetime | None) -> str:
return dt.strftime("%Y-%m-%d") if dt is not None else ""
def _fmt_mp(mp: MergeProposal) -> str:
message = (mp.commit_message or "(no commit message)").strip().splitlines()[0]
return f"[{mp.queue_status}] {message}"
def _format_section(title: str, lines: list[str]) -> str:
if not lines:
return f"{title} (0)\n (none)"
return f"{title} ({len(lines)})\n" + "\n".join(lines)
def format_report(activity: WeeklyActivity) -> str:
out: list[str] = []
out.append(f"Launchpad weekly activity — ~{activity.username}")
out.append(
f"Period: {activity.since.strftime('%Y-%m-%d')} .. "
f"{(activity.until - timedelta(days=1)).strftime('%Y-%m-%d')} (UTC, inclusive)"
)
out.append("")
mp_lines: list[str] = []
for mp in activity.authored_mps:
mp_lines.append(
f" {_fmt_mp(mp)}\n"
f" created {_fmt_date(mp.date_created)} · "
f"merged {_fmt_date(mp.date_merged)}\n"
f" {mp.web_link}"
)
out.append(_format_section("Merge proposals authored", mp_lines))
review_lines: list[str] = []
for mp in activity.new_review_requests:
review_lines.append(
f" {_fmt_mp(mp)}\n"
f" requested {_fmt_date(mp.date_review_requested)}\n"
f" {mp.web_link}"
)
out.append("")
out.append(_format_section("New review requests received", review_lines))
perf_lines: list[str] = []
for mp in activity.performed_reviews:
perf_lines.append(
f" {_fmt_mp(mp)}\n reviewed {_fmt_date(mp.date_reviewed)}\n {mp.web_link}"
)
out.append("")
out.append(_format_section("Reviews performed", perf_lines))
repo_lines: list[str] = []
for repo in activity.git_repositories:
name = repo.display_name or repo.name or repo.web_link
branch = repo.default_branch or ""
repo_lines.append(
f" {name}\n"
f" default branch {branch} · "
f"last modified {_fmt_date(repo.date_last_modified)}\n"
f" {repo.web_link}"
)
out.append("")
out.append(_format_section("Repositories modified", repo_lines))
bug_lines: list[str] = []
for bug in activity.bugs:
roles = "/".join(sorted(bug.roles)) or "?"
bug_id = f"#{bug.bug_id}" if bug.bug_id is not None else "#?"
bug_lines.append(
f" [{bug_id}] {bug.title}\n"
f" {roles} · created {_fmt_date(bug.date_created)} · "
f"updated {_fmt_date(bug.date_last_updated)}\n"
f" {bug.web_link}"
)
out.append("")
out.append(_format_section("Bugs", bug_lines))
if activity.warnings:
out.append("")
out.append("Warnings (partial data — some queries failed):")
for w in activity.warnings:
out.append(f" {w}")
out.append("Note: commit-level history is not exposed by the Launchpad REST API;")
out.append("read the branch directly (bzr/git) for per-commit activity.")
return "\n".join(out)
def main() -> None:
args = parse_args()
since, until = resolve_period(args)
client = LaunchpadClient(
service=args.service,
anonymous=args.anonymous,
application_name="launchpad-weekly-activity",
credentials_file=args.credentials_file,
)
try:
activity = client.collect(args.username, since, until)
except LookupError as exc:
print(f"error: {exc}", file=sys.stderr)
sys.exit(2)
except Exception as exc: # noqa: BLE001 - surface API/network failures cleanly
print(f"error: failed to fetch activity: {exc}", file=sys.stderr)
sys.exit(1)
print(format_report(activity))
if __name__ == "__main__":
main()
+3
View File
@@ -0,0 +1,3 @@
"""Weekly activity report generator for multiple platforms."""
__version__ = "0.2.0"
+102
View File
@@ -0,0 +1,102 @@
"""Command-line entry point: parse args, dispatch to source, print report."""
from __future__ import annotations
import argparse
import sys
from collections.abc import Sequence
from datetime import UTC, datetime
from weekly_activity.model import last_week_window
from weekly_activity.report import format_report
from weekly_activity.sources.github import GitHubSource
from weekly_activity.sources.launchpad import LaunchpadSource
def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
base = argparse.ArgumentParser(add_help=False)
base.add_argument(
"--since",
help="Inclusive start as an ISO date/datetime (UTC). Default: 7 days ago.",
)
base.add_argument(
"--until",
help="Inclusive end as an ISO date/datetime (UTC). Default: today.",
)
parser = argparse.ArgumentParser(
prog="weekly-activity",
description="Summarize a user's activity for last week (or a custom range).",
)
subparsers = parser.add_subparsers(dest="source", required=True)
lp = subparsers.add_parser("launchpad", parents=[base], help="Launchpad activity.")
lp.add_argument("username", help="Launchpad username (the ~name).")
lp.add_argument(
"--anonymous",
action="store_true",
help="Anonymous access: public data only, no OAuth.",
)
lp.add_argument("--credentials-file", default=None, help="OAuth token file path.")
lp.add_argument(
"--service",
choices=["production", "staging"],
default="production",
)
gh = subparsers.add_parser("github", parents=[base], help="GitHub activity.")
gh.add_argument("username", help="GitHub username.")
gh.add_argument("--token", default=None, help="GitHub token (or set GITHUB_TOKEN env var).")
return parser.parse_args(argv)
def _parse_dt(value: str | None) -> datetime | None:
if value is None:
return None
dt = datetime.fromisoformat(value)
if dt.tzinfo is not None:
dt = dt.astimezone(UTC).replace(tzinfo=None)
return dt
def resolve_period(args: argparse.Namespace) -> tuple[datetime, datetime]:
if args.since or args.until:
since = _parse_dt(args.since)
until = _parse_dt(args.until)
if since is None or until is None:
raise SystemExit("error: --since and --until must be provided together as ISO dates.")
if since >= until:
raise SystemExit("error: --since must be earlier than --until.")
return since, until
return last_week_window()
def main() -> None:
args = parse_args()
since, until = resolve_period(args)
if args.source == "launchpad":
source = LaunchpadSource(
service=args.service,
anonymous=args.anonymous,
credentials_file=args.credentials_file,
)
elif args.source == "github":
source = GitHubSource(token=args.token)
else:
raise SystemExit(f"error: unknown source {args.source!r}")
try:
report = source.collect(args.username, since, until)
except LookupError as exc:
print(f"error: {exc}", file=sys.stderr)
sys.exit(2)
except Exception as exc: # noqa: BLE001 - surface API/network failures cleanly
print(f"error: failed to fetch activity: {exc}", file=sys.stderr)
sys.exit(1)
print(format_report(report))
if __name__ == "__main__":
main()
+43
View File
@@ -0,0 +1,43 @@
"""Shared data model for weekly activity reports from multiple sources."""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
@dataclass(frozen=True, slots=True)
class Section:
"""A titled group of formatted entries in the report."""
title: str
entries: list[str]
@dataclass(frozen=True, slots=True)
class ActivityReport:
"""A weekly activity report produced by an ActivitySource."""
source: str
username: str
since: datetime
until: datetime
sections: list[Section]
warnings: list[str] = field(default_factory=list)
note: str | None = None
def last_week_window(now: datetime | None = None) -> tuple[datetime, datetime]:
"""Return the last 7 days ending today (rolling week), naive UTC.
The range is half-open: ``since <= event < until``. ``until`` is midnight
tomorrow, so all of today is included.
"""
today = (
(now or datetime.now(UTC))
.astimezone(UTC)
.replace(tzinfo=None, hour=0, minute=0, second=0, microsecond=0)
)
since = today - timedelta(days=7)
until = today + timedelta(days=1)
return since, until
+34
View File
@@ -0,0 +1,34 @@
"""Source-agnostic report formatter — renders ActivityReport sections to text."""
from __future__ import annotations
from datetime import timedelta
from weekly_activity.model import ActivityReport, Section
def _format_section(section: Section) -> str:
if not section.entries:
return f"{section.title} (0)\n (none)"
return f"{section.title} ({len(section.entries)})\n" + "\n".join(section.entries)
def format_report(report: ActivityReport) -> str:
out: list[str] = []
out.append(f"Activity report — {report.source} / ~{report.username}")
out.append(
f"Period: {report.since.strftime('%Y-%m-%d')} .. "
f"{(report.until - timedelta(days=1)).strftime('%Y-%m-%d')} (UTC, inclusive)"
)
for section in report.sections:
out.append("")
out.append(_format_section(section))
if report.warnings:
out.append("")
out.append("Warnings (partial data — some queries failed):")
for w in report.warnings:
out.append(f" {w}")
if report.note:
out.append("")
out.append(report.note)
return "\n".join(out)
+17
View File
@@ -0,0 +1,17 @@
"""Activity source protocol — each source implements this interface."""
from __future__ import annotations
from datetime import datetime
from typing import Protocol, runtime_checkable
from weekly_activity.model import ActivityReport
@runtime_checkable
class ActivitySource(Protocol):
"""A source of weekly activity data (Launchpad, GitHub, etc.)."""
name: str
def collect(self, username: str, since: datetime, until: datetime) -> ActivityReport: ...
+270
View File
@@ -0,0 +1,270 @@
"""GitHub activity source — pull requests, issues, repositories, and commits."""
from __future__ import annotations
import os
import subprocess
from datetime import UTC, datetime
from typing import Any
import httpx
from weekly_activity.model import ActivityReport, Section
def _discover_token() -> str | None:
"""Discover a GitHub token from GITHUB_TOKEN env var or `gh auth token`."""
token = os.environ.get("GITHUB_TOKEN")
if token:
return token
try:
result = subprocess.run(
["gh", "auth", "token"],
capture_output=True,
text=True,
timeout=5,
)
if result.returncode == 0:
return result.stdout.strip()
except (FileNotFoundError, subprocess.TimeoutExpired):
pass
return None
def _parse_gh_date(s: str | None) -> datetime | None:
if s is None:
return None
try:
dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
except ValueError:
return None
if dt.tzinfo is not None:
return dt.astimezone(UTC).replace(tzinfo=None)
return dt
def _fmt_date(dt: datetime | None) -> str:
return dt.strftime("%Y-%m-%d") if dt is not None else ""
class GitHubSource:
"""GitHub activity source implementing ActivitySource."""
name = "github"
def __init__(self, *, token: str | None = None) -> None:
self._token = token or _discover_token()
headers: dict[str, str] = {
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
"User-Agent": "weekly-activity",
}
if self._token:
headers["Authorization"] = f"Bearer {self._token}"
self._client = httpx.Client(
base_url="https://api.github.com",
headers=headers,
timeout=30.0,
)
def _search_issues(self, query: str) -> list[dict[str, Any]]:
response = self._client.get(
"/search/issues",
params={"q": query, "per_page": 100},
)
response.raise_for_status()
return response.json().get("items", [])
def _get_user_repos(self, username: str) -> list[dict[str, Any]]:
response = self._client.get(
f"/users/{username}/repos",
params={"sort": "updated", "per_page": 100},
)
response.raise_for_status()
return response.json()
def _get_commits(
self, owner: str, repo: str, author: str, since_iso: str, until_iso: str
) -> list[dict[str, Any]]:
response = self._client.get(
f"/repos/{owner}/{repo}/commits",
params={"author": author, "since": since_iso, "until": until_iso, "per_page": 100},
)
response.raise_for_status()
return response.json()
def collect(self, username: str, since: datetime, until: datetime) -> ActivityReport:
since_date = since.strftime("%Y-%m-%d")
until_date = until.strftime("%Y-%m-%d")
iso_since = since.isoformat() + "Z"
iso_until = until.isoformat() + "Z"
warnings: list[str] = []
def safe(label: str, fn: Any) -> Any:
try:
return fn()
except httpx.HTTPStatusError as exc:
msg = f"HTTP {exc.response.status_code}"
remaining = exc.response.headers.get("x-ratelimit-remaining")
if remaining == "0":
msg += " (rate limited)"
warnings.append(f"{label}: {msg}")
return None
except Exception as exc: # noqa: BLE001 - degrade section, keep report
warnings.append(f"{label}: {type(exc).__name__}: {exc}")
return None
# --- Pull requests authored ---
pr_items = safe(
"PRs authored",
lambda: self._search_issues(
f"author:{username} type:pr created:{since_date}..{until_date}"
),
)
pr_entries: list[str] = []
if pr_items:
for item in pr_items:
state = item.get("state", "unknown")
merged_at = item.get("pull_request", {}).get("merged_at")
if merged_at:
state = "merged"
created = _parse_gh_date(item.get("created_at"))
merged = _parse_gh_date(merged_at)
pr_entries.append(
f" [{state}] {item.get('title', '(untitled)')}\n"
f" created {_fmt_date(created)} · merged {_fmt_date(merged)}\n"
f" {item.get('html_url', '')}"
)
# --- Pull requests reviewed ---
review_items = safe(
"PRs reviewed",
lambda: self._search_issues(
f"reviewed-by:{username} type:pr updated:{since_date}..{until_date}"
),
)
review_entries: list[str] = []
if review_items:
for item in review_items:
state = item.get("state", "unknown")
updated = _parse_gh_date(item.get("updated_at"))
review_entries.append(
f" [{state}] {item.get('title', '(untitled)')}\n"
f" updated {_fmt_date(updated)}\n"
f" {item.get('html_url', '')}"
)
# --- Issues (reported/commented/assigned) ---
issues: dict[str, dict[str, Any]] = {}
def add_issue(item: dict[str, Any], role: str) -> None:
url = item.get("html_url", "")
if url in issues:
issues[url]["roles"].add(role)
else:
issues[url] = {
"number": item.get("number"),
"title": item.get("title", "(untitled)"),
"url": url,
"created": _parse_gh_date(item.get("created_at")),
"updated": _parse_gh_date(item.get("updated_at")),
"roles": {role},
}
issue_queries: list[tuple[str, str, str]] = [
("reported", "author", "created"),
("commented", "commenter", "updated"),
("assigned", "assignee", "updated"),
]
for role, qualifier, date_field in issue_queries:
items = safe(
f"issues ({role})",
lambda r=role, q=qualifier, d=date_field: self._search_issues(
f"{q}:{username} type:issue {d}:{since_date}..{until_date}"
),
)
if items:
for item in items:
add_issue(item, role)
issue_entries = [
f" [#{info['number']}] {info['title']}\n"
f" {'/'.join(sorted(info['roles']))} · "
f"created {_fmt_date(info['created'])} · "
f"updated {_fmt_date(info['updated'])}\n"
f" {info['url']}"
for info in issues.values()
]
# --- Repositories modified ---
repo_items = safe("repositories", lambda: self._get_user_repos(username))
repo_entries: list[str] = []
active_repos: list[dict[str, Any]] = []
if repo_items:
for repo in repo_items:
pushed = _parse_gh_date(repo.get("pushed_at"))
if pushed is not None and since <= pushed < until:
full_name = repo.get("full_name", repo.get("name", "(unknown)"))
default_branch = repo.get("default_branch", "")
repo_entries.append(
f" {full_name}\n"
f" default branch {default_branch} · pushed {_fmt_date(pushed)}\n"
f" {repo.get('html_url', '')}"
)
active_repos.append(repo)
# --- Commits pushed ---
all_commits: list[tuple[str, dict[str, Any]]] = []
for repo in active_repos[:20]: # cap to avoid rate-limiting
full_name = repo.get("full_name", "")
if not full_name:
continue
owner, _, name = full_name.partition("/")
if not name:
continue
commits = safe(
f"commits ({full_name})",
lambda o=owner, n=name: self._get_commits(o, n, username, iso_since, iso_until),
)
if commits:
for c in commits:
all_commits.append((full_name, c))
def commit_date(c: dict[str, Any]) -> datetime | None:
return _parse_gh_date(c.get("commit", {}).get("author", {}).get("date"))
all_commits.sort(key=lambda x: commit_date(x[1]) or datetime.min, reverse=True)
commit_entries: list[str] = []
for full_name, c in all_commits:
message = c.get("commit", {}).get("message", "")
first_line = message.splitlines()[0] if message else "(no message)"
sha = c.get("sha", "")[:7]
date = commit_date(c)
commit_entries.append(
f" {full_name}: {first_line}\n"
f" {sha} · {_fmt_date(date)}\n"
f" {c.get('html_url', '')}"
)
if not self._token:
warnings.insert(
0,
"no GitHub token found (set GITHUB_TOKEN or run `gh auth login`); "
"anonymous rate limit is 60 req/hour",
)
return ActivityReport(
source=self.name,
username=username,
since=since,
until=until,
sections=[
Section("Pull requests authored", pr_entries),
Section("Pull requests reviewed", review_entries),
Section("Issues", issue_entries),
Section("Repositories modified", repo_entries),
Section("Commits pushed", commit_entries),
],
warnings=warnings,
)
@@ -1,19 +1,15 @@
"""Typed facade over the (dynamic) launchpadlib web service client.
launchpadlib exposes ``ws.op`` operations as runtime-bound methods, so the
Launchpad entry objects are deliberately typed as :class:`typing.Any`. Every
public function here returns plain dataclasses, keeping the untyped boundary
confined to this module.
"""
"""Launchpad activity source — merge proposals, git repositories, and bugs."""
from __future__ import annotations
from dataclasses import dataclass, field, replace
from datetime import UTC, datetime, timedelta
from datetime import datetime
from typing import Any
from launchpadlib.launchpad import Launchpad
from weekly_activity.model import ActivityReport, Section
# All BranchMergeProposal queue statuses, per the Launchpad web service docs.
MP_STATUSES: tuple[str, ...] = (
"Work in progress",
@@ -27,6 +23,9 @@ MP_STATUSES: tuple[str, ...] = (
)
# --- Source-specific dataclasses (internal) ---
@dataclass(frozen=True, slots=True)
class MergeProposal:
web_link: str
@@ -58,32 +57,7 @@ class BugRef:
roles: frozenset[str] = field(default_factory=frozenset)
@dataclass(frozen=True, slots=True)
class WeeklyActivity:
username: str
since: datetime
until: datetime
authored_mps: list[MergeProposal]
new_review_requests: list[MergeProposal]
performed_reviews: list[MergeProposal]
git_repositories: list[GitRepository]
bugs: list[BugRef]
warnings: list[str] = field(default_factory=list)
def last_week_window(now: datetime | None = None) -> tuple[datetime, datetime]:
"""Return the last 7 days ending today (rolling week), naive UTC.
The range is half-open: ``since <= event < until``. ``until`` is midnight
"""
today = (
(now or datetime.now(UTC))
.astimezone(UTC)
.replace(tzinfo=None, hour=0, minute=0, second=0, microsecond=0)
)
since = today - timedelta(days=7)
until = today + timedelta(days=1)
return since, until
# --- Helpers ---
def _to_naive_utc(value: object) -> datetime | None:
@@ -91,7 +65,7 @@ def _to_naive_utc(value: object) -> datetime | None:
if not isinstance(value, datetime):
return None
if value.tzinfo is not None:
return value.astimezone(UTC).replace(tzinfo=None)
return value.astimezone(value.tzinfo).replace(tzinfo=None)
return value
@@ -156,18 +130,12 @@ def _bug_from(bug: Any, role: str) -> BugRef:
def _summarize_error(exc: BaseException) -> str:
"""Produce a concise warning string from a launchpadlib exception.
launchpadlib's HTTPError.__str__ dumps the full response (headers + HTML
body), which can be thousands of characters. Extract just the status,
reason, and OOPS ID.
"""
"""Produce a concise warning string from a launchpadlib exception."""
response = getattr(exc, "response", None)
if response is not None:
status = getattr(response, "status", None)
reason = getattr(response, "reason", "")
parts = [f"{type(exc).__name__}: HTTP {status} {reason}".rstrip()]
# Extract OOPS ID from response headers (x-lazr-oopsid).
try:
oops = response.get("x-lazr-oopsid") or response.get("X-lazr-oopsid")
except Exception: # noqa: BLE001
@@ -175,20 +143,33 @@ def _summarize_error(exc: BaseException) -> str:
if oops:
parts.append(str(oops) if str(oops).startswith("OOPS") else f"OOPS-{oops}")
return " ".join(parts)
# Fallback: first line of str(exc) to avoid dumping huge payloads.
first_line = str(exc).splitlines()[0] if str(exc) else type(exc).__name__
return f"{type(exc).__name__}: {first_line}"
class LaunchpadClient:
"""Read-only Launchpad client wrapping launchpadlib."""
def _fmt_date(dt: datetime | None) -> str:
return dt.strftime("%Y-%m-%d") if dt is not None else ""
def _fmt_mp(mp: MergeProposal) -> str:
message = (mp.commit_message or "(no commit message)").strip().splitlines()[0]
return f"[{mp.queue_status}] {message}"
# --- Source ---
class LaunchpadSource:
"""Launchpad activity source implementing ActivitySource."""
name = "launchpad"
def __init__(
self,
*,
service: str,
anonymous: bool,
application_name: str,
service: str = "production",
anonymous: bool = False,
application_name: str = "weekly-activity",
credentials_file: str | None = None,
) -> None:
self._anonymous = anonymous
@@ -216,7 +197,7 @@ class LaunchpadClient:
except KeyError as exc:
raise LookupError(f"no Launchpad user named {username!r}") from exc
def collect(self, username: str, since: datetime, until: datetime) -> WeeklyActivity:
def collect(self, username: str, since: datetime, until: datetime) -> ActivityReport:
person = self.get_person(username)
person_link = _attr_str(person, "self_link")
iso_since = since.isoformat()
@@ -224,13 +205,13 @@ class LaunchpadClient:
warnings: list[str] = []
def safe(label: str, fn: Any) -> Any:
"""Run a query; on failure record a warning and return None."""
try:
return fn()
except Exception as exc: # noqa: BLE001 - degrade section, keep report
warnings.append(f"{label}: {_summarize_error(exc)}")
return None
# --- Merge proposals authored ---
authored_mps: list[MergeProposal] = []
result = safe(
"merge proposals",
@@ -241,10 +222,16 @@ class LaunchpadClient:
if result is not None:
authored_mps = [_mp_from(e) for e in result]
mp_entries = [
f" {_fmt_mp(mp)}\n"
f" created {_fmt_date(mp.date_created)} · merged {_fmt_date(mp.date_merged)}\n"
f" {mp.web_link}"
for mp in authored_mps
]
# --- Review requests + performed reviews ---
new_review_requests: list[MergeProposal] = []
performed_reviews: list[MergeProposal] = []
# getRequestedReviews has no server-side date filter — it fetches all
# review requests for the person, which is expensive and may 503.
result = safe(
"review requests",
lambda: list(person.getRequestedReviews(status=list(MP_STATUSES))),
@@ -257,8 +244,20 @@ class LaunchpadClient:
if _in_window(mp.date_reviewed, since, until):
performed_reviews.append(mp)
review_entries = [
f" {_fmt_mp(mp)}\n"
f" requested {_fmt_date(mp.date_review_requested)}\n"
f" {mp.web_link}"
for mp in new_review_requests
]
perf_entries = [
f" {_fmt_mp(mp)}\n reviewed {_fmt_date(mp.date_reviewed)}\n {mp.web_link}"
for mp in performed_reviews
]
# --- Git repositories ---
git_repositories: list[GitRepository] = []
# Query git repositories by owner + modified_since_date.
result = safe(
"git repositories",
lambda: list(
@@ -275,6 +274,15 @@ class LaunchpadClient:
):
git_repositories.append(repo)
repo_entries = [
f" {repo.display_name or repo.name or repo.web_link}\n"
f" default branch {repo.default_branch or ''} · "
f"last modified {_fmt_date(repo.date_last_modified)}\n"
f" {repo.web_link}"
for repo in git_repositories
]
# --- Bugs ---
bugs: dict[str, BugRef] = {}
def add_bug(task: Any, role: str) -> None:
@@ -305,14 +313,28 @@ class LaunchpadClient:
collect_bug_tasks("commented", bug_commenter=person_link, modified_since=iso_since)
collect_bug_tasks("assigned", assignee=person_link, modified_since=iso_since)
return WeeklyActivity(
bug_entries = [
f" [#{bug.bug_id}] {bug.title}\n"
f" {'/'.join(sorted(bug.roles)) or '?'} · "
f"created {_fmt_date(bug.date_created)} · "
f"updated {_fmt_date(bug.date_last_updated)}\n"
f" {bug.web_link}"
for bug in bugs.values()
]
return ActivityReport(
source=self.name,
username=username,
since=since,
until=until,
authored_mps=authored_mps,
new_review_requests=new_review_requests,
performed_reviews=performed_reviews,
git_repositories=git_repositories,
bugs=list(bugs.values()),
sections=[
Section("Merge proposals authored", mp_entries),
Section("New review requests received", review_entries),
Section("Reviews performed", perf_entries),
Section("Repositories modified", repo_entries),
Section("Bugs", bug_entries),
],
warnings=warnings,
note="Note: commit-level history is not exposed by the Launchpad REST API; "
"read the branch directly (bzr/git) for per-commit activity.",
)
Generated
+106 -27
View File
@@ -2,6 +2,19 @@ version = 1
revision = 3
requires-python = ">=3.11"
[[package]]
name = "anyio"
version = "4.14.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "idna" },
{ name = "typing-extensions", marker = "python_full_version < '3.13'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/61/cc/a381afa6efea9f496eff839d4a6a1aed3bfafc7b3ab4b0d1b243a12573dd/anyio-4.14.2.tar.gz", hash = "sha256:cfa139f3ed1a23ee8f88a145ddb5ac7605b8bbfd8592baacd7ce3d8bb4313c7f", size = 260176, upload-time = "2026-07-12T20:29:07.082Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/da/35/f2287558c17e29fafc8ef3daf819bb9834061cfa43bff8014f7df7f63bdc/anyio-4.14.2-py3-none-any.whl", hash = "sha256:9f505dda5ac9f0c8309b5e8bd445a8c2bf7246f3ce950121e45ea15bc41d1494", size = 125813, upload-time = "2026-07-12T20:29:05.763Z" },
]
[[package]]
name = "backports-tarfile"
version = "1.2.0"
@@ -11,6 +24,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/b9/fa/123043af240e49752f1c4bd24da5053b6bd00cad78c2be53c0d1e8b975bc/backports.tarfile-1.2.0-py3-none-any.whl", hash = "sha256:77e284d754527b01fb1e6fa8a1afe577858ebe4e9dad8919e34c862cb399bc34", size = 30181, upload-time = "2024-05-28T17:01:53.112Z" },
]
[[package]]
name = "certifi"
version = "2026.7.22"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/a3/c2/24167ea9858356b47a87a50d39908bfdb72ceeefe0041586e704e5376b3a/certifi-2026.7.22.tar.gz", hash = "sha256:741e2c3b351ddf169a738da9f2c048608ff7f2c5cc02f1ebc6b118bb090d5d55", size = 138112, upload-time = "2026-07-22T03:35:12.644Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/0b/a7/71ac2cff56fec219ed242bb11b8efb69fcc4bec75db06fb7bfe35de520e6/certifi-2026.7.22-py3-none-any.whl", hash = "sha256:62f22742b58a1a33014a2b6b706588a8d7e2a88ae7bd1a6ebe8c992928483775", size = 136983, upload-time = "2026-07-22T03:35:11.276Z" },
]
[[package]]
name = "cffi"
version = "2.1.1"
@@ -131,6 +153,28 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/12/b3/231ffd4ab1fc9d679809f356cebee130ac7daa00d6d6f3206dd4fd137e9e/distro-1.9.0-py3-none-any.whl", hash = "sha256:7bffd925d65168f85027d8da9af6bddab658135b840670a223589bc0c8ef02b2", size = 20277, upload-time = "2023-12-24T09:54:30.421Z" },
]
[[package]]
name = "h11"
version = "0.16.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/01/ee/02a2c011bdab74c6fb3c75474d40b3052059d95df7e73351460c8588d963/h11-0.16.0.tar.gz", hash = "sha256:4e35b956cf45792e4caa5885e69fba00bdbc6ffafbfa020300e549b208ee5ff1", size = 101250, upload-time = "2025-04-24T03:35:25.427Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/04/4b/29cac41a4d98d144bf5f6d33995617b185d14b22401f75ca86f384e87ff1/h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86", size = 37515, upload-time = "2025-04-24T03:35:24.344Z" },
]
[[package]]
name = "httpcore"
version = "1.0.9"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "certifi" },
{ name = "h11" },
]
sdist = { url = "https://files.pythonhosted.org/packages/06/94/82699a10bca87a5556c9c59b5963f2d039dbd239f25bc2a63907a05a14cb/httpcore-1.0.9.tar.gz", hash = "sha256:6e34463af53fd2ab5d807f399a9b45ea31c3dfa2276f15a2c3f00afff6e176e8", size = 85484, upload-time = "2025-04-24T22:06:22.219Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/7e/f5/f66802a942d491edb555dd61e3a9961140fd64c90bce1eafd741609d334d/httpcore-1.0.9-py3-none-any.whl", hash = "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55", size = 78784, upload-time = "2025-04-24T22:06:20.566Z" },
]
[[package]]
name = "httplib2"
version = "0.32.0"
@@ -143,6 +187,30 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/33/a0/550eec327e5f5c7b732531c489f5307efec41f047b0d703bd4ca1e5ad2db/httplib2-0.32.0-py3-none-any.whl", hash = "sha256:dc6705cacdf3fb0a2aba7629fa33c90fd93e30035db0c157325826be177e4816", size = 93148, upload-time = "2026-06-26T10:13:54.985Z" },
]
[[package]]
name = "httpx"
version = "0.28.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "anyio" },
{ name = "certifi" },
{ name = "httpcore" },
{ name = "idna" },
]
sdist = { url = "https://files.pythonhosted.org/packages/b1/df/48c586a5fe32a0f01324ee087459e112ebb7224f646c0b5023f5e79e9956/httpx-0.28.1.tar.gz", hash = "sha256:75e98c5f16b0f35b567856f597f06ff2270a374470a5c2392242528e3e3e42fc", size = 141406, upload-time = "2024-12-06T15:37:23.222Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" },
]
[[package]]
name = "idna"
version = "3.19"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/5f/f7/abb373e5757eaec4b922b92f97ec8d6d7e057cf06778247604fbc4e7c3f3/idna-3.19.tar.gz", hash = "sha256:5e0811a4383b21dc5838069f801c4fb62113b7447663d2530d2bd6e77b49bf15", size = 215237, upload-time = "2026-08-18T05:14:24.27Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/57/b0/0e52c878c53f245edd3a11020f20979b3f490f245af532c7cae3027754b5/idna-3.19-py3-none-any.whl", hash = "sha256:815e7be7a7806d54abb586dc943addc79e8b2ee16915059658cbeff4b1b43bf4", size = 68550, upload-time = "2026-08-18T05:14:22.343Z" },
]
[[package]]
name = "importlib-metadata"
version = "9.0.0"
@@ -218,33 +286,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/81/db/e655086b7f3a705df045bf0933bdd9c2f79bb3c97bfef1384598bb79a217/keyring-25.7.0-py3-none-any.whl", hash = "sha256:be4a0b195f149690c166e850609a477c532ddbfbaed96a404d4e43f8d5e2689f", size = 39160, upload-time = "2025-11-16T16:26:08.402Z" },
]
[[package]]
name = "launchpad-weekly-activity"
version = "0.1.0"
source = { editable = "." }
dependencies = [
{ name = "keyring" },
{ name = "launchpadlib" },
]
[package.dev-dependencies]
dev = [
{ name = "ruff" },
{ name = "ty" },
]
[package.metadata]
requires-dist = [
{ name = "keyring", specifier = ">=25" },
{ name = "launchpadlib", specifier = ">=2.1.0" },
]
[package.metadata.requires-dev]
dev = [
{ name = "ruff", specifier = ">=0.6" },
{ name = "ty", specifier = ">=0.0.1" },
]
[[package]]
name = "launchpadlib"
version = "2.1.0"
@@ -391,6 +432,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/ea/88/25333bbfea6a5dc064371d2002d3d4807db90b84d5448f9106b2712b0fbc/ty-0.0.73-py3-none-win_arm64.whl", hash = "sha256:e47068f8369dea5d641a26a2ad0a947a320b02ff87099b07e95de0323245a4dc", size = 12443573, upload-time = "2026-08-19T03:12:41.449Z" },
]
[[package]]
name = "typing-extensions"
version = "4.16.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/f6/cc/6253133b5bb138fc3306cebfbda2c520f545d36b5be2c7255cc528bb45d6/typing_extensions-4.16.0.tar.gz", hash = "sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5", size = 113555, upload-time = "2026-07-02T08:40:05.92Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/49/d3/b8441a820a491ddfc024b0b0cf0393375b75ea13866d9c66727e54c2fc80/typing_extensions-4.16.0-py3-none-any.whl", hash = "sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8", size = 45571, upload-time = "2026-07-02T08:40:04.659Z" },
]
[[package]]
name = "wadllib"
version = "2.1.0"
@@ -403,6 +453,35 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/ec/f3/00b61e720165f73c6ef3c7a6ea732314c8ca1ba95b9ed1545fd788d9af9b/wadllib-2.1.0-py3-none-any.whl", hash = "sha256:41b58db0bb5fb21e188c7452281ee7b364b2690a4f6c388afea0ceaeafed1132", size = 61997, upload-time = "2026-07-01T10:48:01.37Z" },
]
[[package]]
name = "weekly-activity"
version = "0.2.0"
source = { editable = "." }
dependencies = [
{ name = "httpx" },
{ name = "keyring" },
{ name = "launchpadlib" },
]
[package.dev-dependencies]
dev = [
{ name = "ruff" },
{ name = "ty" },
]
[package.metadata]
requires-dist = [
{ name = "httpx", specifier = ">=0.27" },
{ name = "keyring", specifier = ">=25" },
{ name = "launchpadlib", specifier = ">=2.1.0" },
]
[package.metadata.requires-dev]
dev = [
{ name = "ruff", specifier = ">=0.6" },
{ name = "ty", specifier = ">=0.0.1" },
]
[[package]]
name = "zipp"
version = "4.1.0"