from re import search as re_search, IGNORECASE as re_IGNORECASE
from threading import Thread
from time import sleep
-from typing import Any, Callable, Iterable, Optional, Self
+from typing import Any, Callable, Iterable, Optional
from uuid import UUID, uuid4
# ourselves
from ircplom.db_primitives import (
password: str = ''
+@dataclass
class ChatMessage:
'Collects all we want to know on incoming PRIVMSG or NOTICE chat message.'
content: str = ''
self._remove_membership_prefix(user_id, prefix)
-class _ChatMessage(ChatMessage):
-
- def __init__(self,
- sender: str | NickUserHost = '',
- db: Optional['_ClientDb'] = None
- ) -> None:
- self.sender = sender if isinstance(sender, str) else sender.nick
- self._db = db
-
- def to(self, target: str) -> Self:
- 'Extend self with .target, return self.'
- self.target = target
- return self
-
- def __setattr__(self, key: str, value: str) -> None:
- if key in {'privmsg', 'notice'}:
- assert self._db is not None
- self.is_notice = key == 'notice'
- self.content = value
- self._db.message = self
- # to clean update cache, enabling equal messages in direct sequence
- self._db.message = ChatMessage()
- else:
- super().__setattr__(key, value)
-
-
class _SetNickuserhostMixin:
def __setattr__(self, key: str, value: NickUserHost | str) -> None:
del self.channels[channel]
self.users.purge()
- def messaging(self, src: str | NickUserHost) -> ChatMessage:
- 'Start input chain for chat message data.'
- return _ChatMessage(sender=src, db=self)
+ def _set_message(self,
+ content: str,
+ sender: str | NickUserHost,
+ target: str,
+ is_notice: bool
+ ) -> None:
+ self.message = ChatMessage(
+ content,
+ sender if isinstance(sender, str) else sender.nick,
+ target,
+ is_notice)
+ # to clean update cache, enabling equal messages in direct sequence
+ self.message = ChatMessage()
+
+ def notice(self,
+ content: str,
+ sender: str | NickUserHost = '',
+ target: str = ''
+ ) -> None:
+ 'Set .message from NOTICE.'
+ self._set_message(content, sender, target, is_notice=True)
+
+ def privmsg(self,
+ content: str,
+ sender: str | NickUserHost,
+ target: str = ''
+ ) -> None:
+ 'Set .message from PRIVMSG.'
+ self._set_message(content, sender, target, is_notice=False)
def into_endnode_updates(self, path: tuple[str, ...]
) -> list[tuple[tuple[str, ...], Any]]:
'NOTICE',
_MsgToken.SERVER,
('*',
- (_MsgToken.ANY, 'setattr_db.messaging..to.:notice'))),
+ (_MsgToken.ANY, ':content')),
+ bonus_tasks=('do_db.notice:content',)),
_MsgParseExpectation(
'NOTICE',
_MsgToken.SERVER,
((_MsgToken.NICKNAME, 'setattr_db.users.me:nick'),
- (_MsgToken.ANY, 'setattr_db.messaging..to.:notice'))),
+ (_MsgToken.ANY, ':content')),
+ bonus_tasks=('do_db.notice:content',)),
_MsgParseExpectation(
'NOTICE',
- (_MsgToken.SERVER, ':SERVER'),
- ((_MsgToken.CHANNEL, ':CHANNEL'),
- (_MsgToken.ANY, 'setattr_db.messaging.SERVER.to.CHANNEL:notice'))),
+ (_MsgToken.SERVER, ':sender'),
+ ((_MsgToken.CHANNEL, ':target'),
+ (_MsgToken.ANY, ':content')),
+ bonus_tasks=('do_db.notice:content,sender,target',)),
_MsgParseExpectation(
'NOTICE',
- (_MsgToken.NICK_USER_HOST, ':USER'),
- ((_MsgToken.NICKNAME, 'setattr_db.users.me:nick'),
- (_MsgToken.ANY, 'setattr_db.messaging.USER.to.:notice'))),
+ (_MsgToken.NICK_USER_HOST, ':sender'),
+ ((_MsgToken.NICKNAME, 'setattr_db.users.me:target'),
+ (_MsgToken.ANY, ':content')),
+ bonus_tasks=('do_db.notice:content,sender',)),
_MsgParseExpectation(
'NOTICE',
- (_MsgToken.NICK_USER_HOST, ':USER'),
- ((_MsgToken.CHANNEL, ':CHANNEL'),
- (_MsgToken.ANY, 'setattr_db.messaging.USER.to.CHANNEL:notice'))),
+ (_MsgToken.NICK_USER_HOST, ':sender'),
+ ((_MsgToken.CHANNEL, ':target'),
+ (_MsgToken.ANY, ':content')),
+ bonus_tasks=('do_db.notice:content,sender,target',)),
_MsgParseExpectation(
'PRIVMSG',
- (_MsgToken.NICK_USER_HOST, ':USER'),
- ((_MsgToken.NICKNAME, 'setattr_db.users.me:nick'),
- (_MsgToken.ANY, 'setattr_db.messaging.USER.to.:privmsg'))),
+ (_MsgToken.NICK_USER_HOST, ':sender'),
+ ((_MsgToken.NICKNAME, 'setattr_db.users.me:target'),
+ (_MsgToken.ANY, ':content')),
+ bonus_tasks=('do_db.privmsg:content,sender',)),
_MsgParseExpectation(
'PRIVMSG',
- (_MsgToken.NICK_USER_HOST, ':USER'),
- ((_MsgToken.CHANNEL, ':CHANNEL'),
- (_MsgToken.ANY, 'setattr_db.messaging.USER.to.CHANNEL:privmsg'))),
+ (_MsgToken.NICK_USER_HOST, ':sender'),
+ ((_MsgToken.CHANNEL, ':target'),
+ (_MsgToken.ANY, ':content')),
+ bonus_tasks=('do_db.privmsg:content,sender,target',)),
# connection state