path: root/url_image_preview/url_image_preview.py

# This file is part of Image Preview Gajim Plugin.
#
# Image Preview Gajim Plugin is free software;
# you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published
# by the Free Software Foundation; version 3 only.
#
# Image Preview Gajim Plugin is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Image Preview Gajim Plugin.
# If not, see <http://www.gnu.org/licenses/>.

import os
import logging
import shutil
import mimetypes
from pathlib import Path
from functools import partial
from urllib.parse import urlparse

import gi
from gi.repository import Gtk
from gi.repository import Gdk
from gi.repository import GdkPixbuf
from gi.repository import Gio
from gi.repository import GLib
from gi.repository import Soup
try:
    gi.require_version('Gst', '1.0')
    gi.require_version('GstPbutils', '1.0')
    from gi.repository import Gst
    from gi.repository import GstPbutils
except Exception:
    pass


from gajim.common import app
from gajim.common import configpaths
from gajim.common.helpers import open_file
from gajim.common.helpers import open_uri
from gajim.common.helpers import write_file_async
from gajim.common.helpers import load_file_async
from gajim.common.helpers import get_tls_error_phrase
from gajim.common.helpers import get_user_proxy
from gajim.gui.dialogs import ErrorDialog
from gajim.gui.filechoosers import FileSaveDialog
from gajim.gui.util import get_cursor
from gajim.gui.util import get_monitor_scale_factor
from gajim.gui.util import load_icon

from gajim.plugins import GajimPlugin
from gajim.plugins.helpers import get_builder
from gajim.plugins.plugins_i18n import _

from url_image_preview.config_dialog import UrlImagePreviewConfigDialog
from url_image_preview.mime_types import MIME_TYPES

log = logging.getLogger('gajim.p.preview')

ERROR_MSG = None
try:
    from PIL import Image  # pylint: disable=unused-import
except ImportError:
    Image = None
    log.error('Pillow not available')
    ERROR_MSG = _('Please install python-pillow')

try:
    import cryptography  # pylint: disable=unused-import
except Exception:
    ERROR_MSG = _('Please install python-cryptography')
    log.error('python-cryptography not available')

# pylint: disable=ungrouped-imports
if ERROR_MSG is None:
    from url_image_preview.utils import aes_decrypt
    from url_image_preview.utils import get_image_paths
    from url_image_preview.utils import split_geo_uri
    from url_image_preview.utils import parse_fragment
    from url_image_preview.utils import create_thumbnail
    from url_image_preview.utils import pixbuf_from_data
    from url_image_preview.utils import filename_from_uri
# pylint: enable=ungrouped-imports


def get_previewable_mime_types():
    previewable_mime_types = set()
    for fmt in GdkPixbuf.Pixbuf.get_formats():
        for mime_type in fmt.get_mime_types():
            previewable_mime_types.add(mime_type.lower())
    if Image is not None:
        Image.init()
        for mime_type in Image.MIME.values():
            previewable_mime_types.add(mime_type.lower())
    return tuple(filter(
        lambda mime_type: mime_type.startswith('image'),
        previewable_mime_types
    ))


PREVIEWABLE_MIME_TYPES = get_previewable_mime_types()
mime_types = set(MIME_TYPES)
# Merge both: if it’s a previewable image, it should be allowed
ALLOWED_MIME_TYPES = mime_types.union(PREVIEWABLE_MIME_TYPES)


