forked from projectfluent/python-fluent
-
Notifications
You must be signed in to change notification settings - Fork 0
/
context.py
286 lines (230 loc) · 10.9 KB
/
context.py
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
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
# coding=utf8
from __future__ import unicode_literals
import os
import sys
import codecs
import logging
import fluent.syntax.ast as FTL
from fluent.syntax.parser import FluentParser
from fluent.syntax.serializer import FluentSerializer
from fluent.util import fold
try:
from compare_locales.parser import getParser
except ImportError:
def getParser(path):
raise RuntimeError('compare-locales required')
from .cldr import get_plural_categories
from .transforms import Source
from .merge import merge_resource
from .errors import NotSupportedError, UnreadableReferenceError
class MergeContext(object):
"""Stateful context for merging translation resources.
`MergeContext` must be configured with the target language and the
directory locations of the input data.
The transformation takes four types of input data:
- The en-US FTL reference files which will be used as templates for
message order, comments and sections.
- The current FTL files for the given language.
- The legacy (DTD, properties) translation files for the given
language. The translations from these files will be transformed
into FTL and merged into the existing FTL files for this language.
- A list of `FTL.Message` objects some of whose nodes are special
helper or transform nodes:
helpers: EXTERNAL_ARGUMENT, MESSAGE_REFERENCE
transforms: COPY, REPLACE_IN_TEXT, REPLACE, PLURALS, CONCAT
"""
def __init__(self, lang, reference_dir, localization_dir):
self.fluent_parser = FluentParser(with_spans=False)
self.fluent_serializer = FluentSerializer()
# An iterable of plural category names relevant to the context's
# language. E.g. ('one', 'other') for English.
try:
self.plural_categories = get_plural_categories(lang)
except RuntimeError as e:
print(e.message)
self.plural_categories = 'en'
# Paths to directories with input data, relative to CWD.
self.reference_dir = reference_dir
self.localization_dir = localization_dir
# Parsed input resources stored by resource path.
self.reference_resources = {}
self.localization_resources = {}
# An iterable of `FTL.Message` objects some of whose nodes can be the
# transform operations.
self.transforms = {}
# A dict whose keys are `(path, key)` tuples corresponding to target
# FTL translations, and values are sets of `(path, key)` tuples
# corresponding to localized entities which will be migrated.
self.dependencies = {}
def read_ftl_resource(self, path):
"""Read an FTL resource and parse it into an AST."""
f = codecs.open(path, 'r', 'utf8')
try:
contents = f.read()
except UnicodeDecodeError as err:
logger = logging.getLogger('migrate')
logger.warn('Unable to read file {}: {}'.format(path, err))
raise err
finally:
f.close()
ast = self.fluent_parser.parse(contents)
annots = [
annot
for entry in ast.body
for annot in entry.annotations
]
if len(annots):
logger = logging.getLogger('migrate')
for annot in annots:
msg = annot.message
logger.warn('Syntax error in {}: {}'.format(path, msg))
return ast
def read_legacy_resource(self, path):
"""Read a legacy resource and parse it into a dict."""
parser = getParser(path)
parser.readFile(path)
# Transform the parsed result which is an iterator into a dict.
return {entity.key: entity.val for entity in parser}
def maybe_add_localization(self, path):
"""Add a localization resource to migrate translations from.
Only legacy resources can be added as migration sources. The resource
may be missing on disk.
Uses a compare-locales parser to create a dict of (key, string value)
tuples.
"""
if path.endswith('.ftl'):
error_message = (
'Migrating translations from Fluent files is not supported '
'({})'.format(path))
logging.getLogger('migrate').error(error_message)
raise NotSupportedError(error_message)
try:
fullpath = os.path.join(self.localization_dir, path)
collection = self.read_legacy_resource(fullpath)
except IOError:
logger = logging.getLogger('migrate')
logger.warn('Missing localization file: {}'.format(path))
else:
self.localization_resources[path] = collection
def add_transforms(self, path, reference, transforms):
"""Define transforms for path using reference as template.
Each transform is an extended FTL node with `Transform` nodes as some
values. Transforms are stored in their lazy AST form until
`merge_changeset` is called, at which point they are evaluated to real
FTL nodes with migrated translations.
Each transform is scanned for `Source` nodes which will be used to
build the list of dependencies for the transformed message.
"""
def get_sources(acc, cur):
if isinstance(cur, Source):
acc.add((cur.path, cur.key))
return acc
refpath = os.path.join(self.reference_dir, reference)
try:
ast = self.read_ftl_resource(refpath)
except IOError as err:
error_message = 'Missing reference file: {}'.format(refpath)
logging.getLogger('migrate').error(error_message)
raise UnreadableReferenceError(error_message)
except UnicodeDecodeError as err:
error_message = 'Error reading file {}: {}'.format(refpath, err)
logging.getLogger('migrate').error(error_message)
raise UnreadableReferenceError(error_message)
else:
# The reference file will be used by the merge function as
# a template for serializing the merge results.
self.reference_resources[path] = ast
for node in transforms:
# Scan `node` for `Source` nodes and collect the information they
# store into a set of dependencies.
dependencies = fold(get_sources, node, set())
# Set these sources as dependencies for the current transform.
self.dependencies[(path, node.id.name)] = dependencies
path_transforms = self.transforms.setdefault(path, [])
path_transforms += transforms
if path not in self.localization_resources:
fullpath = os.path.join(self.localization_dir, path)
try:
ast = self.read_ftl_resource(fullpath)
except IOError:
logger = logging.getLogger('migrate')
logger.info(
'Localization file {} does not exist and '
'it will be created'.format(path))
except UnicodeDecodeError:
logger = logging.getLogger('migrate')
logger.warn(
'Localization file {} will be re-created and some '
'translations might be lost'.format(path))
else:
self.localization_resources[path] = ast
def get_source(self, path, key):
"""Get an entity value from a localized legacy source.
Used by the `Source` transform.
"""
resource = self.localization_resources[path]
return resource.get(key, None)
def merge_changeset(self, changeset=None):
"""Return a generator of FTL ASTs for the changeset.
The input data must be configured earlier using the `add_*` methods.
if given, `changeset` must be a set of (path, key) tuples describing
which legacy translations are to be merged.
Given `changeset`, return a dict whose keys are resource paths and
values are `FTL.Resource` instances. The values will also be used to
update this context's existing localization resources.
"""
if changeset is None:
# Merge all known legacy translations. Used in tests.
changeset = {
(path, key)
for path, strings in self.localization_resources.iteritems()
if not path.endswith('.ftl')
for key in strings.iterkeys()
}
for path, reference in self.reference_resources.iteritems():
current = self.localization_resources.get(path, FTL.Resource())
transforms = self.transforms.get(path, [])
def in_changeset(ident):
"""Check if entity should be merged.
If at least one dependency of the entity is in the current
set of changeset, merge it.
"""
message_deps = self.dependencies.get((path, ident), None)
# Don't merge if we don't have a transform for this message.
if message_deps is None:
return False
# As a special case, if a transform exists but has no
# dependecies, it's a hardcoded `FTL.Node` which doesn't
# migrate any existing translation but rather creates a new
# one. Merge it.
if len(message_deps) == 0:
return True
# If the intersection of the dependencies and the current
# changeset is non-empty, merge this message.
return message_deps & changeset
# Merge legacy translations with the existing ones using the
# reference as a template.
snapshot = merge_resource(
self, reference, current, transforms, in_changeset
)
# If none of the transforms is in the given changeset, the merged
# snapshot is identical to the current translation. We compare
# JSON trees rather then use filtering by `in_changeset` to account
# for translations removed from `reference`.
if snapshot.to_json() == current.to_json():
continue
# Store the merged snapshot on the context so that the next merge
# already takes it into account as the existing localization.
self.localization_resources[path] = snapshot
# The result for this path is a complete `FTL.Resource`.
yield path, snapshot
def serialize_changeset(self, changeset):
"""Return a dict of serialized FTLs for the changeset.
Given `changeset`, return a dict whose keys are resource paths and
values are serialized FTL snapshots.
"""
return {
path: self.fluent_serializer.serialize(snapshot)
for path, snapshot in self.merge_changeset(changeset)
}
logging.basicConfig()