remove src/utils/irc/protocol.py

This commit is contained in:
jesopo 2019-10-28 10:56:33 +00:00
parent 1bddc3b37f
commit 7ee65f8f8c
8 changed files with 66 additions and 144 deletions

View file

@ -2,7 +2,7 @@
#--depends-on permissions #--depends-on permissions
import re, string, traceback, typing import re, string, traceback, typing
from src import EventManager, ModuleManager, utils from src import EventManager, IRCLine, ModuleManager, utils
from . import outs from . import outs
COMMAND_METHOD = "command-method" COMMAND_METHOD = "command-method"
@ -172,6 +172,9 @@ class Module(ModuleManager.BaseModule):
if not is_success: if not is_success:
raise utils.EventError(message) raise utils.EventError(message)
def _tagmsg(self, target, tags):
return IRCLine.ParsedLine("TAGMSG", [target], tags=tags)
def command(self, server, target, target_str, is_channel, user, command, def command(self, server, target, target_str, is_channel, user, command,
args_split, tags, hook, **kwargs): args_split, tags, hook, **kwargs):
message_tags = server.has_capability(MESSAGE_TAGS_CAP) message_tags = server.has_capability(MESSAGE_TAGS_CAP)
@ -188,8 +191,8 @@ class Module(ModuleManager.BaseModule):
send_tags["+draft/reply"] = msgid send_tags["+draft/reply"] = msgid
if expect_output: if expect_output:
server.send(utils.irc.protocol.tagmsg(target_str, line = self._tagmsg(target_str, {"+draft/typing": "active"})
{"+draft/typing": "active"}), immediate=True) server.send(line, immediate=True)
stdout = outs.StdOut(server, module_name, target, target_str, send_tags) stdout = outs.StdOut(server, module_name, target, target_str, send_tags)
stderr = outs.StdErr(server, module_name, target, target_str, send_tags) stderr = outs.StdErr(server, module_name, target, target_str, send_tags)
@ -238,8 +241,8 @@ class Module(ModuleManager.BaseModule):
ret = new_event.eaten ret = new_event.eaten
if expect_output and message_tags and not has_out: if expect_output and message_tags and not has_out:
server.send(utils.irc.protocol.tagmsg(target_str, line = self._tagmsg(target_str, {"+draft/typing": "done"})
{"+draft/typing": "done"}), immediate=True) server.send(line, immediate=True)
return ret return ret

View file

@ -6,6 +6,14 @@ STR_MORE_LEN = len(STR_MORE.encode("utf8"))
STR_CONTINUED = "(...continued) " STR_CONTINUED = "(...continued) "
WORD_BOUNDARY = ' ' WORD_BOUNDARY = ' '
def _mesage_factory(command):
if not command in ["PRIVMSG", "NOTICE"]:
raise ValueError("Unknown command method '%s'" % method)
def _(target, message):
return IRCLine.ParsedLine(command, [target, message])
return _
class Out(object): class Out(object):
def __init__(self, server, module_name, target, target_str, tags): def __init__(self, server, module_name, target, target_str, tags):
self.server = server self.server = server
@ -39,15 +47,9 @@ class Out(object):
text = text.replace("\n\n", "\n") text = text.replace("\n\n", "\n")
full_text = "%s%s" % (prefix, text) full_text = "%s%s" % (prefix, text)
line_factory = None message_factory = _mess_factory(method)
if method == "PRIVMSG":
line_factory = utils.irc.protocol.privmsg
elif method == "NOTICE":
line_factory = utils.irc.protocol.notice
else:
raise ValueError("Unknown command method '%s'" % method)
line = line_factory(self._target_str, full_text, tags=self._tags) line = message_factory(self._target_str, full_text, tags=self._tags)
if self._assured: if self._assured:
line.assure() line.assure()

View file

@ -18,7 +18,8 @@ class Module(ModuleManager.BaseModule):
batch = IRCLine.IRCSendBatch("draft/multiline", batch = IRCLine.IRCSendBatch("draft/multiline",
[target]) [target])
for line in lines: for line in lines:
batch.add_line(utils.irc.protocol.privmsg(target, line)) line = IRCLine.ParsedLine("PRIVMSG", [target, line])
batch.add_line(line)
for line in batch.get_lines(): for line in batch.get_lines():
event["server"].send(line) event["server"].send(line)

