21.88%

1"""Weave the content of the changes data file into the output structure (for now LaTeX).""" 

2 

3import os 

4import pathlib 

5from typing import Generator, Union, no_type_check 

6 

7import liitos.gather as gat 

8import liitos.template as tpl 

9import liitos.tools as too 

10from liitos import ENCODING, LOG_SEPARATOR, PathLike, log 

11 

12 

13PUBLISHER_TEMPLATE = os.getenv('LIITOS_PUBLISHER_TEMPLATE', '') 

14PUBLISHER_TEMPLATE_IS_EXTERNAL = bool(PUBLISHER_TEMPLATE) 

15if not PUBLISHER_TEMPLATE: 15 ↛ 18line 15 didn't jump to line 18 because the condition on line 15 was always true

16 PUBLISHER_TEMPLATE = 'templates/publisher.tex.in' 

17 

18PUBLISHER_PATH = pathlib.Path('render/pdf/publisher.tex') 

19TOKEN = r'THE.ISSUE.CODE & THE.REVISION.CODE & THE.AUTHOR.NAME & THE.DESCRIPTION \\' # nosec B105 

20DEFAULT_REVISION = '00' 

21ROW_TEMPLATE = r'issue & revision & author & summary \\' 

22GLUE = '\n\\hline\n' 

23JSON_CHANNEL = 'json' 

24YAML_CHANNEL = 'yaml' 

25COLUMNS_EXPECTED = sorted(['author', 'date', 'issue', 'revision', 'summary']) 

26COLUMNS_MINIMAL = sorted(['author', 'issue', 'summary']) 

27CUT_MARKER_CHANGES_TOP = '% |-- changes - cut - marker - top -->' 

28CUT_MARKER_CHANGES_BOTTOM = '% <-- changes - cut - marker - bottom --|' 

29CUT_MARKER_NOTICES_TOP = '% |-- notices - cut - marker - top -->' 

30CUT_MARKER_NOTICES_BOTTOM = '% <-- notices - cut - marker - bottom --|' 

31TOKEN_ADJUSTED_PUSHDOWN = r'\AdustedPushdown' # nosec B105 

32DEFAULT_ADJUSTED_PUSHDOWN_VALUE = 14 

33 

34NL = '\n' 

35 

36 

37def get_layout(layout_path: PathLike, target_key: str, facet_key: str) -> dict[str, dict[str, dict[str, bool]]]: 

38 """Boolean layout decisions on bookmatter and publisher page conten. 

39 

40 Deprecated as the known use cases evolved into a different direction ... 

41 """ 

42 layout = {'layout': {'global': {'has_approvals': True, 'has_changes': True, 'has_notices': True}}} 

43 if layout_path: 

44 log.info(f'loading layout from {layout_path=} for changes and notices') 

45 return gat.load_layout(facet_key, target_key, layout_path)[0] # type: ignore 

46 

47 log.info('using default layout for approvals') 

48 return layout 

49 

50 

51def derive_model(model_path: PathLike) -> tuple[str, list[str]]: 

52 """Derive the model as channel type and column model from the given path.""" 

53 channel = JSON_CHANNEL if str(model_path).endswith('.json') else YAML_CHANNEL 

54 columns_expected = COLUMNS_MINIMAL if channel == JSON_CHANNEL else COLUMNS_EXPECTED 

55 

56 return channel, columns_expected 

57 

58 

59def columns_are_present(columns_present: list[str], columns_expected: list[str]) -> bool: 

60 """Ensure the needed columns are present.""" 

61 return all(column in columns_expected for column in columns_present) 

62 

63 

64@no_type_check 

65def normalize(changes: object, channel: str, columns_expected: list[str]) -> list[dict[str, str]]: 

66 """Normalize the channel specific topology of the model into a logical model. 

67 

68 On error an empty logical model is returned. 

69 """ 

70 if channel == JSON_CHANNEL: 

71 for slot, change in enumerate(changes[0]['changes'], start=1): 

72 if not set(columns_expected).issubset(set(change)): 

73 log.error('unexpected column model!') 

74 log.error(f'- expected: ({columns_expected})') 

75 log.error(f'- minimal: ({COLUMNS_MINIMAL})') 

76 log.error(f'- but found: ({change}) for entry #{slot}') 

77 return [] 

78 

79 if channel == YAML_CHANNEL: 

80 for slot, change in enumerate(changes[0]['changes'], start=1): 

81 model = sorted(change.keys()) 

82 if not set(COLUMNS_MINIMAL).issubset(set(model)): 

83 log.error('unexpected column model!') 

84 log.error(f'- expected: ({columns_expected})') 

85 log.error(f'- minimal: ({COLUMNS_MINIMAL})') 

86 log.error(f'- but found: ({model}) in slot {slot}') 

