#!/usr/bin/python3
# SPDX-License-Identifier: LicenseRef-Proprietary
"""Version-compatible notification controls for the Gravitas panel."""

import argparse
import html
import json
import os
from pathlib import Path
import re
import stat
import subprocess
import sys
import tempfile
import textwrap


MAKO_CONFIG = Path('/usr/share/gravitas/mako.conf')
SERVICE = 'org.freedesktop.Notifications'
MODE = 'do-not-disturb'
TIMEOUT = 2
MAX_REPLY = 1024 * 1024


class PanelError(Exception):
    pass


def run(command, **kwargs):
    try:
        result = subprocess.run(command, capture_output=True, text=True, timeout=TIMEOUT, **kwargs)
    except (OSError, subprocess.TimeoutExpired) as error:
        raise PanelError(f'{command[0]} is unavailable: {error}') from error
    if result.returncode:
        detail = clean(result.stderr, 250) or f'exit status {result.returncode}'
        raise PanelError(f'{command[0]} failed: {detail}')
    if len(result.stdout) > MAX_REPLY:
        raise PanelError('The notification service returned too much data.')
    return result.stdout


def bus_call(method, signature=None, *arguments, service=SERVICE,
             path='/fr/emersion/Mako', interface='fr.emersion.Mako'):
    command = ['busctl', '--user', '--auto-start=no', '--timeout=2', '--json=short',
               'call', service, path, interface, method]
    if signature:
        command.extend((signature, *arguments))
    output = run(command)
    if not output.strip():
        return None
    try:
        reply = json.loads(output)
    except json.JSONDecodeError as error:
        raise PanelError('The notification service returned invalid JSON.') from error
    if not isinstance(reply, dict) or not isinstance(reply.get('data'), list):
        raise PanelError('The notification service returned an invalid typed reply.')
    return reply


def typed(method, signature, **kwargs):
    reply = bus_call(method, **kwargs)
    if reply is None or reply.get('type') != signature or len(reply['data']) != 1:
        raise PanelError(f'The notification service returned an invalid {method} reply.')
    return reply['data'][0]


def notifications(method):
    values = typed(method, 'aa{sv}')
    if not isinstance(values, list) or not all(isinstance(value, dict) for value in values):
        raise PanelError('The notification service returned an invalid notification list.')
    return values


def modes():
    values = typed('ListModes', 'as')
    if not isinstance(values, list) or not all(isinstance(value, str) for value in values):
        raise PanelError('The notification service returned invalid modes.')
    return values


def config_paths(environment=None):
    home = Path.home() if environment is None else Path(environment['HOME'])
    environment = os.environ if environment is None else environment
    config_home = Path(environment.get('XDG_CONFIG_HOME') or home / '.config')
    return home / '.mako/config', config_home / 'mako/config'


def setup():
    paths = config_paths()
    if any(path.exists() or path.is_symlink() for path in paths):
        return
    target = paths[1]
    target.parent.mkdir(parents=True, exist_ok=True)
    # Linking a completed temporary file publishes the default without replacing a concurrent
    # user-created file or following a newly created symlink.
    with tempfile.NamedTemporaryFile(dir=target.parent, delete=False) as stream:
        temporary = Path(stream.name)
        try:
            stream.write(MAKO_CONFIG.read_bytes())
            stream.flush()
            os.fsync(stream.fileno())
            try:
                os.link(temporary, target)
            except FileExistsError:
                return
        finally:
            temporary.unlink(missing_ok=True)


def dnd_config():
    # A daemon launched with --config can use a different file from the login default.
    reply = bus_call('GetConnectionUnixProcessID', 's', SERVICE, service='org.freedesktop.DBus',
                     path='/org/freedesktop/DBus', interface='org.freedesktop.DBus')
    if reply is None or reply.get('type') != 'u' or len(reply['data']) != 1 or type(reply['data'][0]) is not int:
        raise PanelError('Cannot identify the running notification service.')
    pid = reply['data'][0]
    arguments = Path(f'/proc/{pid}/cmdline').read_bytes().split(b'\0')
    arguments = [os.fsdecode(value) for value in arguments[1:] if value]
    if not arguments:
        environment = {}
        for entry in Path(f'/proc/{pid}/environ').read_bytes().split(b'\0'):
            key, separator, value = entry.partition(b'=')
            if separator and key in (b'HOME', b'XDG_CONFIG_HOME'):
                environment[os.fsdecode(key)] = os.fsdecode(value)
        if not environment.get('HOME'):
            raise PanelError('Cannot identify the running notification service\'s config directory.')
        paths = config_paths(environment)
        path = next((p for p in paths if p.exists() or p.is_symlink()), paths[1])
    elif len(arguments) == 2 and arguments[0] in ('-c', '--config'):
        path = Path(arguments[1])
    elif len(arguments) == 1 and arguments[0].startswith('--config='):
        path = Path(arguments[0].split('=', 1)[1])
    else:
        raise PanelError('Custom mako launch options prevent verifying Do Not Disturb. Use mako with its config file.')
    if not path.is_absolute():
        path = Path(f'/proc/{pid}/cwd').resolve(strict=True) / path
    validate_dnd_config(path)


