Coverage for sources/agentsmgr/instructions.py: 10%

1# vim: set filetype=python fileencoding=utf-8: 

2# -*- coding: utf-8 -*- 

3 

4#============================================================================# 

5# # 

6# Licensed under the Apache License, Version 2.0 (the "License"); # 

7# you may not use this file except in compliance with the License. # 

8# You may obtain a copy of the License at # 

9# # 

10# http://www.apache.org/licenses/LICENSE-2.0 # 

11# # 

12# Unless required by applicable law or agreed to in writing, software # 

13# distributed under the License is distributed on an "AS IS" BASIS, # 

14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # 

15# See the License for the specific language governing permissions and # 

16# limitations under the License. # 

17# # 

18#============================================================================# 

19 

20 

21''' Instruction file population from Git sources. 

22 

23 This module provides functionality for downloading and preprocessing 

24 documentation instruction files from Git repositories, supporting 

25 flexible configuration through Copier answers. 

26''' 

27 

28 

29from . import __ 

30from . import exceptions as _exceptions 

31from . import sources as _sources 

32 

33 

34InstructionSourceConfiguration: __.typx.TypeAlias = ( 

35 __.cabc.Mapping[ str, __.typx.Any ] ) 

36FilePreprocessingConfiguration: __.typx.TypeAlias = ( 

37 __.cabc.Mapping[ str, __.typx.Any ] ) 

38 

39 

40_scribe = __.provide_scribe( __name__ ) 

41 

42 

43def populate_instructions( 

44 sources_configuration: __.cabc.Sequence[ InstructionSourceConfiguration ], 

45 target: __.Path, 

46 tag_prefix: __.Absential[ str ] = __.absent, 

47 simulate: bool = False, 

48) -> tuple[ int, int ]: 

49 ''' Populates instruction files from configured Git sources. 

50 

51 For each source configuration, resolves the Git source, filters 

52 files by configured patterns, applies preprocessing (such as 

53 header stripping), and writes results to the target directory. 

54 

55 Returns tuple of (files_attempted, files_written) across all 

56 sources. 

57 ''' 

58 files_attempted = 0 

59 files_written = 0 

60 for source_config in sources_configuration: 

61 try: source_spec = source_config[ 'source' ] 

62 except KeyError as exception: 

63 raise _exceptions.InstructionSourceFieldAbsence( ) from exception 

64 files_config = source_config.get( 'files', { '*.rst': { } } ) 

65 if not isinstance( files_config, __.cabc.Mapping ): 

66 raise _exceptions.InstructionFilesConfigurationInvalidity( ) 

67 files_mapping: __.cabc.Mapping[ 

68 str, FilePreprocessingConfiguration ] = ( 

69 __.typx.cast( 

70 __.cabc.Mapping[ str, FilePreprocessingConfiguration ], 

71 files_config ) ) 

72 _scribe.info( f"Resolving instruction source: {source_spec}" ) 

73 try: 

74 source_location = _sources.resolve_source_location( 

75 source_spec, tag_prefix ) 

76 except Exception as exception: 

77 _scribe.warning( 

78 f"Failed to resolve instruction source '{source_spec}': " 

79 f"{exception}" ) 

80 continue 

81 attempted, written = _populate_instructions_from_location( 

82 source_location, target, files_mapping, simulate ) 

83 files_attempted += attempted 

84 files_written += written 

85 return ( files_attempted, files_written ) 

86 

87 

88def _populate_instructions_from_location( 

89 source_location: __.Path, 

90 target: __.Path, 

91 files_configuration: __.cabc.Mapping[ 

92 str, FilePreprocessingConfiguration ], 

93 simulate: bool, 

94) -> tuple[ int, int ]: 

95 ''' Populates instructions from resolved source location. 

96 

97 Filters files by configured patterns, applies preprocessing, and 

98 writes to target directory. Returns tuple of (attempted, written). 

99 ''' 

100 files_attempted = 0 

101 files_written = 0 

102 if not source_location.exists( ): 

103 _scribe.warning( 

104 f"Instruction source location does not exist: {source_location}" ) 

105 return ( files_attempted, files_written ) 

106 for pattern, preprocessing_config in files_configuration.items( ): 

107 config: FilePreprocessingConfiguration = ( 

108 preprocessing_config 

109 if isinstance( preprocessing_config, __.cabc.Mapping ) 

110 else { } ) 

111 for file_path in source_location.rglob( pattern ): 

112 if not file_path.is_file( ): continue 

113 files_attempted += 1 

114 was_written = _process_and_write_instruction_file( 

115 file_path, target, config, simulate ) 

116 if was_written: files_written += 1 

117 return ( files_attempted, files_written ) 

118 

119 

120def _process_and_write_instruction_file( 

121 source_file: __.Path, 

122 target_directory: __.Path, 

123 preprocessing_configuration: FilePreprocessingConfiguration, 

124 simulate: bool, 

125) -> bool: 

126 ''' Processes and writes instruction file to target directory. 

127 

128 Reads source file, applies configured preprocessing (such as 

129 header stripping), and writes to target directory. Returns True 

130 if file was written, False otherwise. 

131 ''' 

132 try: content = source_file.read_text( encoding = 'utf-8' ) 

133 except ( OSError, IOError ) as exception: 

134 _scribe.warning( 

135 f"Failed to read instruction file '{source_file}': {exception}" ) 

136 return False 

137 processed_content = _preprocess_content( 

138 content, preprocessing_configuration ) 

139 target_file = target_directory / source_file.name 

140 if simulate: 

141 try: relative_path = target_file.relative_to( __.Path.cwd( ) ) 

142 except ValueError: relative_path = target_file 

143 _scribe.info( f"Would write instruction file: {relative_path}" ) 

144 return True 

145 try: 

146 target_directory.mkdir( parents = True, exist_ok = True ) 

147 target_file.write_text( processed_content, encoding = 'utf-8' ) 

148 except ( OSError, IOError ) as exception: 

149 _scribe.warning( 

150 f"Failed to write instruction file '{target_file}': {exception}" ) 

151 return False 

152 else: 

153 try: relative_path = target_file.relative_to( __.Path.cwd( ) ) 

154 except ValueError: relative_path = target_file 

155 _scribe.debug( f"Wrote instruction file: {relative_path}" ) 

156 return True 

157 

158 

159def _preprocess_content( 

160 content: str, 

161 configuration: FilePreprocessingConfiguration, 

162) -> str: 

163 ''' Applies configured preprocessing transforms to content. 

164 

165 Supports header stripping via strip_header_lines configuration. 

166 Additional preprocessing operations can be added here. 

167 ''' 

168 if not configuration: return content 

169 strip_lines = configuration.get( 'strip_header_lines' ) 

170 if strip_lines is None: return content 

171 if not isinstance( strip_lines, int ): 

172 _scribe.warning( 

173 f"Invalid strip_header_lines value (expected int): {strip_lines}" ) 

174 return content 

175 if strip_lines <= 0: return content 

176 lines = content.splitlines( keepends = True ) 

177 if len( lines ) <= strip_lines: return '' 

178 return ''.join( lines[ strip_lines: ] )