87 return [] 

88 

89 model = [] 

90 if channel == JSON_CHANNEL: 

91 for change in changes[0]['changes']: 

92 issue, author, summary = change['issue'], change['author'], change['summary'] 

93 revision = change.get('revision', DEFAULT_REVISION) 

94 model.append({'issue': issue, 'revision': revision, 'author': author, 'summary': summary}) 

95 return model 

96 

97 for change in changes[0]['changes']: 

98 author = change['author'] 

99 issue = change['issue'] 

100 revision = change.get('revision', DEFAULT_REVISION) 

101 summary = change['summary'] 

102 model.append({'issue': issue, 'revision': revision, 'author': author, 'summary': summary}) 

103 

104 return model 

105 

106 

107def adjust_pushdown_gen(text_lines: list[str], pushdown: float) -> Generator[str, None, None]: 

108 """Update the pushdown line filtering the incoming lines.""" 

109 for line in text_lines: 

110 if TOKEN_ADJUSTED_PUSHDOWN in line: 

111 line = line.replace(TOKEN_ADJUSTED_PUSHDOWN, f'{pushdown}em') 

112 log.info(f'set adjusted pushdown value {pushdown}em') 

113 yield line 

114 

115 

116def weave( 

117 doc_root: Union[str, pathlib.Path], 

118 structure_name: str, 

119 target_key: str, 

120 facet_key: str, 

121 options: dict[str, Union[bool, str]], 

122) -> int: 

123 """Later alligator.""" 

124 log.info(LOG_SEPARATOR) 

125 log.info('entered changes weave function ...') 

126 structure, asset_map = gat.prelude( 

127 doc_root=doc_root, structure_name=structure_name, target_key=target_key, facet_key=facet_key, command='changes' 

128 ) 

129 

130 layout_path = asset_map[target_key][facet_key].get(gat.KEY_LAYOUT, '') 

131 layout = get_layout(layout_path, target_key=target_key, facet_key=facet_key) 

132 log.info(f'{layout=}') 

133 

134 log.info(LOG_SEPARATOR) 

135 changes_path = asset_map[target_key][facet_key][gat.KEY_CHANGES] 

136 channel, columns_expected = derive_model(changes_path) 

137 log.info(f'detected changes channel ({channel}) weaving in from ({changes_path})') 

138 

139 log.info(f'loading changes from {changes_path=}') 

140 changes = gat.load_changes(facet_key, target_key, changes_path) 

141 log.info(f'{changes=}') 

142 

143 log.info(LOG_SEPARATOR) 

144 log.info('plausibility tests for changes ...') 

145 

146 logical_model = normalize(changes, channel=channel, columns_expected=columns_expected) 

147 

148 rows = [ 

149 ROW_TEMPLATE.replace('issue', kv['issue']) 

150 .replace('revision', kv['revision']) 

151 .replace('author', kv['author']) 

152 .replace('summary', kv['summary']) 

153 for kv in logical_model 

154 ] 

155 

156 pushdown = DEFAULT_ADJUSTED_PUSHDOWN_VALUE 

157 log.info(f'calculated adjusted pushdown to be {pushdown}em') 

158 

159 publisher_template = tpl.load_resource(PUBLISHER_TEMPLATE, PUBLISHER_TEMPLATE_IS_EXTERNAL) 

160 lines = [line.rstrip() for line in publisher_template.split(NL)] 

161 

162 if any(TOKEN_ADJUSTED_PUSHDOWN in line for line in lines): 

163 lines = list(adjust_pushdown_gen(lines, pushdown)) 

164 else: 

165 log.error(f'token ({TOKEN_ADJUSTED_PUSHDOWN}) not found - template mismatch') 

166 

167 if not layout['layout']['global']['has_changes']: 

168 log.info('removing changes from document layout') 

169 lines = list(too.remove_target_region_gen(lines, CUT_MARKER_CHANGES_TOP, CUT_MARKER_CHANGES_BOTTOM)) 

170 

171 if not layout['layout']['global']['has_notices']: 

172 log.info('removing notices from document layout') 

173 lines = list(too.remove_target_region_gen(lines, CUT_MARKER_NOTICES_TOP, CUT_MARKER_NOTICES_BOTTOM)) 

174 

175 log.info(LOG_SEPARATOR) 

176 log.info('weaving in the changes from {changes_path} ...') 

177 for n, line in enumerate(lines): 

178 if line.strip() == TOKEN: 

179 lines[n] = GLUE.join(rows) 

180 break 

181 if lines[-1]: 

182 lines.append('\n') 

183 with open(PUBLISHER_PATH, 'wt', encoding=ENCODING) as handle: 

184 handle.write('\n'.join(lines)) 

185 log.info(LOG_SEPARATOR) 

186 

187 return 0