support & in utils.parse.sed, change sed.py to use utils.parse.sed
This commit is contained in:
parent
9d94c55539
commit
e5fdef6726
3 changed files with 65 additions and 87 deletions
|
@ -30,7 +30,7 @@ class Module(ModuleManager.BaseModule):
|
|||
|
||||
filters = self._get_filters(event["server"], target)
|
||||
for filter in filters:
|
||||
sed = utils.parse.sed.parse_sed(filter)
|
||||
sed = utils.parse.sed.parse(filter)
|
||||
type, out = utils.parse.sed.sed(sed, message)
|
||||
|
||||
if type == "m" and out:
|
||||
|
|
|
@ -4,9 +4,7 @@
|
|||
import re, traceback
|
||||
from src import ModuleManager, utils
|
||||
|
||||
REGEX_SPLIT = re.compile("(?<!\\\\)/")
|
||||
REGEX_SED = re.compile("^(?:(\\S+)[:,] )?s/")
|
||||
SED_AMPERSAND = re.compile(r"((?:^|[^\\])(?:\\\\)*)&")
|
||||
|
||||
@utils.export("channelset",
|
||||
utils.BoolSetting("sed","Disable/Enable sed in a channel"))
|
||||
|
@ -21,72 +19,38 @@ class Module(ModuleManager.BaseModule):
|
|||
@utils.kwarg("command", "sed")
|
||||
@utils.kwarg("pattern", REGEX_SED)
|
||||
def channel_message(self, event):
|
||||
sed_split = re.split(REGEX_SPLIT, event["message"], 3)
|
||||
if len(sed_split) > 2:
|
||||
if not self._closest_setting(event, "sed", False):
|
||||
return
|
||||
for_user = event["match"].group(1)
|
||||
sed_s = event["message"]
|
||||
if for_user:
|
||||
sed_s = sed_s.split(" ", 1)[1]
|
||||
if not self._closest_setting(event, "sed", False):
|
||||
return
|
||||
|
||||
regex_flags = 0
|
||||
flags = (sed_split[3:] or [""])[0].split(" ", 1)[0]
|
||||
count = None
|
||||
try:
|
||||
sed = utils.parse.sed.parse(event["message"])
|
||||
except:
|
||||
traceback.print_exc()
|
||||
event["stderr"].write("Invalid regex in pattern")
|
||||
return
|
||||
sed.replace = utils.irc.bold(sed.replace)
|
||||
|
||||
last_flag = ""
|
||||
for flag in flags:
|
||||
if flag.isdigit():
|
||||
if last_flag.isdigit():
|
||||
count = int(str(count) + flag)
|
||||
elif not count:
|
||||
count = int(flag)
|
||||
elif flag == "i":
|
||||
regex_flags |= re.I
|
||||
elif flag == "g":
|
||||
count = 0
|
||||
last_flag = flag
|
||||
if count == None:
|
||||
count = 1
|
||||
if self._closest_setting(event, "sed-sender-only", False):
|
||||
for_user = event["user"].nickname
|
||||
|
||||
try:
|
||||
pattern = re.compile(sed_split[1], regex_flags)
|
||||
except:
|
||||
traceback.print_exc()
|
||||
event["stderr"].write("Invalid regex in pattern")
|
||||
return
|
||||
match_line = None
|
||||
match_message = None
|
||||
with utils.deadline():
|
||||
for line in event["target"].buffer.get_all(for_user):
|
||||
if not line.from_self:
|
||||
match = sed.match(line.message)
|
||||
if not match == line.message:
|
||||
match_line = line
|
||||
match_message = match
|
||||
break
|
||||
|
||||
for_user = event["match"].group(1)
|
||||
if self._closest_setting(event, "sed-sender-only", False):
|
||||
for_user = event["user"].nickname
|
||||
|
||||
match_line = None
|
||||
match = None
|
||||
match_message = None
|
||||
with utils.deadline():
|
||||
for line in event["target"].buffer.get_all(for_user):
|
||||
if not line.from_self:
|
||||
message = line.notes.get("sed-line", line.message)
|
||||
match = pattern.search(message)
|
||||
if match and not REGEX_SED.match(message):
|
||||
match_line = line
|
||||
match = match.group(0)
|
||||
match_message = message
|
||||
break
|
||||
|
||||
if match:
|
||||
replace = sed_split[2]
|
||||
replace = replace.replace("\\/", "/")
|
||||
|
||||
with utils.deadline():
|
||||
for found in SED_AMPERSAND.finditer(replace):
|
||||
found = found.group(1)
|
||||
replace.replace(found, "%s%s" % (found, match))
|
||||
|
||||
replace_color = utils.irc.bold(replace)
|
||||
|
||||
new_message = re.sub(pattern, replace, message, count)
|
||||
new_message_color = re.sub(pattern, utils.irc.bold(replace),
|
||||
message, count)
|
||||
if match_line.action:
|
||||
prefix = "* %s" % match_line.sender
|
||||
else:
|
||||
prefix = "<%s>" % match_line.sender
|
||||
match_line.notes["sed-line"] = new_message
|
||||
event["stdout"].write("%s %s" % (prefix, new_message_color))
|
||||
if match_line:
|
||||
if match_line.action:
|
||||
format = "* %s %s"
|
||||
else:
|
||||
format = "<%s> %s"
|
||||
event["stdout"].write(format % (match_line.sender, match_message))
|
||||
|
|
|
@ -1,5 +1,18 @@
|
|||
import dataclasses, re, typing
|
||||
|
||||
def _tokens(s: str, token: str) -> typing.List[int]:
|
||||
backslash = False
|
||||
tokens = []
|
||||
for i, c in enumerate(s):
|
||||
if not backslash:
|
||||
if c == token:
|
||||
tokens.append(i)
|
||||
elif c == "\\":
|
||||
backslash = True
|
||||
else:
|
||||
backslash = False
|
||||
return tokens
|
||||
|
||||
class Sed(object):
|
||||
type: str
|
||||
def match(self, s: str) -> typing.Optional[str]:
|
||||
|
@ -13,7 +26,17 @@ class SedReplace(Sed):
|
|||
count: int
|
||||
|
||||
def match(self, s):
|
||||
return self.pattern.sub(self.replace, s, self.count)
|
||||
matches = list(self.pattern.finditer(s))
|
||||
if not self.count == 0:
|
||||
matches = matches[:self.count]
|
||||
|
||||
for match in matches:
|
||||
replace_copy = self.replace
|
||||
for token in reversed(_tokens(replace_copy, "&")):
|
||||
replace_copy = (
|
||||
replace_copy[:token]+match.group(0)+replace_copy[token+1:])
|
||||
s = s.replace(match.group(0), replace_copy, 1)
|
||||
return s
|
||||
|
||||
@dataclasses.dataclass
|
||||
class SedMatch(Sed):
|
||||
|
@ -26,24 +49,15 @@ class SedMatch(Sed):
|
|||
return match.group(0)
|
||||
return None
|
||||
|
||||
def _sed_split(s):
|
||||
backslash = False
|
||||
forward_slash = []
|
||||
for i, c in enumerate(s):
|
||||
if not backslash:
|
||||
if c == "/":
|
||||
forward_slash.append(i)
|
||||
if c == "\\":
|
||||
backslash = True
|
||||
else:
|
||||
backslash = False
|
||||
if forward_slash and (not forward_slash[-1] == (len(s)-1)):
|
||||
forward_slash.append(len(s))
|
||||
def _sed_split(s: str) -> typing.List[str]:
|
||||
tokens = _tokens(s, "/")
|
||||
if tokens and (not tokens[-1] == (len(s)-1)):
|
||||
tokens.append(len(s))
|
||||
|
||||
last = 0
|
||||
out = []
|
||||
for i in forward_slash:
|
||||
out.append(s[last:i])
|
||||
for i in tokens:
|
||||
out.append(s[last:i].replace("\\/", "/"))
|
||||
last = i+1
|
||||
return out
|
||||
|
||||
|
@ -56,13 +70,13 @@ def _sed_flags(s: str) -> typing.Tuple[int, int]:
|
|||
re_flags |= re.I
|
||||
return count, re_flags
|
||||
|
||||
def parse_sed(sed_s: str) -> typing.Optional[Sed]:
|
||||
def parse(sed_s: str) -> typing.Optional[Sed]:
|
||||
type, pattern, *args = _sed_split(sed_s)
|
||||
if type == "s":
|
||||
replace, *args = args
|
||||
count, flags = _sed_flags((args or [""])[0])
|
||||
pattern = re.compile(pattern, flags)
|
||||
return SedReplace(type, pattern, replace, count)
|
||||
pattern_re = re.compile(pattern, flags)
|
||||
return SedReplace(type, pattern_re, replace, count)
|
||||
elif type == "m":
|
||||
count, flags = _sed_flags((args or [""])[0])
|
||||
return SedMatch(type, re.compile(pattern, flags))
|
||||
|
|
Loading…
Reference in a new issue