-
Notifications
You must be signed in to change notification settings - Fork 75
/
Copy pathproject.py
1267 lines (1124 loc) · 47.9 KB
/
project.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
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# -*- coding: utf-8 -*-
import getpass
import os
import re
import fnmatch
import datetime
import time
import ssl
try:
import configparser
except ImportError:
import ConfigParser as configparser
from txclib.web import *
from txclib.utils import *
from txclib.packages import urllib3
from txclib.packages.urllib3.packages import six
from txclib.urls import API_URLS
from txclib.config import OrderedRawConfigParser, Flipdict
from txclib.log import logger
from txclib.processors import visit_hostname
from txclib.paths import posix_path, native_path, posix_sep
from txclib.packages.urllib3.exceptions import SSLError
class ProjectNotInit(Exception):
pass
class Project(object):
"""
Represents an association between the local and remote project instances.
"""
def __init__(self, path_to_tx=None, init=True):
"""
Initialize the Project attributes.
"""
if init:
self._init(path_to_tx)
def _init(self, path_to_tx=None):
instructions = "Run 'tx init' to initialize your project first!"
try:
self.root = self._get_tx_dir_path(path_to_tx)
self.config_file = self._get_config_file_path(self.root)
self.config = self._read_config_file(self.config_file)
self.txrc_file = self._get_transifex_file()
local_txrc_file = self._get_transifex_file(os.getcwd())
self.txrc = self._get_transifex_config([self.txrc_file, local_txrc_file])
if os.path.exists(local_txrc_file):
self.txrc_file = local_txrc_file
except ProjectNotInit as e:
logger.error('\n'.join([six.u(str(e)), instructions]))
raise
host = self.config.get('main', 'host')
if host.lower().startswith('https://'):
self.conn = urllib3.connection_from_url(
host,
cert_reqs=ssl.CERT_REQUIRED,
ca_certs=certs_file()
)
else:
self.conn = urllib3.connection_from_url(host)
def _get_config_file_path(self, root_path):
"""Check the .tx/config file exists."""
config_file = os.path.join(root_path, ".tx", "config")
logger.debug("Config file is %s" % config_file)
if not os.path.exists(config_file):
msg = "Cannot find the config file (.tx/config)!"
raise ProjectNotInit(msg)
return config_file
def _get_tx_dir_path(self, path_to_tx):
"""Check the .tx directory exists."""
root_path = path_to_tx or find_dot_tx()
logger.debug("Path to tx is %s." % root_path)
if not root_path:
msg = "Cannot find any .tx directory!"
raise ProjectNotInit(msg)
return root_path
def _read_config_file(self, config_file):
"""Parse the config file and return its contents."""
config = OrderedRawConfigParser()
try:
config.read(config_file)
except Exception as err:
msg = "Cannot open/parse .tx/config file: %s" % err
raise ProjectNotInit(msg)
return config
def _get_transifex_config(self, txrc_files):
"""Read the configuration from the .transifexrc files."""
txrc = OrderedRawConfigParser()
try:
txrc.read(txrc_files)
except Exception as e:
msg = "Cannot read configuration file: %s" % e
raise ProjectNotInit(msg)
self._migrate_txrc_file(txrc)
return txrc
def _migrate_txrc_file(self, txrc):
"""Migrate the txrc file, if needed."""
if not os.path.exists(self.txrc_file):
return txrc
for section in txrc.sections():
orig_hostname = txrc.get(section, 'hostname')
hostname = visit_hostname(orig_hostname)
if hostname != orig_hostname:
msg = "Hostname %s should be changed to %s."
logger.info(msg % (orig_hostname, hostname))
if (sys.stdin.isatty() and sys.stdout.isatty() and
confirm('Change it now? ', default=True)):
txrc.set(section, 'hostname', hostname)
msg = 'Hostname changed'
logger.info(msg)
else:
hostname = orig_hostname
self._save_txrc_file(txrc)
return txrc
def _get_transifex_file(self, directory=None):
"""Fetch the path of the .transifexrc file.
It is in the home directory of the user by default.
"""
if directory is not None:
logger.debug(".transifexrc file is at %s" % directory)
return os.path.join(directory, ".transifexrc")
directory = os.path.expanduser('~')
txrc_file = os.path.join(directory, ".transifexrc")
logger.debug(".transifexrc file is at %s" % directory)
if not os.path.exists(txrc_file):
msg = "%s not found." % (txrc_file)
logger.info(msg)
mask = os.umask(0o077)
open(txrc_file, 'w').close()
os.umask(mask)
return txrc_file
def validate_config(self):
"""
To ensure the json structure is correctly formed.
"""
pass
def getset_host_credentials(self, host, user=None, password=None):
"""
Read .transifexrc and report user,pass for a specific host else ask the
user for input.
"""
try:
username = self.txrc.get(host, 'username')
passwd = self.txrc.get(host, 'password')
except (configparser.NoOptionError, configparser.NoSectionError):
logger.info("No entry found for host %s. Creating..." % host)
username = user or input("Please enter your transifex username: ")
while (not username):
username = input("Please enter your transifex username: ")
passwd = password
while (not passwd):
passwd = getpass.getpass()
logger.info("Updating %s file..." % self.txrc_file)
self.txrc.add_section(host)
self.txrc.set(host, 'username', username)
self.txrc.set(host, 'password', passwd)
self.txrc.set(host, 'token', '')
self.txrc.set(host, 'hostname', host)
return username, passwd
def set_remote_resource(self, resource, source_lang, i18n_type, host,
file_filter="translations<sep>%(proj)s.%(res)s<sep><lang>.%(extension)s"):
"""Method to handle the add/conf of a remote resource."""
if not self.config.has_section(resource):
self.config.add_section(resource)
p_slug, r_slug = resource.split('.', 1)
file_filter = file_filter.replace("<sep>", r"%s" % posix_sep)
self.url_info = {
'host': host,
'project': p_slug,
'resource': r_slug
}
extension = self._extension_for(i18n_type)[1:]
self.config.set(resource, 'source_lang', source_lang)
self.config.set(
resource, 'file_filter',
file_filter % {'proj': p_slug, 'res': r_slug, 'extension': extension}
)
self.config.set(resource, 'type', i18n_type)
if host != self.config.get('main', 'host'):
self.config.set(resource, 'host', host)
def get_resource_host(self, resource):
"""
Returns the host that the resource is configured to use. If there is no
such option we return the default one
"""
return self.config.get('main', 'host')
def get_resource_lang_mapping(self, resource):
"""Get language mappings for a specific resource."""
lang_map = Flipdict()
try:
args = self.config.get("main", "lang_map")
for arg in args.replace(' ', '').split(','):
k,v = arg.split(":")
lang_map.update({k:v})
except configparser.NoOptionError:
pass
except (ValueError, KeyError):
raise Exception("Your lang map configuration is not correct.")
if self.config.has_section(resource):
res_lang_map = Flipdict()
try:
args = self.config.get(resource, "lang_map")
for arg in args.replace(' ', '').split(','):
k,v = arg.split(":")
res_lang_map.update({k:v})
except configparser.NoOptionError:
pass
except (ValueError, KeyError):
raise Exception("Your lang map configuration is not correct.")
# merge the lang maps and return result
lang_map.update(res_lang_map)
return lang_map
def get_source_file(self, resource):
"""
Get source file for a resource.
"""
if self.config.has_section(resource):
source_lang = self.config.get(resource, "source_lang")
source_file = self.get_resource_option(resource, 'source_file') or None
if source_file is None:
try:
file_filter = self.config.get(resource, "file_filter")
filename = file_filter.replace('<lang>', source_lang)
if os.path.exists(filename):
return native_path(filename)
except configparser.NoOptionError:
pass
else:
return native_path(source_file)
def get_resource_files(self, resource):
"""
Get a dict for all files assigned to a resource. First we calculate the
files matching the file expression and then we apply all translation
excpetions. The resulting dict will be in this format:
{ 'en': 'path/foo/en/bar.po', 'de': 'path/foo/de/bar.po', 'es': 'path/exceptions/es.po'}
NOTE: All paths are relative to the root of the project
"""
tr_files = {}
if self.config.has_section(resource):
try:
file_filter = self.config.get(resource, "file_filter")
except configparser.NoOptionError:
file_filter = "$^"
source_lang = self.config.get(resource, "source_lang")
source_file = self.get_source_file(resource)
expr_re = regex_from_filefilter(file_filter, self.root)
expr_rec = re.compile(expr_re)
for f_path in files_in_project(self.root):
match = expr_rec.match(posix_path(f_path))
if match:
lang = match.group(1)
if lang != source_lang:
f_path = os.path.relpath(f_path, self.root)
if f_path != source_file:
tr_files.update({lang: f_path})
for (name, value) in self.config.items(resource):
if name.startswith("trans."):
value = native_path(value)
lang = name.split('.')[1]
# delete language which has same file
if value in list(tr_files.values()):
keys = []
for k, v in six.iteritems(tr_files):
if v == value:
keys.append(k)
if len(keys) == 1:
del tr_files[keys[0]]
else:
raise Exception("Your configuration seems wrong."\
" You have multiple languages pointing to"\
" the same file.")
# Add language with correct file
tr_files.update({lang:value})
return tr_files
return None
def get_resource_option(self, resource, option):
"""
Return the requested option for a specific resource
If there is no such option, we return None
"""
if self.config.has_section(resource):
if self.config.has_option(resource, option):
return self.config.get(resource, option)
return None
def get_resource_list(self, project=None):
"""
Parse config file and return tuples with the following format
[ (project_slug, resource_slug), (..., ...)]
"""
resource_list= []
for r in self.config.sections():
if r == 'main':
continue
p_slug, r_slug = r.split('.', 1)
if project and p_slug != project:
continue
resource_list.append(r)
return resource_list
def save(self):
"""
Store the config dictionary in the .tx/config file of the project.
"""
self._save_tx_config()
self._save_txrc_file()
def _save_tx_config(self, config=None):
"""Save the local config file."""
if config is None:
config = self.config
fh = open(self.config_file,"w")
config.write(fh)
fh.close()
def _save_txrc_file(self, txrc=None):
"""Save the .transifexrc file."""
if txrc is None:
txrc = self.txrc
mask = os.umask(0o077)
fh = open(self.txrc_file, 'w')
txrc.write(fh)
fh.close()
os.umask(mask)
def get_full_path(self, relpath):
if relpath[0] == os.path.sep:
return relpath
else:
return os.path.join(self.root, relpath)
def _get_pseudo_file(self, slang, resource, file_filter):
pseudo_file = file_filter.replace('<lang>', '%s_pseudo' % slang)
return native_path(pseudo_file)
def pull(self, languages=[], resources=[], overwrite=True, fetchall=False,
fetchsource=False, force=False, skip=False, minimum_perc=0, mode=None,
pseudo=False):
"""Pull all translations file from transifex server."""
self.minimum_perc = minimum_perc
resource_list = self.get_chosen_resources(resources)
if mode == 'reviewed':
url = 'pull_reviewed_file'
elif mode == 'translator':
url = 'pull_translator_file'
elif mode == 'developer':
url = 'pull_developer_file'
else:
url = 'pull_file'
for resource in resource_list:
logger.debug("Handling resource %s" % resource)
self.resource = resource
project_slug, resource_slug = resource.split('.', 1)
files = self.get_resource_files(resource)
slang = self.get_resource_option(resource, 'source_lang')
sfile = self.get_source_file(resource)
lang_map = self.get_resource_lang_mapping(resource)
host = self.get_resource_host(resource)
logger.debug("Language mapping is: %s" % lang_map)
if mode is None:
mode = self._get_option(resource, 'mode')
self.url_info = {
'host': host,
'project': project_slug,
'resource': resource_slug
}
logger.debug("URL data are: %s" % self.url_info)
stats = self._get_stats_for_resource()
try:
file_filter = self.config.get(resource, 'file_filter')
except configparser.NoOptionError:
file_filter = None
# Pull source file
pull_languages = set([])
new_translations = set([])
if pseudo:
pseudo_file = self._get_pseudo_file(
slang, resource, file_filter
)
if self._should_download(slang, stats, local_file=pseudo_file):
logger.info("Pulling pseudo file for resource %s (%s)." % (
resource,
color_text(pseudo_file, "RED")
))
self._download_pseudo(
project_slug, resource_slug, pseudo_file
)
if not languages:
continue
if fetchall:
new_translations = self._new_translations_to_add(
files, slang, lang_map, stats, force
)
if new_translations:
msg = "New translations found for the following languages: %s"
logger.info(msg % ', '.join(new_translations))
existing, new = self._languages_to_pull(
languages, files, lang_map, stats, force
)
pull_languages |= existing
new_translations |= new
logger.debug("Adding to new translations: %s" % new)
if fetchsource:
if sfile and slang not in pull_languages:
pull_languages.add(slang)
elif slang not in new_translations:
new_translations.add(slang)
if pull_languages:
logger.debug("Pulling languages for: %s" % pull_languages)
msg = "Pulling translations for resource %s (source: %s)"
logger.info(msg % (resource, sfile))
for lang in pull_languages:
local_lang = lang
if lang in list(lang_map.values()):
remote_lang = lang_map.flip[lang]
else:
remote_lang = lang
if languages and lang not in pull_languages:
logger.debug("Skipping language %s" % lang)
continue
if lang != slang:
local_file = files.get(lang, None) or files[lang_map[lang]]
else:
local_file = sfile
logger.debug("Using file %s" % local_file)
kwargs = {
'lang': remote_lang,
'stats': stats,
'local_file': local_file,
'force': force,
'mode': mode,
}
if not self._should_update_translation(**kwargs):
msg = "Skipping '%s' translation (file: %s)."
logger.info(
msg % (color_text(remote_lang, "RED"), local_file)
)
continue
if not overwrite:
local_file = ("%s.new" % local_file)
logger.warning(
" -> %s: %s" % (color_text(remote_lang, "RED"), local_file)
)
try:
r, charset = self.do_url_request(url, language=remote_lang)
except Exception as e:
if isinstance(e, SSLError) or not skip:
raise
else:
logger.error(e)
continue
base_dir = os.path.split(local_file)[0]
mkdir_p(base_dir)
fd = open(local_file, 'wb')
fd.write(r.encode(charset))
fd.close()
if new_translations:
msg = "Pulling new translations for resource %s (source: %s)"
logger.info(msg % (resource, sfile))
for lang in new_translations:
if lang in list(lang_map.keys()):
local_lang = lang_map[lang]
else:
local_lang = lang
remote_lang = lang
if file_filter:
local_file = os.path.relpath(
os.path.join(
self.root, native_path(
file_filter.replace('<lang>', local_lang)
)
), os.curdir
)
else:
trans_dir = os.path.join(self.root, ".tx", resource)
if not os.path.exists(trans_dir):
os.mkdir(trans_dir)
local_file = os.path.relpath(os.path.join(trans_dir, '%s_translation' %
local_lang, os.curdir))
if lang != slang:
satisfies_min = self._satisfies_min_translated(
stats[remote_lang], mode
)
if not satisfies_min:
msg = "Skipping language %s due to used options."
logger.info(msg % lang)
continue
logger.warning(
" -> %s: %s" % (color_text(remote_lang, "RED"), local_file)
)
r, charset = self.do_url_request(url, language=remote_lang)
base_dir = os.path.split(local_file)[0]
mkdir_p(base_dir)
fd = open(local_file, 'wb')
fd.write(r.encode(charset))
fd.close()
def push(self, source=False, translations=False, force=False, resources=[], languages=[],
skip=False, no_interactive=False):
"""
Push all the resources
"""
resource_list = self.get_chosen_resources(resources)
self.skip = skip
self.force = force
for resource in resource_list:
push_languages = []
project_slug, resource_slug = resource.split('.', 1)
files = self.get_resource_files(resource)
slang = self.get_resource_option(resource, 'source_lang')
sfile = self.get_source_file(resource)
lang_map = self.get_resource_lang_mapping(resource)
host = self.get_resource_host(resource)
logger.debug("Language mapping is: %s" % lang_map)
logger.debug("Using host %s" % host)
self.url_info = {
'host': host,
'project': project_slug,
'resource': resource_slug
}
logger.info("Pushing translations for resource %s:" % resource)
stats = self._get_stats_for_resource()
if force and not no_interactive:
answer = input("Warning: By using --force, the uploaded"
" files will overwrite remote translations, even if they"
" are newer than your uploaded files.\nAre you sure you"
" want to continue? [y/N] ")
if not answer in ["", 'Y', 'y', "yes", 'YES']:
return
if source:
if sfile is None:
logger.error("You don't seem to have a proper source file"
" mapping for resource %s. Try without the --source"
" option or set a source file first and then try again." %
resource)
continue
# Push source file
try:
logger.warning("Pushing source file (%s)" % sfile)
if not self._resource_exists(stats):
logger.info("Resource does not exist. Creating...")
fileinfo = "%s;%s" % (resource_slug, slang)
filename = self.get_full_path(sfile)
self._create_resource(resource, project_slug, fileinfo, filename)
self.do_url_request(
'push_source', multipart=True, method="PUT",
files=[(
"%s;%s" % (resource_slug, slang)
, self.get_full_path(sfile)
)],
)
except Exception as e:
if isinstance(e, SSLError) or not skip:
raise
else:
logger.error(e)
else:
try:
self.do_url_request('resource_details')
except Exception as e:
if isinstance(e, SSLError):
raise
code = getattr(e, 'code', None)
if code == 404:
msg = "Resource %s doesn't exist on the server."
logger.error(msg % resource)
continue
if translations:
# Check if given language codes exist
if not languages:
push_languages = list(files.keys())
else:
push_languages = []
f_langs = list(files.keys())
for l in languages:
if l in list(lang_map.keys()):
l = lang_map[l]
push_languages.append(l)
if l not in f_langs:
msg = "Warning: No mapping found for language code '%s'."
logger.error(msg % color_text(l,"RED"))
logger.debug("Languages to push are %s" % push_languages)
# Push translation files one by one
for lang in push_languages:
local_lang = lang
if lang in list(lang_map.values()):
remote_lang = lang_map.flip[lang]
else:
remote_lang = lang
try:
local_file = files[local_lang]
except KeyError as e:
msg = "No translation file found for language code '%s' in resource '%s'"
logger.error(msg % (color_text(local_lang, "RED"), resource_slug))
if not skip:
raise e
continue
kwargs = {
'lang': remote_lang,
'stats': stats,
'local_file': local_file,
'force': force,
}
if not self._should_push_translation(**kwargs):
msg = "Skipping '%s' translation (file: %s)."
logger.info(msg % (color_text(lang, "RED"), local_file))
continue
msg = "Pushing '%s' translations (file: %s)"
logger.warning(
msg % (color_text(remote_lang, "RED"), local_file)
)
try:
self.do_url_request(
'push_translation', multipart=True, method='PUT',
files=[(
"%s;%s" % (resource_slug, remote_lang),
self.get_full_path(local_file)
)], language=remote_lang
)
logger.debug("Translation %s pushed." % remote_lang)
except HttpNotFound:
if not source:
logger.error("Resource hasn't been created. Try pushing source file.")
except Exception as e:
if isinstance(e, SSLError) or not skip:
raise
else:
logger.error(e)
def delete(self, resources=[], languages=[], skip=False, force=False):
"""Delete translations."""
resource_list = self.get_chosen_resources(resources)
self.skip = skip
self.force = force
if not languages:
delete_func = self._delete_resource
else:
delete_func = self._delete_translations
for resource in resource_list:
project_slug, resource_slug = resource.split('.', 1)
host = self.get_resource_host(resource)
self.url_info = {
'host': host,
'project': project_slug,
'resource': resource_slug
}
logger.debug("URL data are: %s" % self.url_info)
json, _ = self.do_url_request('project_details', project=self)
project_details = parse_json(json)
teams = project_details['teams']
stats = self._get_stats_for_resource()
delete_func(project_details, resource, stats, languages)
def _delete_resource(self, project_details, resource, stats, *args):
"""Delete a resource from Transifex."""
project_slug, resource_slug = resource.split('.', 1)
project_resource_slugs = [
r['slug'] for r in project_details['resources']
]
logger.info("Deleting resource %s:" % resource)
if resource_slug not in project_resource_slugs:
if not self.skip:
msg = "Skipping: %s : Resource does not exist."
logger.info(msg % resource)
return
if not self.force:
slang = self.get_resource_option(resource, 'source_lang')
for language in stats:
if language == slang:
continue
if int(stats[language]['translated_entities']) > 0:
msg = (
"Skipping: %s : Unable to delete resource because it "
"has a not empty %s translation.\nPlease use -f or "
"--force option to delete this resource."
)
logger.info(msg % (resource, language))
return
try:
self.do_url_request('delete_resource', method="DELETE")
self.config.remove_section(resource)
self.save()
msg = "Deleted resource %s of project %s."
logger.info(msg % (resource_slug, project_slug))
except Exception as e:
msg = "Unable to delete resource %s of project %s."
logger.error(msg % (resource_slug, project_slug))
if isinstance(e, SSLError) or not self.skip:
raise
def _delete_translations(self, project_details, resource, stats, languages):
"""Delete the specified translations for the specified resource."""
logger.info("Deleting translations from resource %s:" % resource)
for language in languages:
self._delete_translation(project_details, resource, stats, language)
def _delete_translation(self, project_details, resource, stats, language):
"""Delete a specific translation from the specified resource."""
project_slug, resource_slug = resource.split('.', 1)
if language not in stats:
if not self.skip:
msg = "Skipping %s: Translation does not exist."
logger.warning(msg % (language))
return
if not self.force:
teams = project_details['teams']
if language in teams:
msg = (
"Skipping %s: Unable to delete translation because it is "
"associated with a team.\nPlease use -f or --force option "
"to delete this translation."
)
logger.warning(msg % language)
return
if int(stats[language]['translated_entities']) > 0:
msg = (
"Skipping %s: Unable to delete translation because it "
"is not empty.\nPlease use -f or --force option to delete "
"this translation."
)
logger.warning(msg % language)
return
try:
self.do_url_request(
'delete_translation', language=language, method="DELETE"
)
msg = "Deleted language %s from resource %s of project %s."
logger.info(msg % (language, resource_slug, project_slug))
except Exception as e:
msg = "Unable to delete translation %s"
logger.error(msg % language)
if isinstance(e, SSLError) or not self.skip:
raise
def do_url_request(self, api_call, multipart=False, data=None,
files=[], method="GET", **kwargs):
"""
Issues a url request.
"""
# Read the credentials from the config file (.transifexrc)
host = self.url_info['host']
try:
username = self.txrc.get(host, 'username')
passwd = self.txrc.get(host, 'password')
token = self.txrc.get(host, 'token')
hostname = self.txrc.get(host, 'hostname')
except configparser.NoSectionError:
raise Exception("No user credentials found for host %s. Edit"
" ~/.transifexrc and add the appropriate info in there." %
host)
# Create the Url
kwargs['hostname'] = hostname
kwargs.update(self.url_info)
url = API_URLS[api_call] % kwargs
if multipart:
for info, filename in files:
#FIXME: It works because we only pass to files argument
#only one item
name = os.path.basename(filename)
data = {
"resource": info.split(';')[0],
"language": info.split(';')[1],
"uploaded_file": (name, open(filename, 'rb').read())
}
return make_request(method, hostname, url, username, passwd, data)
def _should_update_translation(self, lang, stats, local_file, force=False,
mode=None):
"""Whether a translation should be udpated from Transifex.
We use the following criteria for that:
- If user requested to force the download.
- If language exists in Transifex.
- If the local file is older than the Transifex's file.
- If the user requested a x% completion.
Args:
lang: The language code to check.
stats: The (global) statistics object.
local_file: The local translation file.
force: A boolean flag.
mode: The mode for the translation.
Returns:
True or False.
"""
return self._should_download(lang, stats, local_file, force)
def _should_add_translation(self, lang, stats, force=False, mode=None):
"""Whether a translation should be added from Transifex.
We use the following criteria for that:
- If user requested to force the download.
- If language exists in Transifex.
- If the user requested a x% completion.
Args:
lang: The language code to check.
stats: The (global) statistics object.
force: A boolean flag.
mode: The mode for the translation.
Returns:
True or False.
"""
return self._should_download(lang, stats, None, force)
def _should_download(self, lang, stats, local_file=None, force=False,
mode=None):
"""Return whether a translation should be downloaded.
If local_file is None, skip the timestamps check (the file does
not exist locally).
"""
try:
lang_stats = stats[lang]
except KeyError as e:
logger.debug("No lang %s in statistics" % lang)
return False
satisfies_min = self._satisfies_min_translated(lang_stats, mode)
if not satisfies_min:
return False
if force:
logger.debug("Downloading translation due to -f")
return True
if local_file is not None:
remote_update = self._extract_updated(lang_stats)
if not self._remote_is_newer(remote_update, local_file):
logger.debug("Local is newer than remote for lang %s" % lang)
return False
return True
def _should_push_translation(self, lang, stats, local_file, force=False):
"""Return whether a local translation file should be
pushed to Trasnifex.
We use the following criteria for that:
- If user requested to force the upload.
- If language exists in Transifex.
- If local file is younger than the remote file.
Args:
lang: The language code to check.
stats: The (global) statistics object.
local_file: The local translation file.
force: A boolean flag.
Returns:
True or False.
"""
if force:
logger.debug("Push translation due to -f.")
return True
try:
lang_stats = stats[lang]
except KeyError as e:
logger.debug("Language %s does not exist in Transifex." % lang)
return True
if local_file is not None:
remote_update = self._extract_updated(lang_stats)
if self._remote_is_newer(remote_update, local_file):
msg = "Remote translation is newer than local file for lang %s"
logger.debug(msg % lang)
return False
return True
def _generate_timestamp(self, update_datetime):
"""Generate a UNIX timestamp from the argument.
Args:
update_datetime: The datetime in the format used by Transifex.
Returns:
A float, representing the timestamp that corresponds to the
argument.
"""
time_format = "%Y-%m-%d %H:%M:%S"
return time.mktime(
datetime.datetime(
*time.strptime(update_datetime, time_format)[0:5]
).utctimetuple()
)
def _get_time_of_local_file(self, path):
"""Get the modified time of the path_.
Args:
path: The path we want the mtime for.
Returns:
The time as a timestamp or None, if the file does not exist
"""
if not os.path.exists(path):
return None
return time.mktime(time.gmtime(os.path.getmtime(path)))
def _satisfies_min_translated(self, stats, mode=None):
"""Check whether a translation fulfills the filter used for
minimum translated percentage.
Args:
perc: The current translation percentage.
Returns:
True or False
"""
cur = self._extract_completed(stats, mode)
option_name = 'minimum_perc'
if self.minimum_perc is not None:
minimum_percent = self.minimum_perc
else:
global_minimum = int(
self.get_resource_option('main', option_name) or 0
)
resource_minimum = int(
self.get_resource_option(
self.resource, option_name
) or global_minimum
)
minimum_percent = resource_minimum
return cur >= minimum_percent
def _remote_is_newer(self, remote_updated, local_file):
"""Check whether the remote translation is newer that the local file.
Args:
remote_updated: The date and time the translation was last
updated remotely.
local_file: The local file.
Returns:
True or False.
"""