feat(parsers): add HTML parser combine it with telegram message entities

aiogram/utils/_parsers.py

@@ -0,0 +1,165 @@
+# Implementation inspired on the similar one from telethon
+# Credits to https://github.com/LonamiWebs/Telethon
+
+from typing import Tuple, List, Dict, Deque, Optional
+import struct
+from collections import deque
+from html.parser import HTMLParser
+
+from aiogram.types.message_entity import (
+    MessageEntity,
+    MessageEntityType,
+)
+
+
+# region Unicode surrogates
+def add_surrogate(text: str) -> str:
+    return ''.join(
+        # SMP -> Surrogate Pairs (Telegram offsets are calculated with these).
+        # See https://en.wikipedia.org/wiki/Plane_(Unicode)#Overview for more.
+        ''.join(chr(y) for y in struct.unpack('<HH', x.encode('utf-16le')))
+        if (0x10000 <= ord(x) <= 0x10FFFF) else x for x in text
+    )
+
+
+def del_surrogate(text: str) -> str:
+    return text.encode('utf-16', 'surrogatepass').decode('utf-16')
+# endregion
+
+
+def strip_text(text: str, entities: List[MessageEntity]) -> str:
+    if not entities:
+        return text.strip()
+
+    while text and text[-1].isspace():
+        e = entities[-1]
+        if e.offset + e.length == len(text):
+            if e.length == 1:
+                del entities[-1]
+                if not entities:
+                    return text.strip()
+            else:
+                e.length -= 1
+        text = text[:-1]
+
+    while text and text[0].isspace():
+        for i in reversed(range(len(entities))):
+            e = entities[i]
+            if e.offset != 0:
+                e.offset -= 1
+                continue
+
+            if e.length == 1:
+                del entities[0]
+                if not entities:
+                    return text.lstrip()
+            else:
+                e.length -= 1
+
+        text = text[1:]
+
+    return text
+
+
+# region HTML parser
+class _HTMLToTelegramParser(HTMLParser):
+    def __init__(self):
+        super(_HTMLToTelegramParser, self).__init__(convert_charrefs=False)
+
+        self.text = ''
+
+        self.entities: List[MessageEntity] = []
+        self._building_entities: Dict[str, MessageEntity] = {}
+
+        self._open_tags: Deque[MessageEntity] = deque()
+        self._open_tags_meta: Deque[Optional[MessageEntity]] = deque()
+
+    def handle_starttag(self, tag, attrs):
+        self._open_tags.appendleft(tag)
+        self._open_tags_meta.appendleft(None)
+
+        attrs = dict(attrs)
+        entity_type = None
+        args = {}
+        if tag == 'strong' or tag == 'b':
+            entity_type = MessageEntityType.BOLD
+        elif tag == 'em' or tag == 'i':
+            entity_type = MessageEntityType.ITALIC
+        elif tag == 'u':
+            entity_type = MessageEntityType.UNDERLINE
+        elif tag == 'del' or tag == 's':
+            entity_type = MessageEntityType.STRIKETHROUGH
+        elif tag == 'code':
+            try:
+                pre = self._building_entities['pre']
+                try:
+                    pre.language = attrs['class'][len('language-'):]
+                except KeyError:
+                    pass
+            except KeyError:
+                entity_type = MessageEntityType.CODE
+        elif tag == 'pre':
+            entity_type = MessageEntityType.PRE
+            args['language'] = ''
+        elif tag == 'a':
+            try:
+                url = attrs['href']
+            except KeyError:
+                return
+            if url.startswith('mailto:'):
+                url = url[len('mailto:'):]
+                entity_type = MessageEntityType.EMAIL
+            else:
+                if self.get_starttag_text() == url:
+                    entity_type = MessageEntityType.URL
+                else:
+                    entity_type = MessageEntityType.TEXT_LINK
+                    args['url'] = url
+                    url = None
+            self._open_tags_meta.popleft()
+            self._open_tags_meta.appendleft(url)
+
+        if entity_type and tag not in self._building_entities:
+            self._building_entities[tag] = MessageEntity(
+                type=entity_type,
+                offset=len(self.text),
+                length=0,
+                **args,)
+
+    def handle_data(self, text):
+        previous_tag = self._open_tags[0] if len(self._open_tags) > 0 else ''
+        if previous_tag == 'a':
+            url = self._open_tags_meta[0]
+            if url:
+                text = url
+
+        for tag, entity in self._building_entities.items():
+            entity.length += len(text)
+
+        self.text += text
+
+    def handle_endtag(self, tag):
+        try:
+            self._open_tags.popleft()
+            self._open_tags_meta.popleft()
+        except IndexError:
+            pass
+        entity = self._building_entities.pop(tag, None)
+        if entity:
+            self.entities.append(entity)
+
+    def error(self, message):
+        raise ValueError(message)
+
+
+def parse_html(html: str) -> Tuple[str, List[MessageEntity]]:
+    if not html:
+        return html, []
+
+    parser = _HTMLToTelegramParser()
+    parser.feed(add_surrogate(html))
+    text = strip_text(parser.text, parser.entities)
+    return del_surrogate(text), parser.entities
+
+# endregion
+