Show unlinked community PRs on the Open-tasks page (#409)

tools/generate_open_tasks.py

@@ -103,6 +103,8 @@
 
 LABEL = "help wanted"
 ORG = "flipperdevices"
+SECTION_SLUGS = {section["slug"] for section in SECTIONS}
+ORG_MEMBER_ASSOCIATIONS = {"MEMBER", "OWNER"}
 SCRIPT_URL = "https://github.com/flipperdevices/flipperone-docs/blob/public-release/tools/generate_open_tasks.py"
 TEMPLATE_DIR = Path(__file__).resolve().parent / "templates"
 TEMPLATE_NAME = "open_tasks.md.j2"
@@ -152,6 +154,30 @@ class _DisplayIssue(TypedDict):
 class _DisplaySection(_Section):
     issues: list[_DisplayIssue]
 
+class _Author(TypedDict, total=False):
+    login: str
+    is_bot: bool
+
+class _CommunityPR(TypedDict, total=False):
+    id: str
+    repository: _Repository
+    number: int
+    title: str
+    url: str
+    isDraft: bool
+    commentsCount: int
+    author: _Author
+    authorAssociation: str
+
+class _DisplayCommunityPR(TypedDict):
+    number: int
+    url: str
+    title: str
+    repo: str
+    author: str
+    comments_count: int
+    is_draft: bool
+
 
 def fetch_issues() -> list[_Issue]:
     """Fetch all open `help wanted` issues across the org, each with its `linkedPRs` var."""
@@ -206,6 +232,73 @@ def fetch_linked_prs(repo_full: str, number: int) -> list[_LinkedPR]:
     return nodes
 
 
+def is_community_pr(pr: _CommunityPR) -> bool:
+    """True if the PR is from a non-org, non-bot author."""
+    if pr.get("author", {}).get("is_bot"):
+        return False
+    return pr.get("authorAssociation", "").upper() not in ORG_MEMBER_ASSOCIATIONS
+
+
+_LINKED_ISSUES_QUERY = """
+query($ids: [ID!]!) {
+  nodes(ids: $ids) {
+    ... on PullRequest {
+      id
+      closingIssuesReferences(first: 1) { totalCount }
+    }
+  }
+}
+"""
+
+
+def fetch_pr_ids_with_linked_issues(pr_ids: list[str]) -> set[str]:
+    """Return IDs of PRs that close or link an issue in one GraphQL request."""
+    if not pr_ids:
+        return set()
+    result = subprocess.run(
+        [
+            "gh", "api", "graphql",
+            "-f", f"query={_LINKED_ISSUES_QUERY}",
+            *(arg for pr_id in pr_ids for arg in ("-F", f"ids[]={pr_id}")),
+        ],
+        capture_output=True, text=True, check=True,
+    )
+    data = json.loads(result.stdout)
+    nodes = data.get("data", {}).get("nodes") or []
+    return {
+        node["id"]
+        for node in nodes
+        if node
+        and ((node.get("closingIssuesReferences") or {}).get("totalCount") or 0) > 0
+    }
+
+
+def fetch_community_prs() -> list[_CommunityPR]:
+    """Fetch open PRs in the sub-project repos that are from community authors and
+    not linked to any issue."""
+    result = subprocess.run(
+        [
+            "gh", "search", "prs",
+            "--owner", ORG,
+            "--state", "open",
+            "--limit", "200",
+            "--json", "id,repository,number,title,url,isDraft,commentsCount,author,authorAssociation",
+        ],
+        capture_output=True, text=True, check=True,
+    )
+    prs: list[_CommunityPR] = json.loads(result.stdout)
+    candidates = [
+        pr
+        for pr in prs
+        if pr["repository"]["nameWithOwner"] in SECTION_SLUGS
+        and is_community_pr(pr)
+    ]
+    linked_pr_ids = fetch_pr_ids_with_linked_issues(
+        [pr["id"] for pr in candidates]
+    )
+    return [pr for pr in candidates if pr["id"] not in linked_pr_ids]
+
+
 def load_issues_from_file(path: Path) -> list[_Issue]:
     """For local testing without `gh`: load a JSON array of issues from a file."""
     return json.loads(path.read_text(encoding="utf-8"))  # type: ignore[no-any-return]
@@ -307,6 +400,25 @@ def _build_template_sections(issues: list[_Issue]) -> list[_DisplaySection]:
     return sections
 
 
+def _build_community_prs(prs: list[_CommunityPR]) -> list[_DisplayCommunityPR]:
+    """Prepare community PR data for the Open Tasks Jinja template."""
+    display: list[_DisplayCommunityPR] = []
+    for pr in prs:
+        repo_full = pr["repository"]["nameWithOwner"]
+        repo_short = repo_full.split("/", 1)[1] if "/" in repo_full else repo_full
+        display.append({
+            "number": pr["number"],
+            "url": html_escape(pr["url"]),
+            "title": html_escape(pr["title"]),
+            "repo": html_escape(repo_short),
+            "author": html_escape(pr.get("author", {}).get("login", "")),
+            "comments_count": pr.get("commentsCount", 0),
+            "is_draft": pr.get("isDraft", False),
+        })
+    display.sort(key=lambda p: (p["repo"], -p["number"]))
+    return display
+
+
 def _render_open_tasks_template(**context: object) -> str:
     env = Environment(
         loader=FileSystemLoader(TEMPLATE_DIR),
@@ -318,7 +430,11 @@ def _render_open_tasks_template(**context: object) -> str:
     return env.get_template(TEMPLATE_NAME).render(**context)
 
 
-def generate_page(issues: list[_Issue], existing_created_at: str | None = None) -> str:
+def generate_page(
+    issues: list[_Issue],
+    existing_created_at: str | None = None,
+    community_prs: list[_CommunityPR] | None = None,
+) -> str:
     now = datetime.now(timezone.utc)
     created_at = existing_created_at or archbee_timestamp(now)
     updated_at = archbee_timestamp(now)
@@ -328,6 +444,7 @@ def generate_page(issues: list[_Issue], existing_created_at: str | None = None)
         updated_at=updated_at,
         script_url=SCRIPT_URL,
         sections=_build_template_sections(issues),
+        community_prs=_build_community_prs(community_prs or []),
     )
     return page.rstrip() + "\n"
 
@@ -349,7 +466,9 @@ def main() -> int:
 
     issues = (load_issues_from_file(args.input_json)
               if args.input_json else fetch_issues())
-    page = generate_page(issues, existing_created_at=existing_created_at)
+    community_prs = [] if args.input_json else fetch_community_prs()
+    page = generate_page(issues, existing_created_at=existing_created_at,
+                         community_prs=community_prs)
 
     if args.out:
         if args.out.exists():

tools/templates/open_tasks.md.j2

@@ -66,3 +66,32 @@ Learn what's going on in the {{ section.title }} sub-project:
 :::
 
 {% endfor %}
+{% if community_prs %}
+***
+
+# 🔀 Community pull requests
+
+Open pull requests from community contributors that aren't linked to any task above. Help review and discuss them.
+
+<table isTableHeaderOn="true" columnWidths="536,125">
+  <tr>
+    <td align="left">
+      <p><strong>Pull request</strong></p>
+    </td>
+    <td align="center">
+      <p><strong>Comments</strong></p>
+    </td>
+  </tr>
+{% for pr in community_prs %}
+  <tr>
+    <td align="left" colSpan="1" rowSpan="1">
+      <p>🔀 <a href="{{ pr.url }}">{{ pr.repo }}#{{ pr.number }}</a> {{ pr.title }}{% if pr.is_draft %} (draft){% endif %}</p>
+      <p>by {{ pr.author }}</p>
+    </td>
+    <td align="center" colSpan="1" rowSpan="1">
+      <p>💬 {{ pr.comments_count }}</p>
+    </td>
+  </tr>
+{% endfor %}
+</table>
+{% endif %}

tools/test_generate_open_tasks.py

@@ -7,7 +7,15 @@
 from unittest import mock
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
-from generate_open_tasks import fetch_linked_prs, generate_page, make_summary, _Issue
+from generate_open_tasks import (
+    fetch_linked_prs,
+    fetch_pr_ids_with_linked_issues,
+    generate_page,
+    is_community_pr,
+    make_summary,
+    _CommunityPR,
+    _Issue,
+)
 
 
 class GenerateOpenTasksTest(unittest.TestCase):
@@ -127,6 +135,78 @@ def test_fetch_linked_prs_handles_missing_issue(self) -> None:
             )
             self.assertEqual(fetch_linked_prs("o/r", 1), [])
 
