Coverage for app/venv/lib/python3.14/site-packages/weblate/machinery/aws.py: 48%

52 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-10-07 07:15 +0000

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

3# SPDX-License-Identifier: GPL-3.0-or-later 

4from __future__ import annotations 

5 

6import operator 

7from typing import TYPE_CHECKING, ClassVar 

8 

9from django.utils.functional import cached_property 

10 

11from .base import ( 

12 GlossaryDoesNotExistError, 

13 GlossaryMachineTranslationMixin, 

14) 

15from .forms import AWSMachineryForm 

16 

17if TYPE_CHECKING: 17 ↛ 18line 17 didn't jump to line 18 because the condition on line 17 was never true

18 from .base import ( 

19 DownloadTranslations, 

20 ) 

21 

22 

23class AWSTranslation(GlossaryMachineTranslationMixin): 

24 """AWS machine translation.""" 

25 

26 name = "Amazon Translate" 

27 max_score = 88 

28 language_map: ClassVar[dict[str, str]] = { 

29 "zh_Hant": "zh-TW", 

30 "zh_Hans": "zh", 

31 "nb_NO": "no", 

32 } 

33 settings_form = AWSMachineryForm 

34 

35 # glossary name must match the pattern ^([A-Za-z0-9-]_?)+$ 

36 glossary_name_format = ( 

37 "weblate_-_{project}_-_{source_language}_-_{target_language}_-_{checksum}" 

38 ) 

39 

40 glossary_count_limit = 100 

41 

42 @classmethod 

43 def get_identifier(cls) -> str: 

44 return "aws" 

45 

46 @cached_property 

47 def client(self): 

48 import boto3 

49 

50 return boto3.client( 

51 service_name="translate", 

52 region_name=self.settings["region"], 

53 aws_access_key_id=self.settings["key"], 

54 aws_secret_access_key=self.settings["secret"], 

55 ) 

56 

57 def map_language_code(self, code): 

58 """Convert language to service specific code.""" 

59 return super().map_language_code(code).replace("_", "-").split("@")[0] 

60 

61 def download_languages(self): 

62 """List of supported languages.""" 

63 result = self.client.list_languages() 

64 return [lang["LanguageCode"] for lang in result["Languages"]] 

65 

66 def download_translations( 

67 self, 

68 source_language, 

69 target_language, 

70 text: str, 

71 unit, 

72 user, 

73 threshold: int = 75, 

74 ) -> DownloadTranslations: 

75 params = { 

76 "Text": text, 

77 "SourceLanguageCode": source_language, 

78 "TargetLanguageCode": target_language, 

79 } 

80 

81 glossary_name: str | None = self.get_glossary_id( 

82 source_language, target_language, unit 

83 ) 

84 if glossary_name: 

85 params["TerminologyNames"] = [glossary_name] 

86 

87 response = self.client.translate_text(**params) 

88 yield { 

89 "text": response["TranslatedText"], 

90 "quality": self.max_score, 

91 "service": self.name, 

92 "source": text, 

93 } 

94 

95 def create_glossary( 

96 self, source_language: str, target_language: str, name: str, tsv: str 

97 ) -> None: 

98 """Create glossary in the service.""" 

99 # add header with source and target languages 

100 tsv = f"{source_language}\t{target_language}\n{tsv}" 

101 

102 # AWS gracefully handles duplicate entries by merging them 

103 self.client.import_terminology( 

104 Name=name, 

105 MergeStrategy="OVERWRITE", 

106 TerminologyData={ 

107 "File": tsv.encode(), 

108 "Format": "TSV", 

109 "Directionality": "UNI", 

110 }, 

111 ) 

112 

113 def is_glossary_supported(self, source_language: str, target_language: str) -> bool: 

114 """Check whether given language combination is supported for glossary.""" 

115 return self.is_supported(source_language, target_language) 

116 

117 def list_glossaries(self) -> dict[str, str]: 

118 """List all glossaries from service.""" 

119 result = ( 

120 self.client.get_paginator("list_terminologies") 

121 .paginate() 

122 .build_full_result() 

123 ) 

124 return { 

125 terminology["Name"]: terminology["Name"] 

126 for terminology in result["TerminologyPropertiesList"] 

127 } 

128 

129 def delete_glossary(self, glossary_id: str) -> None: 

130 """Delete a single glossary from service.""" 

131 try: 

132 self.client.delete_terminology(Name=glossary_id) 

133 except self.client.exceptions.ResourceNotFoundException as error: 

134 raise GlossaryDoesNotExistError from error 

135 

136 def delete_oldest_glossary(self) -> None: 

137 """Delete oldest glossary if any.""" 

138 result = ( 

139 self.client.get_paginator("list_terminologies") 

140 .paginate() 

141 .build_full_result() 

142 ) 

143 glossaries = sorted( 

144 result["TerminologyPropertiesList"], 

145 key=operator.itemgetter("CreatedAt"), 

146 ) 

147 if glossaries: 

148 self.delete_glossary(glossaries[0]["Name"])