579 lines
28 KiB
Python
579 lines
28 KiB
Python
#!/usr/bin/python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import datetime
|
|
from fastapi import FastAPI, Request, Response
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
import json
|
|
from os import mkdir
|
|
from os.path import exists
|
|
from slixmpp import ClientXMPP
|
|
from slixmpp.exceptions import IqError, IqTimeout
|
|
import xml.etree.ElementTree as ET
|
|
#import importlib.resources
|
|
|
|
try:
|
|
import tomllib
|
|
except:
|
|
import tomli as tomllib
|
|
|
|
class XmppInstance(ClientXMPP):
|
|
def __init__(self, jid, password):
|
|
super().__init__(jid, password)
|
|
self.register_plugin('xep_0060')
|
|
self.connect()
|
|
# self.process(forever=False)
|
|
|
|
xmpp = None
|
|
|
|
app = FastAPI()
|
|
|
|
# Mount static graphic, script and stylesheet directories
|
|
app.mount("/css", StaticFiles(directory="css"), name="css")
|
|
app.mount("/data", StaticFiles(directory="data"), name="data")
|
|
app.mount("/graphic", StaticFiles(directory="graphic"), name="graphic")
|
|
app.mount("/script", StaticFiles(directory="script"), name="script")
|
|
app.mount("/xsl", StaticFiles(directory="xsl"), name="xsl")
|
|
|
|
@app.get('/favicon.ico', include_in_schema=False)
|
|
async def favicon():
|
|
return FileResponse('favicon.ico')
|
|
|
|
@app.route('/')
|
|
@app.get('/opml')
|
|
async def view_pubsub_nodes(request: Request):
|
|
global xmpp
|
|
if not xmpp:
|
|
credentials = get_configuration('account')
|
|
xmpp = XmppInstance(credentials['xmpp'], credentials['pass'])
|
|
# xmpp.connect()
|
|
|
|
pubsub = request.query_params.get('pubsub', '')
|
|
settings = get_configuration('settings')
|
|
result = None
|
|
if settings['service']:
|
|
if settings['include'] in pubsub or not settings['include']:
|
|
if pubsub:
|
|
iq = await get_nodes(pubsub)
|
|
if iq:
|
|
link = 'xmpp:{pubsub}'.format(pubsub=pubsub)
|
|
xml_opml = generate_opml(iq)
|
|
result = append_stylesheet(xml_opml, 'opml')
|
|
else:
|
|
text = 'Please ensure that PubSub "{}" (Jabber ID) is valid and accessible.'.format(pubsub)
|
|
xml_atom = error_message(text)
|
|
result = append_stylesheet(xml_atom, 'atom')
|
|
else:
|
|
text = 'The given domain {} is not allowed.'.format(pubsub)
|
|
xml_atom = error_message(text)
|
|
result = append_stylesheet(xml_atom, 'atom')
|
|
default = get_configuration('default')
|
|
if not result:
|
|
if default['pubsub']:
|
|
if not pubsub:
|
|
pubsub = default['pubsub']
|
|
iq = await get_nodes(pubsub)
|
|
link = 'xmpp:{pubsub}'.format(pubsub=pubsub)
|
|
xml_opml = generate_opml(iq)
|
|
result = append_stylesheet(xml_opml, 'opml')
|
|
elif not settings['service']:
|
|
pubsub = default['pubsub']
|
|
link = 'xmpp:{pubsub}'.format(pubsub=pubsub)
|
|
xml_opml = generate_opml(iq)
|
|
result = append_stylesheet(xml_opml, 'opml')
|
|
else:
|
|
text = 'Please contact the administrator and ask him to set default PubSub and Node ID.'
|
|
xml_atom = error_message(text)
|
|
result = append_stylesheet(xml_atom, 'atom')
|
|
response = Response(content=result, media_type="application/xml")
|
|
return response
|
|
|
|
@app.get('/atom')
|
|
async def view_node_items(request: Request):
|
|
global xmpp
|
|
if not xmpp:
|
|
credentials = get_configuration('account')
|
|
xmpp = XmppInstance(credentials['xmpp'], credentials['pass'])
|
|
# xmpp.connect()
|
|
|
|
pubsub = request.query_params.get('pubsub', '')
|
|
node = request.query_params.get('node', '')
|
|
item_id = request.query_params.get('item', '')
|
|
settings = get_configuration('settings')
|
|
result = None
|
|
if settings['service']:
|
|
if settings['include'] in pubsub or not settings['include']:
|
|
if pubsub and node and item_id:
|
|
iq = await get_node_item(pubsub, node, item_id)
|
|
if iq:
|
|
link = form_an_item_link(pubsub, node, item_id)
|
|
xml_atom = generate_atom_comment(iq, link) if 'urn:xmpp:microblog:0:comments/' in node else generate_atom_post(iq, link)
|
|
iq = await get_node_items(pubsub, node)
|
|
if not '/' in node:
|
|
if iq:
|
|
generate_json(iq)
|
|
else:
|
|
operator = get_configuration('settings')['operator']
|
|
json_data = [{'title' : 'Error retrieving node items.',
|
|
'link' : ('javascript:alert("Rivista has experienced an error '
|
|
'while attempting to retrieve the list of items for '
|
|
'Node {} of PubSub {}.")')
|
|
.format(node, pubsub)},
|
|
{'title' : 'Contact the operator.',
|
|
'link' : ('xmpp:{}?message;subject=Rivista;body=Greetings! '
|
|
'I am contacting you to inform you that there is an error listing '
|
|
'node items for Node {} on PubSub {}.').format(operator, node, pubsub)}]
|
|
filename = 'data/{}.json'.format(node)
|
|
with open(filename, 'w', encoding='utf-8') as f:
|
|
json.dump(json_data, f, ensure_ascii=False, indent=4)
|
|
else:
|
|
text = 'Please ensure that PubSub node "{}" and item "{}" are valid and accessible.'.format(node, item_id)
|
|
xml_atom = error_message(text)
|
|
result = append_stylesheet(xml_atom, 'atom')
|
|
|
|
# try:
|
|
# iq = await get_node_items(pubsub, node)
|
|
# generate_json(iq, node)
|
|
# except:
|
|
# operator = get_configuration('settings')['operator']
|
|
# json_data = [{'title' : 'Timeout retrieving node items from {}'.format(node),
|
|
# 'link' : 'xmpp:{}?message'.format(operator)}]
|
|
# filename = 'data/{}.json'.format(node)
|
|
# with open(filename, 'w', encoding='utf-8') as f:
|
|
# json.dump(json_data, f, ensure_ascii=False, indent=4)
|
|
elif pubsub and node:
|
|
iq = await get_node_items(pubsub, node)
|
|
if iq:
|
|
link = form_a_node_link(pubsub, node)
|
|
xml_atom = generate_atom_comment(iq, link) if 'urn:xmpp:microblog:0:comments/' in node else generate_atom_post(iq, link)
|
|
else:
|
|
text = 'Please ensure that PubSub node "{}" is valid and accessible.'.format(node)
|
|
xml_atom = error_message(text)
|
|
result = append_stylesheet(xml_atom, 'atom')
|
|
elif pubsub:
|
|
text = 'Node parameter is missing.'
|
|
xml_atom = error_message(text)
|
|
result = append_stylesheet(xml_atom, 'atom')
|
|
elif node:
|
|
text = 'PubSub parameter is missing.'
|
|
xml_atom = error_message(text)
|
|
result = append_stylesheet(xml_atom, 'atom')
|
|
# else:
|
|
# text = ('Mandatory parameter PubSub and '
|
|
# 'optional parameter Node are missing.')
|
|
# xml_atom = error_message(text)
|
|
# result = append_stylesheet(xml_atom, 'atom')
|
|
else:
|
|
text = 'The given domain {} is not allowed.'.format(pubsub)
|
|
xml_atom = error_message(text)
|
|
result = append_stylesheet(xml_atom, 'atom')
|
|
if not result:
|
|
default = get_configuration('default')
|
|
if default['pubsub'] and default['nodeid']:
|
|
if not pubsub and not node:
|
|
pubsub = default['pubsub']
|
|
node = default['nodeid']
|
|
iq = await get_node_items(pubsub, node)
|
|
if iq:
|
|
link = form_a_node_link(pubsub, node)
|
|
xml_atom = generate_atom_post(iq, link)
|
|
else:
|
|
text = 'Please ensure that PubSub node "{}" is valid and accessible.'.format(node)
|
|
xml_atom = error_message(text)
|
|
elif not settings['service']:
|
|
pubsub = default['pubsub']
|
|
node = default['nodeid']
|
|
iq = await get_node_items(pubsub, node)
|
|
if iq:
|
|
link = form_a_node_link(pubsub, node)
|
|
xml_atom = generate_atom_post(iq, link)
|
|
else:
|
|
text = 'Please ensure that PubSub node "{}" is valid and accessible.'.format(node)
|
|
xml_atom = error_message(text)
|
|
result = append_stylesheet(xml_atom, 'atom')
|
|
else:
|
|
text = 'Please contact the administrator and ask him to set default PubSub and Node ID.'
|
|
xml_atom = error_message(text)
|
|
result = append_stylesheet(xml_atom, 'atom')
|
|
response = Response(content=result, media_type="application/xml")
|
|
return response
|
|
|
|
def get_configuration(section):
|
|
with open('configuration.toml', mode="rb") as configuration:
|
|
result = tomllib.load(configuration)[section]
|
|
return result
|
|
|
|
#@timeout(5)
|
|
async def get_node_item(pubsub, node, item_id):
|
|
try:
|
|
iq = await xmpp.plugin['xep_0060'].get_item(pubsub, node, item_id, timeout=5)
|
|
return iq
|
|
except (IqError, IqTimeout) as e:
|
|
print(e)
|
|
|
|
async def get_node_items(pubsub, node):
|
|
try:
|
|
iq = await xmpp.plugin['xep_0060'].get_items(pubsub, node, timeout=5)
|
|
return iq
|
|
except (IqError, IqTimeout) as e:
|
|
print(e)
|
|
|
|
async def get_nodes(pubsub):
|
|
try:
|
|
iq = await xmpp.plugin['xep_0060'].get_nodes(pubsub, timeout=5)
|
|
return iq
|
|
except (IqError, IqTimeout) as e:
|
|
print(e)
|
|
|
|
def form_a_node_link(pubsub, node):
|
|
link = 'xmpp:{pubsub}?;node={node}'.format(pubsub=pubsub, node=node)
|
|
return link
|
|
|
|
def form_an_item_link(pubsub, node, item_id):
|
|
link = 'xmpp:{pubsub}?;node={node};item={item}'.format(
|
|
pubsub=pubsub, node=node, item=item_id)
|
|
return link
|
|
|
|
def error_message(text):
|
|
"""Error message in RFC 4287: The Atom Syndication Format."""
|
|
title = 'Rivista'
|
|
subtitle = 'XMPP Journal Publisher'
|
|
description = ('This is a syndication feed generated with Rivista, an XMPP '
|
|
'Journal Publisher, which conveys XEP-0060: Publish-'
|
|
'Subscribe nodes to standard RFC 4287: The Atom Syndication '
|
|
'Format.')
|
|
language = 'en'
|
|
feed = ET.Element("feed")
|
|
feed.set('xmlns', 'http://www.w3.org/2005/Atom')
|
|
ET.SubElement(feed, 'title', {'type': 'text'}).text = title
|
|
ET.SubElement(feed, 'subtitle', {'type': 'text'}).text = subtitle
|
|
ET.SubElement(feed, 'author', {'name':'Rivista','email':'rivista@schimon.i2p'})
|
|
ET.SubElement(feed, 'generator', {
|
|
'uri': 'https://git.xmpp-it.net/sch/Rivista',
|
|
'version': '0.1'}).text = 'Rivista XJP'
|
|
ET.SubElement(feed, 'updated').text = datetime.datetime.now(datetime.UTC).isoformat()
|
|
entry = ET.SubElement(feed, 'entry')
|
|
ET.SubElement(entry, 'title').text = 'Error'
|
|
ET.SubElement(entry, 'id').text = 'rivista-error'
|
|
ET.SubElement(entry, 'updated').text = datetime.datetime.now(datetime.UTC).isoformat()
|
|
ET.SubElement(entry, 'published').text = datetime.datetime.now(datetime.UTC).isoformat()
|
|
# ET.SubElement(entry, 'summary', {'type': summary_type_text}).text = summary_text
|
|
ET.SubElement(entry, 'content', {'type': 'text'}).text = text
|
|
return ET.tostring(feed, encoding='unicode')
|
|
|
|
# generate_rfc_4287
|
|
def generate_atom_post(iq, link):
|
|
"""Generate an Atom Syndication Format (RFC 4287) from a Publish-Subscribe (XEP-0060) node items."""
|
|
pubsub = iq['from'].bare
|
|
node = iq['pubsub']['items']['node']
|
|
title = node
|
|
link = link
|
|
# link = form_a_node_link(pubsub, node)
|
|
# subtitle = 'XMPP PubSub Syndication Feed'
|
|
subtitle = pubsub
|
|
description = ('This is a syndication feed generated with Rivista, an XMPP '
|
|
'Journal Publisher, which conveys XEP-0060: Publish-'
|
|
'Subscribe nodes to standard RFC 4287: The Atom Syndication '
|
|
'Format.')
|
|
language = iq['pubsub']['items']['lang']
|
|
items = iq['pubsub']['items']
|
|
e_feed = ET.Element("feed")
|
|
e_feed.set('xmlns', 'http://www.w3.org/2005/Atom')
|
|
ET.SubElement(e_feed, 'title', {'type': 'text'}).text = title
|
|
ET.SubElement(e_feed, 'subtitle', {'type': 'text'}).text = subtitle
|
|
ET.SubElement(e_feed, 'link', {'rel': 'self', 'href': link})
|
|
ET.SubElement(e_feed, 'generator', {
|
|
'uri': 'https://git.xmpp-it.net/sch/Rivista',
|
|
'version': '0.1'}).text = 'Rivista XJP'
|
|
ET.SubElement(e_feed, 'updated').text = datetime.datetime.now(datetime.UTC).isoformat()
|
|
for item in list(items)[::-1]:
|
|
item_id = item['id']
|
|
item_payload = item['payload']
|
|
namespace = '{http://www.w3.org/2005/Atom}'
|
|
title = item_payload.find(namespace + 'title')
|
|
links = item_payload.find(namespace + 'link')
|
|
if (not isinstance(title, ET.Element) and
|
|
not isinstance(links, ET.Element)): continue
|
|
e_entry = ET.SubElement(e_feed, 'entry')
|
|
title_text = None if title == None else title.text
|
|
ET.SubElement(e_entry, 'title').text = title_text
|
|
if isinstance(links, ET.Element):
|
|
for link in item_payload.findall(namespace + 'link'):
|
|
link_href = link.attrib['href'] if 'href' in link.attrib else ''
|
|
link_type = link.attrib['type'] if 'type' in link.attrib else ''
|
|
link_rel = link.attrib['rel'] if 'rel' in link.attrib else ''
|
|
ET.SubElement(e_entry, 'link', {'href': link_href, 'rel': link_rel, 'type': link_type})
|
|
else:
|
|
ET.SubElement(e_entry, 'content', {'href': ''})
|
|
link_xmpp = form_an_item_link(pubsub, node, item_id)
|
|
ET.SubElement(e_entry, 'link', {'href': link_xmpp, 'rel': 'alternate', 'type': 'x-scheme-handler/xmpp'})
|
|
contents = item_payload.find(namespace + 'content')
|
|
if isinstance(contents, ET.Element):
|
|
for content in item_payload.findall(namespace + 'content'):
|
|
if not content.text: continue
|
|
content_text = content.text
|
|
content_type = content.attrib['type'] if 'type' in content.attrib else 'html'
|
|
content_type_text = 'html' if 'html' in content_type else 'text'
|
|
ET.SubElement(e_entry, 'content', {'type': content_type_text}).text = content_text
|
|
else:
|
|
summary = item_payload.find(namespace + 'summary')
|
|
summary_text = summary.text if summary else None
|
|
if summary_text:
|
|
summary_type = summary.attrib['type'] if 'type' in summary.attrib else 'html'
|
|
summary_type_text = 'html' if 'html' in summary_type else 'text'
|
|
ET.SubElement(e_entry, 'content', {'type': summary_type_text}).text = summary_text
|
|
else:
|
|
ET.SubElement(e_entry, 'content').text = 'No content.'
|
|
published = item_payload.find(namespace + 'published')
|
|
published_text = None if published == None else published.text
|
|
ET.SubElement(e_entry, 'published').text = published_text
|
|
updated = item_payload.find(namespace + 'updated')
|
|
updated_text = None if updated == None else updated.text
|
|
ET.SubElement(e_entry, 'updated').text = updated_text
|
|
authors = item_payload.find(namespace + 'author')
|
|
if isinstance(authors, ET.Element):
|
|
for author in item_payload.findall(namespace + 'author'):
|
|
e_author = ET.SubElement(e_entry, 'author')
|
|
author_email = author.find(namespace + 'email')
|
|
if author_email is not None:
|
|
author_email_text = author_email.text
|
|
if author_email_text: ET.SubElement(e_author, 'email').text = author_email.text
|
|
else:
|
|
author_email_text = None
|
|
author_uri = author.find(namespace + 'uri')
|
|
if author_uri is not None:
|
|
author_uri_text = author_uri.text
|
|
if author_uri_text: ET.SubElement(e_author, 'uri').text = author_uri.text
|
|
else:
|
|
author_uri_text = None
|
|
author_name = author.find(namespace + 'name')
|
|
if author_name is not None and author_name.text:
|
|
author_name_text = author_name.text
|
|
else:
|
|
author_name_text = author_uri_text or author_email_text
|
|
ET.SubElement(e_author, 'name').text = author_name_text
|
|
for uri in item_payload.iter(namespace + 'author'):
|
|
uri_text = uri.text
|
|
if uri_text:
|
|
ET.SubElement(e_entry, 'uri').text = uri_text
|
|
# if not e_author:
|
|
# ET.SubElement(e_author, 'name').text = uri_text
|
|
# ET.SubElement(e_author, 'uri').text = uri_text
|
|
categories = item_payload.find(namespace + 'category')
|
|
if isinstance(categories, ET.Element):
|
|
for category in item_payload.findall(namespace + 'category'):
|
|
if 'term' in category.attrib and category.attrib['term']:
|
|
category_term = category.attrib['term']
|
|
ET.SubElement(e_entry, 'category', {'term': category_term})
|
|
|
|
|
|
identifier = item_payload.find(namespace + 'id')
|
|
if identifier and identifier.attrib: print(identifier.attrib)
|
|
|
|
identifier_text = None if identifier == None else identifier.text
|
|
ET.SubElement(e_entry, 'id').text = identifier_text
|
|
# ET.SubElement(e_entry, 'summary', {'type': summary_type_text}).text = summary_text
|
|
return ET.tostring(e_feed, encoding='unicode')
|
|
|
|
# generate_rfc_4287
|
|
def generate_atom_comment(iq, link):
|
|
"""Generate an Atom Syndication Format (RFC 4287) from a Publish-Subscribe (XEP-0060) node items."""
|
|
pubsub = iq['from'].bare
|
|
node = iq['pubsub']['items']['node']
|
|
title = node
|
|
link = link
|
|
# link = form_a_node_link(pubsub, node)
|
|
# subtitle = 'XMPP PubSub Syndication Feed'
|
|
subtitle = pubsub
|
|
description = ('This is a syndication feed generated with Rivista, an XMPP '
|
|
'Journal Publisher, which conveys XEP-0060: Publish-'
|
|
'Subscribe nodes to standard RFC 4287: The Atom Syndication '
|
|
'Format.')
|
|
language = iq['pubsub']['items']['lang']
|
|
items = iq['pubsub']['items']
|
|
e_feed = ET.Element("feed")
|
|
e_feed.set('xmlns', 'http://www.w3.org/2005/Atom')
|
|
ET.SubElement(e_feed, 'title', {'type': 'text'}).text = title
|
|
ET.SubElement(e_feed, 'subtitle', {'type': 'text'}).text = subtitle
|
|
ET.SubElement(e_feed, 'link', {'rel': 'self', 'href': link})
|
|
ET.SubElement(e_feed, 'generator', {
|
|
'uri': 'https://git.xmpp-it.net/sch/Rivista',
|
|
'version': '0.1'}).text = 'Rivista XJP'
|
|
ET.SubElement(e_feed, 'updated').text = datetime.datetime.now(datetime.UTC).isoformat()
|
|
for item in list(items)[::-1]:
|
|
item_id = item['id']
|
|
item_payload = item['payload']
|
|
namespace = '{http://www.w3.org/2005/Atom}'
|
|
authors = item_payload.find(namespace + 'author')
|
|
links = item_payload.find(namespace + 'link')
|
|
if (not isinstance(authors, ET.Element) and
|
|
not isinstance(links, ET.Element)): continue
|
|
e_entry = ET.SubElement(e_feed, 'entry')
|
|
author_text = None
|
|
for author in item_payload.findall(namespace + 'author'):
|
|
author_email = author.find(namespace + 'email')
|
|
if author_email is not None:
|
|
author_text = author_email.text
|
|
if not author_text:
|
|
author_uri = author.find(namespace + 'uri')
|
|
if author_uri is not None:
|
|
author_text = author_uri.text
|
|
if not author_text:
|
|
author_name = author.find(namespace + 'name')
|
|
if author_name is not None and author_name.text:
|
|
author_text = author_name.text
|
|
if not author_text:
|
|
for uri in item_payload.iter(namespace + 'author'):
|
|
author_text = uri.text
|
|
if author_text:
|
|
ET.SubElement(e_entry, 'title').text = author_text
|
|
break
|
|
if isinstance(links, ET.Element):
|
|
for link in item_payload.findall(namespace + 'link'):
|
|
link_href = link.attrib['href'] if 'href' in link.attrib else ''
|
|
link_type = link.attrib['type'] if 'type' in link.attrib else ''
|
|
link_rel = link.attrib['rel'] if 'rel' in link.attrib else ''
|
|
ET.SubElement(e_entry, 'link', {'href': link_href, 'rel': link_rel, 'type': link_type})
|
|
else:
|
|
ET.SubElement(e_entry, 'content', {'href': ''})
|
|
link_xmpp = form_an_item_link(pubsub, node, item_id)
|
|
ET.SubElement(e_entry, 'link', {'href': link_xmpp, 'rel': 'alternate', 'type': 'x-scheme-handler/xmpp'})
|
|
contents = item_payload.find(namespace + 'content')
|
|
if isinstance(contents, ET.Element):
|
|
for content in item_payload.findall(namespace + 'content'):
|
|
if not content.text: continue
|
|
content_text = content.text
|
|
content_type = content.attrib['type'] if 'type' in content.attrib else 'html'
|
|
content_type_text = 'html' if 'html' in content_type else 'text'
|
|
ET.SubElement(e_entry, 'content', {'type': content_type_text}).text = content_text
|
|
else:
|
|
summary = item_payload.find(namespace + 'summary')
|
|
summary_text = summary.text if summary else None
|
|
if summary_text:
|
|
summary_type = summary.attrib['type'] if 'type' in summary.attrib else 'html'
|
|
summary_type_text = 'html' if 'html' in summary_type else 'text'
|
|
ET.SubElement(e_entry, 'content', {'type': summary_type_text}).text = summary_text
|
|
else:
|
|
title = item_payload.find(namespace + 'title')
|
|
title_text = None if title == None else title.text
|
|
if title_text:
|
|
ET.SubElement(e_entry, 'content').text = title_text
|
|
else:
|
|
ET.SubElement(e_entry, 'content').text = 'No content.'
|
|
published = item_payload.find(namespace + 'published')
|
|
published_text = None if published == None else published.text
|
|
ET.SubElement(e_entry, 'published').text = published_text
|
|
updated = item_payload.find(namespace + 'updated')
|
|
updated_text = None if updated == None else updated.text
|
|
ET.SubElement(e_entry, 'updated').text = updated_text
|
|
authors = item_payload.find(namespace + 'author')
|
|
if isinstance(authors, ET.Element):
|
|
contact_text = None
|
|
for author in item_payload.findall(namespace + 'author'):
|
|
author_email = author.find(namespace + 'email')
|
|
if author_email is not None:
|
|
contact_text = author_email.text
|
|
if not contact_text:
|
|
author_uri = author.find(namespace + 'uri')
|
|
if author_uri is not None:
|
|
contact_text = author_uri.text
|
|
if not contact_text:
|
|
for uri in item_payload.iter(namespace + 'author'):
|
|
contact_text = uri.text
|
|
if contact_text:
|
|
if contact_text.startswith('xmpp:'):
|
|
contact_type = 'x-scheme-handler/xmpp'
|
|
elif contact_text.startswith('mailto:'):
|
|
contact_type = 'x-scheme-handler/mailto'
|
|
else:
|
|
contact_type = None
|
|
if contact_type:
|
|
ET.SubElement(e_entry, 'link', {'href': contact_text, 'rel': 'contact', 'type': contact_type})
|
|
else:
|
|
ET.SubElement(e_entry, 'link', {'href': contact_text, 'rel': 'contact'})
|
|
break
|
|
categories = item_payload.find(namespace + 'category')
|
|
if isinstance(categories, ET.Element):
|
|
for category in item_payload.findall(namespace + 'category'):
|
|
if 'term' in category.attrib and category.attrib['term']:
|
|
category_term = category.attrib['term']
|
|
ET.SubElement(e_entry, 'category', {'term': category_term})
|
|
|
|
|
|
identifier = item_payload.find(namespace + 'id')
|
|
if identifier and identifier.attrib: print(identifier.attrib)
|
|
|
|
identifier_text = None if identifier == None else identifier.text
|
|
ET.SubElement(e_entry, 'id').text = identifier_text
|
|
# ET.SubElement(e_entry, 'summary', {'type': summary_type_text}).text = summary_text
|
|
return ET.tostring(e_feed, encoding='unicode')
|
|
|
|
def generate_json(iq):
|
|
"""Create a JSON file from node items."""
|
|
json_data = []
|
|
pubsub = iq['from'].bare
|
|
node = iq['pubsub']['items']['node']
|
|
entries = iq['pubsub']['items']
|
|
for entry in entries:
|
|
item_id = entry['id']
|
|
item_payload = entry['payload']
|
|
namespace = '{http://www.w3.org/2005/Atom}'
|
|
title = item_payload.find(namespace + 'title')
|
|
title_text = '*** No Title ***' if title == None else title.text
|
|
# updated = item.find(namespace + 'updated')
|
|
# updated = None if updated == None else updated.text
|
|
# if updated: updated = datetime.datetime.fromisoformat(updated)
|
|
link_href = form_an_item_link(pubsub, node, item_id)
|
|
# link = item.find(namespace + 'link')
|
|
# link_href = '' if link == None else link.attrib['href']
|
|
json_data_entry = {'title' : title_text,
|
|
'link' : link_href}
|
|
json_data.append(json_data_entry)
|
|
#if len(json_data) > 6: break
|
|
directory = 'data/{}/'.format(pubsub)
|
|
if not exists(directory):
|
|
mkdir(directory)
|
|
filename = 'data/{}/{}.json'.format(pubsub, node)
|
|
|
|
with open(filename, 'w', encoding='utf-8') as f:
|
|
json.dump(json_data, f, ensure_ascii=False, indent=4)
|
|
|
|
"""Patch function to append XSLT reference to XML"""
|
|
"""Why is not this a built-in function of ElementTree or LXML"""
|
|
def append_stylesheet(xml_data, type):
|
|
# Register namespace in order to avoide ns0:
|
|
if type == 'atom': ET.register_namespace('', 'http://www.w3.org/2005/Atom')
|
|
# Load XML from string
|
|
tree = ET.fromstring(xml_data)
|
|
# The following direction removes the XML declaration
|
|
xml_data_without_a_declaration = ET.tostring(tree, encoding='unicode')
|
|
# Add XML declaration and stylesheet
|
|
xml_data_declaration = (
|
|
'<?xml version="1.0" encoding="utf-8"?>'
|
|
'<?xml-stylesheet type="text/xsl" href="xsl/{}.xsl"?>'.format(type) +
|
|
xml_data_without_a_declaration)
|
|
return xml_data_declaration
|
|
|
|
def generate_opml(iq):
|
|
pubsub = iq['from'].bare
|
|
items = iq['disco_items']['items']
|
|
opml = ET.Element("opml")
|
|
opml.set("version", "1.0")
|
|
head = ET.SubElement(opml, "head")
|
|
ET.SubElement(head, "title").text = 'An OPML of ' + pubsub
|
|
ET.SubElement(head, "description").text = (
|
|
"PubSub Nodes of {}").format(pubsub)
|
|
ET.SubElement(head, "generator").text = 'Rivista'
|
|
ET.SubElement(head, "urlPublic").text = 'https://git.xmpp-it.net/sch/Rivista'
|
|
time_stamp = datetime.datetime.now(datetime.UTC).isoformat()
|
|
ET.SubElement(head, "dateCreated").text = time_stamp
|
|
ET.SubElement(head, "dateModified").text = time_stamp
|
|
body = ET.SubElement(opml, "body")
|
|
for item in items:
|
|
pubsub, node, title = item
|
|
uri = form_a_node_link(pubsub, node)
|
|
outline = ET.SubElement(body, "outline")
|
|
outline.set("text", title or node)
|
|
outline.set("xmlUrl", uri)
|
|
return ET.tostring(opml, encoding='unicode')
|