+    def test_is_community_pr_filters_members_and_bots(self) -> None:
+        cases: list[tuple[_CommunityPR, bool]] = [
+            ({"authorAssociation": "MEMBER", "author": {"login": "a"}}, False),
+            ({"authorAssociation": "OWNER", "author": {"login": "b"}}, False),
+            ({"authorAssociation": "NONE", "author": {"is_bot": True}}, False),
+            ({"authorAssociation": "NONE", "author": {"login": "c"}}, True),
+            ({"authorAssociation": "CONTRIBUTOR", "author": {"login": "d"}}, True),
+        ]
+        for pr, expected in cases:
+            with self.subTest(pr=pr):
+                self.assertEqual(is_community_pr(pr), expected)
+
+    def test_fetch_pr_ids_with_linked_issues_batches_prs(self) -> None:
+        payload = {
+            "data": {
+                "nodes": [
+                    {
+                        "id": "PR_1",
+                        "closingIssuesReferences": {"totalCount": 0},
+                    },
+                    {
+                        "id": "PR_2",
+                        "closingIssuesReferences": {"totalCount": 2},
+                    },
+                    None,
+                ]
+            }
+        }
+        with mock.patch("generate_open_tasks.subprocess.run") as run:
+            run.return_value = mock.Mock(stdout=json.dumps(payload))
+            linked_ids = fetch_pr_ids_with_linked_issues(
+                ["PR_1", "PR_2", "PR_MISSING"]
+            )
+
+        self.assertEqual(linked_ids, {"PR_2"})
+        run.assert_called_once()
+        args = run.call_args[0][0]
+        self.assertIn("ids[]=PR_1", args)
+        self.assertIn("ids[]=PR_2", args)
+        self.assertIn("ids[]=PR_MISSING", args)
+
+    def test_fetch_pr_ids_with_linked_issues_skips_empty_request(self) -> None:
+        with mock.patch("generate_open_tasks.subprocess.run") as run:
+            self.assertEqual(fetch_pr_ids_with_linked_issues([]), set())
+            run.assert_not_called()
+
+    def test_generate_page_renders_community_prs(self) -> None:
+        community_prs: list[_CommunityPR] = [
+            {
+                "repository": {"nameWithOwner": "flipperdevices/flipperone-docs"},
+                "number": 356,
+                "title": 'Fix "typo" & stuff',
+                "url": "https://github.com/flipperdevices/flipperone-docs/pull/356",
+                "isDraft": False,
+                "commentsCount": 3,
+                "author": {"login": "someone"},
+                "authorAssociation": "NONE",
+            }
+        ]
+
+        page = generate_page([], community_prs=community_prs)
+
+        self.assertIn("# 🔀 Community pull requests", page)
+        self.assertIn(
+            '<p>🔀 <a href="https://github.com/flipperdevices/flipperone-docs/pull/356">flipperone-docs#356</a> Fix &quot;typo&quot; &amp; stuff</p>',
+            page,
+        )
+        self.assertIn("<p>by someone</p>", page)
+
+    def test_generate_page_omits_community_section_when_empty(self) -> None:
+        self.assertNotIn("# 🔀 Community pull requests", generate_page([]))
+
 
 if __name__ == "__main__":
     unittest.main()