Add functionality to export to markdown

This commit is contained in:
Schimon Jehudah 2024-01-03 12:17:39 +00:00
parent a3d68f6a8f
commit 9386e5de49
4 changed files with 76 additions and 36 deletions

View file

@ -22,6 +22,20 @@ def now():
return date return date
def current_date():
"""
Print MM DD, YYYY (Weekday Time) timestamp.
Returns
-------
date : str
MM DD, YYYY (Weekday Time) timestamp.
"""
now = datetime.now()
time = now.strftime("%B %d, %Y (%A %T)")
return time
def current_time(): def current_time():
""" """
Print HH:MM:SS timestamp. Print HH:MM:SS timestamp.

18
slixfeed/export.py Normal file
View file

@ -0,0 +1,18 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from slixfeed.datetime import current_date
def markdown(jid, filename, results):
with open(filename, 'w') as file:
file.write(
'# Subscriptions for {}\n'.format(jid))
file.write(
'## Set of feeds exported with Slixfeed\n')
for result in results:
file.write(
'- [{}]({})\n'.format(result[0], result[1]))
file.write(
'\n\n* * *\n\nThis list was saved on {} from xmpp:{} using '
'[Slixfeed](https://gitgud.io/sjehuda/slixfeed)\n'.format(
current_date(), jid))

View file

@ -33,8 +33,6 @@
'version': 'opml2' 'version': 'opml2'
} }
"""
import listparser import listparser
import lxml import lxml
@ -48,17 +46,19 @@ async def import_from_file(db_file, opml_doc):
# tags = feed['tags'] # tags = feed['tags']
# await datahandler.add_feed_no_check(db_file, [url, title]) # await datahandler.add_feed_no_check(db_file, [url, title])
"""
from slixfeed.datetime import current_time from slixfeed.datetime import current_time
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
# NOTE Use OPyML or LXML # NOTE Use OPyML or LXML
async def export_to_file(jid, filename, results): def export_to_file(jid, filename, results):
root = ET.Element("opml") root = ET.Element("opml")
root.set("version", "1.0") root.set("version", "1.0")
head = ET.SubElement(root, "head") head = ET.SubElement(root, "head")
ET.SubElement(head, "title").text = "Subscriptions for {}".format(jid) ET.SubElement(head, "title").text = "Subscriptions for {}".format(jid)
ET.SubElement(head, "description").text = ( ET.SubElement(head, "description").text = (
"Set of feeds exported with Slixfeed.") "Set of feeds exported with Slixfeed")
ET.SubElement(head, "generator").text = "Slixfeed" ET.SubElement(head, "generator").text = "Slixfeed"
ET.SubElement(head, "urlPublic").text = ( ET.SubElement(head, "urlPublic").text = (
"https://gitgud.io/sjehuda/slixfeed") "https://gitgud.io/sjehuda/slixfeed")

View file

@ -26,6 +26,7 @@ from slixfeed.config import (
get_pathname_to_database, get_pathname_to_database,
remove_from_list) remove_from_list)
from slixfeed.datetime import current_time, timestamp from slixfeed.datetime import current_time, timestamp
import slixfeed.export as export
import slixfeed.fetch as fetcher import slixfeed.fetch as fetcher
import slixfeed.opml as opml import slixfeed.opml as opml
import slixfeed.sqlite as sqlite import slixfeed.sqlite as sqlite
@ -38,6 +39,7 @@ import slixfeed.xmpp.muc as groupchat
import slixfeed.xmpp.status as status import slixfeed.xmpp.status as status
import slixfeed.xmpp.text as text import slixfeed.xmpp.text as text
import slixfeed.xmpp.upload as upload import slixfeed.xmpp.upload as upload
from slixfeed.xmpp.utility import jid_type
async def event(self, event): async def event(self, event):
@ -326,38 +328,43 @@ async def message(self, message):
response = "Missing keywords." response = "Missing keywords."
send_reply_message(self, message, response) send_reply_message(self, message, response)
case _ if message_lowercase.startswith("export "): case _ if message_lowercase.startswith("export "):
valid = 1
key = message_text[7:] key = message_text[7:]
if key in ("opml", "html", "md", "xbel"):
status_type = "dnd"
status_message = (
"📂️ Procesing request to export feeds into {} ..."
).format(key)
send_status_message(
self, jid, status_type, status_message)
data_dir = get_default_dbdir() data_dir = get_default_dbdir()
if not os.path.isdir(data_dir): if not os.path.isdir(data_dir):
os.mkdir(data_dir) os.mkdir(data_dir)
if not os.path.isdir(data_dir + '/' + key):
os.mkdir(data_dir + '/' + key)
filename = os.path.join( filename = os.path.join(
data_dir, "opml", "slixfeed_" + timestamp() + "." + key) data_dir, key, "slixfeed_" + timestamp() + "." + key)
db_file = get_pathname_to_database(jid) db_file = get_pathname_to_database(jid)
results = await sqlite.get_feeds(db_file) results = await sqlite.get_feeds(db_file)
match key: match key:
case "opml":
status_type = "dnd"
status_message = (
"📂️ Procesing request to export feeds into OPML ...")
send_status_message(self, jid, status_type, status_message)
await opml.export_to_file(
jid, filename, results)
url = await upload.start(self, jid, filename)
response = (
"Feeds exported successfully to an OPML "
"Outline Syndication.\n{}").format(url)
await task.start_tasks_xmpp(self, jid, ["status"])
case "html": case "html":
response = "Not yet implemented." response = "Not yet implemented."
case "markdown": case "md":
response = "Not yet implemented" export.markdown(
case _: jid, filename, results)
response = "Unsupported filetype." case "opml":
valid = 0 opml.export_to_file(
if valid: jid, filename, results)
case "xbel":
response = "Not yet implemented."
url = await upload.start(self, jid, filename)
response = (
"Feeds exported successfully to {}.\n{}"
).format(key, url)
# send_oob_reply_message(message, url, response) # send_oob_reply_message(message, url, response)
send_oob_message(self, jid, url) await send_oob_message(self, jid, url)
await task.start_tasks_xmpp(self, jid, ["status"])
else:
response = "Unsupported filetype."
send_reply_message(self, message, response) send_reply_message(self, message, response)
case _ if (message_lowercase.startswith("gemini") or case _ if (message_lowercase.startswith("gemini") or
message_lowercase.startswith("gopher")): message_lowercase.startswith("gopher")):
@ -775,11 +782,12 @@ def send_oob_reply_message(message, url, response):
reply.send() reply.send()
def send_oob_message(self, jid, url): async def send_oob_message(self, jid, url):
chat_type = await jid_type(self, jid)
html = ( html = (
f'<body xmlns="http://www.w3.org/1999/xhtml">' f'<body xmlns="http://www.w3.org/1999/xhtml">'
f'<a href="{url}">{url}</a></body>') f'<a href="{url}">{url}</a></body>')
message = self.make_message(mto=jid, mbody=url, mhtml=html) message = self.make_message(mto=jid, mbody=url, mhtml=html, mtype=chat_type)
message['oob']['url'] = url message['oob']['url'] = url
message.send() message.send()