Update parse_quit, now it returns MUser object and the reason.

This commit is contained in:
adator
2025-11-02 21:28:44 +01:00
parent 5a1432c1e6
commit 79c1b94a92
4 changed files with 28 additions and 22 deletions

View File

@@ -1196,25 +1196,26 @@ class Inspircd(IProtocol):
uid = scopy[2] uid = scopy[2]
return self._User.get_user(uid) return self._User.get_user(uid)
def parse_quit(self, server_msg: list[str]) -> dict[str, str]: def parse_quit(self, server_msg: list[str]) -> tuple[Optional['MUser'], str]:
"""Parse quit and return dictionary. """Parse quit and return dictionary.
>>> [':97KAAAAAB', 'QUIT', ':Quit:', 'this', 'is', 'my', 'reason', 'to', 'quit'] >>> [':97KAAAAAB', 'QUIT', ':Quit:', 'this', 'is', 'my', 'reason', 'to', 'quit']
Args: Args:
server_msg (list[str]): The server message to parse server_msg (list[str]): The server message to parse
Returns: Returns:
dict[str, str]: The dictionary. tuple[MUser, str]: The User Who Quit Object and the reason.
""" """
scopy = server_msg.copy() scopy = server_msg.copy()
if scopy[0].startswith('@'): if scopy[0].startswith('@'):
scopy.pop(0) scopy.pop(0)
user_obj = self._User.get_user(self._Utils.clean_uid(scopy[0]))
tmp_reason = scopy[3:]
tmp_reason[0] = tmp_reason[0].replace(':', '')
reason = ' '.join(tmp_reason)
response = { return user_obj, reason
"uid": scopy[0].replace(':', ''),
"reason": " ".join(scopy[3:])
}
return response
def parse_nick(self, server_msg: list[str]) -> dict[str, str]: def parse_nick(self, server_msg: list[str]) -> dict[str, str]:
"""Parse nick changes. """Parse nick changes.

View File

@@ -327,14 +327,14 @@ class IProtocol(ABC):
""" """
@abstractmethod @abstractmethod
def parse_quit(self, serverMsg: list[str]) -> dict[str, str]: def parse_quit(self, serverMsg: list[str]) -> tuple[Optional['MUser'], str]:
"""Parse quit and return dictionary. """Parse quit and return dictionary.
>>> [':97KAAAAAB', 'QUIT', ':Quit:', 'this', 'is', 'my', 'reason', 'to', 'quit'] >>> [':97KAAAAAB', 'QUIT', ':Quit:', 'this', 'is', 'my', 'reason', 'to', 'quit']
Args: Args:
serverMsg (list[str]): The server message to parse serverMsg (list[str]): The server message to parse
Returns: Returns:
dict[str, str]: The response as dictionary. tuple[MUser, str]: The User Who Quit Object and the reason.
""" """
@abstractmethod @abstractmethod

View File