class UrlImagePreviewPlugin(GajimPlugin):
    def init(self):
        # pylint: disable=attribute-defined-outside-init
        if ERROR_MSG:
            self.activatable = False
            self.available_text = ERROR_MSG
            self.config_dialog = None
            return

        self.description = _('Displays a preview of image links.')
        self.config_dialog = partial(UrlImagePreviewConfigDialog, self)

        self.gui_extension_points = {
            'chat_control_base': (self._on_connect_chat_control_base,
                                  self._on_disconnect_chat_control_base),
            'history_window': (self._on_connect_history_window,
                               self._on_disconnect_history_window),
            'print_real_text': (self._print_real_text, None),
        }

        self.config_default_values = {
            'PREVIEW_SIZE': (150, 'Preview size (100-1000)'),
            'MAX_FILE_SIZE': ('10485760', 'Max file size for image preview'),
            'ALLOW_ALL_IMAGES': (False, ''),
            'LEFTCLICK_ACTION': ('open_menuitem', 'Open'),
            'ANONYMOUS_MUC': (False, ''),
            'VERIFY': (True, ''),
        }

        self._textviews = {}
        self._sessions = {}

        self._orig_dir = Path(configpaths.get('MY_DATA')) / 'downloads'
        self._thumb_dir = Path(configpaths.get('MY_CACHE')) / 'downloads.thumb'

        if GLib.mkdir_with_parents(str(self._orig_dir), 0o700) != 0:
            log.error('Failed to create: %s', self._orig_dir)

        if GLib.mkdir_with_parents(str(self._thumb_dir), 0o700) != 0:
            log.error('Failed to create: %s', self._thumb_dir)

        if app.settings.get('use_kib_mib'):
            self._units = GLib.FormatSizeFlags.IEC_UNITS
        else:
            self._units = GLib.FormatSizeFlags.DEFAULT

        self._migrate_config()
        self._load_css()

    def _migrate_config(self):
        action = self.config['LEFTCLICK_ACTION']
        if action.endswith('_menuitem'):
            self.config['LEFTCLICK_ACTION'] = action[:-9]

    @staticmethod
    def _load_css():
        path = Path(__file__).parent / 'preview.css'
        try:
            with path.open('r') as file:
                css = file.read()
        except Exception as exc:
            log.error('Error loading css: %s', exc)
            return

        try:
            provider = Gtk.CssProvider()
            provider.load_from_data(bytes(css.encode('utf-8')))
            Gtk.StyleContext.add_provider_for_screen(Gdk.Screen.get_default(),
                                                     provider, 610)
        except Exception:
            log.exception('Error loading application css')

    def _on_connect_chat_control_base(self, chat_control):
        account = chat_control.account
        if account not in self._sessions:
            self._sessions[account] = self._create_session(account)
        self._textviews[chat_control.control_id] = chat_control.conv_textview

    def _on_disconnect_chat_control_base(self, chat_control):
        self._textviews.pop(chat_control.control_id, None)

    def _on_connect_history_window(self, history_window):
        account = history_window.account
        if (account is not None and account not in self._sessions):
            self._sessions[account] = self._create_session(account)
        self._textviews[id(history_window)] = history_window.history_textview

    def _on_disconnect_history_window(self, history_window):
        self._textviews.pop(id(history_window), None)

    def _get_control_id(self, textview):
        for control_id, textview_ in self._textviews.items():
            if textview == textview_:
                return control_id

    @staticmethod
    def _create_session(account):
        session = Soup.Session()
        session.add_feature_by_type(Soup.ContentSniffer)
        session.props.https_aliases = ['aesgcm']
        session.props.ssl_strict = False

        proxy = get_user_proxy(account)
        if proxy is None:
            resolver = None
        else:
            resolver = proxy.get_resolver()

        session.props.proxy_resolver = resolver
        return session, resolver

    def _get_session(self, account):
        return self._sessions[account][0]

    def _print_real_text(self, textview, text, _text_tags, _graphics,
                         iter_, additional_data):

        if len(text.split(' ')) > 1:
            # urlparse doesn't recognise spaces as URL delimiter
            log.debug('Text is not an uri: %s...', text[:15])
            return

        uri = text
        urlparts = urlparse(uri)
        if not self._accept_uri(urlparts, uri, additional_data):
            return

        textview.plugin_modified = True
        control_id = self._get_control_id(textview)

        start_mark, end_mark = self._print_text(textview.tv.get_buffer(),
                                                iter_,
                                                uri)

        if uri.startswith('geo:'):
            preview = self._process_geo_uri(uri,
                                            start_mark,
                                            end_mark,
                                            control_id,
                                            textview.account)
            if preview is None:
                return
            pixbuf = load_icon('map',
                               size=preview.size,
                               scale=get_monitor_scale_factor(),
                               pixbuf=True)
            self._update_textview(preview, pixbuf)
            return

        preview = self._process_web_uri(uri,
                                        urlparts,
                                        start_mark,
                                        end_mark,
                                        control_id,
                                        textview.account)

        if not preview.orig_exists():
            self._download_content(preview)

        elif not preview.thumb_exists():
            load_file_async(preview.orig_path,
                            self._on_orig_load_finished,
                            preview)

        else:
            load_file_async(preview.thumb_path,
                            self._on_thumb_load_finished,
                            preview)

    @staticmethod
    def _print_text(buffer_, iter_, text):
        if not iter_:
            iter_ = buffer_.get_end_iter()

        start_mark = buffer_.create_mark(None, iter_, True)
        buffer_.insert_with_tags_by_name(iter_, text, 'url')
        end_mark = buffer_.create_mark(None, iter_, True)
        return start_mark, end_mark

    def _accept_uri(self, urlparts, uri, additional_data):
        try:
            oob_url = additional_data['gajim']['oob_url']
        except (KeyError, AttributeError):
            oob_url = None

        # geo
        if urlparts.scheme == 'geo':
            return True

        if not urlparts.netloc:
            log.info('No netloc found in URL: %s', uri)
            return False

        # aesgcm
        if urlparts.scheme == 'aesgcm':
            return True

        # http/https
        if urlparts.scheme in ('https', 'http'):
            if self.config['ALLOW_ALL_IMAGES']:
                return True

            if oob_url is None:
                log.info('No oob url for: %s', uri)
                return False

            if uri != oob_url:
                log.info('uri != oob url: %s != %s', uri, oob_url)
                return False
            return True

        log.info('Unsupported URI scheme: %s', uri)
        return False

    @staticmethod
    def _process_geo_uri(uri,
                         start_mark,
                         end_mark,
                         control_id,
                         account):
        try:
            split_geo_uri(uri)
        except Exception as err:
            log.error(uri)
            log.error(err)
            return

        return Preview(uri,
                       None,
                       None,
                       None,
                       start_mark,
                       end_mark,
                       96,
                       control_id,
                       account)

    def _process_web_uri(self,
                         uri,
                         urlparts,
                         start_mark,
                         end_mark,
                         control_id,
                         account):

        size = self.config['PREVIEW_SIZE']
        orig_path, thumb_path = get_image_paths(uri,
                                                urlparts,
                                                size,
                                                self._orig_dir,
                                                self._thumb_dir)
        return Preview(uri,
                       urlparts,
                       orig_path,
                       thumb_path,
                       start_mark,
                       end_mark,
                       size,
                       control_id,
                       account)

    def _on_orig_load_finished(self, data, error, preview):
        if data is None:
            log.error('%s: %s', preview.orig_path.name, error)
            return

        preview.mime_type = self._guess_mime_type(preview.orig_path)
        preview.file_size = os.path.getsize(preview.orig_path)
        if preview.is_previewable:
            if preview.create_thumbnail(data):
                write_file_async(preview.thumb_path,
                                 preview.thumbnail,
                                 self._on_thumb_write_finished,
                                 preview)
        else:
            self._update_textview(preview, None)

    def _on_thumb_load_finished(self, data, error, preview):
        if data is None:
            log.error('%s: %s', preview.thumb_path.name, error)
            return

        preview.thumbnail = data
        preview.mime_type = self._guess_mime_type(preview.orig_path)
        preview.file_size = os.path.getsize(preview.orig_path)

        try:
            pixbuf = pixbuf_from_data(preview.thumbnail)
        except Exception as err:
            log.error('Unable to load: %s, %s',
                      preview.thumb_path.name,
                      err)
            return
        self._update_textview(preview, pixbuf)

    def _download_content(self, preview, force=False):
        if preview.account is None:
            # History Window can be opened without account context
            # This means we can not apply proxy settings
            return
        log.info('Start downloading: %s', preview.request_uri)
        message = Soup.Message.new('GET', preview.request_uri)
        message.connect('starting', self._check_certificate, preview)
        message.connect(
            'content-sniffed', self._on_content_sniffed, preview, force)

        session = self._get_session(preview.account)
        session.queue_message(message, self._on_finished, preview)

    def _check_certificate(self, message, preview):
        _https_used, _tls_certificate, tls_errors = message.get_https_status()

        if not self.config['VERIFY']:
            return

        if tls_errors:
            phrase = get_tls_error_phrase(tls_errors)
            log.warning('TLS verification failed: %s', phrase)
            session = self._get_session(preview.account)
            session.cancel_message(message, Soup.Status.CANCELLED)
            return

    def _on_content_sniffed(self, message, type_, _params, preview, force):
        file_size = message.props.response_headers.get_content_length()
        uri = message.props.uri.to_string(False)
        session = self._get_session(preview.account)
        preview.mime_type = type_
        preview.file_size = file_size

        if type_ not in ALLOWED_MIME_TYPES:
            log.info('Not an allowed content type: %s, %s', type_, uri)
            session.cancel_message(message, Soup.Status.CANCELLED)
            return

        if file_size == 0 or file_size > int(self.config['MAX_FILE_SIZE']):
            log.info(
                'File size (%s) too big or unknown (zero) for URL: \'%s\'',
                file_size, uri)
            if not force:
                session.cancel_message(message, Soup.Status.CANCELLED)

        self._update_textview(preview, None)

    def _on_finished(self, _session, message, preview):
        if message.status_code != Soup.Status.OK:
            log.warning('Download failed: %s', preview.request_uri)
            log.warning(Soup.Status.get_phrase(message.status_code))
            return

        data = message.props.response_body_data.get_data()
        if data is None:
            return

        if preview.is_aes_encrypted:
            data = aes_decrypt(preview, data)

        if preview.mime_type == 'application/octet-stream':
            preview.mime_type = self._guess_mime_type(preview.orig_path, data)

        write_file_async(preview.orig_path,
                         data,
                         self._on_orig_write_finished,
                         preview)

        if preview.is_previewable:
            if preview.create_thumbnail(data):
                write_file_async(preview.thumb_path,
                                 preview.thumbnail,
                                 self._on_thumb_write_finished,
                                 preview)

    def _on_orig_write_finished(self, _result, error, preview):
        if error is not None:
            log.error('%s: %s', preview.orig_path.name, error)
            return

        log.info('File stored: %s', preview.orig_path.name)
        preview.file_size = os.path.getsize(preview.orig_path)
        if not preview.is_previewable:
            # Don’t update preview if thumb is already displayed,
            # but update preview for audio files
            self._update_textview(preview, None)

    def _on_thumb_write_finished(self, _result, error, preview):
        if error is not None:
            log.error('%s: %s', preview.thumb_path.name, error)
            return

        log.info('Thumbnail stored: %s ', preview.thumb_path.name)

        try:
            pixbuf = pixbuf_from_data(preview.thumbnail)
        except Exception as err:
            log.error('Unable to load: %s, %s',
                      preview.thumb_path.name,
                      err)
            return
        self._update_textview(preview, pixbuf)

    @staticmethod
    def _guess_mime_type(file_path, data=None):
        mime_type, _ = mimetypes.MimeTypes().guess_type(str(file_path))
        if mime_type is None:
            # Try to guess MIME type by file name
            mime_type, _ = Gio.content_type_guess(str(file_path), data)
        log.debug('Guessed MIME type: %s', str(mime_type))
        return mime_type

    @staticmethod
    def _get_icon_for_mime_type(mime_type):
        if mime_type is None:
            return Gio.Icon.new_for_string('mail-attachment')
        return Gio.content_type_get_icon(mime_type)

    def _update_textview(self, preview, data):
        textview = self._textviews.get(preview.control_id)
        if textview is None:
            # Control closed
            return

        buffer_ = preview.start_mark.get_buffer()
        iter_ = buffer_.get_iter_at_mark(preview.start_mark)
        buffer_.insert(iter_, '\n')
        anchor = buffer_.create_child_anchor(iter_)
        anchor.plaintext = preview.uri

        preview_widget = self._create_preview_widget(preview, data)

        textview.tv.add_child_at_anchor(preview_widget, anchor)
        buffer_.delete(iter_,
                       buffer_.get_iter_at_mark(preview.end_mark))

        if textview.autoscroll:
            textview.scroll_to_end()

    def _create_preview_widget(self, preview, data):
        if isinstance(data, GdkPixbuf.PixbufAnimation):
            image = Gtk.Image.new_from_animation(data)
        elif isinstance(data, GdkPixbuf.Pixbuf):
            image = Gtk.Image.new_from_pixbuf(data)
        else:
            icon = self._get_icon_for_mime_type(preview.mime_type)
            image = Gtk.Image.new_from_gicon(icon, Gtk.IconSize.DIALOG)

        path = self.local_file_path('preview.ui')
        ui = get_builder(path)
        ui.connect_signals(self)

        ui.download_button.set_no_show_all(True)
        ui.download_button.connect('clicked', self._on_download, preview)
        ui.save_as_button.set_no_show_all(True)
        ui.save_as_button.connect('clicked', self._on_save_as, preview)
        ui.open_folder_button.set_no_show_all(True)
        ui.open_folder_button.connect('clicked', self._on_open_folder, preview)

        ui.image_button.add(image)
        ui.image_button.set_tooltip_text(preview.filename)
        ui.image_button.connect('clicked',
                                self._on_image_button_clicked,
                                preview)
        ui.image_button.connect('button_press_event',
                                self._on_button_press_event,
                                preview)

        ui.preview_box.show_all()

        if preview.is_geo_uri:
            ui.file_name.set_text(_('Click to view location'))
            ui.save_as_button.hide()
            ui.open_folder_button.hide()
            ui.download_button.hide()
            location = split_geo_uri(preview.uri)
            ui.file_size.set_text(_('Lat: %s Lon: %s') % (
                location.lat, location.lon))
            ui.image_button.set_tooltip_text(
                _('Location at Lat: %s Lon: %s') % (
                    location.lat, location.lon))
            ui.image_event_box.set_halign(Gtk.Align.CENTER)
            ui.preview_box.set_size_request(160, -1)
            return ui.preview_box

        if preview.is_previewable and preview.orig_exists():
            ui.image_event_box.set_halign(Gtk.Align.CENTER)
        else:
            image.set_property('pixel-size', 64)

        if preview.orig_exists():
            ui.download_button.hide()
            if (preview.is_audio and app.is_installed('GST') and
                    self._contains_audio_streams(preview.orig_path)):
                audio_widget = AudioWidget(preview.orig_path)
                ui.preview_box.pack_start(audio_widget, True, True, 0)
        else:
            ui.save_as_button.hide()
            ui.open_folder_button.hide()

        file_size_string = _('File size unknown')
        if preview.file_size != 0:
            file_size_string = GLib.format_size_full(
                preview.file_size, self._units)
        ui.file_size.set_text(file_size_string)

        ui.preview_box.set_size_request(300, -1)
        ui.file_name.set_text(preview.filename)
        ui.file_name.set_tooltip_text(preview.filename)

        return ui.preview_box

    def _get_context_menu(self, preview):
        def destroy(menu, _pspec):
            visible = menu.get_property('visible')
            if not visible:
                GLib.idle_add(menu.destroy)

        path = self.local_file_path('context_menu.ui')
        ui = get_builder(path)

        ui.download.connect(
            'activate', self._on_download, preview)
        ui.open.connect(
            'activate', self._on_open, preview)
        ui.save_as.connect(
            'activate', self._on_save_as, preview)
        ui.open_folder.connect(
            'activate', self._on_open_folder, preview)
        ui.open_link_in_browser.connect(
            'activate', self._on_open_link_in_browser, preview)
        ui.copy_link_location.connect(
            'activate', self._on_copy_link_location, preview)
        ui.context_menu.connect('notify::visible', destroy)

        if preview.is_aes_encrypted:
            ui.open_link_in_browser.hide()

        if preview.is_geo_uri:
            ui.download.hide()
            ui.open_link_in_browser.hide()
            ui.save_as.hide()
            ui.open_folder.hide()
            return ui.context_menu

        if preview.orig_exists():
            ui.download.hide()
        else:
            ui.open.hide()
            ui.save_as.hide()
            ui.open_folder.hide()

        return ui.context_menu

    def _on_download(self, _menu, preview):
        if not preview.orig_exists():
            self._download_content(preview, force=True)

    def _on_open(self, _menu, preview):
        if preview.is_geo_uri:
            open_uri(preview.uri)
            return

        if not preview.orig_exists():
            self._download_content(preview, force=True)
            return

        open_file(preview.orig_path)

    def _on_save_as(self, _menu, preview):
        def on_ok(target_path):
            dirname = Path(target_path).parent
            if not os.access(dirname, os.W_OK):
                ErrorDialog(
                    _('Directory \'%s\' is not writable') % dirname,
                    _('You do not have the proper permissions to '
                      'create files in this directory.'),
                    transient_for=app.app.get_active_window())
                return
            shutil.copyfile(str(preview.orig_path), target_path)

        if not preview.orig_exists():
            self._download_content(preview, force=True)
            return

        FileSaveDialog(on_ok,
                       path=app.settings.get('last_save_dir'),
                       file_name=preview.filename,
                       transient_for=app.app.get_active_window())

    def _on_open_folder(self, _menu, preview):
        if not preview.orig_exists():
            self._download_content(preview, force=True)
            return
        open_file(preview.orig_path.parent)

    @staticmethod
    def _on_copy_link_location(_menu, preview):
        clipboard = Gtk.Clipboard.get_default(Gdk.Display.get_default())
        clipboard.set_text(preview.uri, -1)

    @staticmethod
    def _on_open_link_in_browser(_menu, preview):
        if preview.is_aes_encrypted:
            if preview.is_geo_uri:
                open_uri(preview.uri)
                return
            open_file(preview.orig_path)
        else:
            open_uri(preview.uri)

    def _on_image_button_clicked(self, _button, preview):
        action = self.config['LEFTCLICK_ACTION']
        method = getattr(self, '_on_%s' % action)
        method(None, preview)

    def _on_button_press_event(self, _button, event, preview):
        if event.type == Gdk.EventType.BUTTON_PRESS and event.button == 3:
            # Right klick
            menu = self._get_context_menu(preview)
            menu.popup_at_pointer(event)

    @staticmethod
    def _contains_audio_streams(file_path):
        # Check if it is really an audio file
        has_audio = False
        discoverer = GstPbutils.Discoverer()
        try:
            info = discoverer.discover_uri(f'file://{file_path}')
            has_audio = bool(info.get_audio_streams())
        except GLib.Error as err:
            log.error('Error while reading %s: %s', str(file_path), err)
            return False
        if not has_audio:
            log.warning('File does not contain audio stream: %s',
                        str(file_path))
        return has_audio

    @staticmethod
    def _on_realize(event_box):
        event_box.get_window().set_cursor(get_cursor('pointer'))


