iartamonov/mautrix-telegram
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch 'master' into authless-relaybot-portals

Browse Source
This commit is contained in:
Tulir Asokan
2018-02-23 16:02:18 +02:00
parent 2d2fe86757 a26f2c2c36
commit c2e4f5596c
2 changed files with 114 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mautrix_telegram/formatter/from_matrix.py
+113 -67
View File
@@ -17,6 +17,7 @@
from html import unescape
from html.parser import HTMLParser
from collections import deque
import math
import re
import logging
@@ -32,6 +33,10 @@ log = logging.getLogger("mau.fmt.mx")
class MatrixParser(HTMLParser):
mention_regex = re.compile("https://matrix.to/#/(@.+:.+)")
room_regex = re.compile("https://matrix.to/#/(#.+:.+)")
block_tags = ("br", "p", "pre", "blockquote",
"ol", "ul", "li",
"h1", "h2", "h3", "h4", "h5", "h6"
"div", "hr", "table")
def __init__(self):
super().__init__()
@@ -41,112 +46,151 @@ class MatrixParser(HTMLParser):
self._list_counter = 0
self._open_tags = deque()
self._open_tags_meta = deque()
self._previous_ended_line = True
self._line_is_new = True
self._list_entry_is_new = False
def _parse_url(self, url, args):
mention = self.mention_regex.match(url)
if mention:
mxid = mention.group(1)
user = pu.Puppet.get_by_mxid(mxid, create=False)
if not user:
user = u.User.get_by_mxid(mxid, create=False)
if not user:
return None, None
if user.username:
entity_type = MessageEntityMention
url = f"@{user.username}"
else:
entity_type = MessageEntityMentionName
args["user_id"] = user.tgid
return url, entity_type
room = self.room_regex.match(url)
if room:
username = po.Portal.get_username_from_mx_alias(room.group(1))
portal = po.Portal.find_by_username(username)
if portal and portal.username:
return f"@{portal.username}", MessageEntityMention
if url.startswith("mailto:"):
return MessageEntityEmail, url[len("mailto:"):]
if self.get_starttag_text() == url:
return MessageEntityUrl, url
else:
args["url"] = url
return MessageEntityTextUrl, None
def handle_starttag(self, tag, attrs):
self._open_tags.appendleft(tag)
self._open_tags_meta.appendleft(0)
attrs = dict(attrs)
entity_type = None
args = {}
if tag == "strong" or tag == "b":
if tag in ("strong", "b"):
entity_type = MessageEntityBold
elif tag == "em" or tag == "i":
elif tag in ("em", "i"):
entity_type = MessageEntityItalic
elif tag == "code":
try:
pre = self._building_entities["pre"]
try:
# Pre tag and language found, add language to MessageEntityPre
pre.language = attrs["class"][len("language-"):]
except KeyError:
# Pre tag found, but language not found, keep pre as-is
pass
except KeyError:
# No pre tag found, this is inline code
entity_type = MessageEntityCode
elif tag == "pre":
entity_type = MessageEntityPre
args["language"] = ""
elif tag == "li":
self._list_entry_is_new = True
elif tag == "a":
try:
url = attrs["href"]
except KeyError:
return
mention = self.mention_regex.match(url)
room = self.room_regex.match(url)
if mention:
mxid = mention.group(1)
user = pu.Puppet.get_by_mxid(mxid, create=False)
if not user:
user = u.User.get_by_mxid(mxid, create=False)
if not user:
return
if user.username:
entity_type = MessageEntityMention
url = f"@{user.username}"
else:
entity_type = MessageEntityMentionName
args["user_id"] = user.tgid
elif room:
username = po.Portal.get_username_from_mx_alias(room.group(1))
portal = po.Portal.find_by_username(username)
if portal and portal.username:
url = f"@{portal.username}"
entity_type = MessageEntityMention
elif url.startswith("mailto:"):
url = url[len("mailto:"):]
entity_type = MessageEntityEmail
else:
if self.get_starttag_text() == url:
entity_type = MessageEntityUrl
else:
entity_type = MessageEntityTextUrl
args["url"] = url
url = None
entity_type, url = self._parse_url(url, args)
self._open_tags_meta.popleft()
self._open_tags_meta.appendleft(url)
if tag in self.block_tags:
self._newline()
if entity_type and tag not in self._building_entities:
offset = len(self.text)
self._building_entities[tag] = entity_type(offset=offset, length=0, **args)
def _list_depth(self):
depth = 0
for tag in self._open_tags:
if tag == "ol" or tag == "ul":
depth += 1
return depth
@property
def _list_indent(self):
indent = 0
first_skipped = False
for index, tag in enumerate(self._open_tags):
if not first_skipped and tag in ("ol", "ul"):
# The first list level isn't indented, so skip it.
first_skipped = True
continue
if tag == "ol":
n = self._open_tags_meta[index]
extra_length_for_long_index = (int(math.log(n, 10)) - 1) * 3
indent += 4 + extra_length_for_long_index
elif tag == "ul":
indent += 3
return indent
def _newline(self, allow_multi=False):
if self._line_is_new or allow_multi:
return
self.text += "\n"
self._line_is_new = True
for entity in self._building_entities.values():
entity.length += 1
def handle_data(self, text):
text = unescape(text)
previous_tag = self._open_tags[0] if len(self._open_tags) > 0 else ""
list_format_offset = 0
extra_offset = 0
if previous_tag == "a":
url = self._open_tags_meta[0]
if url:
text = url
elif len(self._open_tags) > 1 and self._previous_ended_line and previous_tag == "li":
list_type = self._open_tags[1]
indent = (self._list_depth() - 1) * 4 * " "
text = text.strip("\n")
if len(text) == 0:
return
elif list_type == "ul":
text = f"{indent}* {text}"
list_format_offset = len(indent) + 2
elif list_type == "ol":
n = self._open_tags_meta[1]
n += 1
self._open_tags_meta[1] = n
text = f"{indent}{n}. {text}"
list_format_offset = len(indent) + 3
list_entry_handled_once = False
# In order to maintain order of things like blockquotes in lists or lists in blockquotes,
# we can't just have ifs/elses and we need to actually loop through the open tags in order.
for index, tag in enumerate(self._open_tags):
if tag == "blockquote" and self._line_is_new:
text = f"> {text}"
extra_offset += 2
elif tag == "li" and not list_entry_handled_once:
list_type_index = index + 1
list_type = self._open_tags[list_type_index]
indent = self._list_indent * " " if self._line_is_new else ""
if list_type == "ol":
n = self._open_tags_meta[list_type_index]
if self._list_entry_is_new:
n += 1
self._open_tags_meta[list_type_index] = n
prefix = f"{n}. "
else:
prefix = int(math.log(n, 10)) * 3 * " " + 4 * " "
else:
prefix = "* " if self._list_entry_is_new else 3 * " "
if not self._list_entry_is_new and not self._line_is_new:
prefix = ""
extra_offset = len(indent) + len(prefix)
text = indent + prefix + text
self._list_entry_is_new = False
list_entry_handled_once = True
for tag, entity in self._building_entities.items():
entity.length += len(text.strip("\n"))
entity.offset += list_format_offset
if text.endswith("\n"):
self._previous_ended_line = True
else:
self._previous_ended_line = False
entity.length += len(text) - extra_offset
entity.offset += extra_offset
self._line_is_new = False
self.text += text
def handle_endtag(self, tag):
@@ -155,8 +199,10 @@ class MatrixParser(HTMLParser):
self._open_tags_meta.popleft()
except IndexError:
pass
if (tag == "ul" or tag == "ol") and self.text.endswith("\n"):
self.text = self.text[:-1]
if tag in self.block_tags:
self._newline(allow_multi=tag == "br")
entity = self._building_entities.pop(tag, None)
if entity:
self.entities.append(entity)
@@ -165,8 +211,8 @@ class MatrixParser(HTMLParser):
def matrix_to_telegram(html):
try:
parser = MatrixParser()
parser.feed(add_surrogates(html))
return remove_surrogates(parser.text), parser.entities
parser.feed(add_surrogates(html.replace("\n", "")))
return remove_surrogates(parser.text.strip()), parser.entities
except Exception:
log.exception("Failed to convert Matrix format:\nhtml=%s", html)
setup.py
+1 -1
View File
@@ -28,7 +28,7 @@ setuptools.setup(
],
dependency_links=[
("https://github.com/LonamiWebs/Telethon/tarball/7998fd59f709ae1cd959c5cc4ab107982307f4a6#egg=Telethon"
if sys.version_info > (3, 5)
if sys.version_info >= (3, 6)
else "https://github.com/tulir/Telethon/tarball/ca08fe28800d74fd6c19fd6f473e12fbf2c258de#egg=Telethon")
],