feat(opml): process feed title when export/import

Signed-off-by: Rongrong <[email protected]>

src/command/inner/sub.py

@@ -283,6 +283,7 @@ async def export_opml(user_id: int) -> Optional[bytes]:
         outline = opml.new_tag(name='outline', attrs=dict(
             type='rss',
             text=_sub.title or _sub.feed.title,
+            title=_sub.feed.title,
             xmlUrl=_sub.feed.link
         ))
         opml.body.append(outline)

src/command/opml.py

@@ -91,9 +91,14 @@ async def opml_import(event: Union[events.NewMessage.Event, Message],
         logger.warning(f'Failed to parse opml file from {chat_id}')
         return
 
-    import_result = await inner.sub.subs(chat_id,
-                                         tuple((feed.url, feed.title) for feed in opml_d.feeds),
-                                         lang=lang)
+    import_result = await inner.sub.subs(
+        chat_id,
+        tuple(
+            (feed.url, feed.text) if feed.text and feed.text != feed.title_orig else feed.url
+            for feed in opml_d.feeds
+        ),
+        lang=lang
+    )
     logger.info(f'Imported feed(s) for {chat_id}')
     msg = await send_success_and_failure_msg(reply, **import_result, lang=lang, edit=True)
 

src/compat.py

@@ -8,8 +8,10 @@
 
 from typing import Callable
 
+import copy
 import functools
-
+import listparser.opml
+import listparser.common
 from aiohttp import ClientResponse
 from bs4 import BeautifulSoup
 from cachetools.keys import hashkey
@@ -174,3 +176,78 @@ def bozo_exception_removal_wrapper(func: Callable, *args, **kwargs):
         del ret['bozo_exception']
 
     return ret
+
+
+class OpmlMixin(listparser.opml.OpmlMixin):
+    """
+    Monkey-patching `listparser.opml.OpmlMixin` to support `text` and `title_orig`
+    https://github.com/kurtmckee/listparser/issues/71
+    Copied and modified from
+    https://github.com/kurtmckee/listparser/blob/1910c45232f679e63294c1bc1bcc1520a10b0383/src/listparser/opml.py#L20-L82
+    """
+
+    def start_opml_outline(self, attrs):
+        url = None
+        # Find an appropriate title in @text or @title (else empty)
+        text = attrs.get('text', '').strip()
+        title_orig = attrs.get('title', '').strip()
+        title = text or title_orig
+
+        # Search for the URL regardless of xmlUrl's case
+        for k, v in attrs.items():
+            if k.lower() == 'xmlurl':
+                url = v.strip()
+                break
+
+        append_to = None
+
+        # Determine whether the outline is a feed or subscription list
+        if url is not None:
+            # It's a feed
+            append_to = 'feeds'
+            if attrs.get('type', '').strip().lower() == 'source':
+                # Actually, it's a subscription list!
+                append_to = 'lists'
+        elif attrs.get('type', '').lower() in ('link', 'include'):
+            # It's a subscription list
+            append_to = 'lists'
+            url = attrs.get('url', '').strip()
+        elif title:
+            # Assume that this is a grouping node
+            self.hierarchy.append(title)
+            return
+        # Look for an opportunity URL
+        if not url and 'htmlurl' in (k.lower() for k in attrs.keys()):
+            for k, v in attrs.items():
+                if k.lower() == 'htmlurl':
+                    url = v.strip()
+            append_to = 'opportunities'
+        if not url:
+            # Maintain the hierarchy
+            self.hierarchy.append('')
+            return
+        if url not in self.found_urls and append_to:
+            # This is a brand new URL
+            obj = listparser.common.SuperDict({'url': url, 'title': title, 'text': text, 'title_orig': title_orig})
+            self.found_urls[url] = (append_to, obj)
+            self.harvest[append_to].append(obj)
+        else:
+            obj = self.found_urls[url][1]
+
+        # Handle categories and tags
+        obj.setdefault('categories', [])
+        if 'category' in attrs.keys():
+            for i in attrs['category'].split(','):
+                tmp = [j.strip() for j in i.split('/') if j.strip()]
+                if tmp and tmp not in obj['categories']:
+                    obj['categories'].append(tmp)
+        # Copy the current hierarchy into `categories`
+        if self.hierarchy and self.hierarchy not in obj['categories']:
+            obj['categories'].append(copy.copy(self.hierarchy))
+        # Copy all single-element `categories` into `tags`
+        obj['tags'] = [i[0] for i in obj['categories'] if len(i) == 1]
+
+        self.hierarchy.append('')
+
+
+listparser.opml.OpmlMixin.start_opml_outline = OpmlMixin.start_opml_outline