class Preview:
    def __init__(self, uri, urlparts, orig_path, thumb_path,
                 start_mark, end_mark, size, control_id, account):
        self._uri = uri
        self._urlparts = urlparts
        self._filename = filename_from_uri(self._uri)

        self.size = size
        self.control_id = control_id
        self.orig_path = orig_path
        self.thumb_path = thumb_path
        self.start_mark = start_mark
        self.end_mark = end_mark
        self.account = account

        self.thumbnail = None
        self.mime_type = None
        self.file_size = 0

        self.key, self.iv = None, None
        if self.is_aes_encrypted:
            self.key, self.iv = parse_fragment(urlparts.fragment)

    @property
    def is_geo_uri(self):
        return self._uri.startswith('geo:')

    @property
    def is_web_uri(self):
        return not self.is_geo_uri

    @property
    def is_previewable(self):
        return self.mime_type in PREVIEWABLE_MIME_TYPES

    @property
    def is_audio(self):
        is_allowed = bool(self.mime_type in ALLOWED_MIME_TYPES)
        return is_allowed and self.mime_type.startswith('audio/')

    @property
    def uri(self):
        return self._uri

    @property
    def filename(self):
        return self._filename

    @property
    def request_uri(self):
        if self.is_aes_encrypted:
            # Remove fragments so we dont transmit it to the server
            urlparts = self._urlparts._replace(scheme='https', fragment='')
            return urlparts.geturl()
        return self._urlparts.geturl()

    @property
    def is_aes_encrypted(self):
        if self._urlparts is None:
            return False
        return self._urlparts.scheme == 'aesgcm'

    def thumb_exists(self):
        return self.thumb_path.exists()

    def orig_exists(self):
        return self.orig_path.exists()

    def create_thumbnail(self, data):
        self.thumbnail = create_thumbnail(data, self.size)
        if self.thumbnail is None:
            log.warning('Creating thumbnail failed for: %s', self.orig_path)
            return False
        return True


