Welcome to mirror list, hosted at ThFree Co, Russian Federation.

history_export.py « gtk « gajim - dev.gajim.org/gajim/gajim.git - Unnamed repository; edit this file 'description' to name the repository.
summaryrefslogtreecommitdiff
blob: ce615dcd4426f2f637254a1654c2db939abf6a4d (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
# This file is part of Gajim.
#
# Gajim 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.
#
# Gajim 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 Gajim. If not, see <http://www.gnu.org/licenses/>.

from __future__ import annotations

from typing import cast
from typing import Literal
from typing import Optional
from typing import overload

import logging
import time
from datetime import datetime
from pathlib import Path

from gi.repository import Gtk

from gajim.common import app
from gajim.common import configpaths
from gajim.common.const import KindConstant
from gajim.common.helpers import filesystem_path_from_uri
from gajim.common.helpers import make_path_from_jid
from gajim.common.i18n import _
from gajim.common.storage.archive import MessageExportRow

from gajim.gtk.assistant import Assistant
from gajim.gtk.assistant import ErrorPage
from gajim.gtk.assistant import Page
from gajim.gtk.builder import get_builder

log = logging.getLogger('gajim.gtk.history_export')


class HistoryExport(Assistant):
    def __init__(self, account: Optional[str] = None) -> None:
        Assistant.__init__(self)

        self.account = account

        self.add_button('export',
                        _('Export'),
                        complete=True,
                        css_class='suggested-action')
        self.add_button('close', _('Close'))
        self.add_button('back', _('Back'))
        self.set_button_visible_func(self._visible_func)

        self.add_pages({'start': SelectAccountDir(account)})

        progress_page = self.add_default_page('progress')
        progress_page.set_title(_('Exporting History...'))
        progress_page.set_text(_('Exporting your messages...'))

        success_page = self.add_default_page('success')
        success_page.set_title(_('Export Finished'))
        success_page.set_text(
            _('Your messages have been exported successfully'))

        error_page = self.add_default_page('error')
        error_page.set_title(_('Error while Exporting'))
        error_page.set_text(
            _('An error occurred while exporting your messages'))

        self.connect('button-clicked', self._on_button_clicked)
        self.show_all()

    @overload
    def get_page(self, name: Literal['error']) -> ErrorPage: ...

    @overload
    def get_page(self, name: Literal['start']) -> SelectAccountDir: ...

    def get_page(self, name: str) -> Page:
        return self._pages[name]

    @staticmethod
    def _visible_func(_assistant: Assistant, page_name: str) -> list[str]:
        if page_name == 'start':
            return ['close', 'export']

        if page_name == 'progress':
            return []

        if page_name == 'success':
            return ['back', 'close']

        if page_name == 'error':
            return ['back', 'close']
        raise ValueError(f'page {page_name} unknown')

    def _on_button_clicked(self,
                           _assistant: Assistant,
                           button_name: str
                           ) -> None:
        if button_name == 'export':
            self.show_page('progress', Gtk.StackTransitionType.SLIDE_LEFT)
            self._on_export()

        elif button_name == 'back':
            self.show_page('start', Gtk.StackTransitionType.SLIDE_RIGHT)

        elif button_name == 'close':
            self.destroy()

    def _on_export(self) -> None:
        start_page = self.get_page('start')
        account, directory = start_page.get_account_and_directory()

        current_time = datetime.now()
        time_str = current_time.strftime('%Y-%m-%d-%H-%M-%S')
        export_dir = Path(directory) / f'export_{time_str}'

        jids = app.storage.archive.get_conversation_jids(account)

        for jid in jids:
            messages = app.storage.archive.get_messages_for_export(account, jid)
            if not messages:
                continue

            file_path = make_path_from_jid(export_dir, jid)
            try:
                file_path.mkdir(parents=True, exist_ok=True)
            except OSError as err:
                self.get_page('error').set_text(
                    _('An error occurred while trying to create a '
                      'file at %(path)s: %(error)s') % {
                          'path': file_path,
                          'error': str(err)})
                self.show_page('error', Gtk.StackTransitionType.SLIDE_LEFT)
                return

            with open(file_path / 'history.txt', 'w', encoding='utf-8') as file:
                file.write(f'History for {jid}\n\n')
                for message in messages:
                    file.write(self._get_export_line(message))

        self.show_page('success', Gtk.StackTransitionType.SLIDE_LEFT)

    def _get_export_line(self, message: MessageExportRow) -> str:
        if message.kind in (KindConstant.SINGLE_MSG_RECV,
                            KindConstant.CHAT_MSG_RECV):
            name = message.jid
        elif message.kind in (KindConstant.SINGLE_MSG_SENT,
                              KindConstant.CHAT_MSG_SENT):
            name = _('You')
        elif message.kind == KindConstant.GC_MSG:
            name = message.contact_name
        else:
            raise ValueError('Unknown kind: %s' % message.kind)

        timestamp = ''
        try:
            timestamp = time.strftime(
                '%Y-%m-%d %H:%M:%S', time.localtime(message.time))
        except ValueError:
            pass

        return f'{timestamp} {name}: {message.message}\n'


class SelectAccountDir(Page):
    def __init__(self, account: Optional[str]) -> None:
        Page.__init__(self)
        self._account = account
        self._export_directory = str(configpaths.get('MY_DATA'))

        self.title = _('Export Chat History')

        self._ui = get_builder('history_export.ui')
        self.add(self._ui.select_account_box)
        self._ui.connect_signals(self)

        accounts = app.get_enabled_accounts_with_labels()
        liststore = cast(Gtk.ListStore, self._ui.account_combo.get_model())
        for acc in accounts:
            liststore.append(acc)

        if self._account is not None:
            self._ui.account_combo.set_active_id(self._account)
        else:
            self._ui.account_combo.set_active(0)

        self._ui.file_chooser_button.set_current_folder(self._export_directory)

        self._set_complete()

        self.show_all()

    def _on_account_changed(self, combobox: Gtk.ComboBox) -> None:
        account = combobox.get_active_id()
        self._account = account
        self._set_complete()

    def _set_complete(self) -> None:
        self.complete = bool(self._account is not None)
        self.update_page_complete()

    def _on_file_set(self, button: Gtk.FileChooserButton) -> None:
        uri = button.get_uri()
        assert uri is not None
        path = filesystem_path_from_uri(uri)
        assert path is not None
        self._export_directory = str(path)

    def get_account_and_directory(self) -> tuple[str, str]:
        assert self._account is not None
        assert self._export_directory is not None
        return self._account, self._export_directory