#!/usr/bin/python # -*- coding: utf-8 -*- from email.utils import parseaddr from fastapi import FastAPI, Form, HTTPException, Request, Response from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, HTMLResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from jabbercard.config import Cache, Settings, Data from jabbercard.utilities.graphics import Graphics from jabbercard.utilities.toml import Toml from jabbercard.utilities.xml import Syndication from jabbercard.xmpp.utilities import XmppUtilities import os from starlette.responses import RedirectResponse from urllib.parse import urlsplit class HttpInstance: def __init__(self): directory_settings = Settings.get_directory() filename_settings = os.path.join(directory_settings, 'settings.toml') settings = Toml.open_file_toml(filename_settings) account = settings['account'] jabber_id = account['xmpp'] password = account['pass'] alias = account['alias'] brand = settings['brand'] brand_name = brand['name'] brand_site = brand['site'] chat_client = brand['chat'] news_client = brand['news'] 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_img = os.path.join(directory_data, 'img') filename_favicon = os.path.join(directory_data, 'img', 'favicon.ico') directory_data_template = os.path.join(directory_data, 'template') directory_cache = Cache.get_directory() directory_cache_qr = os.path.join(directory_cache, 'qr') directory_cache_photo = os.path.join(directory_cache, 'photo') self.app = FastAPI() templates = Jinja2Templates(directory=directory_data_template) # TODO # 1) Mount at the same mountpoint /img. # 2) Image filename to be constant, i.e. /img/photo.png and /img/qr.png. self.app.mount('/photo', StaticFiles(directory=directory_cache_photo), name='photo') self.app.mount('/qr', StaticFiles(directory=directory_cache_qr), name='qr') self.app.mount('/css', StaticFiles(directory=directory_data_css), name='css') self.app.mount('/img', StaticFiles(directory=directory_data_img), name='img') # @self.app.get(filename_favicon, include_in_schema=False) # def favicon_get(): # return FileResponse('graphic/hermes.ico') # @self.app.get('/hermes.svg') # def logo_get(): # return FileResponse('graphic/hermes.svg') @self.app.get('/v/{jid}') async def view_jid(request: Request, jid): """View recent messages of a conference""" jid_path = urlsplit(jid).path if parseaddr(jid_path)[1] == jid_path: jid_bare = jid_path.lower() else: jid_bare = jid note = 'Jabber ID appears to be malformed' if jid_bare == jabber_id: raise HTTPException(status_code=403, detail='access-denied') #try: if True: exception = jid_vcard = messages_10 = note = node_title = \ node_note = number_of_pages = page_number = previous = \ selection = services_sorted = subject = None link_href = 'xmpp:{}?join'.format(jid_bare) link_text = 'Join' xmpp_uri = '{}'.format(jid_bare) filename = os.path.join(directory_cache, 'details', jid_bare + '.toml') if os.path.exists(filename) and os.path.getsize(filename) > 0: jid_details = Toml.open_file_toml(filename) else: jid_details = await XmppUtilities.cache_jid_data( jabber_id, password, jid_bare, alias=alias) count = jid_details['count'] items = jid_details['items'] jid_info = { 'error' : jid_details['error'], 'text' : jid_details['error_text'], 'condition' : jid_details['error_condition']} jid_kind = jid_details['kind'] jid_vcard = { 'name' : jid_details['name'], 'note' : jid_details['note'], 'type' : jid_details['image_type']} messages = jid_details['messages'] nodes = jid_details['nodes'] note = jid_details['note'] subject = jid_details['subject'] title = jid_details['name'] # Group chat messages # NOTE TODO page_number = request.query_params.get('page', '') if page_number: try: page_number = int(page_number) ix = (page_number -1) * 10 except: ix = 0 page_number = 1 else: ix = 0 page_number = 1 messages_10 = messages[ix:][:10] number_of_pages = int(len(messages) / 10) if number_of_pages < len(messages) / 10: number_of_pages += 1 if jid_kind: # Action and instance type action, instance = XmppUtilities.set_action_instance_type(jid_kind) else: # jid_info['error'] action = 'Contact' instance = view_href = '' message = '{}: {} (XEP-0030)'.format(jid_info['text'], jid_info['condition']) xmpp_uri = jid_bare # Query URI links print('Query URI links') links = XmppUtilities.get_query_uri_links(jid_bare, jid_kind) link_href = XmppUtilities.get_link_href(jid_bare, jid_kind) view_href = XmppUtilities.get_view_href(jid_bare, jid_kind) xmpp_uri = XmppUtilities.get_xmpp_uri(jid_bare, jid_kind) # Graphic files filename, filepath, filetype, selection = Graphics.handle_photo( jid_bare, jid_vcard, link_href) #except Exception as e: else: exception = str(e) action = 'Error' title = 'Slixmpp error' xmpp_uri = note = jid filename = jid_bare = link_href = link_tex = node_note = \ node_title = number_of_pages = page_number = previous = \ selection = services = services_sorted = url = None #if title == 'remote-server-timeout': # raise HTTPException(status_code=408, detail='remote-server-timeout') #else: template_file = 'conference.xhtml' template_dict = { 'action' : action, 'brand_name' : brand_name, 'brand_site' : brand_site, 'chat_client' : chat_client, 'exception' : exception, 'filename' : filename, 'jid_bare' : jid, 'jid_note' : note, 'jid_title' : title, 'links' : links, 'messages' : messages_10, 'node_title' : node_title, 'node_note' : node_note, 'number_of_pages' : number_of_pages, 'page_number' : page_number, 'previous' : previous, 'request' : request, 'selection' : selection, 'subject' : subject, 'title' : title, 'url' : request.url._url, 'xmpp_uri' : xmpp_uri} response = templates.TemplateResponse(template_file, template_dict) response.headers['Content-Type'] = 'application/xhtml+xml' return response @self.app.get('/c/{jid}') async def c_jid_get(request: Request, jid): """Display entries of a vCard4""" jid_path = urlsplit(jid).path if parseaddr(jid_path)[1] == jid_path: jid_bare = jid_path.lower() else: jid_bare = jid note = 'Jabber ID appears to be malformed' if jid_bare == jabber_id: raise HTTPException(status_code=403, detail='access-denied') node_name_vcard4 = 'urn:xmpp:vcard4' item_id_vcard4 = 'current' #try: if True: entries = [] exception = jid_vcard = note = node_items = node_note = \ number_of_pages = page_number = previous = selection = \ title = None filename = os.path.join(directory_cache, 'xep_0060', jid_bare, node_name_vcard4, item_id_vcard4 + '.xml') if os.path.exists(filename) and os.path.getsize(filename) > 0: xml_data = Toml.open_file_xml(filename) else: await XmppUtilities.cache_vcard_data( jabber_id, password, jid_bare, node_name_vcard4, item_id_vcard4) xml_data = Toml.open_file_xml(filename) root_element = xml_data.getroot() child_element = root_element[0] #vcard_info = Syndication.extract_vcard_items(child_element) vcard_info = Syndication.extract_vcard4_items(child_element) # Action and instance type action = 'Profile' filename = os.path.join(directory_cache, 'details', jid_bare + '.toml') if os.path.exists(filename) and os.path.getsize(filename) > 0: jid_details = Toml.open_file_toml(filename) else: jid_details = await XmppUtilities.cache_jid_data( jabber_id, password, jid_bare, alias=alias) # Set node name to 'urn:xmpp:microblog:0' jid_kind = jid_details['kind'] nodes = jid_details['nodes'] if (jid_kind not in ('conference', 'mix', 'muc') and '@' in jid_bare and 'urn:xmpp:microblog:0' in nodes): node_name = 'urn:xmpp:microblog:0' # Query URI links print('Query URI links') jid_kind = 'account' link_href = XmppUtilities.get_link_href(jid_bare, jid_kind) xmpp_uri = XmppUtilities.get_xmpp_uri(jid_bare, jid_kind, node_name_vcard4) links = XmppUtilities.get_query_uri_links(jid_bare, jid_kind, node_name) # Graphic files filename, filepath, filetype, selection = Graphics.handle_photo( jid_bare, jid_vcard, link_href) #except Exception as e: else: exception = str(e) action = 'Error' title = 'Slixmpp error' xmpp_uri = note = jid filename = jid_bare = link_href = link_tex = node_note = \ node_title = number_of_pages = page_number = previous = \ selection = url = None if 'fn' in vcard_info and vcard_info['fn']: title = vcard_info['fn'] elif 'alias' in vcard_info and vcard_info['alias']: title = vcard_info['alias'] else: title = jid_bare.split('@')[0] if 'alias' in vcard_info and vcard_info['alias']: alias = vcard_info['alias'] else: alias = jid_bare.split('@')[0] #if title == 'remote-server-timeout': # raise HTTPException(status_code=408, detail='remote-server-timeout') #else: template_file = 'vcard.xhtml' template_dict = { 'action' : action, 'alias' : alias, 'brand_name' : brand_name, 'brand_site' : brand_site, 'chat_client' : chat_client, 'entries' : entries, 'exception' : exception, 'filename' : filename, 'jid_bare' : jid, 'jid_note' : note, #'jid_title' : title, #'node_title' : node_title, 'links' : links, 'node_name' : node_name_vcard4, 'number_of_pages' : number_of_pages, 'page_number' : page_number, 'previous' : previous, 'request' : request, 'selection' : selection, 'title' : title, 'url' : request.url._url, 'vcard_info' : vcard_info, 'xmpp_uri' : xmpp_uri} response = templates.TemplateResponse(template_file, template_dict) response.headers['Content-Type'] = 'application/xhtml+xml' return response @self.app.get('/b/{jid}') async def b_jid_get(request: Request, jid): response = await browse_jid_node_get(request, jid, 'urn:xmpp:microblog:0') return response # TODO Change to /p/ for pubsub @self.app.get('/d/{jid}/{node_name}') @self.app.get('/d/{jid}/{node_name}/{item_id}') async def d_jid_node_get(request: Request, jid, node_name, item_id=None): response = await browse_jid_node_get(request, jid, node_name, item_id) return response async def browse_jid_node_get(request: Request, jid, node_name, item_id=None): """Browse items of a pubsub node""" jid_path = urlsplit(jid).path if parseaddr(jid_path)[1] == jid_path: jid_bare = jid_path.lower() else: jid_bare = jid note = 'Jabber ID appears to be malformed' if jid_bare == jabber_id: raise HTTPException(status_code=403, detail='access-denied') #try: if True: exception = jid_vcard = note = node_items = node_note = \ number_of_pages = page_number = previous = selection = None filename = os.path.join(directory_cache, 'details', jid_bare + '.toml') if os.path.exists(filename) and os.path.getsize(filename) > 0: jid_details = Toml.open_file_toml(filename) else: jid_details = await XmppUtilities.cache_jid_data( jabber_id, password, jid_bare, node_name, item_id) # Node item IDs nodes = jid_details['nodes'] #items = jid_details['items'] # for item in items: # if item[1] == node_name: # nodes[node_name]['title'] = item[2] # break supdirectory = os.path.join(directory_cache, 'xep_0060', jid_bare) if not os.path.exists(supdirectory): os.mkdir(supdirectory) directory = os.path.join(directory_cache, 'xep_0060', jid_bare, node_name) if not os.path.exists(directory): os.mkdir(directory) await XmppUtilities.cache_node_data( jabber_id, password, jid_bare, node_name) count = jid_details['count'] jid_info = { 'error' : jid_details['error'], 'text' : jid_details['error_text'], 'condition' : jid_details['error_condition']} jid_kind = jid_details['kind'] jid_vcard = { 'name' : jid_details['name'], 'note' : jid_details['note'], 'type' : jid_details['image_type']} messages = jid_details['messages'] #node_title = nodes[node_name]['title'] if 'title' in nodes[node_name] else jid_details['name'] node_title = node_name note = jid_details['note'] #title = nodes[node_name]['title'] if node_name else jid_details['name'] title = jid_details['name'] #link_href = 'xmpp:{}?pubsub;node={};action=subscribe'.format( # jid_bare, node_name) #link_text = 'Subscribe' #xmpp_uri = '{}?;node={}'.format(jid_bare, node_name) # TODO Support viewing of a single item # Node items entries = [] if item_id: node_items = item_ids_10 = [item_id + '.xml'] else: node_items = os.listdir(directory) if 'urn:xmpp:avatar:metadata.xml' in node_items: node_items.remove('urn:xmpp:avatar:metadata.xml') page_number = request.query_params.get('page', '') if page_number: try: page_number = int(page_number) ix = (page_number -1) * 10 except: ix = 0 page_number = 1 else: ix = 0 page_number = 1 item_ids_10 = node_items[ix:][:10] number_of_pages = int(len(node_items) / 10) if number_of_pages < len(node_items) / 10: number_of_pages += 1 if node_items: for item in item_ids_10: filename = os.path.join(directory, item) xml_data = Toml.open_file_xml(filename) root_element = xml_data.getroot() child_element = root_element[0] entry = Syndication.extract_atom_items(child_element) if entry: filename_without_file_extension = item[:len(item)-4] entry['id'] = filename_without_file_extension entries.append(entry) #if len(entries) > 10: break if jid_kind: # Action and instance type action, instance = XmppUtilities.set_action_instance_type(jid_kind, node_name) else: # jid_info['error'] action = 'Contact' instance = view_href = '' message = '{}: {} (XEP-0030)'.format(jid_info['text'], jid_info['condition']) xmpp_uri = jid_bare # Query URI links print('Query URI links') links = XmppUtilities.get_query_uri_links(jid_bare, jid_kind, node_name, item_id) link_href = XmppUtilities.get_link_href(jid_bare, jid_kind, node_name) view_href = XmppUtilities.get_view_href(jid_bare, jid_kind, node_name) xmpp_uri = XmppUtilities.get_xmpp_uri(jid_bare, jid_kind, node_name) node_note = xmpp_uri # Graphic files filename, filepath, filetype, selection = Graphics.handle_photo( jid_bare, jid_vcard, link_href) #except Exception as e: else: exception = str(e) action = 'Error' title = 'Slixmpp error' xmpp_uri = note = jid filename = jid_bare = link_href = link_tex = node_note = \ node_title = number_of_pages = page_number = previous = \ selection = url = None #if title == 'remote-server-timeout': # raise HTTPException(status_code=408, detail='remote-server-timeout') #else: template_file = 'node.xhtml' template_dict = { 'action' : action, 'brand_name' : brand_name, 'brand_site' : brand_site, 'chat_client' : chat_client, 'entries' : entries, 'exception' : exception, 'filename' : filename, 'jid_bare' : jid, 'jid_note' : note, 'jid_title' : title, 'links' : links, 'node_title' : node_title, 'node_note' : node_note, 'node_name' : node_name, 'number_of_pages' : number_of_pages, 'page_number' : page_number, 'previous' : previous, 'request' : request, 'selection' : selection, 'title' : node_title, 'url' : request.url._url, 'xmpp_uri' : xmpp_uri} response = templates.TemplateResponse(template_file, template_dict) response.headers['Content-Type'] = 'application/xhtml+xml' return response @self.app.get('/d/{jid}') async def discover_jid_get(request: Request, jid): """View items of a selected service""" jid_path = urlsplit(jid).path if parseaddr(jid_path)[1] == jid_path: jid_bare = jid_path.lower() else: jid_bare = jid note = 'Jabber ID appears to be malformed' if jid_bare == jabber_id: raise HTTPException(status_code=403, detail='access-denied') #try: if True: exception = note = selection = services_sorted = None title = 'Services' link_href = xmpp_uri = jid_bare link_text = 'Reload' # Start an XMPP instance and retrieve information xmpp_instance = XmppInstance(jabber_id, password, jid_bare) xmpp_instance.connect() # JID services action = 'Discover' jid_info = await XmppXep0030.get_jid_info(xmpp_instance, jid_bare) iq = jid_info['iq'] if iq: jid_kind = jid_info['kind'] iq_disco_info = iq['disco_info'] for identity in iq_disco_info['identities']: if jid_kind == identity[0] and identity[3]: note = identity[3] if not note: note = jid_bare jid_items = await XmppXep0030.get_jid_items(xmpp_instance, jid_bare) iq = jid_items['iq'] iq_disco_items = iq['disco_items'] iq_disco_items_items = iq_disco_items['items'] services = {} #services_sorted = {} category = 'unsorted' for item in iq_disco_items_items: jid_bare = item[0] if len(iq_disco_items_items) > 20 or jid_kind and jid_kind in ('pubsub'): identity = sub_jid_info = sub_jid_info_iq = '' if jid_kind and jid_kind in ('conference', 'mix', 'muc'): category = 'conference' if jid_kind and jid_kind in ('pubsub'): category = 'pubsub' else: sub_jid_info = await XmppXep0030.get_jid_info(xmpp_instance, jid_bare) sub_jid_info_iq = sub_jid_info['iq'] try: for identity_item in sub_jid_info_iq['disco_info']['identities']: identity = identity_item break if sub_jid_info_iq: category = identity[0] if (identity, list) and identity[0] else 'other' except: identity = None category = 'unavailable' sub_jid_kind = sub_jid_info['kind'] if 'kind' in sub_jid_info else None if category not in services: services[category] = [] services[category].append( {'identity' : identity, 'info' : sub_jid_info, 'jid' : jid_bare, 'kind' : sub_jid_kind, 'name' : item[2] or item[1] or item[0], 'node' : item[1]}) services_sorted = {k: v for k, v in services.items() if k != 'unavailable'} if 'unavailable' in services: services_sorted['unavailable'] = services['unavailable'] else: message = '{}: {} (XEP-0030)'.format(jid_info['condition'], jid_info['text']) services = services_sorted = None xmpp_instance.disconnect() #except Exception as e: else: exception = str(e) action = 'Error' title = 'Slixmpp error' xmpp_uri = note = jid filename = jid_bare = link_href = link_text = selection = services = services_sorted = url = None #if title == 'remote-server-timeout': # raise HTTPException(status_code=408, detail='remote-server-timeout') #else: template_file = 'disco.xhtml' template_dict = { 'action' : action, 'filename' : 'default.svg', 'brand_name' : brand_name, 'brand_site' : brand_site, 'chat_client' : chat_client, 'exception' : exception, 'jid_bare' : jid, 'note' : note, 'request' : request, 'services' : services_sorted, 'title' : title, 'url' : request.url._url, 'link_href' : link_href, 'link_text' : link_text, 'xmpp_uri' : xmpp_uri} response = templates.TemplateResponse(template_file, template_dict) response.headers['Content-Type'] = 'application/xhtml+xml' return response @self.app.get('/j/{jid}/{node_name}') async def jid_node_get(request: Request, jid, node_name): response = await main_jid_node_get(request, jid, node_name) return response @self.app.get('/j/{jid}') async def jid_get(request: Request, jid): node_name = request.query_params.get('node', '') if node_name: response = RedirectResponse(url='/j/{}/{}'.format(jid, node_name)) else: response = await main_jid_node_get(request, jid) return response async def main_jid_node_get(request: Request, jid, node_name=None): jid_bare = jid jid_path = urlsplit(jid).path if parseaddr(jid_path)[1] == jid_path: jid_bare = jid_path.lower() else: jid_bare = jid note = 'Jabber ID appears to be malformed' if jid_bare == jabber_id: raise HTTPException(status_code=403, detail='access-denied') #try: if True: action = alias = count_item = count_message = exception = \ instance = jid_vcard = jid_info = link_href = message = note = \ selection = title = vcard4 = view_href = xmpp_uri = None #node_name = 'urn:xmpp:microblog:0' filename = os.path.join(directory_cache, 'details', jid_bare + '.toml') if os.path.exists(filename) and os.path.getsize(filename) > 0: jid_details = Toml.open_file_toml(filename) else: jid_details = await XmppUtilities.cache_jid_data( jabber_id, password, jid_bare, node_name, alias=alias) # Set node name to 'urn:xmpp:microblog:0' jid_kind = jid_details['kind'] nodes = jid_details['nodes'] count_message = jid_details['messages'] if (jid_kind not in ('conference', 'mix', 'muc') and '@' in jid_bare and not node_name and 'urn:xmpp:microblog:0' in nodes): node_name = 'urn:xmpp:microblog:0' items = jid_details['items'] jid_info = { 'error' : jid_details['error'], 'text' : jid_details['error_text'], 'condition' : jid_details['error_condition']} jid_vcard = { 'name' : jid_details['name'], 'note' : jid_details['note'], 'type' : jid_details['image_type']} messages = jid_details['messages'] #note = nodes[node_name]['title'] if node_name in nodes else jid_details['note'] #note = jid_details['note'] # vCard4 node_name_vcard4 = 'urn:xmpp:vcard4' item_id_vcard4 = 'current' directory = os.path.join(directory_cache, 'xep_0060', jid_bare, node_name_vcard4) filename = os.path.join(directory, item_id_vcard4 + '.xml') if os.path.exists(filename) and os.path.getsize(filename) > 0: xml_data = Toml.open_file_xml(filename) root_element = xml_data.getroot() child_element = root_element[0] #vcard_info = Syndication.extract_vcard_items(child_element) vcard_info = Syndication.extract_vcard4_items(child_element) title = vcard_info['fn'] alias = vcard_info['alias'] #note = vcard_info['note'] else: await XmppUtilities.cache_vcard_data( jabber_id, password, jid_bare, node_name_vcard4, item_id_vcard4) if os.path.exists(filename) and os.path.getsize(filename) > 0: vcard4 = True # Node item IDs supdirectory = os.path.join(directory_cache, 'xep_0060', jid_bare) if not os.path.exists(supdirectory): os.mkdir(supdirectory) if node_name: directory = os.path.join(directory_cache, 'xep_0060', jid_bare, node_name) if not os.path.exists(directory): os.mkdir(directory) await XmppUtilities.cache_node_data( jabber_id, password, jid_bare, node_name) # JID or node items if jid_kind in ('mix', 'muc', 'conference', 'server'): count_item = jid_details['count'] elif jid_kind in ('account', 'pubsub'): node_items = os.listdir(directory) if 'urn:xmpp:avatar:metadata.xml' in node_items: node_items.remove('urn:xmpp:avatar:metadata.xml') count_item = len(node_items) # if ('@' in jid_bare and # 'urn:xmpp:microblog:0' not in nodes and # jid_kind not in ('conference', 'mix', 'muc')): # count_item = 0 # else: # count_item = len(node_items) if jid_kind == 'pubsub' and node_name: items = jid_details['items'] for item in items: if item[1] == node_name: #nodes[node_name]['title'] = item[2] title = item[2] break if not title: title = node_name else: title = jid_details['name'] # TODO Consider also the existence of a node /j/pubsub.movim.eu/i2p if jid_kind: # Action and instance type action, instance = XmppUtilities.set_action_instance_type(jid_kind, node_name) view_href = XmppUtilities.get_view_href(jid_bare, jid_kind, node_name) xmpp_uri = XmppUtilities.get_xmpp_uri(jid_bare, jid_kind, node_name) else: # jid_info['error'] action = 'Contact' instance = view_href = '' if jid_info['condition']: message = '{}: {} (XEP-0030)'.format(jid_info['text'], jid_info['condition']) xmpp_uri = jid_bare link_href = XmppUtilities.get_link_href(jid_bare, jid_kind, node_name) # Query URI links print('Query URI links') links = XmppUtilities.get_query_uri_links(jid_bare, jid_kind, node_name) # Graphic files filename, filepath, filetype, selection = Graphics.handle_photo( jid_bare, jid_vcard, link_href) #except Exception as e: else: exception = str(e) print(exception) action = 'Error' title = 'Slixmpp error' xmpp_uri = jid alias = count_item = count_message = filename = jid_bare = \ jid_vcard = jid_kind = links = message = selection = url = \ vcard4 = None #note_500 = note[:500] #note = note_500 + ' …' if note_500 < note else note_500 # NOTE Handling of variables "title" and "note" in case of '/j/{jid}/{node_name}' is confusing. # TODO Add new keys that are of 'node' and be utilized for nodes, instead of reusing a variable for several roles. # FIXME If no title be provided to 'node name', use 'node name' itself as title (to be done at XmppUtilities.cache_jid_data). template_file = 'jid.xhtml' template_dict = { 'action' : action, 'alias' : alias, 'brand_name' : brand_name, 'brand_site' : brand_site, 'chat_client' : chat_client, 'count_item' : count_item, 'count_message' : count_message, 'instance' : instance, 'exception' : exception, 'filename' : filename, 'jid_bare' : jid_bare, 'jid_kind' : jid_kind, 'links' : links, 'message' : message, 'news_client' : news_client, 'note' : note, # TODO node_note or title of PubSub JID 'request' : request, 'selection' : selection, 'title' : title, # TODO node_title 'url' : request.url._url, 'vcard4' : vcard4, 'view_href' : view_href, 'xmpp_uri' : xmpp_uri} response = templates.TemplateResponse(template_file, template_dict) response.headers['Content-Type'] = 'application/xhtml+xml' return response @self.app.get('/selection') async def selection_get(request: Request): filename = os.path.join(directory_data, 'systems.toml') software = Toml.open_file_toml(filename)['systems'] template_file = 'software.xhtml' template_dict = { 'brand_name' : brand_name, 'brand_site' : brand_site, 'chat_client' : chat_client, 'request' : request, 'software' : software, 'url' : request.url._url} response = templates.TemplateResponse(template_file, template_dict) response.headers['Content-Type'] = 'application/xhtml+xml' return response #@self.app.get('/download/select') #async def download_select_get(request, software=None): @self.app.get('/download/{software}') async def download_software_get(request: Request, software): response = await download_get(request, featured=True, software=software) return response @self.app.get('/download/{software}/all') async def download_software_all_get(request: Request, software): response = await download_get(request, featured=False, software=software) return response @self.app.get('/download') async def download_get(request: Request, featured=True, software=None): # TODO # Fearured clients '/download/{software}' # All clients '/download/{software}/all' # Select software '/download/select' skipped = False if not software: user_agent = request.headers.get("user-agent") user_agent_lower = user_agent.lower() match user_agent_lower: case _ if 'bsd' in user_agent_lower: software = 'bsd' case _ if 'linux' in user_agent_lower: software = 'linux' case _ if 'haiku' in user_agent_lower: software = 'haiku' case _ if 'android' in user_agent_lower: software = 'android' case _ if 'reactos' in user_agent_lower or 'windows' in user_agent_lower: software = 'windows' case _ if 'ios' in user_agent_lower or 'macos' in user_agent_lower: software = 'apple' name = software.title() if software == 'bsd': name = 'BSD' if software == 'posix': name = 'POSIX' if software == 'ubports': name = 'UBports' if name.endswith('os'): name = name.replace('os', 'OS') filename_clients = os.path.join(directory_data, 'clients.toml') clients = Toml.open_file_toml(filename_clients) client_selection = [] clients_software = 0 for client in clients: if software in clients[client]: clients_software += 1 if featured and 'featured' not in clients[client]['properties']: skipped = True continue client_selected = { 'name' : clients[client]['title'], 'about' : clients[client]['about'], 'href' : clients[client][software], 'iden' : client, 'properties' : clients[client]['properties'], 'resources' : clients[client]['resources'] if 'resources' in clients[client] else ''} client_selection.append(client_selected) skipped = False if len(client_selection) == clients_software else True template_file = 'download.xhtml' template_dict = { 'brand_name' : brand_name, 'brand_site' : brand_site, 'chat_client' : chat_client, 'client_selection' : client_selection, 'featured' : featured, 'skipped' : skipped, 'request' : request, 'software' : software, 'title' : name, 'url' : request.url._url} response = templates.TemplateResponse(template_file, template_dict) response.headers['Content-Type'] = 'application/xhtml+xml' return response @self.app.exception_handler(403) def access_denied_exception_handler(request: Request, exc: HTTPException): action = 'Warning' title = 'Access Denied' return result_get(request, action, title) @self.app.exception_handler(404) def not_found_exception_handler(request: Request, exc: HTTPException): action = 'Warning' title = 'Not Found' return result_get(request, action, title) @self.app.exception_handler(500) def internal_error_exception_handler(request: Request, exc: HTTPException): action = 'Error' title = 'Internal Server Error' return result_get(request, action, title) @self.app.exception_handler(504) def time_out_exception_handler(request: Request, exc: HTTPException): action = 'Warning' title = 'Time Out' return result_get(request, action, title) def result_get(request: Request, action: str, title: str): template_file = 'result.xhtml' template_dict = { 'action' : action, 'brand_name' : brand_name, 'brand_site' : brand_site, 'chat_client' : chat_client, 'request' : request, 'title' : title, 'url' : request.url._url} response = templates.TemplateResponse(template_file, template_dict) response.headers['Content-Type'] = 'application/xhtml+xml' return response @self.app.get('/') async def main_get(request: Request): jabber_id = request.query_params.get('jid', '') if jabber_id: response = RedirectResponse(url='/j/' + jabber_id) else: template_file = 'main.xhtml' template_dict = { 'brand_name' : brand_name, 'brand_site' : brand_site, 'chat_client' : chat_client, 'request' : request, 'url' : request.url._url} response = templates.TemplateResponse(template_file, template_dict) response.headers['Content-Type'] = 'application/xhtml+xml' return response