Coverage for src/dmxfy/client/client.py: 100%

1import http.client 

2import json 

3from typing import Any 

4 

5from dmxfy.config.config import Config 

6from dmxfy.languages.languages import LANGUAGES 

7from dmxfy.exceptions.exceptions import TranslationError 

8 

9 

10class TranslationClient: 

11 """Client for handling translations with the DashScope API""" 

12 

13 def __init__(self) -> None: 

14 self.config = Config() 

15 self._conn: http.client.HTTPSConnection = http.client.HTTPSConnection( 

16 "dashscope.aliyuncs.com" 

17 ) 

18 

19 def translate( 

20 self, text: str, source_lang: str, target_lang: str 

21 ) -> str: 

22 """ 

23 Translate text using the DashScope API 

24 

25 Args: 

26 text: The text to translate 

27 source_lang: The source language name (in Chinese) 

28 target_lang: The target language name (in Chinese) 

29 

30 Returns: 

31 The translated text 

32 

33 Raises: 

34 TranslationError: If there's an error during translation 

35 """ 

36 if not text.strip(): 

37 return "请输入要翻译的文本" 

38 

39 try: 

40 # Prepare the request data 

41 json_data = { 

42 "model": "qwen-mt-turbo", 

43 "messages": [ 

44 { 

45 "role": "user", 

46 "content": text, 

47 }, 

48 ], 

49 "translation_options": { 

50 "source_lang": LANGUAGES.get(source_lang, "auto"), 

51 "target_lang": LANGUAGES.get(target_lang, "English"), 

52 }, 

53 } 

54 

55 # Make the request 

56 self._conn.request( 

57 "POST", 

58 "/compatible-mode/v1/chat/completions", 

59 json.dumps(json_data), 

60 self.config.headers, 

61 ) 

62 

63 # Get and parse the response 

64 response = self._conn.getresponse() 

65 resp_str = response.read().decode("utf-8") 

66 resp_json: dict[str, Any] = json.loads(resp_str) 

67 

68 return str(resp_json["choices"][0]["message"]["content"]) 

69 

70 except Exception as e: 

71 raise TranslationError(f"Translation failed: {str(e)}") from e