View file

@ -46,13 +46,14 @@ class Module(ModuleManager.BaseModule):
relay_message = "[%s%s] %s" % (server_name, relay_message = "[%s%s] %s" % (server_name,
relay_prefix_channel, event["line"]) relay_prefix_channel, event["line"])
message = utils.irc.protocol.privmsg(relay_channel.name, self.bot.trigger(self._send_factory(server, relay_channel.name,
relay_message) relay_message))
server._relay_ignore.append(message.id)
self.bot.trigger(self._send_factory(server, message))
def _send_factory(self, server, message): def _send_factory(self, server, channel_name, message):
return lambda: server.send(message) def _():
message = server.send_message(channel_name, message)
server._relay_ignore.append(message.parsed_line.id)
return _
@utils.hook("formatted.message.channel") @utils.hook("formatted.message.channel")
@utils.hook("formatted.notice.channel") @utils.hook("formatted.notice.channel")

View file

@ -1,5 +1,5 @@
import signal, sys import signal, sys
from src import Config, ModuleManager, utils from src import Config, IRCLine, ModuleManager, utils
class Module(ModuleManager.BaseModule): class Module(ModuleManager.BaseModule):
def on_load(self): def on_load(self):
@ -24,7 +24,7 @@ class Module(ModuleManager.BaseModule):
if server.connected: if server.connected:
server.socket.clear_send_buffer() server.socket.clear_send_buffer()
line = utils.irc.protocol.quit("Shutting down") line = IRCLine.ParsedLine("QUIT", ["Shutting down"])
sent_line = server.send(line, immediate=True) sent_line = server.send(line, immediate=True)
sent_line.events.on("send").hook(self._make_hook(server)) sent_line.events.on("send").hook(self._make_hook(server))

View file