@@ -1,4 +1,5 @@
from base64 import b64decode from base64 import b64decode
from optparse import Option
from re import match, findall, search from re import match, findall, search
from datetime import datetime from datetime import datetime
from typing import TYPE_CHECKING, Any, Optional from typing import TYPE_CHECKING, Any, Optional
@@ -637,30 +638,31 @@ class Unrealircd6(IProtocol):
serverMsg (list[str]): The UID ircd response serverMsg (list[str]): The UID ircd response
""" """
scopy = serverMsg.copy() scopy = serverMsg.copy()
if '@' in scopy[0]: if scopy[0].startswith('@'):
scopy.pop(0) scopy.pop(0)
uid = scopy[7] uid = scopy[7]
return self._User.get_user(uid) return self._User.get_user(uid)
def parse_quit(self, serverMsg: list[str]) -> dict[str, str]: def parse_quit(self, serverMsg: list[str]) -> tuple[Optional['MUser'], str]:
"""Parse quit and return dictionary. """Parse quit and return dictionary.
>>> # ['@unrealtag...', ':001JKNY0N', 'QUIT', ':Quit:', '....'] >>> # ['@unrealtag...', ':001JKNY0N', 'QUIT', ':Quit:', '....']
Args: Args:
serverMsg (list[str]): The server message to parse serverMsg (list[str]): The server message to parse
Returns: Returns:
dict[str, str]: The dictionary. tuple[MUser, str]: The User Who Quit Object and the reason.
""" """
scopy = serverMsg.copy() scopy = serverMsg.copy()
if scopy[0].startswith('@'): if scopy[0].startswith('@'):
scopy.pop(0) scopy.pop(0)
user_obj = self._User.get_user(self._Utils.clean_uid(scopy[0]))
tmp_reason = scopy[3:]
tmp_reason[0] = tmp_reason[0].replace(':', '')
reason = ' '.join(tmp_reason)
response = { return user_obj, reason
"uid": scopy[0].replace(':', ''),
"reason": " ".join(scopy[3:])
}
return response
def parse_nick(self, serverMsg: list[str]) -> dict[str, str]: def parse_nick(self, serverMsg: list[str]) -> dict[str, str]:
"""Parse nick changes and return dictionary. """Parse nick changes and return dictionary.
@@ -705,7 +707,7 @@ class Unrealircd6(IProtocol):
channel = self._Channel.get_channel(scopy[2]) if self._Channel.is_valid_channel(scopy[2]) else None channel = self._Channel.get_channel(scopy[2]) if self._Channel.is_valid_channel(scopy[2]) else None
tmp_message = scopy[3:] tmp_message = scopy[3:]
tmp_message = tmp_message[0].replace(':', '') tmp_message[0] = tmp_message[0].replace(':', '')
message = ' '.join(tmp_message) message = ' '.join(tmp_message)
return sender, reciever, channel, message return sender, reciever, channel, message

View File

@@ -184,14 +184,17 @@ def handle_on_quit(uplink: 'Defender', srvmsg: list[str]):
srvmsg (list[str]): The Server MSG srvmsg (list[str]): The Server MSG
""" """
p = uplink.Protocol p = uplink.Protocol
parser = p.parse_quit(srvmsg) userobj, reason = p.parse_quit(srvmsg)
confmodel = uplink.ModConfig confmodel = uplink.ModConfig
if userobj is None:
uplink.Logs.error("Error when parsing message QUIT", exc_info=True)
return None
ban_all_chan = uplink.Base.int_if_possible(confmodel.reputation_ban_all_chan) ban_all_chan = uplink.Base.int_if_possible(confmodel.reputation_ban_all_chan)
final_UID = uplink.Loader.Utils.clean_uid(str(parser.get('uid', None)))
jail_salon = uplink.Config.SALON_JAIL jail_salon = uplink.Config.SALON_JAIL
service_id = uplink.Config.SERVICE_ID service_id = uplink.Config.SERVICE_ID
get_user_reputation = uplink.Reputation.get_reputation(final_UID) get_user_reputation = uplink.Reputation.get_reputation(userobj.uid)
if get_user_reputation is not None: if get_user_reputation is not None:
final_nickname = get_user_reputation.nickname final_nickname = get_user_reputation.nickname
@@ -200,7 +203,7 @@ def handle_on_quit(uplink: 'Defender', srvmsg: list[str]):
p.send2socket(f":{service_id} MODE {chan.name} -b {final_nickname}!*@*") p.send2socket(f":{service_id} MODE {chan.name} -b {final_nickname}!*@*")
uplink.Logs.debug(f"Mode -b {final_nickname} on channel {chan.name}") uplink.Logs.debug(f"Mode -b {final_nickname} on channel {chan.name}")
uplink.Reputation.delete(final_UID) uplink.Reputation.delete(userobj.uid)
uplink.Logs.debug(f"Client {get_user_reputation.nickname} has been removed from Reputation local DB") uplink.Logs.debug(f"Client {get_user_reputation.nickname} has been removed from Reputation local DB")
def handle_on_uid(uplink: 'Defender', srvmsg: list[str]): def handle_on_uid(uplink: 'Defender', srvmsg: list[str]):