diff --git a/.gitignore b/.gitignore index b245e56..72496ab 100644 --- a/.gitignore +++ b/.gitignore @@ -10,4 +10,5 @@ dist/ .streamlit/secrets.toml *.local.toml *.local.env -.playwright-mcp/ \ No newline at end of file +.playwright-mcp/ +exports/ diff --git a/README.md b/README.md index 4d7b38a..352b8ce 100644 --- a/README.md +++ b/README.md @@ -34,10 +34,20 @@ API keys are stored in `.env` on your machine only. They are never logged or sen - Missing assignees. - Items not linked to a project board (optional). - Issues or PRs without a development link. +- Item state shown with GitHub colors β€” open (green), closed (red), merged (purple), draft (gray). +- Stale branches β€” separate **Branches** tab lists every branch with age, last committer, + and pull request status; the stale threshold is adjustable and nothing is ever deleted. - Optional date range β€” last 30 days or custom `from / to`. - One project, many projects, or all projects in an organization. - One repo, many repos, or all organization repos. +## Offline Export + +Both tabs offer **🌐 Download offline HTML report** β€” a single self-contained `.html` +file with the full scan embedded. Anyone can open it by double-click in a browser and +search, filter (repository, state, assignee, missing field, …), and sort β€” no Streamlit, +terminal, network, or setup needed. The Excel download is still available next to it. + ## Setup Guide β€” Step by Step You don't need any programming experience. The whole setup is: install one helper diff --git a/app.py b/app.py index 397f40e..b94d40d 100644 --- a/app.py +++ b/app.py @@ -18,6 +18,7 @@ describe_pending_write, resolve_created_item_ids, ) +from github_audit.branches import fetch_branches from github_audit.config import Settings from github_audit.discovery import discover_all, discover_repositories from github_audit.github_client import GitHubClient, GitHubError @@ -25,16 +26,18 @@ AddToProjectPlan, ApplyPlan, AuditFinding, + BranchInfo, PendingWrite, ProjectFieldDefinition, ) +from github_audit.offline_export import OfflineColumn, render_offline_html from github_audit.project_fields import ( fetch_assignable_users, fetch_repo_labels, fetch_repo_milestones, search_items, ) -from github_audit.scanner import scan_all +from github_audit.scanner import parse_github_date, scan_all st.set_page_config(page_title="GitHub Audit", page_icon="πŸ”", layout="wide") @@ -48,10 +51,13 @@ class FindingRow(TypedDict): - project: int + # None when the item is not on the scanned board β€” a number here would be + # whichever project scan happened to win deduplication, i.e. noise. + project: int | None project_title: str repository: str item_type: str + state: str number: int updated_at: str title: str @@ -60,6 +66,17 @@ class FindingRow(TypedDict): url: str +class BranchRow(TypedDict): + repository: str + branch: str + age_days: int + last_commit: str + committer: str + pr_state: str + prs: str + url: str + + type ScanStats = dict[str, int] @@ -123,6 +140,7 @@ def _date_label(value: str | None) -> str: "project_title": "Project title", "repository": "Repository", "item_type": "Type", + "state": "State", "number": "#", "updated_at": "Updated", "title": "Title", @@ -219,6 +237,9 @@ def _bool(key: str, fallback: str = "false") -> bool: "stats": None, "limitations": list[str](), "scan_time": None, + "branches": None, + "branches_error": None, + "branches_time": None, "project_ids_by_number": None, "project_fields_by_number": None, "agent_messages": [], @@ -686,11 +707,13 @@ def _run_scan() -> None: best: dict[tuple[str, str, int], tuple[FindingRow, AuditFinding]] = {} for r in results: for f in r.findings: + on_board = f.project_item_id is not None row: FindingRow = { - "project": f.project_number or 0, - "project_title": f.project_title or "", + "project": f.project_number if on_board else None, + "project_title": (f.project_title or "") if on_board else "", "repository": f.repository.split("/")[-1], "item_type": "PR" if f.item_type == "pull_request" else "Issue", + "state": f.display_state, "number": f.number, "updated_at": _date_label(f.updated_at), "title": f.title, @@ -1187,199 +1210,537 @@ def _render_agent_assistant(visible_rows: list[FindingRow]) -> None: st.rerun() -# ── main content ────────────────────────────────────────────────────────────── -_h1, _h2 = st.columns([7, 1]) -with _h1: - st.title("πŸ” GitHub Audit") -with _h2: - st.write("") - # Filled after the table filters run so the AI dropdown mirrors the table. - _ai_popover = st.popover("AI", use_container_width=True, help="Open AI assistant") +# ── state styling ───────────────────────────────────────────────────────────── +# GitHub conventions: open = green, closed = red, merged = purple, draft = gray. +_STATE_DOTS = {"Open": "🟒", "Draft": "βšͺ", "Merged": "🟣", "Closed": "πŸ”΄"} +_PR_STATE_LABELS = { + "open": "🟒 Open PR", + "merged": "🟣 Merged PR", + "closed": "πŸ”΄ Closed PR", + "none": "βšͺ No PR", +} -if st.session_state.error: - st.error(st.session_state.error) -if st.session_state.rows is None: - with _ai_popover: - _render_agent_assistant([]) - st.info( - "Configure settings in the sidebar, then click **β–Ά Run Scan** to audit " - "your GitHub Projects for missing fields and workflow gaps." +def _state_cell(state: str) -> str: + dot = _STATE_DOTS.get(state) + return f"{dot} {state}" if dot else state + + +# ── offline HTML export ─────────────────────────────────────────────────────── +_EXPORTS_DIR = Path(__file__).parent / "exports" + + +def _save_export_copy(file_name: str, content: str) -> None: + """Keep a reliable copy on disk β€” some browsers (webviews, Brave) mangle + the downloaded file name or extension.""" + try: + _EXPORTS_DIR.mkdir(exist_ok=True) + path = _EXPORTS_DIR / file_name + path.write_text(content, encoding="utf-8") + st.toast(f"Copy saved to {path}", icon="πŸ’Ύ") + except OSError as exc: + st.toast(f"Could not save a local copy: {exc}", icon="⚠️") + + +def _findings_offline_html(all_rows: list[FindingRow]) -> str: + columns = [ + OfflineColumn("repository", "Repository", filterable=True), + OfflineColumn("item_type", "Type", "badge", filterable=True), + OfflineColumn("state", "State", "badge", filterable=True), + OfflineColumn("number", "#", "number"), + OfflineColumn("updated_at", "Updated"), + OfflineColumn("title", "Title"), + OfflineColumn("assignees", "Assignees", filterable=True, split=True), + OfflineColumn("missing_fields", "Missing fields", filterable=True, split=True), + OfflineColumn("url", "Link", "link"), + ] + export_rows: list[dict[str, str | int]] = [ + { + "repository": row["repository"], + "item_type": row["item_type"], + "state": row["state"], + "number": row["number"], + "updated_at": row["updated_at"], + "title": row["title"], + "assignees": row["assignees"], + "missing_fields": row["missing_fields"], + "url": row["url"], + } + for row in all_rows + ] + subtitle = ( + f"{org or 'GitHub'} Β· scanned {st.session_state.scan_time or '?'} Β· " + f"{len(all_rows)} findings Β· filters work fully offline" ) - with st.expander("What does this tool check?"): - st.markdown(""" + return render_offline_html("GitHub Audit β€” Issues & PRs", subtitle, columns, export_rows) + + +def _branches_offline_html(branch_rows: list[BranchRow], threshold: int) -> str: + columns = [ + OfflineColumn("repository", "Repository", filterable=True), + OfflineColumn("branch", "Branch"), + OfflineColumn("status", "Status", "badge", filterable=True), + OfflineColumn("age_days", "Age (days)", "number", min_filter=True), + OfflineColumn("last_commit", "Last commit"), + OfflineColumn("committer", "Committer", filterable=True), + OfflineColumn("pr_state", "PR status", "badge", filterable=True), + OfflineColumn("prs", "Pull requests"), + OfflineColumn("url", "Link", "link"), + ] + export_rows: list[dict[str, str | int]] = [ + { + "repository": row["repository"], + "branch": row["branch"], + "status": "Stale" if _is_stale(row, threshold) else "Active", + "age_days": row["age_days"], + "last_commit": row["last_commit"], + "committer": row["committer"], + "pr_state": row["pr_state"], + "prs": row["prs"], + "url": row["url"], + } + for row in branch_rows + ] + subtitle = ( + f"{org or 'GitHub'} Β· scanned {st.session_state.branches_time or '?'} Β· " + f"{len(branch_rows)} branches Β· stale = no open PR and no commit " + f"in {threshold} days Β· filters work fully offline" + ) + return render_offline_html("GitHub Audit β€” Branches", subtitle, columns, export_rows) + + +# ── branch audit ────────────────────────────────────────────────────────────── +def _is_stale(row: BranchRow, threshold: int) -> bool: + return row["age_days"] >= threshold and row["pr_state"] != "open" + + +def _pr_label(state: str, is_draft: bool) -> str: + return "draft" if is_draft and state.upper() == "OPEN" else state.lower() + + +def _branch_rows(branches: list[BranchInfo]) -> list[BranchRow]: + today = date.today() + branch_rows: list[BranchRow] = [] + for branch in branches: + commit_date = parse_github_date(branch.last_commit_date) + prs = ", ".join( + f"#{pr.number} ({_pr_label(pr.state, pr.is_draft)})" for pr in branch.pull_requests + ) + if branch.pull_requests_total > len(branch.pull_requests): + prs += f", +{branch.pull_requests_total - len(branch.pull_requests)} more" + branch_rows.append( + BranchRow( + repository=branch.repository.split("/")[-1], + branch=branch.name, + age_days=(today - commit_date).days if commit_date else -1, + last_commit=_date_label(branch.last_commit_date), + committer=branch.last_committer or "(unknown)", + pr_state=branch.pr_state, + prs=prs or "(none)", + url=branch.url, + ) + ) + branch_rows.sort(key=lambda r: r["age_days"], reverse=True) + return branch_rows + + +def _run_branch_scan() -> None: + try: + settings = Settings.model_validate( + { + "github_token": token, + "github_org": org, + # branch audit needs no project boards β€” skip project validation + "my_work_mode": True, + "require_target_assignee": False, + "github_include_all_repositories": inc_all_repos, + "github_repository_allowlist_raw": "" if inc_all_repos else _csv(repo_allowlist), + "github_repository_denylist_raw": _csv(repo_denylist), + } + ) + except ValidationError as exc: + msgs = "; ".join(e["msg"] for e in exc.errors()) + st.session_state.branches_error = f"Configuration error: {msgs}" + return + try: + with GitHubClient(settings.github_token) as client: + repositories = discover_repositories(client, settings) + branches = fetch_branches(client, repositories) + except GitHubError as exc: + st.session_state.branches_error = str(exc) + return + # Default branches (main/master) are never cleanup candidates. + st.session_state.branches = _branch_rows([b for b in branches if not b.is_default]) + st.session_state.branches_time = datetime.now().strftime("%Y-%m-%d %H:%M") + st.session_state.branches_error = None + + +def _render_branch_tab() -> None: + bc1, bc2, bc3 = st.columns([1.2, 1, 2.8]) + with bc1: + st.write("") + branch_btn = st.button("β–Ά Scan Branches", type="primary", width="stretch") + with bc2: + threshold = int( + st.number_input( + "Stale after (days)", + min_value=1, + max_value=3650, + value=30, + help=( + "A branch counts as stale when its last commit is older than this " + "and it has no open pull request. Adjust freely β€” nothing is deleted." + ), + ) + ) + with bc3: + st.write("") + st.caption( + "Audits branches in the configured repository scope. Default branches are " + "excluded. Decision support only β€” this never deletes anything." + ) + + if branch_btn: + if not token.strip() or not org.strip(): + st.session_state.branches_error = "GitHub token and organization are required." + elif not inc_all_repos and not repo_allowlist.strip(): + st.session_state.branches_error = ( + "Either enable 'All org repositories' or enter a repository allowlist." + ) + else: + with st.spinner("Fetching branches from GitHub…"): + _run_branch_scan() + + if st.session_state.branches_error: + st.error(st.session_state.branches_error) + branch_rows = cast(list[BranchRow] | None, st.session_state.branches) + if branch_rows is None: + st.info( + "Click **β–Ά Scan Branches** to list all branches with their age, last " + "committer, and pull request status β€” and spot stale ones." + ) + return + + stale_rows = [row for row in branch_rows if _is_stale(row, threshold)] + no_pr = sum(1 for row in branch_rows if row["pr_state"] == "none") + m1, m2, m3 = st.columns(3) + m1.metric("Branches", len(branch_rows)) + m2.metric(f"Stale (>{threshold}d, no open PR)", len(stale_rows)) + m3.metric("Without any PR", no_pr) + if st.session_state.branches_time: + st.caption(f"Last branch scan: {st.session_state.branches_time}") + + all_branch_repos = sorted({row["repository"] for row in branch_rows}) + all_committers = sorted({row["committer"] for row in branch_rows}) + all_pr_states = [ + s + for s in ("open", "merged", "closed", "none") + if any(r["pr_state"] == s for r in branch_rows) + ] + for _fk, _fopts in ( + ("branch_filter_repos", all_branch_repos), + ("branch_filter_committers", all_committers), + ("branch_filter_pr_states", all_pr_states), + ): + if _fk in st.session_state: + st.session_state[_fk] = [v for v in st.session_state[_fk] if v in _fopts] + bf1, bf2, bf3, bf4 = st.columns([1, 1, 1, 0.8]) + with bf1: + sel_branch_repos = st.multiselect("Repository", all_branch_repos, key="branch_filter_repos") + with bf2: + sel_committers = st.multiselect("Committer", all_committers, key="branch_filter_committers") + with bf3: + sel_pr_states = st.multiselect( + "PR status", + all_pr_states, + key="branch_filter_pr_states", + format_func=lambda s: _PR_STATE_LABELS.get(str(s), str(s)), + ) + with bf4: + st.markdown('
', unsafe_allow_html=True) + stale_only = st.toggle("Stale only", value=False) + + visible = [ + row + for row in branch_rows + if (not sel_branch_repos or row["repository"] in sel_branch_repos) + and (not sel_committers or row["committer"] in sel_committers) + and (not sel_pr_states or row["pr_state"] in sel_pr_states) + and (not stale_only or _is_stale(row, threshold)) + ] + st.caption(f"Showing **{len(visible)}** of {len(branch_rows)} branches") + st.dataframe( + [ + { + "Repository": row["repository"], + "Branch": row["branch"], + "Stale": "🟠 Stale" if _is_stale(row, threshold) else "", + "Age (days)": row["age_days"], + "Last commit": row["last_commit"], + "Committer": row["committer"], + "PR status": _PR_STATE_LABELS.get(row["pr_state"], row["pr_state"]), + "Pull requests": row["prs"], + "URL": row["url"], + } + for row in visible + ], + use_container_width=True, + hide_index=True, + height=min(600, 100 + len(visible) * 35), + column_config={ + "URL": st.column_config.LinkColumn("Link", display_text="Open β†—"), + "Age (days)": st.column_config.NumberColumn("Age (days)", format="%d", width="small"), + "Stale": st.column_config.TextColumn("Stale", width="small"), + "Last commit": st.column_config.TextColumn("Last commit", width="small"), + }, + ) + branches_html = _branches_offline_html(branch_rows, threshold) + st.download_button( + "🌐 Download offline HTML report", + branches_html, + "github-audit-branches.html", + "text/html", + on_click=_save_export_copy, + args=("github-audit-branches.html", branches_html), + help=( + "Self-contained page with all scanned branches β€” open and filter anywhere. " + "A copy is also saved to the app's exports folder." + ), + ) + + +# ── findings tab ────────────────────────────────────────────────────────────── +def _render_findings_tab() -> None: + if st.session_state.rows is None: + with _ai_popover: + _render_agent_assistant([]) + st.info( + "Configure settings in the sidebar, then click **β–Ά Run Scan** to audit " + "your GitHub Projects for missing fields and workflow gaps." + ) + with st.expander("What does this tool check?"): + st.markdown(""" - **Required fields** β€” Estimate, Priority, Iteration (sprint), Difficulty, Status (configurable) - **Assignees** β€” whether items are assigned and to your target users - **Development links** β€” whether issues have a linked PR or branch - **Project board membership** β€” optional check for items missing from the V2 board - """) - st.stop() + """) + return -rows = cast(list[FindingRow], st.session_state.rows) -stats = cast(ScanStats, st.session_state.stats) + rows = cast(list[FindingRow], st.session_state.rows) + stats = cast(ScanStats, st.session_state.stats) -c1, c2, c3 = st.columns(3) -c1.metric("Issues scanned", stats["issues"]) -c2.metric("PRs scanned", stats["prs"]) -c3.metric("Findings", stats["findings"]) + c1, c2, c3 = st.columns(3) + c1.metric("Issues scanned", stats["issues"]) + c2.metric("PRs scanned", stats["prs"]) + c3.metric("Findings", stats["findings"]) -if not rows: - with _ai_popover: - _render_agent_assistant([]) - st.success("βœ… No findings β€” everything looks good!") - st.stop() + if not rows: + with _ai_popover: + _render_agent_assistant([]) + st.success("βœ… No findings β€” everything looks good!") + return -# ── filters ─────────────────────────────────────────────────────────────────── -st.subheader("Filters") -fc1, fc2, fc3, fc4, fc5, fc6 = st.columns([1, 1, 1, 1, 1, 0.22]) + # ── filters ─────────────────────────────────────────────────────────────── + st.subheader("Filters") + fc1, fc2, fc3, fc4, fc5, fc6, fc7 = st.columns([1, 1, 1, 1, 1, 1, 0.22]) -all_repos = sorted({row["repository"] for row in rows}) -all_missing = sorted( - {f.strip() for row in rows for f in row["missing_fields"].split(",") if f.strip()} -) -all_assignees = sorted( - { - a.strip() - for row in rows - for a in row["assignees"].split(",") - if a.strip() and a.strip() != "(none)" - } -) -all_types = sorted({row["item_type"] for row in rows}) -proj_labels: dict[int, str] = {} -for row in rows: - p = row["project"] - if p and p not in proj_labels: - t = row["project_title"] - proj_labels[p] = f"{p} - {t}" if t else str(p) -all_proj_options = [proj_labels[p] for p in sorted(proj_labels)] - -_date_active = bool( - st.session_state.get("filter_date_from") or st.session_state.get("filter_date_to") -) -_date_btn_label = "πŸ“… ●" if _date_active else "πŸ“…" - -# Explicit keys keep selections alive across rescans; prune values that no -# longer exist in the fresh options (a stale value would raise otherwise). -for _fk, _fopts in ( - ("filter_repos", all_repos), - ("filter_missing", all_missing), - ("filter_assignees", all_assignees), - ("filter_types", all_types), - ("filter_projects", all_proj_options), -): - if _fk in st.session_state: - st.session_state[_fk] = [v for v in st.session_state[_fk] if v in _fopts] - -with fc1: - sel_repos = st.multiselect("Repository", all_repos, key="filter_repos") -with fc2: - sel_missing = st.multiselect("Missing field", all_missing, key="filter_missing") -with fc3: - sel_assignees = st.multiselect("Assignee", all_assignees, key="filter_assignees") -with fc4: - sel_types = st.multiselect("Type", all_types, key="filter_types") -with fc5: - sel_proj_labels = st.multiselect("Project", all_proj_options, key="filter_projects") -with fc6: - st.markdown('
', unsafe_allow_html=True) - with st.popover(_date_btn_label, use_container_width=True, help="Filter by last updated date"): - _dc1, _dc2 = st.columns(2) - with _dc1: - date_from = st.date_input("From", value=None, key="filter_date_from") - with _dc2: - date_to = st.date_input("To", value=None, key="filter_date_to") - -sel_proj_nums = {p for p, lbl in proj_labels.items() if lbl in sel_proj_labels} - -filtered: list[FindingRow] = [] -for row in rows: - if sel_repos and row["repository"] not in sel_repos: - continue - # exact membership, not substring: "assignee" must not match "target assignee", - # login "jan" must not match "janedoe" - row_missing = {part.strip() for part in row["missing_fields"].split(",")} - if sel_missing and not any(f in row_missing for f in sel_missing): - continue - row_assignees = {part.strip() for part in row["assignees"].split(",")} - if sel_assignees and not any(a in row_assignees for a in sel_assignees): - continue - if sel_types and row["item_type"] not in sel_types: - continue - if sel_proj_nums and row["project"] not in sel_proj_nums: - continue - if date_from or date_to: - raw = row["updated_at"] - try: - row_date = date.fromisoformat(raw[:10]) if raw else None - except ValueError: - row_date = None - if row_date is None: + all_repos = sorted({row["repository"] for row in rows}) + all_missing = sorted( + {f.strip() for row in rows for f in row["missing_fields"].split(",") if f.strip()} + ) + all_assignees = sorted( + { + a.strip() + for row in rows + for a in row["assignees"].split(",") + if a.strip() and a.strip() != "(none)" + } + ) + all_types = sorted({row["item_type"] for row in rows}) + all_states = [s for s in _STATE_DOTS if any(row["state"] == s for row in rows)] + proj_labels: dict[int, str] = {} + for row in rows: + p = row["project"] + if p and p not in proj_labels: + t = row["project_title"] + proj_labels[p] = f"{p} - {t}" if t else str(p) + all_proj_options = [proj_labels[p] for p in sorted(proj_labels)] + + _date_active = bool( + st.session_state.get("filter_date_from") or st.session_state.get("filter_date_to") + ) + _date_btn_label = "πŸ“… ●" if _date_active else "πŸ“…" + + # Explicit keys keep selections alive across rescans; prune values that no + # longer exist in the fresh options (a stale value would raise otherwise). + for _fk, _fopts in ( + ("filter_repos", all_repos), + ("filter_missing", all_missing), + ("filter_assignees", all_assignees), + ("filter_types", all_types), + ("filter_states", all_states), + ("filter_projects", all_proj_options), + ): + if _fk in st.session_state: + st.session_state[_fk] = [v for v in st.session_state[_fk] if v in _fopts] + + with fc1: + sel_repos = st.multiselect("Repository", all_repos, key="filter_repos") + with fc2: + sel_missing = st.multiselect("Missing field", all_missing, key="filter_missing") + with fc3: + sel_assignees = st.multiselect("Assignee", all_assignees, key="filter_assignees") + with fc4: + sel_types = st.multiselect("Type", all_types, key="filter_types") + with fc5: + sel_states = st.multiselect( + "State", all_states, key="filter_states", format_func=_state_cell + ) + with fc6: + sel_proj_labels = st.multiselect("Project", all_proj_options, key="filter_projects") + with fc7: + st.markdown('
', unsafe_allow_html=True) + with st.popover( + _date_btn_label, use_container_width=True, help="Filter by last updated date" + ): + _dc1, _dc2 = st.columns(2) + with _dc1: + date_from = st.date_input("From", value=None, key="filter_date_from") + with _dc2: + date_to = st.date_input("To", value=None, key="filter_date_to") + + sel_proj_nums = {p for p, lbl in proj_labels.items() if lbl in sel_proj_labels} + + filtered: list[FindingRow] = [] + for row in rows: + if sel_repos and row["repository"] not in sel_repos: continue - if date_from and row_date < date_from: + # exact membership, not substring: "assignee" must not match "target assignee", + # login "jan" must not match "janedoe" + row_missing = {part.strip() for part in row["missing_fields"].split(",")} + if sel_missing and not any(f in row_missing for f in sel_missing): continue - if date_to and row_date > date_to: + row_assignees = {part.strip() for part in row["assignees"].split(",")} + if sel_assignees and not any(a in row_assignees for a in sel_assignees): continue - filtered.append(row) - -st.caption(f"Showing **{len(filtered)}** of {len(rows)} findings") + if sel_types and row["item_type"] not in sel_types: + continue + if sel_states and row["state"] not in sel_states: + continue + if sel_proj_nums and row["project"] not in sel_proj_nums: + continue + if date_from or date_to: + raw = row["updated_at"] + try: + row_date = date.fromisoformat(raw[:10]) if raw else None + except ValueError: + row_date = None + if row_date is None: + continue + if date_from and row_date < date_from: + continue + if date_to and row_date > date_to: + continue + filtered.append(row) -with _ai_popover: - _render_agent_assistant(filtered) + st.caption(f"Showing **{len(filtered)}** of {len(rows)} findings") -# ── results table ───────────────────────────────────────────────────────────── -st.dataframe( - [ - { - "Project": row["project"], - "Repository": row["repository"], - "Type": row["item_type"], - "#": row["number"], - "Missing": row["missing_fields"], - "Updated": row["updated_at"], - "Title": row["title"], - "Assignees": row["assignees"], - "URL": row["url"], - } - for row in filtered - ], - use_container_width=True, - hide_index=True, - height=min(600, 100 + len(filtered) * 35), - column_config={ - "URL": st.column_config.LinkColumn("Link", display_text="Open β†—"), - "#": st.column_config.NumberColumn("#", format="%d", width="small"), - "Project": st.column_config.NumberColumn("Project", format="%d", width="small"), - "Type": st.column_config.TextColumn("Type", width="small"), - "Missing": st.column_config.TextColumn("Missing", width="large"), - "Updated": st.column_config.TextColumn("Updated", width="small"), - }, -) + with _ai_popover: + _render_agent_assistant(filtered) -st.download_button( - "⬇️ Download filtered Excel", - _xlsx_bytes(filtered), - "findings.xlsx", - _XLSX_MIME, - on_click="ignore", -) + # ── results table ───────────────────────────────────────────────────────── + st.dataframe( + [ + { + "Project": row["project"], + "Repository": row["repository"], + "Type": row["item_type"], + "State": _state_cell(row["state"]), + "#": row["number"], + "Missing": row["missing_fields"], + "Updated": row["updated_at"], + "Title": row["title"], + "Assignees": row["assignees"], + "URL": row["url"], + } + for row in filtered + ], + use_container_width=True, + hide_index=True, + height=min(600, 100 + len(filtered) * 35), + column_config={ + "URL": st.column_config.LinkColumn("Link", display_text="Open β†—"), + "#": st.column_config.NumberColumn("#", format="%d", width="small"), + "Project": st.column_config.NumberColumn("Project", format="%d", width="small"), + "Type": st.column_config.TextColumn("Type", width="small"), + "State": st.column_config.TextColumn("State", width="small"), + "Missing": st.column_config.TextColumn("Missing", width="large"), + "Updated": st.column_config.TextColumn("Updated", width="small"), + }, + ) -with st.expander("πŸ“Š Missing field breakdown"): - field_counts: dict[str, int] = {} - for row in rows: - for _f in row["missing_fields"].split(","): - _f = _f.strip() - if _f: - field_counts[_f] = field_counts.get(_f, 0) + 1 - st.bar_chart( - [{"Field": _f, "Count": c} for _f, c in sorted(field_counts.items(), key=lambda x: x[1])], - x="Field", - y="Count", - horizontal=True, + dl1, dl2, _ = st.columns([1, 1, 1.6]) + dl1.download_button( + "⬇️ Download filtered Excel", + _xlsx_bytes(filtered), + "findings.xlsx", + _XLSX_MIME, + on_click="ignore", + ) + findings_html = _findings_offline_html(rows) + dl2.download_button( + "🌐 Download offline HTML report", + findings_html, + "github-audit-report.html", + "text/html", + on_click=_save_export_copy, + args=("github-audit-report.html", findings_html), + help=( + "Self-contained page with all scanned findings β€” anyone can open it " + "in a browser and filter, no install needed. A copy is also saved " + "to the app's exports folder." + ), ) -limitations = cast(list[str], st.session_state.limitations) -if limitations: - with st.expander(f"⚠️ {len(limitations)} scan limitation(s)"): - for lim in limitations: - st.warning(lim, icon="⚠️") + with st.expander("πŸ“Š Missing field breakdown"): + field_counts: dict[str, int] = {} + for row in rows: + for _f in row["missing_fields"].split(","): + _f = _f.strip() + if _f: + field_counts[_f] = field_counts.get(_f, 0) + 1 + st.bar_chart( + [ + {"Field": _f, "Count": c} + for _f, c in sorted(field_counts.items(), key=lambda x: x[1]) + ], + x="Field", + y="Count", + horizontal=True, + ) + + limitations = cast(list[str], st.session_state.limitations) + if limitations: + with st.expander(f"⚠️ {len(limitations)} scan limitation(s)"): + for lim in limitations: + st.warning(lim, icon="⚠️") + + +# ── main content ────────────────────────────────────────────────────────────── +_h1, _h2 = st.columns([7, 1]) +with _h1: + st.title("πŸ” GitHub Audit") +with _h2: + st.write("") + # Filled after the table filters run so the AI dropdown mirrors the table. + _ai_popover = st.popover("AI", use_container_width=True, help="Open AI assistant") + +if st.session_state.error: + st.error(st.session_state.error) + +_tab_findings, _tab_branches = st.tabs(["πŸ“‹ Issues & PRs", "🌿 Branches"]) +with _tab_findings: + _render_findings_tab() +with _tab_branches: + _render_branch_tab() diff --git a/src/github_audit/branches.py b/src/github_audit/branches.py new file mode 100644 index 0000000..6af24f1 --- /dev/null +++ b/src/github_audit/branches.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +from urllib.parse import quote + +from github_audit.github_client import ( + GitHubClient, + JsonObject, + as_list, + as_object, + optional_str, + required_int, + required_str, +) +from github_audit.models import BranchInfo, BranchPullRequest + +BRANCHES_QUERY = """ +query RepoBranches($owner: String!, $name: String!, $after: String) { + repository(owner: $owner, name: $name) { + defaultBranchRef { name } + refs(refPrefix: "refs/heads/", first: 100, after: $after) { + nodes { + name + target { + __typename + ... on Commit { + committedDate + messageHeadline + author { + user { login } + name + } + } + } + associatedPullRequests(first: 5, orderBy: {field: UPDATED_AT, direction: DESC}) { + totalCount + nodes { + number + state + isDraft + title + url + } + } + } + pageInfo { hasNextPage endCursor } + } + } +} +""" + +_BRANCH_MAX_WORKERS = 8 + + +def fetch_branches(client: GitHubClient, repositories: list[str]) -> list[BranchInfo]: + """Fetch all branches for the given "owner/name" repositories, concurrently.""" + if not repositories: + return [] + with ThreadPoolExecutor(max_workers=min(_BRANCH_MAX_WORKERS, len(repositories))) as pool: + futures = [pool.submit(fetch_repo_branches, client, repo) for repo in repositories] + return [branch for future in futures for branch in future.result()] + + +def fetch_repo_branches(client: GitHubClient, repository: str) -> list[BranchInfo]: + owner, _, name = repository.partition("/") + branches: list[BranchInfo] = [] + after: str | None = None + while True: + data = client.graphql(BRANCHES_QUERY, {"owner": owner, "name": name, "after": after}) + repo = as_object(data.get("repository"), "repository") + default_ref = repo.get("defaultBranchRef") + default_name = ( + optional_str(as_object(default_ref, "defaultBranchRef").get("name")) + if default_ref is not None + else None + ) + connection = as_object(repo.get("refs"), "repository.refs") + for node in as_list(connection.get("nodes"), "repository.refs.nodes"): + branches.append(parse_branch(as_object(node, "branch"), repository, default_name)) + page_info = as_object(connection.get("pageInfo"), "repository.refs.pageInfo") + if page_info.get("hasNextPage") is not True: + break + after = optional_str(page_info.get("endCursor")) + return branches + + +def parse_branch(raw: JsonObject, repository: str, default_name: str | None) -> BranchInfo: + name = required_str(raw.get("name"), "branch name") + target = raw.get("target") + commit = as_object(target, "branch target") if isinstance(target, dict) else {} + committer: str | None = None + author_raw = commit.get("author") + if isinstance(author_raw, dict): + user_raw = author_raw.get("user") + committer = ( + optional_str(as_object(user_raw, "commit author user").get("login")) + if isinstance(user_raw, dict) + else None + ) or optional_str(author_raw.get("name")) + prs_raw = as_object(raw.get("associatedPullRequests"), "associatedPullRequests") + pull_requests = [ + BranchPullRequest( + number=required_int(pr.get("number"), "pr number"), + state=required_str(pr.get("state"), "pr state"), + is_draft=pr.get("isDraft") is True, + title=required_str(pr.get("title"), "pr title"), + url=required_str(pr.get("url"), "pr url"), + ) + for node in as_list(prs_raw.get("nodes"), "associatedPullRequests.nodes") + for pr in (as_object(node, "associated pull request"),) + ] + return BranchInfo( + repository=repository, + name=name, + url=f"https://github.com/{repository}/tree/{quote(name, safe='/')}", + is_default=name == default_name, + last_commit_date=optional_str(commit.get("committedDate")), + last_committer=committer, + last_commit_message=optional_str(commit.get("messageHeadline")) or "", + pull_requests=pull_requests, + pull_requests_total=required_int( + prs_raw.get("totalCount"), "associatedPullRequests.totalCount" + ), + ) diff --git a/src/github_audit/models.py b/src/github_audit/models.py index 3886373..a45bcfd 100644 --- a/src/github_audit/models.py +++ b/src/github_audit/models.py @@ -61,6 +61,7 @@ class ProjectItem(BaseModel): milestone: str | None = None updated_at: str | None = None state: str = "" + is_draft: bool = False field_values: dict[str, ProjectFieldValue] = Field(default_factory=dict) linked_pull_requests_count: int = 0 closing_issues_count: int = 0 @@ -94,6 +95,7 @@ class GitHubPullRequest(BaseModel): title: str url: str state: str + is_draft: bool = False body: str assignees: list[str] = Field(default_factory=list) labels: list[str] = Field(default_factory=list) @@ -194,6 +196,8 @@ class AuditFinding(BaseModel): comments: list[GitHubComment] = Field(default_factory=_empty_comments, exclude=True) comments_total_count: int = Field(default=0, exclude=True) url: str + state: str = "" + is_draft: bool = False assignees: list[str] labels: list[str] = Field(default_factory=list) milestone: str | None = None @@ -205,6 +209,13 @@ class AuditFinding(BaseModel): llm_suggestion: LLMSuggestion | None = None apply_status: str = "not_planned" + @property + def display_state(self) -> str: + """Human state label: Open, Draft, Merged, or Closed.""" + if self.is_draft and self.state.upper() == "OPEN": + return "Draft" + return self.state.capitalize() if self.state else "" + class AuditResult(BaseModel): model_config = ConfigDict(extra="forbid") @@ -388,6 +399,39 @@ class BrowserScanResult(BaseModel): limitations: list[str] = Field(default_factory=list) +class BranchPullRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + number: int + state: str + is_draft: bool = False + title: str + url: str + + +class BranchInfo(BaseModel): + model_config = ConfigDict(extra="forbid") + + repository: str + name: str + url: str + is_default: bool = False + last_commit_date: str | None = None + last_committer: str | None = None + last_commit_message: str = "" + pull_requests: list[BranchPullRequest] = Field(default_factory=list[BranchPullRequest]) + pull_requests_total: int = 0 + + @property + def pr_state(self) -> str: + """Best associated-PR state: open beats merged beats closed.""" + states = {pr.state.upper() for pr in self.pull_requests} + for state, label in (("OPEN", "open"), ("MERGED", "merged"), ("CLOSED", "closed")): + if state in states: + return label + return "none" + + class MyWorkItem(BaseModel): model_config = ConfigDict(extra="forbid") diff --git a/src/github_audit/offline_export.py b/src/github_audit/offline_export.py new file mode 100644 index 0000000..b63352c --- /dev/null +++ b/src/github_audit/offline_export.py @@ -0,0 +1,312 @@ +"""Self-contained offline HTML export. + +Produces a single .html file with the audit data embedded as JSON and a small +vanilla-JS filter UI (search, per-column multi-select, sorting). Opens by +double-click in any browser β€” no Streamlit, terminal, or network needed. +""" + +from __future__ import annotations + +import html +import json +from collections.abc import Mapping, Sequence +from dataclasses import asdict, dataclass +from typing import Literal + +CellValue = str | int + + +@dataclass(frozen=True) +class OfflineColumn: + key: str + label: str + kind: Literal["text", "badge", "link", "number"] = "text" + filterable: bool = False + # Multi-valued cells ("a, b") contribute each part as its own filter option. + split: bool = False + # Numeric column gets a "β‰₯ N" input filter (e.g. minimum branch age). + min_filter: bool = False + + +def render_offline_html( + title: str, + subtitle: str, + columns: Sequence[OfflineColumn], + rows: Sequence[Mapping[str, CellValue]], +) -> str: + for row in rows: + for column in columns: + if column.key not in row: + msg = f"row is missing column {column.key!r}" + raise ValueError(msg) + # " blocks ("" would + # terminate them mid-JSON); "<\/" is identical after JSON.parse. + columns_json = json.dumps([asdict(c) for c in columns], ensure_ascii=False) + rows_json = json.dumps([dict(r) for r in rows], ensure_ascii=False) + return ( + _TEMPLATE.replace("__TITLE__", html.escape(title)) + .replace("__SUBTITLE__", html.escape(subtitle)) + .replace("__COLUMNS__", columns_json.replace(" + + + + +__TITLE__ + + + +

__TITLE__

+

__SUBTITLE__

+
+ + +
+
+ + +
+ + + + + +""" diff --git a/src/github_audit/project_fields.py b/src/github_audit/project_fields.py index e9d3470..3610e9e 100644 --- a/src/github_audit/project_fields.py +++ b/src/github_audit/project_fields.py @@ -122,6 +122,7 @@ title url state + isDraft updatedAt body repository { nameWithOwner } @@ -288,6 +289,7 @@ title url state + isDraft updatedAt body repository { nameWithOwner } @@ -695,6 +697,7 @@ def parse_project_item(raw: JsonObject) -> ProjectItem: milestone=parsed_content.milestone, updated_at=parsed_content.updated_at, state=parsed_content.state, + is_draft=isinstance(parsed_content, GitHubPullRequest) and parsed_content.is_draft, field_values=field_values, linked_pull_requests_count=( parsed_content.linked_pull_requests_count @@ -736,6 +739,7 @@ def parse_content(raw: JsonObject) -> GitHubContent | None: title=required_str(raw.get("title"), "pull request title"), url=required_str(raw.get("url"), "pull request url"), state=required_str(raw.get("state"), "pull request state"), + is_draft=raw.get("isDraft") is True, updated_at=optional_str(raw.get("updatedAt")), body=optional_str(raw.get("body")) or "", assignees=parse_named_nodes(raw, "assignees", "login"), diff --git a/src/github_audit/rules.py b/src/github_audit/rules.py index e5d719b..fa8071f 100644 --- a/src/github_audit/rules.py +++ b/src/github_audit/rules.py @@ -54,6 +54,8 @@ def evaluate_item( comments=content.comments, comments_total_count=content.comments_total_count, url=content.url, + state=content.state, + is_draft=not isinstance(content, GitHubIssue) and content.is_draft, assignees=content.assignees, labels=content.labels, milestone=content.milestone, diff --git a/src/github_audit/scanner.py b/src/github_audit/scanner.py index 2638b5d..29b0156 100644 --- a/src/github_audit/scanner.py +++ b/src/github_audit/scanner.py @@ -173,6 +173,7 @@ def content_from_project_item( title=item.title, url=item.url, state=item.state, + is_draft=item.is_draft, body=item.body, assignees=item.assignees, labels=item.labels, diff --git a/tests/test_branches.py b/tests/test_branches.py new file mode 100644 index 0000000..a710f11 --- /dev/null +++ b/tests/test_branches.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +from github_audit.branches import parse_branch +from github_audit.github_client import JsonObject, JsonValue +from github_audit.models import AuditFinding, BranchInfo, BranchPullRequest + + +def _raw_branch(name: str, prs: list[JsonValue]) -> JsonObject: + return { + "name": name, + "target": { + "__typename": "Commit", + "committedDate": "2026-06-01T10:00:00Z", + "messageHeadline": "fix: something", + "author": {"user": {"login": "majkey"}, "name": "Michal"}, + }, + "associatedPullRequests": {"totalCount": len(prs), "nodes": prs}, + } + + +def test_parse_branch_full() -> None: + raw = _raw_branch( + "feat/x", + [ + { + "number": 7, + "state": "MERGED", + "isDraft": False, + "title": "Feat x", + "url": "https://github.com/o/r/pull/7", + } + ], + ) + branch = parse_branch(raw, "o/r", "main") + assert branch.name == "feat/x" + assert branch.is_default is False + assert branch.last_commit_date == "2026-06-01T10:00:00Z" + assert branch.last_committer == "majkey" + assert branch.url == "https://github.com/o/r/tree/feat/x" + assert branch.pull_requests[0].state == "MERGED" + + +def test_parse_branch_default_and_url_quoting() -> None: + raw = _raw_branch("fix/#12 test", []) + raw["name"] = "fix/#12 test" + branch = parse_branch(raw, "o/r", "fix/#12 test") + assert branch.is_default is True + assert branch.url == "https://github.com/o/r/tree/fix/%2312%20test" + + +def test_parse_branch_missing_commit_author_falls_back_to_name() -> None: + raw = _raw_branch("b", []) + target = raw["target"] + assert isinstance(target, dict) + target["author"] = {"user": None, "name": "Ext Committer"} + branch = parse_branch(raw, "o/r", "main") + assert branch.last_committer == "Ext Committer" + + +def test_pr_state_priority() -> None: + def info(states: list[str]) -> BranchInfo: + return BranchInfo( + repository="o/r", + name="b", + url="https://github.com/o/r/tree/b", + pull_requests=[ + BranchPullRequest(number=i, state=s, title="t", url="u") + for i, s in enumerate(states, start=1) + ], + ) + + assert info([]).pr_state == "none" + assert info(["CLOSED"]).pr_state == "closed" + assert info(["CLOSED", "MERGED"]).pr_state == "merged" + assert info(["MERGED", "OPEN"]).pr_state == "open" + + +def test_display_state_labels() -> None: + def finding(state: str, *, is_draft: bool = False) -> AuditFinding: + return AuditFinding( + repository="o/r", + item_type="pull_request", + number=1, + title="t", + url="u", + state=state, + is_draft=is_draft, + assignees=[], + missing_fields=["assignee"], + development_status="closing_issues=0", + ) + + assert finding("OPEN").display_state == "Open" + assert finding("OPEN", is_draft=True).display_state == "Draft" + assert finding("MERGED").display_state == "Merged" + assert finding("CLOSED", is_draft=True).display_state == "Closed" + assert finding("").display_state == "" diff --git a/tests/test_offline_export.py b/tests/test_offline_export.py new file mode 100644 index 0000000..d60da5c --- /dev/null +++ b/tests/test_offline_export.py @@ -0,0 +1,41 @@ +from __future__ import annotations + +import pytest + +from github_audit.offline_export import OfflineColumn, render_offline_html + +_COLUMNS = [ + OfflineColumn("repo", "Repository", filterable=True), + OfflineColumn("state", "State", "badge", filterable=True), + OfflineColumn("age", "Age", "number", min_filter=True), + OfflineColumn("url", "Link", "link"), +] + + +def test_render_embeds_rows_and_columns() -> None: + html = render_offline_html( + "Title ", + "Sub & title", + _COLUMNS, + [{"repo": "r1", "state": "Open", "age": 3, "url": "https://github.com/o/r"}], + ) + assert "Title <x>" in html + assert "Sub & title" in html + assert '"repo": "r1"' in html + assert '"min_filter": true' in html + + +def test_render_neutralizes_script_terminator_in_data() -> None: + html = render_offline_html( + "T", + "S", + _COLUMNS, + [{"repo": "", "state": "", "age": 0, "url": ""}], + ) + assert "