Add !checkping command to manually trigger check pings

bubbles/commands/check_ping.py

@@ -0,0 +1,17 @@
+from utonium import Payload, Plugin
+
+from bubbles.commands.periodic.transcription_check_ping import transcription_check_ping
+from bubbles.utils import parse_user
+
+
+def check_ping(payload: Payload) -> None:
+    """!check_ping [user] - Manually trigger a transcription check ping."""
+    tokens = payload.cleaned_text.split()
+
+    user_raw = tokens.get(1)
+    user_filter = parse_user(user_raw) if user_raw is not None else None
+
+    transcription_check_ping(payload.get_channel(), user_filter=user_filter, start_now=True)
+
+
+PLUGIN = Plugin(func=check_ping, regex=r"^checkping")

bubbles/commands/periodic/transcription_check_ping.py

@@ -5,6 +5,8 @@
 from enum import Enum
 from typing import Dict, List, Optional, Tuple, TypedDict
 
+from slack_sdk.web import SlackResponse
+
 from bubbles.commands.helper_functions_history.extract_author import extract_author
 from bubbles.commands.periodic import (
     TRANSCRIPTION_CHECK_CHANNEL,
@@ -267,9 +269,15 @@ def _get_check_fragment(check: CheckData) -> str:
     return f"<{link}|u/{user}>" if link else f"{user} (LINK NOT FOUND)"
 
 
-def _get_check_reminder(aggregate: List) -> str:
+def _get_check_reminder(aggregate: List, user_filter: Optional[str] = None) -> str:
     """Get the reminder text for the checks."""
-    reminder = "*Pending Transcription Checks:*\n\n"
+    if user_filter is not None:
+        reminder = (
+            f"*Pending Transcription Checks for "
+            f"*<https://reddit.com/u/{user_filter}|u/{user_filter}>:*\n\n"
+        )
+    else:
+        reminder = "*Pending Transcription Checks:*\n\n"
 
     for time_str, mod_aggregate in aggregate:
         reminder += f"*{time_str}*:\n"
@@ -295,10 +303,22 @@ def _get_check_reminder(aggregate: List) -> str:
     return reminder
 
 
-def transcription_check_ping_callback() -> None:
+def transcription_check_ping(
+    channel: str,
+    user_filter: Optional[str] = None,
+    start_now: bool = False,
+) -> Optional[SlackResponse]:
+    """Send a reminder about open transcription checks.
+
+    :param channel: The channel to send the reminder into.
+    :param user_filter: Only include checks of the given user (case-insensitive).
+    :param start_now: If set to true, checks will be included in the reminders immediately.
+    Otherwise, they are only included after a delay.
+    :returns: The Slack response of sending the message, or None if something went wrong.
+    """
     now = datetime.now(tz=timezone.utc)
 
-    start_time = now - CHECK_SEARCH_START_DELTA
+    start_time = now if start_now else now - CHECK_SEARCH_START_DELTA
     end_time = now - CHECK_SEARCH_END_DELTA
 
     messages_response = app.client.conversations_history(
@@ -309,17 +329,27 @@ def transcription_check_ping_callback() -> None:
     )
     if not messages_response.get("ok"):
         logging.error(f"Failed to get check messages!\n{messages_response}")
-        return
+        return None
 
     # Get the reminder for the checks
     messages = messages_response["messages"]
     checks = _extract_open_checks(messages)
+
+    # Only consider checks for the given user, if specified
+    if user_filter is not None:
+
+        def matches_filter(check: CheckData) -> bool:
+            check_user = check["user"]
+            return check_user and user_filter.lower() == check_user.lower()
+
+        checks = [check for check in checks if matches_filter(check)]
+
     aggregate = _aggregate_checks_by_time(checks)
-    reminder = _get_check_reminder(aggregate)
+    reminder = _get_check_reminder(aggregate, user_filter=user_filter)
 
     # Post the reminder in Slack
     reminder_response = app.client.chat_postMessage(
-        channel=rooms_list[TRANSCRIPTION_CHECK_PING_CHANNEL],
+        channel=channel,
         link_names=1,
         text=reminder,
         unfurl_links=False,
@@ -328,3 +358,9 @@ def transcription_check_ping_callback() -> None:
     )
     if not reminder_response.get("ok"):
         logging.error(f"Failed to send reminder message!\n{reminder_response}")
+
+    return reminder_response
+
+
+def transcription_check_ping_callback() -> None:
+    transcription_check_ping(channel=rooms_list[TRANSCRIPTION_CHECK_CHANNEL])

bubbles/utils.py

@@ -1,7 +1,7 @@
 import re
 import subprocess
 from datetime import datetime, timedelta
-from typing import List, Optional
+from typing import List, Optional, Match
 
 # First an amount and then a unit
 import pytz as pytz
@@ -20,6 +20,18 @@
     "years": re.compile(r"^y(?:ears?)?$"),
 }
 
+# find a link in the slack format, then strip out the text at the end.
+# they're formatted like this: <https://example.com|Text!>
+SLACK_TEXT_EXTRACTOR = re.compile(
+    # Allow long lines for the regex
+    # flake8: noqa: E501
+    r"<(?P<url>(?:https?://)?[\w-]+(?:\.[\w-]+)+\.?(?::\d+)?(?:/[^\s|]*)?)(?:\|(?P<text>[^>]+))?>"
+)
+
+BOLD_REGEX = re.compile(r"\*(?P<content>[^*]+)\*")
+
+USERNAME_REGEX = re.compile(r"(?:/?u/)?(?P<username>\S+)")
+
 
 class TimeParseError(RuntimeError):
     """Exception raised when a time string is invalid."""
@@ -163,3 +175,53 @@ def parse_time_constraints(
     time_str = f"from {after_time_str} until {before_time_str}"
 
     return after_time, before_time, time_str
+
+
+def extract_text_from_link(text: str) -> str:
+    """Strip link out of auto-generated slack fancy URLS and return the text only."""
+    results = [_ for _ in re.finditer(SLACK_TEXT_EXTRACTOR, text)]
+    # we'll replace things going backwards so that we don't mess up indexing
+    results.reverse()
+
+    def extract_text(mx: Match) -> str:
+        return mx["text"] or mx["url"]
+
+    for match in results:
+        text = text[: match.start()] + extract_text(match) + text[match.end() :]
+    return text
+
+
+def extract_url_from_link(text: str) -> str:
+    """Strip link out of auto-generated slack fancy URLS and return the link only."""
+    results = [_ for _ in re.finditer(SLACK_TEXT_EXTRACTOR, text)]
+    # we'll replace things going backwards so that we don't mess up indexing
+    results.reverse()
+
+    def extract_link(m: Match) -> str:
+        return m["url"]
+
+    for match in results:
+        text = text[: match.start()] + extract_link(match) + text[match.end() :]
+    return text
+
+
+def parse_user(text: str) -> Optional[str]:
+    """Parse a username argument of a Slack command to a user object.
+
+    This takes care of link formatting, bold formatting and the u/ prefix.
+    Returns `None` if the user couldn't be found.
+    """
+    # Remove link formatting
+    username = extract_text_from_link(text)
+
+    # Remove bold formatting
+    bold_match = BOLD_REGEX.match(username)
+    if bold_match:
+        username = bold_match.group("content")
+
+    # Remove u/ prefix
+    prefix_match = USERNAME_REGEX.match(username)
+    if prefix_match:
+        username = prefix_match.group("username")
+
+    return username