@ -296,14 +296,18 @@ class Server(IRCObject.Object):
def send_raw(self, line: str): def send_raw(self, line: str):
return self.send(IRCLine.parse_line(line)) return self.send(IRCLine.parse_line(line))
def _line(self, command, args, tags={}):
return IRCLine.ParsedLine(command,
[arg for arg in args if not arg == None], tags=tags)
def send_user(self, username: str, realname: str def send_user(self, username: str, realname: str
) -> typing.Optional[IRCLine.SentLine]: ) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.user(username, realname)) return self.send(self._line("USER",[username, "0", "*", realname]))
def send_nick(self, nickname: str) -> typing.Optional[IRCLine.SentLine]: def send_nick(self, nickname: str) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.nick(nickname)) return self.send(self._line("NICK", [nickname]))
def send_capibility_ls(self) -> typing.Optional[IRCLine.SentLine]: def send_capibility_ls(self) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.capability_ls()) return self.send(self._line("CAP", ["LS", "302"]))
def send_capability_queue(self): def send_capability_queue(self):
# textwrap works here because in ASCII, all chars are 1 bytes: # textwrap works here because in ASCII, all chars are 1 bytes:
capabilities = list(self.capability_queue.keys()) capabilities = list(self.capability_queue.keys())
@ -319,13 +323,13 @@ class Server(IRCObject.Object):
for capability_batch in capability_batches: for capability_batch in capability_batches:
self.send_capability_request(capability_batch) self.send_capability_request(capability_batch)
def send_capability_request(self, capability: str def send_capability_request(self, capabilities: str
) -> typing.Optional[IRCLine.SentLine]: ) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.capability_request(capability)) return self.send(self._line("CAP", ["REQ", capabilities]))
def send_capability_end(self) -> typing.Optional[IRCLine.SentLine]: def send_capability_end(self) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.capability_end()) return self.send(self._line("CAP", ["END"]))
def send_authenticate(self, text: str) -> typing.Optional[IRCLine.SentLine]: def send_authenticate(self, text: str) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.authenticate(text)) return self.send(self._line("AUTHENTICATE", [text]))
def has_capability(self, capability: utils.irc.Capability) -> bool: def has_capability(self, capability: utils.irc.Capability) -> bool:
return bool(self.available_capability(capability)) return bool(self.available_capability(capability))
def has_capability_str(self, capability: str) -> bool: def has_capability_str(self, capability: str) -> bool:
@ -347,65 +351,67 @@ class Server(IRCObject.Object):
self._capabilities_waiting.clear() self._capabilities_waiting.clear()
def send_pass(self, password: str) -> typing.Optional[IRCLine.SentLine]: def send_pass(self, password: str) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.password(password)) return self.send(self._line("PASS", [password]))
def send_ping(self, nonce: str="hello" def send_ping(self, token: str="hello"
) -> typing.Optional[IRCLine.SentLine]: ) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.ping(nonce), immediate=True) return self.send(self._line("PING", [token]))
def send_pong(self, nonce: str) -> typing.Optional[IRCLine.SentLine]: def send_pong(self, token: str) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.pong(nonce), immediate=True) return self.send(self._line("PONG", [token]))
def send_join(self, channel_name: str, keys: typing.List[str]=None def send_join(self, channel_name: str, key: typing.List[str]=None
) -> typing.Optional[IRCLine.SentLine]: ) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.join(channel_name, keys)) return self.send(self._line("JOIN", [channel_name, key]))
def send_joins(self, channel_names: typing.List[str], def send_joins(self, channel_names: typing.List[str],
keys: typing.List[str]=None): keys: typing.List[str]=None):
return self.send(utils.irc.protocol.join(",".join(channel_names), return self.send(self._line("JOIN",
keys)) [",".join(channel_names)]+(keys or [])))
def send_part(self, channel_name: str, reason: str=None def send_part(self, channel_name: str, reason: str=None
) -> typing.Optional[IRCLine.SentLine]: ) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.part(channel_name, reason)) return self.send(self._line("PART", [channel_name, reason]))
def send_quit(self, reason: str="Leaving" def send_quit(self, reason: str="Leaving"
) -> typing.Optional[IRCLine.SentLine]: ) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.quit(reason)) return self.send(self._line("QUIT", [reason]))
def send_message(self, target: str, message: str, tags: dict={} def send_message(self, target: str, message: str, tags: dict={}
) -> typing.Optional[IRCLine.SentLine]: ) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.privmsg(target, message, tags)) return self.send(self._line("PRIVMSG", [target, message], tags=tags))
def send_notice(self, target: str, message: str, tags: dict={} def send_notice(self, target: str, message: str, tags: dict={}
) -> typing.Optional[IRCLine.SentLine]: ) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.notice(target, message, tags)) return self.send(self._line("NOTICE", [target, message], tags=tags))
def send_tagmsg(self, target: str, tags: dict): def send_tagmsg(self, target: str, tags: dict):
return self.send(utils.irc.protocol.tagmsg(target, tags)) return self.send(self._line("TARGMSG", [], tags=tags))
def send_mode(self, target: str, mode: str=None, args: typing.List[str]=None def send_mode(self, target: str, mode: str=None, args: typing.List[str]=None
) -> typing.Optional[IRCLine.SentLine]: ) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.mode(target, mode, args)) return self.send(self._line("MODE", [target, mode, args]))
def send_topic(self, channel_name: str, topic: str def send_topic(self, channel_name: str, topic: str
) -> typing.Optional[IRCLine.SentLine]: ) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.topic(channel_name, topic)) return self.send(self._line("TOPIC", [channel_name, topic]))
def send_kick(self, channel_name: str, target: str, reason: str=None def send_kick(self, channel_name: str, target: str, reason: str=None
) -> typing.Optional[IRCLine.SentLine]: ) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.kick(channel_name, target, reason)) return self.send(self._line("KICK", [channel_name, target, reason]))
def send_names(self, channel_name: str) -> typing.Optional[IRCLine.SentLine]: def send_names(self, channel_name: str) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.names(channel_name)) return self.send(self._line("NAMES", [channel_name]))
def send_list(self, search_for: str=None def send_list(self, search_for: str=None
) -> typing.Optional[IRCLine.SentLine]: ) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.list(search_for)) return self.send(self._line("LIST", [search_for]))
def send_invite(self, channel_name: str, target: str def send_invite(self, channel_name: str, target: str
) -> typing.Optional[IRCLine.SentLine]: ) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.invite(channel_name, target)) return self.send(self._line("INVITE", [target, channel_name]))
def send_whois(self, target: str) -> typing.Optional[IRCLine.SentLine]: def send_whois(self, target: str) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.whois(target)) return self.send(self._line("WHOIS", [target]))
def send_whowas(self, target: str, amount: int=None, server: str=None def send_whowas(self, target: str, amount: int=None, server: str=None
) -> typing.Optional[IRCLine.SentLine]: ) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.whowas(target, amount, server)) amount_str = str(amount) if not amount == None else None
return self.send(self._line("WHOWAS", [target, amount_str, server]))
def send_who(self, filter: str=None) -> typing.Optional[IRCLine.SentLine]: def send_who(self, filter: str=None) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.who(filter)) return self.send(self._line("WHO", [filter]))
def send_whox(self, mask: str, filter: str, fields: str, label: str=None def send_whox(self, mask: str, filter: str, fields: str, label: str=None
) -> typing.Optional[IRCLine.SentLine]: ) -> typing.Optional[IRCLine.SentLine]:
return self.send(utils.irc.protocol.whox(mask, filter, fields, label)) flags = "%s%%%s%s" % (filter, fields, ","+label if label else "")
return self.send(self._line("WHO", [mask, flags]))

