Coverage for src/prosemark/app/materialize_all_placeholders.py: 100%
88 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-09-30 23:09 +0000
1"""MaterializeAllPlaceholders use case for bulk conversion of placeholders to nodes."""
3import time
4from collections.abc import Callable
5from pathlib import Path
6from typing import TYPE_CHECKING
8from prosemark.domain.batch_materialize_result import BatchMaterializeResult
9from prosemark.domain.materialize_failure import MaterializeFailure
10from prosemark.domain.materialize_result import MaterializeResult
11from prosemark.domain.models import Binder, BinderItem
12from prosemark.domain.placeholder_summary import PlaceholderSummary
14if TYPE_CHECKING: # pragma: no cover
15 from prosemark.app.materialize_node import MaterializeNode
16 from prosemark.ports.binder_repo import BinderRepo
17 from prosemark.ports.clock import Clock
18 from prosemark.ports.id_generator import IdGenerator
19 from prosemark.ports.logger import Logger
20 from prosemark.ports.node_repo import NodeRepo
23class MaterializeAllPlaceholders:
24 """Materialize all placeholder items in a binder to actual content nodes."""
26 def __init__(
27 self,
28 *,
29 materialize_node_use_case: 'MaterializeNode',
30 binder_repo: 'BinderRepo',
31 node_repo: 'NodeRepo',
32 id_generator: 'IdGenerator',
33 clock: 'Clock',
34 logger: 'Logger',
35 ) -> None:
36 """Initialize the MaterializeAllPlaceholders use case.
38 Args:
39 materialize_node_use_case: Use case for individual node materialization
40 binder_repo: Repository for binder operations
41 node_repo: Repository for node operations
42 id_generator: Generator for unique node IDs
43 clock: Clock for timestamps
44 logger: Logger port
46 """
47 self.materialize_node_use_case = materialize_node_use_case
48 self.binder_repo = binder_repo
49 self.node_repo = node_repo
50 self.id_generator = id_generator
51 self.clock = clock
52 self.logger = logger
54 def execute(
55 self,
56 *,
57 binder: Binder | None = None,
58 project_path: Path | None = None,
59 progress_callback: Callable[[str], None] | None = None,
60 ) -> BatchMaterializeResult:
61 """Materialize all placeholders in the binder.
63 Args:
64 binder: Optional binder to process (if not provided, loads from repo)
65 project_path: Project directory path
66 progress_callback: Optional callback for progress reporting
68 Returns:
69 BatchMaterializeResult containing success/failure information
71 """
72 start_time = time.time()
73 project_path = project_path or Path.cwd()
75 # Load binder if not provided
76 if binder is None:
77 binder = self.binder_repo.load()
79 self.logger.info('Starting batch materialization of all placeholders')
81 # Discover all placeholders
82 placeholders = self._discover_placeholders(binder)
83 total_placeholders = len(placeholders)
85 self.logger.info('Found %d placeholders to materialize', total_placeholders)
86 if progress_callback:
87 progress_callback(f'Found {total_placeholders} placeholders to materialize...')
89 # If no placeholders, return early
90 if total_placeholders == 0:
91 execution_time = time.time() - start_time
92 return BatchMaterializeResult(
93 total_placeholders=0,
94 successful_materializations=[],
95 failed_materializations=[],
96 execution_time=execution_time,
97 )
99 # Process each placeholder
100 successful_materializations: list[MaterializeResult] = []
101 failed_materializations: list[MaterializeFailure] = []
103 for i, placeholder in enumerate(placeholders, 1):
104 try:
105 # Attempt materialization using existing use case
106 result = self._materialize_single_placeholder(placeholder=placeholder, project_path=project_path)
107 successful_materializations.append(result)
109 # Report progress
110 if progress_callback:
111 progress_callback(f"✓ Materialized '{result.display_title}' → {result.node_id.value}")
113 self.logger.info(
114 'Successfully materialized placeholder %d/%d: %s',
115 i,
116 total_placeholders,
117 placeholder.display_title,
118 )
120 except Exception as e:
121 # Create failure record
122 failure = MaterializeAllPlaceholders._create_failure_record(placeholder, e)
123 failed_materializations.append(failure)
125 # Report progress
126 if progress_callback:
127 progress_callback(f"✗ Failed to materialize '{placeholder.display_title}': {failure.error_message}")
129 self.logger.exception(
130 'Failed to materialize placeholder %d/%d: %s',
131 i,
132 total_placeholders,
133 placeholder.display_title,
134 )
136 # Check if we should stop the batch on critical errors
137 if failure.should_stop_batch:
138 self.logger.exception('Critical error encountered, stopping batch operation')
139 break
141 execution_time = time.time() - start_time
143 # Create final result
144 batch_result = BatchMaterializeResult(
145 total_placeholders=total_placeholders,
146 successful_materializations=successful_materializations,
147 failed_materializations=failed_materializations,
148 execution_time=execution_time,
149 )
151 self.logger.info(
152 'Batch materialization complete: %d successes, %d failures in %.2f seconds',
153 len(successful_materializations),
154 len(failed_materializations),
155 execution_time,
156 )
158 return batch_result
160 def _discover_placeholders(self, binder: Binder) -> list[PlaceholderSummary]:
161 """Discover all items in the binder hierarchy that need processing.
163 This includes both placeholders (items without node IDs) and materialized nodes
164 (items with node IDs that may have missing files). The enhanced MaterializeNode
165 can handle both cases appropriately.
167 Args:
168 binder: Binder to scan for items to process
170 Returns:
171 List of item summaries in hierarchical order
173 """
174 placeholders: list[PlaceholderSummary] = []
175 self._collect_placeholders_recursive(binder.roots, placeholders, parent_title=None, depth=0, position_path=[])
176 return placeholders
178 def _collect_placeholders_recursive(
179 self,
180 items: list[BinderItem],
181 placeholders: list[PlaceholderSummary],
182 parent_title: str | None,
183 depth: int,
184 position_path: list[int] | None = None,
185 ) -> None:
186 """Recursively collect all items that need processing from binder hierarchy.
188 Collects both placeholders and materialized nodes since the enhanced
189 MaterializeNode can handle both cases (creating new nodes or fixing missing files).
191 Args:
192 items: Current level items to process
193 placeholders: List to append discovered items to
194 parent_title: Title of parent item (if any)
195 depth: Current nesting depth
196 position_path: Hierarchical position path from root
198 """
199 for i, item in enumerate(items):
200 # Create position string based on index
201 position_path = position_path or []
202 position = '[' + ']['.join(str(idx) for idx in [*position_path, i]) + ']'
204 # Collect both placeholders (node_id is None) and materialized nodes (node_id exists)
205 # The enhanced MaterializeNode can handle both cases:
206 # - Placeholders: creates new nodes with both .md and .notes.md files
207 # - Materialized nodes: creates missing .notes.md files if needed
208 placeholder = PlaceholderSummary(
209 display_title=item.display_title,
210 position=position,
211 parent_title=parent_title,
212 depth=depth,
213 )
214 placeholders.append(placeholder)
216 # Recursively process children
217 if item.children:
218 self._collect_placeholders_recursive(
219 items=item.children,
220 placeholders=placeholders,
221 parent_title=item.display_title,
222 depth=depth + 1,
223 position_path=[*position_path, i],
224 )
226 def _materialize_single_placeholder(
227 self, *, placeholder: PlaceholderSummary, project_path: Path
228 ) -> MaterializeResult:
229 """Materialize a single placeholder using the existing MaterializeNode use case.
231 Args:
232 placeholder: Placeholder to materialize
233 project_path: Project directory path
235 Returns:
236 MaterializeResult with the outcome
238 Raises:
239 Various exceptions from the underlying materialization process
241 """
242 # Use the existing MaterializeNode use case
243 result = self.materialize_node_use_case.execute(title=placeholder.display_title, project_path=project_path)
245 # Create file paths
246 file_paths = [f'{result.node_id.value}.md', f'{result.node_id.value}.notes.md']
248 return MaterializeResult(
249 display_title=placeholder.display_title,
250 node_id=result.node_id,
251 file_paths=file_paths,
252 position=placeholder.position,
253 )
255 @staticmethod
256 def _create_failure_record(placeholder: PlaceholderSummary, error: Exception) -> MaterializeFailure:
257 """Create a MaterializeFailure record from an exception.
259 Args:
260 placeholder: Placeholder that failed to materialize
261 error: Exception that occurred during materialization
263 Returns:
264 MaterializeFailure record with categorized error information
266 """
267 # Categorize the error type based on exception
268 error_type = MaterializeAllPlaceholders._categorize_error(error)
269 error_message = str(error)
271 return MaterializeFailure(
272 display_title=placeholder.display_title,
273 error_type=error_type,
274 error_message=error_message,
275 position=placeholder.position,
276 )
278 @staticmethod
279 def _categorize_error(error: Exception) -> str:
280 """Categorize an exception into a known error type.
282 Args:
283 error: Exception to categorize
285 Returns:
286 Error type string from MaterializeFailure.VALID_ERROR_TYPES
288 """
289 error_type_name = type(error).__name__
291 # Map common exception types to our error categories
292 if 'FileSystem' in error_type_name or 'Permission' in error_type_name or 'OSError' in error_type_name:
293 return 'filesystem'
294 if 'Validation' in error_type_name or 'ValueError' in error_type_name:
295 return 'validation'
296 if 'AlreadyMaterialized' in error_type_name:
297 return 'already_materialized'
298 if 'Binder' in error_type_name or 'Integrity' in error_type_name:
299 return 'binder_integrity'
300 if 'UUID' in error_type_name or 'Id' in error_type_name:
301 return 'id_generation'
302 # Default to filesystem for unknown errors
303 return 'filesystem'