Coverage for pds_crawler/report.py: 62%

1# -*- coding: utf-8 -*- 

2# pds-crawler - ETL to index PDS data to pdssp 

3# Copyright (C) 2023 - CNES (Jean-Christophe Malapert for Pôle Surfaces Planétaires) 

4# This file is part of pds-crawler <https://github.com/pdssp/pds_crawler> 

5# SPDX-License-Identifier: LGPL-3.0-or-later 

6import logging 

7import os 

8from dataclasses import asdict 

9from dataclasses import dataclass 

10from json import dumps 

11from typing import Any 

12from typing import IO 

13from typing import Optional 

14 

15from .load import Database 

16from .utils import Observer 

17 

18logger = logging.getLogger(__name__) 

19 

20 

21@dataclass(frozen=True) 

22class MessageModel: 

23 resource: str 

24 explanation: Any 

25 

26 @classmethod 

27 def from_dict(cls, env): 

28 """Converts dictionary to MessageModel object. 

29 

30 Args: 

31 env (dict): Dictionary containing resource and explanation fields. 

32 

33 Returns: 

34 MessageModel: Object containing the fields. 

35 """ 

36 return cls(**{k: v for k, v in env.items()}) 

37 

38 @property 

39 def __dict__(self): 

40 """Returns the dictionary representation of the object. 

41 

42 Returns: 

43 dict: Dictionary representation of the object. 

44 """ 

45 return asdict(self) 

46 

47 @property 

48 def json(self): 

49 """Returns the json representation of the object. 

50 

51 Returns: 

52 str: Json representation of the object. 

53 """ 

54 return dumps(self.__dict__, indent=None) 

55 

56 def __repr__(self) -> str: 

57 """Returns the printable representation of the object. 

58 

59 Returns: 

60 str: Printable representation of the object. 

61 """ 

62 return f"MessageModel({self.resource}, {self.explanation})" 

63 

64 

65class CrawlerReport(Observer): 

66 def __init__(self, db: Database): 

67 """Initializes the CrawlerReport object. 

68 

69 Args: 

70 db (Database): Database object. 

71 """ 

72 self.__db = db 

73 self.__name: str = "default_report" 

74 self.__file: Optional[IO] = None 

75 

76 @property 

77 def name(self): 

78 """Returns the name of the report. 

79 

80 Returns: 

81 str: The name of the report. 

82 """ 

83 return self.__name 

84 

85 @name.setter 

86 def name(self, value: str): 

87 """Sets the name of the report. 

88 

89 Args: 

90 value (str): The name of the report. 

91 """ 

92 self.__name = value 

93 

94 def start_report(self, mode: str = "a"): 

95 """Starts the report and writes the header. 

96 

97 Args: 

98 mode (str, optional): File opening mode. Defaults to "a". 

99 """ 

100 self.__file = open( 

101 os.path.join(self.__db.base_directory, self.name), mode=mode 

102 ) 

103 self._write_header() 

104 

105 def _write_header(self): 

106 """Writes the header for the report.""" 

107 header = """ 

108 | Resource | Explanation | 

109 | ----------------|-------------------| 

110 """ 

111 if self.__file is None: 

112 logger.error(header) 

113 else: 

114 self.__file.write(header) 

115 

116 def close_report(self): 

117 """Closes the report file.""" 

118 if self.__file is not None: 

119 self.__file.flush() 

120 self.__file.close() 

121 del self.__file 

122 self.__file = None 

123 

124 def notify(self, observable, *args, **kwargs): 

125 """Receives the notification. 

126 

127 Args: 

128 observable ([type]): Observable 

129 """ 

130 message: MessageModel = args[0] 

131 if self.__file is None: 

132 logger.error(f"| {message.resource} | {message.explanation} |") 

133 else: 

134 self.__file.write( 

135 f"| {message.resource} | {message.explanation} |\n" 

136 )