feat: add --mentions option and mentioning in notifications

Add --mentions option that accepts member ids and their locations as
hashtags in CSV format for mentioning in notifications. The tool expects
two columns `member_id` and `hashtag`. See location hashtags in the
official channel.

Any valid string path is acceptable. The string could be a URL. Valid
URL schemes include http, ftp, s3, gs and file. For file URLs, a host is
expected. A local file could be: file://localhost/path/to/table.csv.

raid/__init__.py

@@ -1,8 +1,10 @@
 import sys
+from collections import defaultdict
 from copy import copy
 from functools import partial
-from typing import Any, Dict
+from typing import Any, Optional
 
+import pandas as pd
 import typer
 from loguru import logger
 from slack_sdk.http_retry.builtin_async_handlers import (
@@ -27,7 +29,10 @@
 
 
 async def notify_of_alert(
-    event: Message, webhook_client: AsyncWebhookClient, locale: Locale
+    event: Message,
+    webhook_client: AsyncWebhookClient,
+    members_by_hashtag: Optional[dict[str, list[str]]],
+    locale: Locale,
 ) -> None:
     event_logger = logger.bind(event_id=event.id)
     event_logger.debug("event.received", raw_text=event.raw_text)
@@ -39,8 +44,18 @@ async def notify_of_alert(
         event_logger.exception("event.not_parsed", raw_text=event.raw_text)
         return
 
-    message = await format_alert(alert, locale)
-    event_logger.debug("alert.formatted", message=message, locale=locale.value)
+    members = (
+        members_by_hashtag[alert["hashtag"]] if members_by_hashtag is not None else []
+    )
+
+    if members_by_hashtag is not None and not members:
+        event_logger.info("alert.skipped", hashtag=alert["hashtag"])
+        return
+
+    message = await format_alert(alert, members, locale)
+    event_logger.debug(
+        "alert.formatted", message=message, members=members, locale=locale.value
+    )
 
     try:
         response = await webhook_client.send(text=message)
@@ -69,6 +84,14 @@ def main(
     webhook_url: str = typer.Argument(
         ..., envvar="RAID_WEBHOOK_URL", help="Slack incoming webhook."
     ),
+    members: Optional[str] = typer.Option(
+        None,
+        help=(
+            "Member ids and their locations as hashtags in CSV format for mentioning in"
+            " notifications. The tool expects two columns `member_id` and `hashtag`."
+            " See location hashtags in the official channel."
+        ),
+    ),
     chat_id: int = typer.Option(
         default=1766138888,
         help=(
@@ -127,20 +150,42 @@ def main(
         ],
     )
 
+    members_by_hashtag: Optional[dict[str, list[str]]] = None
+    if members is not None:
+        members_df = pd.read_csv(
+            members, names=["member_id", "hashtag"], skiprows=1
+        ).drop_duplicates()
+
+        members_by_hashtag = defaultdict(
+            list,
+            members_df.groupby("hashtag")["member_id"].apply(list).to_dict(),
+        )
+
+        logger.info(
+            "members.loaded",
+            members=members_df["member_id"].nunique(),
+            hashtags=len(members_by_hashtag),
+        )
+
     load_translations(locale)
 
     client = TelegramClient("raid", api_id, api_hash, auto_reconnect=True)
     with client:
         logger.info("client.connected", api_id=api_id, chat_id=chat_id)
 
         client.add_event_handler(
-            partial(notify_of_alert, webhook_client=webhook_client, locale=locale),
+            partial(
+                notify_of_alert,
+                webhook_client=webhook_client,
+                members_by_hashtag=members_by_hashtag,
+                locale=locale,
+            ),
             events.NewMessage(chats=[chat_id]),
         )
         client.run_until_disconnected()
 
 
-def _patch_record(record: Dict[str, Any]) -> None:
+def _patch_record(record: dict[str, Any]) -> None:
     if "alert" in record["extra"]:
         alert = copy(record["extra"]["alert"])
         alert["status"] = str(alert["status"])

raid/formatter.py

@@ -1,6 +1,8 @@
 import asyncio
+from datetime import datetime
 from enum import Enum
 from functools import lru_cache
+from typing import Iterable
 
 from deep_translator import GoogleTranslator
 from i18n import set as _set_i18n_option
@@ -22,19 +24,28 @@ def load_translations(locale: Locale) -> None:
     __import__(f"raid.locales.{locale}")
 
 
-async def format_alert(alert: Alert, locale: Locale) -> str:
+async def format_alert(alert: Alert, member_ids: Iterable[str], locale: Locale) -> str:
     _set_i18n_option("locale", locale)
 
     fields = {
         "status": alert["status"],
-        "time": alert["time"].strftime(_ALERT_TIME_FORMAT),
+        "time": _format_alert_time(alert["time"]),
         "threat": alert["threat"],
         "location": await _translate_alert_location(alert["location"], locale),
+        "mentions": f" {_format_mentions(member_ids)}" if member_ids else "",
     }
 
     return translate(f"{alert['threat']}.{alert['status']}", **fields)
 
 
+def _format_alert_time(time: datetime) -> str:
+    return time.strftime(_ALERT_TIME_FORMAT)
+
+
+def _format_mentions(member_ids: Iterable[str]) -> str:
+    return ", ".join(map(lambda x: f"<@{x}>", set(member_ids))) if member_ids else ""
+
+
 async def _translate_alert_location(location: str, locale: Locale) -> str:
     if locale == _ALERT_LOCATION_SOURCE_LOCALE:
         return location

raid/locales/en.py

@@ -6,22 +6,25 @@
 
 _(
     f"{Threat.AirRaid}.{AlertStatus.Active}",
-    "%{status}%{threat} %{time} Air raid alert in %{location}.",
+    "%{status}%{threat} %{time} Air raid alert in %{location}.%{mentions}",
     __locale__,
 )
 _(
     f"{Threat.AirRaid}.{AlertStatus.Inactive}",
-    "%{status}%{threat} %{time} Air raid alert in %{location} cancelled.",
+    "%{status}%{threat} %{time} Air raid alert in %{location} cancelled.%{mentions}",
     __locale__,
 )
 
 _(
     f"{Threat.ArtilleryShelling}.{AlertStatus.Active}",
-    "%{status}%{threat} %{time} Artillery shelling in %{location}.",
+    "%{status}%{threat} %{time} Artillery shelling in %{location}.%{mentions}",
     __locale__,
 )
 _(
     f"{Threat.ArtilleryShelling}.{AlertStatus.Inactive}",
-    "%{status}%{threat} %{time} Artillery shelling in %{location} cancelled.",
+    (
+        "%{status}%{threat} %{time} Artillery shelling in %{location} cancelled.",
+        "%{mentions}",
+    ),
     __locale__,
 )

raid/locales/uk.py

@@ -6,22 +6,22 @@
 
 _(
     f"{Threat.AirRaid}.{AlertStatus.Active}",
-    "%{status}%{threat} %{time} Повітряна тривога в %{location}.",
+    "%{status}%{threat} %{time} Повітряна тривога в %{location}.%{mentions}",
     __locale__,
 )
 _(
     f"{Threat.AirRaid}.{AlertStatus.Inactive}",
-    "%{status}%{threat} %{time} Повітряна тривога в %{location}.",
+    "%{status}%{threat} %{time} Повітряна тривога в %{location}.%{mentions}",
     __locale__,
 )
 
 _(
     f"{Threat.ArtilleryShelling}.{AlertStatus.Active}",
-    "%{status}%{threat} %{time} Загроза артобстрілу в %{location}.",
+    "%{status}%{threat} %{time} Загроза артобстрілу в %{location}.%{mentions}",
     __locale__,
 )
 _(
     f"{Threat.ArtilleryShelling}.{AlertStatus.Inactive}",
-    "%{status}%{threat} %{time} Відбій загрози артобстрілу в %{location}.",
+    "%{status}%{threat} %{time} Відбій загрози артобстрілу в %{location}.%{mentions}",
     __locale__,
 )

raid/parsing.py

@@ -11,7 +11,8 @@
         r"^(?P<status>.)"
         r".*(?:тривог[аи] в|Зараз у|артобстрілу в)\s"
         r"(?P<location>.*)"
-        r"\s(?:Слідкуйте|Зверніть|артилерійський).*$"
+        r"\s(?:Слідкуйте|Зверніть|артилерійський).*"
+        r"(?P<hashtag>#.*)$"
     ),
     re.DOTALL,
 )
@@ -44,6 +45,7 @@ class Alert(TypedDict):
     time: datetime
     threat: Threat
     location: str
+    hashtag: str
 
 
 def parse_alert(text: str, time: datetime) -> Alert:
@@ -56,6 +58,7 @@ def parse_alert(text: str, time: datetime) -> Alert:
         "time": _normalize_alert_time(time),
         "threat": _parse_alert_threat(text),
         "location": _normalize_alert_location(match["location"]),
+        "hashtag": match["hashtag"],
     }