Refactored apprise_parser.py

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