Coverage for pyPhotoAlbum/merge_manager.py: 72%
202 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-20 12:55 +0000
1"""
2Merge manager for handling project merge conflicts
4This module provides functionality for:
5- Detecting when two projects should be merged vs. concatenated
6- Finding conflicts between two project versions
7- Resolving conflicts based on user input or automatic strategies
8"""
10import copy
11from typing import Dict, Any, List, Optional, Tuple
12from enum import Enum
13from dataclasses import dataclass
14from datetime import datetime, timezone
17class ConflictType(Enum):
18 """Types of merge conflicts"""
20 # Page-level conflicts
21 PAGE_MODIFIED_BOTH = "page_modified_both" # Page modified in both versions
22 PAGE_DELETED_ONE = "page_deleted_one" # Page deleted in one version, modified in other
23 PAGE_ADDED_BOTH = "page_added_both" # Same page number added in both (rare)
25 # Element-level conflicts
26 ELEMENT_MODIFIED_BOTH = "element_modified_both" # Element modified in both versions
27 ELEMENT_DELETED_ONE = "element_deleted_one" # Element deleted in one, modified in other
29 # Project-level conflicts
30 SETTINGS_MODIFIED_BOTH = "settings_modified_both" # Project settings changed in both
33class MergeStrategy(Enum):
34 """Automatic merge resolution strategies"""
36 LATEST_WINS = "latest_wins" # Most recent last_modified wins
37 OURS = "ours" # Always use our version
38 THEIRS = "theirs" # Always use their version
39 MANUAL = "manual" # Require manual resolution
42@dataclass
43class ConflictInfo:
44 """Information about a single merge conflict"""
46 conflict_type: ConflictType
47 page_uuid: Optional[str] # UUID of the page (if page-level conflict)
48 element_uuid: Optional[str] # UUID of the element (if element-level conflict)
49 our_version: Any # Our version of the conflicted item
50 their_version: Any # Their version of the conflicted item
51 description: str # Human-readable description
54class MergeManager:
55 """Manages merge operations between two project versions"""
57 def __init__(self):
58 self.conflicts: List[ConflictInfo] = []
60 def should_merge_projects(self, project_a_data: Dict[str, Any], project_b_data: Dict[str, Any]) -> bool:
61 """
62 Determine if two projects should be merged or concatenated.
64 Projects with the same project_id should be merged (conflict resolution).
65 Projects with different project_ids should be concatenated (combine content).
67 Args:
68 project_a_data: First project's serialized data
69 project_b_data: Second project's serialized data
71 Returns:
72 True if projects should be merged, False if concatenated
73 """
74 project_a_id = project_a_data.get("project_id")
75 project_b_id = project_b_data.get("project_id")
77 # If either project lacks a project_id (v2.0 or earlier), assume different projects
78 if not project_a_id or not project_b_id:
79 print("MergeManager: One or both projects lack project_id, assuming concatenation")
80 return False
82 return bool(project_a_id == project_b_id)
84 def detect_conflicts(
85 self, our_project_data: Dict[str, Any], their_project_data: Dict[str, Any]
86 ) -> List[ConflictInfo]:
87 """
88 Detect conflicts between two versions of the same project.
90 Args:
91 our_project_data: Our version of the project (serialized)
92 their_project_data: Their version of the project (serialized)
94 Returns:
95 List of conflicts found
96 """
97 self.conflicts = []
99 # Detect project-level conflicts
100 self._detect_project_settings_conflicts(our_project_data, their_project_data)
102 # Detect page-level conflicts
103 self._detect_page_conflicts(our_project_data, their_project_data)
105 return self.conflicts
107 def _detect_project_settings_conflicts(self, our_data: Dict[str, Any], their_data: Dict[str, Any]):
108 """Detect conflicts in project-level settings."""
109 # Settings that can conflict
110 settings_keys = [
111 "name",
112 "page_size_mm",
113 "working_dpi",
114 "export_dpi",
115 "has_cover",
116 "paper_thickness_mm",
117 "cover_bleed_mm",
118 "binding_type",
119 ]
121 our_modified = our_data.get("last_modified")
122 their_modified = their_data.get("last_modified")
124 for key in settings_keys:
125 our_value = our_data.get(key)
126 their_value = their_data.get(key)
128 # If values differ, it's a conflict
129 if our_value != their_value:
130 self.conflicts.append(
131 ConflictInfo(
132 conflict_type=ConflictType.SETTINGS_MODIFIED_BOTH,
133 page_uuid=None,
134 element_uuid=None,
135 our_version={key: our_value, "last_modified": our_modified},
136 their_version={key: their_value, "last_modified": their_modified},
137 description=f"Project setting '{key}' modified in both versions",
138 )
139 )
141 def _detect_page_conflicts(self, our_data: Dict[str, Any], their_data: Dict[str, Any]):
142 """Detect conflicts at page level."""
143 our_pages = {page["uuid"]: page for page in our_data.get("pages", [])}
144 their_pages = {page["uuid"]: page for page in their_data.get("pages", [])}
146 # Check all pages that exist in our version
147 for page_uuid, our_page in our_pages.items():
148 their_page = their_pages.get(page_uuid)
150 if their_page is None:
151 # Page exists in ours but not theirs - check if deleted
152 if our_page.get("deleted"):
153 continue # Both deleted, no conflict
154 # We have it, they don't (might have deleted it)
155 # This could be a conflict if we modified it after they deleted it
156 continue
158 # Page exists in both - check for modifications
159 self._detect_page_modification_conflicts(page_uuid, our_page, their_page)
161 # Check for pages that exist only in their version
162 for page_uuid, their_page in their_pages.items():
163 if page_uuid not in our_pages:
164 # They have a page we don't - this is fine, add it
165 # Unless we deleted it
166 pass
168 def _detect_page_modification_conflicts(self, page_uuid: str, our_page: Dict[str, Any], their_page: Dict[str, Any]):
169 """Detect conflicts in a specific page."""
170 our_modified = our_page.get("last_modified")
171 their_modified = their_page.get("last_modified")
173 # Check if both deleted
174 if our_page.get("deleted") and their_page.get("deleted"):
175 return # No conflict
177 # Check if one deleted, one modified
178 if our_page.get("deleted") != their_page.get("deleted"):
179 self.conflicts.append(
180 ConflictInfo(
181 conflict_type=ConflictType.PAGE_DELETED_ONE,
182 page_uuid=page_uuid,
183 element_uuid=None,
184 our_version=our_page,
185 their_version=their_page,
186 description=f"Page deleted in one version but modified in the other",
187 )
188 )
189 return
191 # Check page-level properties
192 page_props = ["page_number", "is_cover", "is_double_spread"]
193 page_modified = False
194 for prop in page_props:
195 if our_page.get(prop) != their_page.get(prop):
196 page_modified = True
197 break
199 # Only flag as conflict if properties differ AND timestamps are identical
200 # (See element conflict detection for detailed explanation of this strategy)
201 if page_modified and our_modified == their_modified:
202 self.conflicts.append(
203 ConflictInfo(
204 conflict_type=ConflictType.PAGE_MODIFIED_BOTH,
205 page_uuid=page_uuid,
206 element_uuid=None,
207 our_version=our_page,
208 their_version=their_page,
209 description=f"Page properties modified with same timestamp (possible conflict)",
210 )
211 )
213 # Check element-level conflicts
214 self._detect_element_conflicts(page_uuid, our_page, their_page)
216 def _detect_element_conflicts(self, page_uuid: str, our_page: Dict[str, Any], their_page: Dict[str, Any]):
217 """Detect conflicts in elements within a page."""
218 our_layout = our_page.get("layout", {})
219 their_layout = their_page.get("layout", {})
221 our_elements = {elem["uuid"]: elem for elem in our_layout.get("elements", [])}
222 their_elements = {elem["uuid"]: elem for elem in their_layout.get("elements", [])}
224 # Check all elements in our version
225 for elem_uuid, our_elem in our_elements.items():
226 their_elem = their_elements.get(elem_uuid)
228 if their_elem is None:
229 # Element exists in ours but not theirs
230 if our_elem.get("deleted"):
231 continue # Both deleted, no conflict
232 # We have it, they don't
233 continue
235 # Element exists in both - check for modifications
236 self._detect_element_modification_conflicts(page_uuid, elem_uuid, our_elem, their_elem)
238 def _detect_element_modification_conflicts(
239 self, page_uuid: str, elem_uuid: str, our_elem: Dict[str, Any], their_elem: Dict[str, Any]
240 ):
241 """Detect conflicts in a specific element."""
242 our_modified = our_elem.get("last_modified")
243 their_modified = their_elem.get("last_modified")
245 # Check if both deleted
246 if our_elem.get("deleted") and their_elem.get("deleted"):
247 return # No conflict
249 # Check if one deleted, one modified
250 if our_elem.get("deleted") != their_elem.get("deleted"):
251 self.conflicts.append(
252 ConflictInfo(
253 conflict_type=ConflictType.ELEMENT_DELETED_ONE,
254 page_uuid=page_uuid,
255 element_uuid=elem_uuid,
256 our_version=our_elem,
257 their_version=their_elem,
258 description=f"Element deleted in one version but modified in the other",
259 )
260 )
261 return
263 # Check element properties
264 elem_props = ["position", "size", "rotation", "z_index"]
266 # Add type-specific properties
267 elem_type = our_elem.get("type")
268 if elem_type == "image":
269 elem_props.extend(["image_path", "crop_info", "pil_rotation_90"])
270 elif elem_type == "textbox":
271 elem_props.extend(["text_content", "font_settings", "alignment"])
273 # Check if any properties differ
274 props_modified = False
275 for prop in elem_props:
276 if our_elem.get(prop) != their_elem.get(prop):
277 props_modified = True
278 break
280 # Without a 3-way merge (base version), we cannot reliably detect if BOTH versions
281 # modified an element vs only ONE version modifying it.
282 #
283 # Strategy: Only flag as conflict when we have strong evidence of concurrent modification:
284 # - Properties differ AND timestamps are identical → suspicious, possible conflict
285 # - Properties differ AND timestamps differ → one version modified it, auto-merge by timestamp
286 #
287 # If timestamps differ, _merge_non_conflicting_changes will handle it by using the newer version.
288 if props_modified and our_modified == their_modified:
289 # Properties differ but timestamps match - this is unusual and might indicate
290 # that both versions modified it at exactly the same time, or there's data corruption.
291 # Flag as conflict to be safe.
292 self.conflicts.append(
293 ConflictInfo(
294 conflict_type=ConflictType.ELEMENT_MODIFIED_BOTH,
295 page_uuid=page_uuid,
296 element_uuid=elem_uuid,
297 our_version=our_elem,
298 their_version=their_elem,
299 description=f"Element modified with same timestamp (possible conflict)",
300 )
301 )
303 # Note: If timestamps differ, we assume one version modified it and the other didn't.
304 # The _merge_non_conflicting_changes method will automatically use the newer version.
306 def auto_resolve_conflicts(self, strategy: MergeStrategy = MergeStrategy.LATEST_WINS) -> Dict[int, str]:
307 """
308 Automatically resolve conflicts based on a strategy.
310 Args:
311 strategy: The resolution strategy to use
313 Returns:
314 Dictionary mapping conflict index to resolution choice ("ours" or "theirs")
315 """
316 resolutions = {}
318 for i, conflict in enumerate(self.conflicts):
319 if strategy == MergeStrategy.LATEST_WINS:
320 # Compare timestamps
321 our_modified = self._get_timestamp(conflict.our_version)
322 their_modified = self._get_timestamp(conflict.their_version)
324 if our_modified and their_modified:
325 resolutions[i] = "ours" if our_modified >= their_modified else "theirs"
326 else:
327 resolutions[i] = "ours" # Default to ours if timestamps missing
329 elif strategy == MergeStrategy.OURS:
330 resolutions[i] = "ours"
332 elif strategy == MergeStrategy.THEIRS:
333 resolutions[i] = "theirs"
335 # MANUAL strategy leaves resolutions empty
337 return resolutions
339 def _get_timestamp(self, version_data: Any) -> Optional[str]:
340 """Extract timestamp from version data."""
341 if isinstance(version_data, dict):
342 return version_data.get("last_modified")
343 return None
345 def apply_resolutions(
346 self, our_project_data: Dict[str, Any], their_project_data: Dict[str, Any], resolutions: Dict[int, str]
347 ) -> Dict[str, Any]:
348 """
349 Apply conflict resolutions to create merged project.
351 Args:
352 our_project_data: Our version of the project
353 their_project_data: Their version of the project
354 resolutions: Dictionary mapping conflict index to choice ("ours" or "theirs")
356 Returns:
357 Merged project data
358 """
359 # Start with a copy of our project
360 merged_data = copy.deepcopy(our_project_data)
362 # Apply resolutions
363 for conflict_idx, choice in resolutions.items():
364 if conflict_idx >= len(self.conflicts):
365 continue
367 conflict = self.conflicts[conflict_idx]
369 if choice == "theirs":
370 # Apply their version
371 self._apply_their_version(merged_data, conflict)
372 # If choice is "ours", no need to do anything
374 # Add pages/elements from their version that we don't have
375 self._merge_non_conflicting_changes(merged_data, their_project_data)
377 return merged_data
379 def _apply_their_version(self, merged_data: Dict[str, Any], conflict: ConflictInfo):
380 """Apply their version for a specific conflict."""
381 if conflict.conflict_type == ConflictType.SETTINGS_MODIFIED_BOTH:
382 # Update project setting
383 for key, value in conflict.their_version.items():
384 if key != "last_modified":
385 merged_data[key] = value
387 elif conflict.conflict_type in [ConflictType.PAGE_MODIFIED_BOTH, ConflictType.PAGE_DELETED_ONE]:
388 # Replace entire page
389 for i, page in enumerate(merged_data.get("pages", [])):
390 if page.get("uuid") == conflict.page_uuid:
391 merged_data["pages"][i] = conflict.their_version
392 break
394 elif conflict.conflict_type in [ConflictType.ELEMENT_MODIFIED_BOTH, ConflictType.ELEMENT_DELETED_ONE]:
395 # Replace element within page
396 for page in merged_data.get("pages", []):
397 if page.get("uuid") == conflict.page_uuid:
398 layout = page.get("layout", {})
399 for i, elem in enumerate(layout.get("elements", [])):
400 if elem.get("uuid") == conflict.element_uuid:
401 layout["elements"][i] = conflict.their_version
402 break
403 break
405 def _merge_non_conflicting_changes(self, merged_data: Dict[str, Any], their_data: Dict[str, Any]):
406 """Add non-conflicting pages and elements from their version."""
407 self._add_missing_pages(merged_data, their_data)
408 self._merge_page_elements(merged_data, their_data)
410 def _add_missing_pages(self, merged_data: Dict[str, Any], their_data: Dict[str, Any]):
411 """Add pages that exist only in their version."""
412 our_page_uuids = {page["uuid"] for page in merged_data.get("pages", [])}
414 for their_page in their_data.get("pages", []):
415 if their_page["uuid"] not in our_page_uuids:
416 merged_data["pages"].append(their_page)
418 def _merge_page_elements(self, merged_data: Dict[str, Any], their_data: Dict[str, Any]):
419 """For pages that exist in both versions, merge their elements."""
420 their_pages = {page["uuid"]: page for page in their_data.get("pages", [])}
422 for our_page in merged_data.get("pages", []):
423 their_page = their_pages.get(our_page["uuid"])
424 if not their_page:
425 continue
427 our_elements = {elem["uuid"]: elem for elem in our_page.get("layout", {}).get("elements", [])}
429 for their_elem in their_page.get("layout", {}).get("elements", []):
430 self._merge_element(
431 our_page=our_page, page_uuid=our_page["uuid"], their_elem=their_elem, our_elements=our_elements
432 )
434 def _merge_element(
435 self, our_page: Dict[str, Any], page_uuid: str, their_elem: Dict[str, Any], our_elements: Dict[str, Any]
436 ):
437 """Merge a single element from their version into our page."""
438 elem_uuid = their_elem["uuid"]
440 # Add new elements that we don't have
441 if elem_uuid not in our_elements:
442 our_page["layout"]["elements"].append(their_elem)
443 return
445 # Element exists in both - check if already resolved as conflict
446 if self._is_element_in_conflict(elem_uuid, page_uuid):
447 return
449 # No conflict - use the more recently modified version
450 self._merge_by_timestamp(our_page, elem_uuid, their_elem, our_elements[elem_uuid])
452 def _is_element_in_conflict(self, elem_uuid: str, page_uuid: str) -> bool:
453 """Check if element was part of a conflict that was already resolved."""
454 return any(c.element_uuid == elem_uuid and c.page_uuid == page_uuid for c in self.conflicts)
456 def _merge_by_timestamp(
457 self, our_page: Dict[str, Any], elem_uuid: str, their_elem: Dict[str, Any], our_elem: Dict[str, Any]
458 ):
459 """Use the more recently modified version of an element."""
460 our_modified = our_elem.get("last_modified")
461 their_modified = their_elem.get("last_modified")
463 # Their version is newer
464 if not their_modified or (our_modified and their_modified <= our_modified):
465 return
467 # Replace with their newer version
468 for i, elem in enumerate(our_page["layout"]["elements"]):
469 if elem["uuid"] == elem_uuid:
470 our_page["layout"]["elements"][i] = their_elem
471 break
474def concatenate_projects(project_a_data: Dict[str, Any], project_b_data: Dict[str, Any]) -> Dict[str, Any]:
475 """
476 Concatenate two projects with different project_ids.
478 This combines the pages from both projects into a single project.
480 Args:
481 project_a_data: First project data
482 project_b_data: Second project data
484 Returns:
485 Combined project data
486 """
487 # Start with project A as base
488 merged_data = copy.deepcopy(project_a_data)
490 # Append all pages from project B
491 merged_data["pages"].extend(copy.deepcopy(project_b_data.get("pages", [])))
493 # Update project name to indicate merge
494 merged_data["name"] = f"{project_a_data.get('name', 'Untitled')} + {project_b_data.get('name', 'Untitled')}"
496 # Keep project A's ID and settings
497 # Update last_modified to now
498 merged_data["last_modified"] = datetime.now(timezone.utc).isoformat()
500 print(
501 f"Concatenated projects: {len(project_a_data.get('pages', []))} + {len(project_b_data.get('pages', []))} = {len(merged_data['pages'])} pages"
502 )
504 return merged_data