Add mention decorator for GitHub command handling

src/django_github_app/mentions.py

@@ -0,0 +1,193 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from enum import Enum
+from typing import NamedTuple
+
+from gidgethub import sansio
+
+
+class EventAction(NamedTuple):
+    event: str
+    action: str
+
+
+class MentionScope(str, Enum):
+    COMMIT = "commit"
+    ISSUE = "issue"
+    PR = "pr"
+
+    def get_events(self) -> list[EventAction]:
+        match self:
+            case MentionScope.ISSUE:
+                return [
+                    EventAction("issue_comment", "created"),
+                ]
+            case MentionScope.PR:
+                return [
+                    EventAction("issue_comment", "created"),
+                    EventAction("pull_request_review_comment", "created"),
+                    EventAction("pull_request_review", "submitted"),
+                ]
+            case MentionScope.COMMIT:
+                return [
+                    EventAction("commit_comment", "created"),
+                ]
+
+    @classmethod
+    def all_events(cls) -> list[EventAction]:
+        return list(
+            dict.fromkeys(
+                event_action for scope in cls for event_action in scope.get_events()
+            )
+        )
+
+    @classmethod
+    def from_event(cls, event: sansio.Event) -> MentionScope | None:
+        if event.event == "issue_comment":
+            issue = event.data.get("issue", {})
+            is_pull_request = (
+                "pull_request" in issue and issue["pull_request"] is not None
+            )
+            return cls.PR if is_pull_request else cls.ISSUE
+
+        for scope in cls:
+            scope_events = scope.get_events()
+            if any(event_action.event == event.event for event_action in scope_events):
+                return scope
+
+        return None
+
+
+@dataclass
+class RawMention:
+    match: re.Match[str]
+    username: str
+    position: int
+    end: int
+
+
+CODE_BLOCK_PATTERN = re.compile(r"```[\s\S]*?```", re.MULTILINE)
+INLINE_CODE_PATTERN = re.compile(r"`[^`]+`")
+BLOCKQUOTE_PATTERN = re.compile(r"^\s*>.*$", re.MULTILINE)
+# GitHub username rules:
+# - 1-39 characters long
+# - Can only contain alphanumeric characters or hyphens
+# - Cannot start or end with a hyphen
+# - Cannot have multiple consecutive hyphens
+GITHUB_MENTION_PATTERN = re.compile(
+    r"(?:^|(?<=\s))@([a-z\d](?:[a-z\d]|-(?=[a-z\d])){0,38})",
+    re.MULTILINE | re.IGNORECASE,
+)
+
+
+def extract_all_mentions(text: str) -> list[RawMention]:
+    # replace all code blocks, inline code, and blockquotes with spaces
+    # this preserves linenos and postitions while not being able to
+    # match against anything in them
+    processed_text = CODE_BLOCK_PATTERN.sub(lambda m: " " * len(m.group(0)), text)
+    processed_text = INLINE_CODE_PATTERN.sub(
+        lambda m: " " * len(m.group(0)), processed_text
+    )
+    processed_text = BLOCKQUOTE_PATTERN.sub(
+        lambda m: " " * len(m.group(0)), processed_text
+    )
+    return [
+        RawMention(
+            match=match,
+            username=match.group(1),
+            position=match.start(),
+            end=match.end(),
+        )
+        for match in GITHUB_MENTION_PATTERN.finditer(processed_text)
+    ]
+
+
+class LineInfo(NamedTuple):
+    lineno: int
+    text: str
+
+    @classmethod
+    def for_mention_in_comment(cls, comment: str, mention_position: int):
+        lines = comment.splitlines()
+        text_before = comment[:mention_position]
+        line_number = text_before.count("\n") + 1
+
+        line_index = line_number - 1
+        line_text = lines[line_index] if line_index < len(lines) else ""
+
+        return cls(lineno=line_number, text=line_text)
+
+
+@dataclass
+class ParsedMention:
+    username: str
+    position: int
+    line_info: LineInfo
+    previous_mention: ParsedMention | None = None
+    next_mention: ParsedMention | None = None
+
+
+def matches_pattern(text: str, pattern: str | re.Pattern[str]) -> bool:
+    match pattern:
+        case re.Pattern():
+            return pattern.fullmatch(text) is not None
+        case str():
+            return text.strip().lower() == pattern.strip().lower()
+
+
+def extract_mentions_from_event(
+    event: sansio.Event, username_pattern: str | re.Pattern[str] | None = None
+) -> list[ParsedMention]:
+    comment_key = "comment" if event.event != "pull_request_review" else "review"
+    comment = event.data.get(comment_key, {}).get("body", "")
+
+    if not comment:
+        return []
+
+    mentions: list[ParsedMention] = []
+    potential_mentions = extract_all_mentions(comment)
+    for raw_mention in potential_mentions:
+        if username_pattern and not matches_pattern(
+            raw_mention.username, username_pattern
+        ):
+            continue
+
+        mentions.append(
+            ParsedMention(
+                username=raw_mention.username,
+                position=raw_mention.position,
+                line_info=LineInfo.for_mention_in_comment(
+                    comment, raw_mention.position
+                ),
+                previous_mention=None,
+                next_mention=None,
+            )
+        )
+
+    for i, mention in enumerate(mentions):
+        if i > 0:
+            mention.previous_mention = mentions[i - 1]
+        if i < len(mentions) - 1:
+            mention.next_mention = mentions[i + 1]
+
+    return mentions
+
+
+@dataclass
+class Mention:
+    mention: ParsedMention
+    scope: MentionScope | None
+
+    @classmethod
+    def from_event(
+        cls,
+        event: sansio.Event,
+        *,
+        username: str | re.Pattern[str] | None = None,
+        scope: MentionScope | None = None,
+    ):
+        mentions = extract_mentions_from_event(event, username)
+        for mention in mentions:
+            yield cls(mention=mention, scope=scope)

