From 6da1d3b2f61fc8883097ad02fe2f1dc1914fdc58 Mon Sep 17 00:00:00 2001 From: CasVT Date: Mon, 18 Aug 2025 16:27:10 +0200 Subject: [PATCH] Refactored apprise_parser.py --- backend/implementations/apprise_parser.py | 283 ++++++++++++---------- 1 file changed, 161 insertions(+), 122 deletions(-) diff --git a/backend/implementations/apprise_parser.py b/backend/implementations/apprise_parser.py index 2d99322..1a3b4c2 100644 --- a/backend/implementations/apprise_parser.py +++ b/backend/implementations/apprise_parser.py @@ -1,5 +1,10 @@ # -*- coding: utf-8 -*- +""" +Process apprise.Apprise().details() output for URL builder. +""" + +from itertools import chain from re import compile from typing import Any, Dict, List, Tuple, Union @@ -8,10 +13,25 @@ from apprise import Apprise from backend.base.helpers import when_not_none remove_named_groups = compile(r'(?<=\()\?P<\w+>') -IGNORED_ARGS = ('cto', 'format', 'overflow', 'rto', 'verify') +IGNORED_ARGS = {'cto', 'format', 'overflow', 'rto', 'verify'} +CUSTOM_URL_SCHEMA = { + "service_name": "Custom URL", + "setup_url": "https://github.com/caronc/apprise#supported-notifications", + "details": { + "templates": ("{url}",), + "tokens": { + "url": { + "name": "Apprise URL", + "type": "string", + "required": True + } + }, + "args": {} + } +} -def process_regex( +def _process_regex( regex: Union[Tuple[str, str], None] ) -> Union[Tuple[str, str], None]: return when_not_none( @@ -20,6 +40,99 @@ def process_regex( ) +def _process_list( + token_name: str, + token_details: Dict[str, Any], + all_tokens: Dict[str, Dict[str, Any]] +) -> Dict[str, Any]: + list_entry = { + 'name': token_details['name'], + 'map_to': token_name, + 'required': token_details['required'], + 'type': 'list', + 'delim': token_details['delim'][0], + 'content': [] + } + + for content in token_details['group']: + token = all_tokens[content] + list_entry['content'].append({ + 'name': token['name'], + 'required': token['required'], + 'type': token['type'], + 'prefix': token.get('prefix'), + 'regex': _process_regex(token.get('regex')) + }) + + return list_entry + + +def _process_normal_token( + token_name: str, + token_details: Dict[str, Any] +) -> Dict[str, Any]: + normal_entry = { + 'name': token_details['name'], + 'map_to': token_name, + 'required': token_details['required'], + 'type': token_details['type'].split(':')[0] + } + + if token_details['type'].startswith('choice'): + normal_entry.update({ + 'options': token_details.get('values'), + 'default': token_details.get('default') + }) + + else: + normal_entry.update({ + 'prefix': token_details.get('prefix'), + 'min': token_details.get('min'), + 'max': token_details.get('max'), + 'regex': _process_regex(token_details.get('regex')) + }) + + return normal_entry + + +def _process_arg( + arg_name: str, + arg_details: Dict[str, Any] +) -> Dict[str, Any]: + args_entry = { + 'name': arg_details.get('name', arg_name), + 'map_to': arg_name, + 'required': arg_details.get('required', False), + 'type': arg_details['type'].split(':')[0], + } + + if arg_details['type'].startswith('list'): + args_entry.update({ + 'delim': arg_details['delim'][0], + 'content': [] + }) + + elif arg_details['type'].startswith('choice'): + args_entry.update({ + 'options': arg_details['values'], + 'default': arg_details.get('default') + }) + + elif arg_details['type'] == 'bool': + args_entry.update({ + 'default': arg_details['default'] + }) + + else: + args_entry.update({ + 'min': arg_details.get('min'), + 'max': arg_details.get('max'), + 'regex': _process_regex(arg_details.get('regex')) + }) + + return args_entry + + def _sort_tokens(t: Dict[str, Any]) -> List[int]: result = [ int(not t['required']) @@ -41,138 +154,64 @@ def _sort_tokens(t: Dict[str, Any]) -> List[int]: def get_apprise_services() -> List[Dict[str, Any]]: - apprise_services = [] + """Get a list of all Apprise services, their URL schemas, tokens and + arguments. - raw = Apprise().details()['schemas'] - for entry in raw: - result = { - 'name': str(entry['service_name']), - 'doc_url': entry['setup_url'], + Returns: + List[Dict[str, Any]]: The list. + """ + result: List[Dict[str, Any]] = [] + + schemas = Apprise().details()['schemas'] + for schema in chain((CUSTOM_URL_SCHEMA,), schemas): + entry = { + 'name': str(schema['service_name']), + 'doc_url': schema['setup_url'], 'details': { - 'templates': entry['details']['templates'], + 'templates': schema['details']['templates'], 'tokens': [], 'args': [] } } + # Process lists and tokens they contain first handled_tokens = set() - for k, v in entry['details']['tokens'].items(): - if not v['type'].startswith('list:'): + for token_name, token_details in schema['details']['tokens'].items(): + if not token_details['type'].startswith('list:'): continue - list_entry = { - 'name': v['name'], - 'map_to': k, - 'required': v['required'], - 'type': 'list', - 'delim': v['delim'][0], - 'content': [] - } + list_entry = _process_list( + token_name, token_details, schema['details']['tokens'] + ) + entry['details']['tokens'].append(list_entry) + handled_tokens.add(token_name) + handled_tokens.update(token_details['group']) - for content in v['group']: - token = entry['details']['tokens'][content] - list_entry['content'].append({ - 'name': token['name'], - 'required': token['required'], - 'type': token['type'], - 'prefix': token.get('prefix'), - 'regex': process_regex(token.get('regex')) - }) - handled_tokens.add(content) + # Process all other tokens + entry['details']['tokens'] += [ + _process_normal_token(token_name, token_details) + for token_name, token_details in schema['details']['tokens'].items() + if token_name not in handled_tokens + ] - result['details']['tokens'].append(list_entry) - handled_tokens.add(k) + # Process args + entry['details']['args'] += [ + _process_arg(arg_name, arg_details) + for arg_name, arg_details in schema['details']['args'].items() + if not ( + arg_details.get('alias_of') is not None + or arg_name in IGNORED_ARGS + ) + ] - for k, v in entry['details']['tokens'].items(): - if k in handled_tokens: - continue + # Sort tokens and args + entry['details']['tokens'].sort(key=_sort_tokens) + entry['details']['args'].sort(key=_sort_tokens) + result.append(entry) - normal_entry = { - 'name': v['name'], - 'map_to': k, - 'required': v['required'], - 'type': v['type'].split(':')[0] - } + result.sort(key=lambda s: ( + int(s["name"] != "Custom URL"), + s["name"].lower() + )) - if v['type'].startswith('choice'): - normal_entry.update({ - 'options': v.get('values'), - 'default': v.get('default') - }) - - else: - normal_entry.update({ - 'prefix': v.get('prefix'), - 'min': v.get('min'), - 'max': v.get('max'), - 'regex': process_regex(v.get('regex')) - }) - - result['details']['tokens'].append(normal_entry) - - for k, v in entry['details']['args'].items(): - if ( - v.get('alias_of') is not None - or k in IGNORED_ARGS - ): - continue - - args_entry = { - 'name': v.get('name', k), - 'map_to': k, - 'required': v.get('required', False), - 'type': v['type'].split(':')[0], - } - - if v['type'].startswith('list'): - args_entry.update({ - 'delim': v['delim'][0], - 'content': [] - }) - - elif v['type'].startswith('choice'): - args_entry.update({ - 'options': v['values'], - 'default': v.get('default') - }) - - elif v['type'] == 'bool': - args_entry.update({ - 'default': v['default'] - }) - - else: - args_entry.update({ - 'min': v.get('min'), - 'max': v.get('max'), - 'regex': process_regex(v.get('regex')) - }) - - result['details']['args'].append(args_entry) - - result['details']['tokens'].sort(key=_sort_tokens) - result['details']['args'].sort(key=_sort_tokens) - apprise_services.append(result) - - apprise_services.sort(key=lambda s: s['name'].lower()) - - apprise_services.insert(0, { - 'name': 'Custom URL', - 'doc_url': 'https://github.com/caronc/apprise#supported-notifications', - 'details': { - 'templates': ['{url}'], - 'tokens': [{ - 'name': 'Apprise URL', - 'map_to': 'url', - 'required': True, - 'type': 'string', - 'prefix': None, - 'min': None, - 'max': None, - 'regex': None - }], - 'args': [] - } - }) - - return apprise_services + return result