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
|
# Copyright (C) 2007, One Laptop Per Child
#
# This program 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; either version 2 of the License, or
# (at your option) any later version.
#
# This program 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 this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
import logging
import os
import tempfile
import dbus
try:
import json
except ImportError:
import simplejson as json
from sugar.bundle.bundle import Bundle, MalformedBundleException
from jarabe.journal import model
class JournalEntryBundle(Bundle):
"""A Journal entry bundle
See http://wiki.laptop.org/go/Journal_entry_bundles for details
"""
# TODO: migrate to SL wiki
MIME_TYPE = 'application/vnd.olpc-journal-entry'
_METADATA_JSON_NAME = '_metadata.json'
_zipped_extension = '.xoj'
_unzipped_extension = None
_infodir = None
def __init__(self, path):
Bundle.__init__(self, path)
def _check_zip_bundle(self):
# potentially expensive, but avoids trouble during unpacking
if self._zip_file.testzip() is not None:
raise MalformedBundleException('Corrupt zip file')
file_names = self._zip_file.namelist()
if len(file_names) == 0:
raise MalformedBundleException('Empty zip file')
metadata_seen = False
for name in file_names:
for part in name.split('/'):
if part.startswith('.'):
raise MalformedBundleException(
'Path component starts with dot: %r', name)
if name.split('/')[-1] == self._METADATA_JSON_NAME:
metadata_seen = True
if not metadata_seen:
raise MalformedBundleException('No metadata file found')
def install(self):
for object_id, file_paths in self._get_directories().items():
if len(object_id) < 36:
logging.warning('Ignoring unknown directory %r', object_id)
continue
if self._METADATA_JSON_NAME not in file_paths:
logging.warning('Ignoring directory %r without %s',
object_id, self._METADATA_JSON_NAME)
continue
try:
self._install_entry(object_id, file_paths)
except Exception:
logging.exception('Error installing Journal entry %r:',
object_id)
def _install_entry(self, object_id, file_paths):
file_paths.remove(self._METADATA_JSON_NAME)
metadata = self._read_metadata(object_id)
data_file_name = ''
if object_id in file_paths:
file_paths.remove(object_id)
data_file_name = self._read_data(object_id)
for path in file_paths:
components = path.split('/')
if len(components) != 2 or components[1] != object_id:
logging.warning('Ignoring unknown file %r', path)
name = components[0]
value = self._zip_file.read(os.path.join(object_id, path))
metadata[name] = dbus.ByteArray(value)
model.write(metadata, data_file_name, transfer_ownership=True)
def get_bundle_id(self):
return None
def _read_data(self, object_id):
data_fd, data_file_name = tempfile.mkstemp(prefix='JEB')
data_file = os.fdopen(data_fd, 'w')
try:
# TODO: handle large files better
# TODO: predict disk-full
data_file.write(self._zip_file.read(
os.path.join(object_id, object_id)))
return data_file_name
finally:
data_file.close()
def _read_metadata(self, object_id):
metadata_path = os.path.join(object_id, self._METADATA_JSON_NAME)
json_data = self._zip_file.read(metadata_path)
return json.loads(json_data)
def _get_directories(self):
"""Return the names of all top-level directories and their contents.
"""
contents = {}
for path in self._zip_file.namelist():
if path.endswith('/'):
continue
directory, file_name = path.lstrip('/').split('/', 1)
contents.setdefault(directory, []).append(file_name)
return contents
def is_installed(self):
# These bundles can be reinstalled as many times as desired.
return False
|