class AudioWidget(Gtk.Box):
    def __init__(self, file_path):
        Gtk.Box.__init__(self, orientation=Gtk.Orientation.HORIZONTAL,
                         spacing=6)
        self._playbin = None
        self._query = None
        self._has_timeout = False

        self._build_audio_widget()
        self._setup_audio_player(file_path)

    def _build_audio_widget(self):
        play_button = Gtk.Button()
        play_button.get_style_context().add_class('flat')
        play_button.get_style_context().add_class('preview-button')
        play_button.set_tooltip_text(_('Start/stop playback'))
        self._play_icon = Gtk.Image.new_from_icon_name(
            'media-playback-start-symbolic',
            Gtk.IconSize.BUTTON)
        play_button.add(self._play_icon)
        play_button.connect('clicked', self._on_play_clicked)
        event_box = Gtk.EventBox()
        event_box.connect('realize', self._on_realize)
        event_box.add(play_button)
        self.add(event_box)

        self._seek_bar = Gtk.Scale(
            orientation=Gtk.Orientation.HORIZONTAL)
        self._seek_bar.set_range(0.0, 1.0)
        self._seek_bar.set_hexpand(True)
        self._seek_bar.set_value_pos(Gtk.PositionType.RIGHT)
        self._seek_bar.connect('change-value', self._on_seek)
        self._seek_bar.connect(
            'format-value', self._format_audio_timestamp)
        event_box = Gtk.EventBox()
        event_box.connect('realize', self._on_realize)
        event_box.add(self._seek_bar)
        self.add(event_box)

        self.connect('destroy', self._on_destroy)
        self.show_all()

    def _setup_audio_player(self, file_path):
        self._playbin = Gst.ElementFactory.make('playbin', 'bin')
        if self._playbin is None:
            return
        self._playbin.set_property(
            'uri', f'file://{file_path}')
        state_return = self._playbin.set_state(Gst.State.PAUSED)
        if state_return == Gst.StateChangeReturn.FAILURE:
            return

        self._query = Gst.Query.new_position(Gst.Format.TIME)
        bus = self._playbin.get_bus()
        bus.add_signal_watch()
        bus.connect('message', self._on_bus_message)

    def _on_bus_message(self, _bus, message):
        if message.type == Gst.MessageType.EOS:
            self._set_pause(True)
            self._playbin.seek_simple(
                Gst.Format.TIME, Gst.SeekFlags.FLUSH, 0)
        elif message.type == Gst.MessageType.STATE_CHANGED:
            _success, duration = self._playbin.query_duration(
                Gst.Format.TIME)
            if duration > 0:
                self._seek_bar.set_range(0.0, duration)

            is_paused = self._get_paused()
            if (duration > 0 and not is_paused and
                    not self._has_timeout):
                GLib.timeout_add(500, self._update_seek_bar)
                self._has_timeout = True

    def _on_seek(self, _range, _scroll, value):
        self._playbin.seek_simple(
            Gst.Format.TIME, Gst.SeekFlags.FLUSH, value)
        return False

    def _on_play_clicked(self, _button):
        self._set_pause(not self._get_paused())

    def _on_destroy(self, _widget):
        self._playbin.set_state(Gst.State.NULL)

    def _get_paused(self):
        _, state, _ = self._playbin.get_state(20)
        return state == Gst.State.PAUSED

    def _set_pause(self, paused):
        if paused:
            self._playbin.set_state(Gst.State.PAUSED)
            self._play_icon.set_from_icon_name(
                'media-playback-start-symbolic',
                Gtk.IconSize.BUTTON)
        else:
            self._playbin.set_state(Gst.State.PLAYING)
            self._play_icon.set_from_icon_name(
                'media-playback-pause-symbolic',
                Gtk.IconSize.BUTTON)

    def _update_seek_bar(self):
        if self._get_paused():
            self._has_timeout = False
            return False

        if self._playbin.query(self._query):
            _fmt, cur_pos = self._query.parse_position()
            self._seek_bar.set_value(cur_pos)
        return True

    @staticmethod
    def _format_audio_timestamp(_widget, ns):
        seconds = ns / 1000000000
        minutes = seconds / 60
        hours = minutes / 60

        i_seconds = int(seconds) % 60
        i_minutes = int(minutes) % 60
        i_hours = int(hours)

        if i_hours > 0:
            return f'{i_hours:d}:{i_minutes:02d}:{i_seconds:02d}'
        return f'{i_minutes:d}:{i_seconds:02d}'

    @staticmethod
    def _on_realize(event_box):
        event_box.get_window().set_cursor(get_cursor('pointer'))