src/django_github_app/routing.py

@@ -1,36 +1,115 @@
 from __future__ import annotations
 
+import re
+from asyncio import iscoroutinefunction
 from collections.abc import Awaitable
 from collections.abc import Callable
+from functools import wraps
 from typing import Any
+from typing import Protocol
 from typing import TypeVar
+from typing import cast
 
 from django.utils.functional import classproperty
 from gidgethub import sansio
 from gidgethub.routing import Router as GidgetHubRouter
 
 from ._typing import override
+from .github import AsyncGitHubAPI
+from .github import SyncGitHubAPI
+from .mentions import Mention
+from .mentions import MentionScope
 
 AsyncCallback = Callable[..., Awaitable[None]]
 SyncCallback = Callable[..., None]
 
 CB = TypeVar("CB", AsyncCallback, SyncCallback)
 
 
+class AsyncMentionHandler(Protocol):
+    async def __call__(
+        self, event: sansio.Event, *args: Any, **kwargs: Any
+    ) -> None: ...
+
+
+class SyncMentionHandler(Protocol):
+    def __call__(self, event: sansio.Event, *args: Any, **kwargs: Any) -> None: ...
+
+
+MentionHandler = AsyncMentionHandler | SyncMentionHandler
+
+
 class GitHubRouter(GidgetHubRouter):
     _routers: list[GidgetHubRouter] = []
 
     def __init__(self, *args) -> None:
         super().__init__(*args)
         GitHubRouter._routers.append(self)
 
+    @override
+    def add(
+        self, func: AsyncCallback | SyncCallback, event_type: str, **data_detail: Any
+    ) -> None:
+        # Override to accept both async and sync callbacks.
+        super().add(cast(AsyncCallback, func), event_type, **data_detail)
+
     @classproperty
     def routers(cls):
         return list(cls._routers)
 
     def event(self, event_type: str, **kwargs: Any) -> Callable[[CB], CB]:
         def decorator(func: CB) -> CB:
-            self.add(func, event_type, **kwargs)  # type: ignore[arg-type]
+            self.add(func, event_type, **kwargs)
+            return func
+
+        return decorator
+
+    def mention(
+        self,
+        *,
+        username: str | re.Pattern[str] | None = None,
+        scope: MentionScope | None = None,
+        **kwargs: Any,
+    ) -> Callable[[CB], CB]:
+        def decorator(func: CB) -> CB:
+            @wraps(func)
+            async def async_wrapper(
+                event: sansio.Event, gh: AsyncGitHubAPI, *args: Any, **kwargs: Any
+            ) -> None:
+                event_scope = MentionScope.from_event(event)
+                if scope is not None and event_scope != scope:
+                    return
+
+                for mention in Mention.from_event(
+                    event, username=username, scope=event_scope
+                ):
+                    await func(event, gh, *args, context=mention, **kwargs)  # type: ignore[func-returns-value]
+
+            @wraps(func)
+            def sync_wrapper(
+                event: sansio.Event, gh: SyncGitHubAPI, *args: Any, **kwargs: Any
+            ) -> None:
+                event_scope = MentionScope.from_event(event)
+                if scope is not None and event_scope != scope:
+                    return
+
+                for mention in Mention.from_event(
+                    event, username=username, scope=event_scope
+                ):
+                    func(event, gh, *args, context=mention, **kwargs)
+
+            wrapper: MentionHandler
+            if iscoroutinefunction(func):
+                wrapper = cast(AsyncMentionHandler, async_wrapper)
+            else:
+                wrapper = cast(SyncMentionHandler, sync_wrapper)
+
+            events = scope.get_events() if scope else MentionScope.all_events()
+            for event_action in events:
+                self.add(
+                    wrapper, event_action.event, action=event_action.action, **kwargs
+                )
+
             return func
 
         return decorator