View file

@ -1,6 +1,5 @@
import json, string, re, typing, uuid import json, string, re, typing, uuid
from src import utils from src import utils
from . import protocol
ASCII_UPPER = string.ascii_uppercase ASCII_UPPER = string.ascii_uppercase
ASCII_LOWER = string.ascii_lowercase ASCII_LOWER = string.ascii_lowercase

View file

@ -1,90 +0,0 @@
import typing
from src import IRCLine, utils
def user(username: str, realname: str) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("USER", [username, "0", "*", realname])
def nick(nickname: str) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("NICK", [nickname])
def capability_ls() -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("CAP", ["LS", "302"])
def capability_request(capability: str) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("CAP", ["REQ", capability])
def capability_end() -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("CAP", ["END"])
def authenticate(text: str) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("AUTHENTICATE", [text])
def password(password: str) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("PASS", [password])
def ping(nonce: str="hello") -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("PING", [nonce])
def pong(nonce: str="hello") -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("PONG", [nonce])
def join(channel_name: str, keys: typing.List[str]=None
) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("JOIN", [channel_name]+(
keys if keys else []))
def part(channel_name: str, reason: str=None) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("PART", [channel_name]+(
[reason] if reason else []))
def quit(reason: str=None) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("QUIT", [reason] if reason else [])
def privmsg(target: str, message: str, tags: typing.Dict[str, str]={}
) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("PRIVMSG", [target, message], tags=tags)
def notice(target: str, message: str, tags: typing.Dict[str, str]={}
) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("NOTICE", [target, message], tags=tags)
def tagmsg(target, tags: dict) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("TAGMSG", [target], tags=tags)
def mode(target: str, mode: str=None, args: typing.List[str]=None
) -> IRCLine.ParsedLine:
command_args = [target]
if mode:
command_args.append(mode)
if args:
command_args = command_args+args
return IRCLine.ParsedLine("MODE", command_args)
def topic(channel_name: str, topic: str) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("TOPIC", [channel_name, topic])
def kick(channel_name: str, target: str, reason: str=None
) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("KICK", [channel_name, target]+(
[reason] if reason else []))
def names(channel_name: str) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("NAMES", [channel_name])
def list(search_for: str=None) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("LIST", [search_for] if search_for else [])
def invite(channel_name: str, target: str) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("INVITE", [target, channel_name])
def whois(target: str) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("WHOIS", [target])
def whowas(target: str, amount: int=None, server: str=None
) -> IRCLine.ParsedLine:
command_args = [target]
if amount:
command_args.append(str(amount))
if server:
command_args.append(server)
return IRCLine.ParsedLine("WHOWAS", command_args)
def who(filter: str=None) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("WHO", [filter] if filter else [])
def whox(mask: str, filter: str, fields: str, label: str=None
) -> IRCLine.ParsedLine:
flags = "%s%%%s%s" % (filter, fields, ","+label if label else "")
return IRCLine.ParsedLine("WHO", [mask, flags])
def batch_start(identifier: str, batch_type: str, tags: typing.Dict[str, str]={}
) -> IRCLine.ParsedLine:
return IRCLine.ParsedLine("BATCH", ["+%s" % identifier, batch_type],
tags=tags)
def batch_end(identifier: str, tags: typing.Dict[str, str]={}):
return IRCLine.ParsedLine("BATCH", ["-%s" % identifier], tags=tags)