def validate_dnd_config(path):
    descriptor = os.open(path, os.O_RDONLY | os.O_NONBLOCK)
    with os.fdopen(descriptor, encoding='utf-8') as stream:
        if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode):
            raise PanelError('The mako config must be a regular file to verify Do Not Disturb.')
        text = stream.read(128 * 1024 + 1)
    if len(text) > 128 * 1024:
        raise PanelError('The mako config is too large to verify Do Not Disturb.')
    section = None
    invisible = None
    for line in text.splitlines():
        line = line.strip()
        if not line or line.startswith('#'):
            continue
        if re.match(r'include\s*=', line):
            raise PanelError('Include-based mako settings are preserved; Do Not Disturb cannot be verified by the panel.')
        if line.startswith('['):
            section, invisible = line, None
        elif line.split('=', 1)[0].strip() == 'invisible':
            invisible = line.split('=', 1)[-1].strip()
    if section != '[mode=do-not-disturb]' or invisible != '1':
        raise PanelError(f'Your mako settings are preserved. Add a final [mode=do-not-disturb] section with invisible=1 to {path}.')


def clean(value, limit=700):
    value = html.unescape(re.sub(r'<[^>]*>', '', str(value)))
    value = ' '.join(''.join(c if c.isprintable() else ' ' for c in value).split())
    return value[:limit]


def field(notification, key):
    value = notification.get(key, {})
    if not isinstance(value, dict) or value.get('type') != 's' or not isinstance(value.get('data'), str):
        raise PanelError(f'The notification service returned an invalid {key}.')
    return clean(value['data'])


def status():
    current = notifications('ListNotifications')
    current_modes = modes()
    dnd_config()
    quiet = MODE in current_modes
    return {
        'text': f'{"◌" if quiet else "◉"} {len(current)}',
        'tooltip': f'Notifications · {len(current)} active\nDo Not Disturb: {"on" if quiet else "off"}\nClick: history · Right-click: toggle Do Not Disturb',
        'class': 'dnd' if quiet else ('active' if current else 'normal'),
        'count': len(current),
        'quiet': quiet,
    }


def toggle():
    current = modes()
    if MODE not in current:
        dnd_config()
        bus_call('Reload')
    changed = [mode for mode in current if mode != MODE] if MODE in current else [*current, MODE]
    bus_call('SetModes', 'as', str(len(changed)), *changed)
    if (MODE in modes()) != (MODE in changed):
        raise PanelError('The notification service did not apply Do Not Disturb.')


def menu(rows, prompt):
    try:
        result = subprocess.run(['fuzzel', '--dmenu', '--prompt', prompt,
                                 '--lines', str(min(12, len(rows))), '--width', '80', '--log-level=error'],
                                input='\n'.join(rows) + '\n', text=True, capture_output=True)
    except OSError as error:
        raise PanelError(f'Cannot open the notification viewer: {error}') from error
    # Fuzzel 1.9 uses 1 for both cancellation and failure; errors log to stderr. Newer
    # releases distinguish cancellation with 2. Suppress informational startup logs above.
    cancelled = result.returncode == 2 or (result.returncode == 1 and not result.stderr.strip())
    if result.returncode != 0 and not cancelled:
        raise PanelError(f'Cannot open the notification viewer: {clean(result.stderr, 250)}')


def history():
    rows = []
    for label, method in [('Active', 'ListNotifications'), ('History', 'ListHistory')]:
        for notification in notifications(method)[:50]:
            app, summary, body = (field(notification, key) for key in ('app-name', 'summary', 'body'))
            rows.append(f'{label} · {app}: {summary}' + (f' — {body}' if body else ''))
    menu(rows or ['No notifications in this session.'], 'Notifications › ')


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('module', choices=['notifications'])
    parser.add_argument('action', choices=['setup', 'status', 'toggle', 'history'])
    args = parser.parse_args()
    try:
        result = globals()[args.action]()
        if args.action == 'status':
            print(json.dumps(result, ensure_ascii=False))
    except (PanelError, OSError, UnicodeError) as error:
        message = clean(error, 600)
        print(f'Gravitas notifications: {message}', file=sys.stderr)
        if args.action == 'status':
            print(json.dumps({'text': '◉ !', 'tooltip': message, 'class': 'error'}, ensure_ascii=False))
        elif args.action != 'setup':
            try:
                menu(textwrap.wrap(message, width=78), 'Notifications unavailable › ')
            except PanelError as display_error:
                print(str(display_error), file=sys.stderr)
            return 1
    return 0


if __name__ == '__main__':
    sys.exit(main())
