feat(parser): add tag access fields to Message objects (#308)

Author: linuxdaemon

irclib/parser.py

@@ -3,7 +3,9 @@
 Backported from async-irc (https://github.com/snoonetIRC/async-irc.git)
 """
 
+import datetime
 import re
+import warnings
 from abc import ABCMeta, abstractmethod
 from collections.abc import Iterable, Iterator, Sequence
 from typing import Final, Literal, Optional, Union, cast
@@ -56,6 +58,17 @@
 }
 
 
+def parse_server_time(
+    value: Optional[str], default: datetime.datetime
+) -> datetime.datetime:
+    if value:
+        return datetime.datetime.strptime(
+            value, "%Y-%m-%dT%H:%M:%S.%fZ"
+        ).replace(tzinfo=datetime.timezone.utc)
+
+    return default
+
+
 class Parseable(metaclass=ABCMeta):
     """Abstract class for parseable objects."""
 
@@ -597,13 +610,59 @@ def __init__(
         prefix: Union[str, Prefix, None, Iterable[str]],
         command: str,
         *parameters: Union[str, list[str], ParamList],
+        time: Optional[datetime.datetime] = None,
     ) -> None:
         """Construct message object."""
         self._tags = _parse_tags(tags)
         self._prefix = _parse_prefix(prefix)
         self._command = command
         self._parameters = _parse_params(parameters)
 
+        if time is None:
+            time = datetime.datetime.now(datetime.timezone.utc)
+
+        if time.tzinfo is None:
+            time = time.replace(tzinfo=datetime.timezone.utc)
+            warnings.warn(
+                "Timezone-naive datetime is deprecated for 'time'.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
+        self._time = time
+
+    def has_tag(self, name: str) -> bool:
+        """Return whether this message has a particular message tag."""
+        if not self.tags:
+            return False
+
+        return name in self.tags
+
+    def get_tag_value(self, name: str) -> Optional[str]:
+        """Get value for a message tag, or None if not set."""
+        if self.tags and name in self.tags:
+            return self.tags[name].value
+
+        return None
+
+    @property
+    def time(self) -> datetime.datetime:
+        """Time the message was sent/received.
+
+        Uses server-time tag if available.
+        """
+        return parse_server_time(self.get_tag_value("time"), self._time)
+
+    @property
+    def message_id(self) -> Optional[str]:
+        """Unique message ID provided by server."""
+        return self.get_tag_value("msgid")
+
+    @property
+    def batch_id(self) -> Optional[str]:
+        """Batch ID provided by server."""
+        return self.get_tag_value("batch")
+
     @property
     def tags(self) -> MsgTagList:
         """IRC tag list."""
@@ -629,7 +688,12 @@ def as_tuple(self) -> MessageTuple:
         return self.tags, self.prefix, self.command, self.parameters
 
     @classmethod
-    def parse(cls, text: Union[str, bytes]) -> Self:
+    def parse(
+        cls,
+        text: Union[str, bytes],
+        *,
+        time: Optional[datetime.datetime] = None,
+    ) -> Self:
         """Parse an IRC message in to objects."""
         if isinstance(text, memoryview):
             text = text.tobytes().decode(errors="ignore")
@@ -652,7 +716,7 @@ def parse(cls, text: Union[str, bytes]) -> Self:
         prefix_obj = Prefix.parse(prefix[1:]) if prefix else None
         command = command.upper()
         param_obj = ParamList.parse(params)
-        return cls(tags_obj, prefix_obj, command, param_obj)
+        return cls(tags_obj, prefix_obj, command, param_obj, time=time)
 
     def __eq__(self, other: object) -> bool:
         """Compare to another message which can be str, bytes, or a Message object."""

tests/parser_test.py

@@ -1,5 +1,6 @@
 """Test IRC parser."""
 
+import datetime
 from typing import Optional, TypedDict
 
 import parser_tests.data
@@ -826,6 +827,100 @@ def test_bool_false(self, obj: Message) -> None:
         """Test all the cases where bool(Message) should return False."""
         assert not obj
 
+    @pytest.mark.parametrize(
+        ("msg", "expected"),
+        [
+            (
+                "@time=2025-09-01T00:11:22.123Z FOO #bar :baz blah",
+                datetime.datetime(
+                    year=2025,
+                    month=9,
+                    day=1,
+                    hour=0,
+                    minute=11,
+                    second=22,
+                    microsecond=123000,
+                    tzinfo=datetime.timezone.utc,
+                ),
+            ),
+            (
+                "FOO #bar :baz blah",
+                datetime.datetime(
+                    year=2006,
+                    month=1,
+                    day=2,
+                    hour=3,
+                    minute=4,
+                    second=5,
+                    microsecond=123456,
+                    tzinfo=datetime.timezone(datetime.timedelta(hours=-7)),
+                ),
+            ),
+        ],
+    )
+    def test_parse_server_time(
+        self, msg: str, expected: datetime.datetime
+    ) -> None:
+        """Test server time parsing."""
+        default_time = datetime.datetime(
+            year=2006,
+            month=1,
+            day=2,
+            hour=3,
+            minute=4,
+            second=5,
+            microsecond=123456,
+            tzinfo=datetime.timezone(datetime.timedelta(hours=-7)),
+        )
+
+        assert Message.parse(msg, time=default_time).time == expected
+
+    @pytest.mark.parametrize(
+        ("msg", "expected"),
+        [
+            ("@msgid=foo FOO #bar :baz blah", "foo"),
+            ("FOO #bar :baz blah", None),
+        ],
+    )
+    def test_parse_msgid(self, msg: str, expected: Optional[str]) -> None:
+        """Ensure message IDs are retrieved."""
+        assert Message.parse(msg).message_id == expected
+
+    @pytest.mark.parametrize(
+        ("msg", "expected"),
+        [
+            ("@batch=foo FOO #bar :baz blah", "foo"),
+            ("FOO #bar :baz blah", None),
+        ],
+    )
+    def test_parse_batch(self, msg: str, expected: Optional[str]) -> None:
+        """Ensure batch IDs are retrieved."""
+        assert Message.parse(msg).batch_id == expected
+
+    def test_naive_datetime_warning(self) -> None:
+        """Ensure warning about use of naive datetimes is present."""
+        with pytest.warns(DeprecationWarning, match=".*naive.*"):
+            Message(
+                None,
+                None,
+                "cmd",
+                time=datetime.datetime(year=1, month=1, day=1),  # noqa: DTZ001
+            )
+
+    @pytest.mark.parametrize(
+        ("msg", "tag", "present"),
+        [
+            ("", "", False),
+            ("@foo=bar FOO", "foo", True),
+            ("@foo=bar FOO", "bar", False),
+            ("@foo= FOO", "foo", True),
+            ("@foo= FOO", "bar", False),
+        ],
+    )
+    def test_has_tag(self, msg: str, tag: str, present: bool) -> None:
+        """Test has_tag."""
+        assert Message.parse(msg).has_tag(tag) == present
+
 
 def test_trail() -> None:
     """Ensure this parser does not have the same issue as https://github.com/hexchat/hexchat/issues/2271."""

tests/string_test.py

@@ -51,6 +51,21 @@ def test_comparisons() -> None:
     assert "B" >= str2
     assert not "B" <= str2
 
+    with pytest.raises(TypeError):
+        assert not str2 < 1  # type: ignore[operator]
+
+    with pytest.raises(TypeError):
+        assert not str2 <= 1  # type: ignore[operator]
+
+    with pytest.raises(TypeError):
+        assert not str2 > 1  # type: ignore[operator]
+
+    with pytest.raises(TypeError):
+        assert not str2 >= 1  # type: ignore[operator]
+
+    assert not str2 == 1
+    assert str2 != 1
+
     assert str2 != "B"
 
     assert str1 != 5