#!/usr/bin/python # -*- coding: utf-8 -*- from fastapi import FastAPI, Request, Response from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles import os from rivista.config import Cache, Data, Settings from rivista.gmi.post import GmiPost from rivista.json.index import JsonIndex from rivista.xml.atom import XmlAtom from rivista.xml.opml import XmlOpml from rivista.xml.xhtml import XmlXhtml from rivista.xml.xslt import XmlXslt from rivista.xmpp.instance import XmppInstance from rivista.xmpp.utilities import XmppUtilities from rivista.xmpp.xep_0060 import XmppXep0060 class HttpInstance: def __init__(self): directory_settings = Settings.get_directory() filename_settings = os.path.join(directory_settings, 'settings.toml') credentials = Settings.get_setting(filename_settings, 'account') settings = Settings.get_setting(filename_settings, 'settings') default = Settings.get_setting(filename_settings, 'default') operator = Settings.get_setting(filename_settings, 'settings')['operator'] directory_data = Data.get_directory() directory_data_css = os.path.join(directory_data, 'css') directory_data_graphic = os.path.join(directory_data, 'graphic') directory_data_script = os.path.join(directory_data, 'script') directory_data_xsl = os.path.join(directory_data, 'xsl') filename_favicon = os.path.join(directory_data, 'img', 'favicon.ico') directory_cache = Cache.get_directory() directory_cache_json = os.path.join(directory_cache, 'json') self.app = FastAPI() # Mount static graphic, script and stylesheet directories self.app.mount("/css", StaticFiles(directory=directory_data_css), name="css") self.app.mount("/data", StaticFiles(directory=directory_cache_json), name="data") self.app.mount("/graphic", StaticFiles(directory=directory_data_graphic), name="graphic") self.app.mount("/script", StaticFiles(directory=directory_data_script), name="script") self.app.mount("/xsl", StaticFiles(directory=directory_data_xsl), name="xsl") @self.app.get('/favicon.ico', include_in_schema=False) async def favicon(): return FileResponse(filename_favicon) @self.app.route('/') @self.app.get('/opml') async def view_pubsub_nodes(request: Request): xmpp = XmppInstance(credentials['xmpp'], credentials['pass']) # xmpp.connect() pubsub = request.query_params.get('pubsub', '') result = None if settings['service']: if not settings['include'] or settings['include'] in pubsub: if pubsub: iq = await XmppXep0060.get_nodes(xmpp, pubsub) if iq: link = 'xmpp:{pubsub}'.format(pubsub=pubsub) xml_opml = XmlOpml.generate_opml(iq) result = XmlXslt.append_stylesheet(xml_opml, 'opml') else: text = 'Please ensure that PubSub "{}" (Jabber ID) is valid and accessible.'.format(pubsub) xml_atom = XmlAtom.error_message(text) result = XmlXslt.append_stylesheet(xml_atom, 'atom') else: text = 'The given domain {} is not allowed.'.format(pubsub) xml_atom = XmlAtom.error_message(text) result = XmlXslt.append_stylesheet(xml_atom, 'atom') if not result: if default['pubsub']: if not pubsub: pubsub = default['pubsub'] iq = await XmppXep0060.get_nodes(xmpp, pubsub) link = 'xmpp:{pubsub}'.format(pubsub=pubsub) xml_opml = XmlOpml.generate_opml(iq) result = XmlXslt.append_stylesheet(xml_opml, 'opml') elif not settings['service']: pubsub = default['pubsub'] link = 'xmpp:{pubsub}'.format(pubsub=pubsub) xml_opml = XmlOpml.generate_opml(iq) result = XmlXslt.append_stylesheet(xml_opml, 'opml') else: text = 'Please contact the administrator and ask him to set default PubSub and Node ID.' xml_atom = XmlAtom.error_message(text) result = XmlXslt.append_stylesheet(xml_atom, 'atom') xmpp.disconnect() response = Response(content=result, media_type="application/xml") return response @self.app.get('/atom') async def view_node_items(request: Request): 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', '') result = None if settings['service']: if not settings['include'] or settings['include'] in pubsub: if pubsub and node and item_id: iq = await XmppXep0060.get_node_item(xmpp, pubsub, node, item_id) if iq: link = XmppUtilities.form_an_item_link(pubsub, node, item_id) if 'urn:xmpp:microblog:0:comments/' in node: atom = XmlAtom.extract_atom(iq) xml_atom = XmlAtom.generate_atom_comment(atom, pubsub, node, link) else: atom = XmlAtom.extract_atom(iq) xml_atom = XmlAtom.generate_atom_post(atom, pubsub, node, link) gmi_text = GmiPost.generate_gmi(atom) iq = await XmppXep0060.get_node_items(xmpp, pubsub, node) if not '/' in node: if iq: JsonIndex.generate_json(iq, directory_cache_json) else: 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 = os.path.join(directory_cache_json, 'json', f'{node}.json') 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 = XmlAtom.error_message(text) result = XmlXslt.append_stylesheet(xml_atom, 'atom') # try: # iq = await XmppXep0060.get_node_items(xmpp, pubsub, node) # JsonIndex.generate_json(iq, node) # except: # 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 XmppXep0060.get_node_items(xmpp, pubsub, node) if iq: link = XmppUtilities.form_a_node_link(pubsub, node) if 'urn:xmpp:microblog:0:comments/' in node: atom = XmlAtom.extract_atom(iq) xml_atom = XmlAtom.generate_atom_comment(atom, pubsub, node, link) else: atom = XmlAtom.extract_atom(iq) xml_atom = XmlAtom.generate_atom_post(atom, pubsub, node, link) gmi_text = GmiPost.generate_gmi(atom) else: text = 'Please ensure that PubSub node "{}" is valid and accessible.'.format(node) xml_atom = XmlAtom.error_message(text) result = XmlXslt.append_stylesheet(xml_atom, 'atom') elif pubsub: text = 'Node parameter is missing.' xml_atom = XmlAtom.error_message(text) result = XmlXslt.append_stylesheet(xml_atom, 'atom') elif node: text = 'PubSub parameter is missing.' xml_atom = XmlAtom.error_message(text) result = XmlXslt.append_stylesheet(xml_atom, 'atom') # else: # text = ('Mandatory parameter PubSub and ' # 'optional parameter Node are missing.') # xml_atom = XmlAtom.error_message(text) # result = XmlXslt.append_stylesheet(xml_atom, 'atom') else: text = 'The given domain {} is not allowed.'.format(pubsub) xml_atom = XmlAtom.error_message(text) result = XmlXslt.append_stylesheet(xml_atom, 'atom') if not result: if default['pubsub'] and default['nodeid']: if not pubsub and not node: pubsub = default['pubsub'] node = default['nodeid'] iq = await XmppXep0060.get_node_items(xmpp, pubsub, node) if iq: link = XmppUtilities.form_a_node_link(pubsub, node) atom = XmlAtom.extract_atom(iq) xml_atom = XmlAtom.generate_atom_post(atom, pubsub, node, link) gmi_text = GmiPost.generate_gmi(atom) else: text = 'Please ensure that PubSub node "{}" is valid and accessible.'.format(node) xml_atom = XmlAtom.error_message(text) elif not settings['service']: pubsub = default['pubsub'] node = default['nodeid'] iq = await XmppXep0060.get_node_items(xmpp, pubsub, node) if iq: link = XmppUtilities.form_a_node_link(pubsub, node) atom = XmlAtom.extract_atom(iq) xml_atom = XmlAtom.generate_atom_post(atom, pubsub, node, link) gmi_text = GmiPost.generate_gmi(atom) else: text = 'Please ensure that PubSub node "{}" is valid and accessible.'.format(node) xml_atom = XmlAtom.error_message(text) result = XmlXslt.append_stylesheet(xml_atom, 'atom') else: text = 'Please contact the administrator and ask him to set default PubSub and Node ID.' xml_atom = XmlAtom.error_message(text) result = XmlXslt.append_stylesheet(xml_atom, 'atom') xmpp.disconnect() response = Response(content=result, media_type="application/xml") return response