""" Token Merge Module Merge tokens from multiple sources with conflict resolution strategies. """ from dataclasses import dataclass, field from datetime import datetime from enum import Enum from typing import List, Dict, Optional, Callable, Tuple from .base import DesignToken, TokenCollection, TokenCategory class MergeStrategy(str, Enum): """Token merge conflict resolution strategies.""" # Simple strategies FIRST = "first" # Keep first occurrence LAST = "last" # Keep last occurrence (override) ERROR = "error" # Raise error on conflict # Value-based strategies PREFER_FIGMA = "prefer_figma" # Prefer Figma source PREFER_CODE = "prefer_code" # Prefer code sources (CSS, SCSS) PREFER_SPECIFIC = "prefer_specific" # Prefer more specific values # Smart strategies MERGE_METADATA = "merge_metadata" # Merge metadata, keep latest value INTERACTIVE = "interactive" # Require user decision @dataclass class MergeConflict: """Represents a token name conflict during merge.""" token_name: str existing: DesignToken incoming: DesignToken resolution: Optional[str] = None resolved_token: Optional[DesignToken] = None @dataclass class MergeResult: """Result of a token merge operation.""" collection: TokenCollection conflicts: List[MergeConflict] = field(default_factory=list) stats: Dict[str, int] = field(default_factory=dict) warnings: List[str] = field(default_factory=list) def __post_init__(self): if not self.stats: self.stats = { "total_tokens": 0, "new_tokens": 0, "updated_tokens": 0, "conflicts_resolved": 0, "conflicts_unresolved": 0, } class TokenMerger: """ Merge multiple token collections with conflict resolution. Usage: merger = TokenMerger(strategy=MergeStrategy.LAST) result = merger.merge([collection1, collection2, collection3]) """ # Source priority for PREFER_* strategies SOURCE_PRIORITY = { "figma": 100, "css": 80, "scss": 80, "tailwind": 70, "json": 60, } def __init__( self, strategy: MergeStrategy = MergeStrategy.LAST, custom_resolver: Optional[Callable[[MergeConflict], DesignToken]] = None ): """ Initialize merger. Args: strategy: Default conflict resolution strategy custom_resolver: Optional custom conflict resolver function """ self.strategy = strategy self.custom_resolver = custom_resolver def merge( self, collections: List[TokenCollection], normalize_names: bool = True ) -> MergeResult: """ Merge multiple token collections. Args: collections: List of TokenCollections to merge normalize_names: Whether to normalize token names before merging Returns: MergeResult with merged collection and conflict information """ result = MergeResult( collection=TokenCollection( name="Merged Tokens", sources=[], ) ) # Track tokens by normalized name tokens_by_name: Dict[str, DesignToken] = {} for collection in collections: result.collection.sources.extend(collection.sources) for token in collection.tokens: # Normalize name if requested name = token.normalize_name() if normalize_names else token.name if name in tokens_by_name: # Conflict detected existing = tokens_by_name[name] conflict = MergeConflict( token_name=name, existing=existing, incoming=token, ) # Resolve conflict resolved = self._resolve_conflict(conflict) conflict.resolved_token = resolved if resolved: tokens_by_name[name] = resolved result.stats["conflicts_resolved"] += 1 result.stats["updated_tokens"] += 1 else: result.stats["conflicts_unresolved"] += 1 result.warnings.append( f"Unresolved conflict for token: {name}" ) result.conflicts.append(conflict) else: # New token tokens_by_name[name] = token result.stats["new_tokens"] += 1 # Build final collection result.collection.tokens = list(tokens_by_name.values()) result.stats["total_tokens"] = len(result.collection.tokens) return result def _resolve_conflict(self, conflict: MergeConflict) -> Optional[DesignToken]: """Resolve a single conflict based on strategy.""" # Try custom resolver first if self.custom_resolver: return self.custom_resolver(conflict) # Apply strategy if self.strategy == MergeStrategy.FIRST: conflict.resolution = "kept_first" return conflict.existing elif self.strategy == MergeStrategy.LAST: conflict.resolution = "used_last" return self._update_token(conflict.incoming, conflict.existing) elif self.strategy == MergeStrategy.ERROR: conflict.resolution = "error" raise ValueError( f"Token conflict: {conflict.token_name} " f"(existing: {conflict.existing.source}, " f"incoming: {conflict.incoming.source})" ) elif self.strategy == MergeStrategy.PREFER_FIGMA: return self._prefer_source(conflict, "figma") elif self.strategy == MergeStrategy.PREFER_CODE: return self._prefer_code_source(conflict) elif self.strategy == MergeStrategy.PREFER_SPECIFIC: return self._prefer_specific_value(conflict) elif self.strategy == MergeStrategy.MERGE_METADATA: return self._merge_metadata(conflict) elif self.strategy == MergeStrategy.INTERACTIVE: # For interactive, we can't resolve automatically conflict.resolution = "needs_input" return None return conflict.incoming def _update_token( self, source: DesignToken, base: DesignToken ) -> DesignToken: """Create updated token preserving some base metadata.""" # Create new token with source's value but enhanced metadata updated = DesignToken( name=source.name, value=source.value, type=source.type, description=source.description or base.description, source=source.source, source_file=source.source_file, source_line=source.source_line, original_name=source.original_name, original_value=source.original_value, category=source.category, tags=list(set(source.tags + base.tags)), deprecated=source.deprecated or base.deprecated, deprecated_message=source.deprecated_message or base.deprecated_message, version=source.version, updated_at=datetime.now(), extensions={**base.extensions, **source.extensions}, ) return updated def _prefer_source( self, conflict: MergeConflict, preferred_source: str ) -> DesignToken: """Prefer token from specific source type.""" existing_source = conflict.existing.source.split(':')[0] incoming_source = conflict.incoming.source.split(':')[0] if incoming_source == preferred_source: conflict.resolution = f"preferred_{preferred_source}" return self._update_token(conflict.incoming, conflict.existing) elif existing_source == preferred_source: conflict.resolution = f"kept_{preferred_source}" return conflict.existing else: # Neither is preferred, use last conflict.resolution = "fallback_last" return self._update_token(conflict.incoming, conflict.existing) def _prefer_code_source(self, conflict: MergeConflict) -> DesignToken: """Prefer code sources (CSS, SCSS) over design sources.""" code_sources = {"css", "scss", "tailwind"} existing_source = conflict.existing.source.split(':')[0] incoming_source = conflict.incoming.source.split(':')[0] existing_is_code = existing_source in code_sources incoming_is_code = incoming_source in code_sources if incoming_is_code and not existing_is_code: conflict.resolution = "preferred_code" return self._update_token(conflict.incoming, conflict.existing) elif existing_is_code and not incoming_is_code: conflict.resolution = "kept_code" return conflict.existing else: # Both or neither are code, use priority return self._prefer_by_priority(conflict) def _prefer_by_priority(self, conflict: MergeConflict) -> DesignToken: """Choose based on source priority.""" existing_source = conflict.existing.source.split(':')[0] incoming_source = conflict.incoming.source.split(':')[0] existing_priority = self.SOURCE_PRIORITY.get(existing_source, 0) incoming_priority = self.SOURCE_PRIORITY.get(incoming_source, 0) if incoming_priority > existing_priority: conflict.resolution = "higher_priority" return self._update_token(conflict.incoming, conflict.existing) else: conflict.resolution = "kept_priority" return conflict.existing def _prefer_specific_value(self, conflict: MergeConflict) -> DesignToken: """Prefer more specific/concrete values.""" existing_value = str(conflict.existing.value).lower() incoming_value = str(conflict.incoming.value).lower() # Prefer concrete values over variables/references existing_is_var = existing_value.startswith('var(') or \ existing_value.startswith('$') or \ existing_value.startswith('{') incoming_is_var = incoming_value.startswith('var(') or \ incoming_value.startswith('$') or \ incoming_value.startswith('{') if incoming_is_var and not existing_is_var: conflict.resolution = "kept_concrete" return conflict.existing elif existing_is_var and not incoming_is_var: conflict.resolution = "preferred_concrete" return self._update_token(conflict.incoming, conflict.existing) # Prefer hex colors over named colors existing_is_hex = existing_value.startswith('#') incoming_is_hex = incoming_value.startswith('#') if incoming_is_hex and not existing_is_hex: conflict.resolution = "preferred_hex" return self._update_token(conflict.incoming, conflict.existing) elif existing_is_hex and not incoming_is_hex: conflict.resolution = "kept_hex" return conflict.existing # Default to last conflict.resolution = "fallback_last" return self._update_token(conflict.incoming, conflict.existing) def _merge_metadata(self, conflict: MergeConflict) -> DesignToken: """Merge metadata from both tokens, keep latest value.""" conflict.resolution = "merged_metadata" # Use incoming value but merge all metadata merged_tags = list(set( conflict.existing.tags + conflict.incoming.tags )) merged_extensions = { **conflict.existing.extensions, **conflict.incoming.extensions } # Track both sources merged_extensions['dss'] = merged_extensions.get('dss', {}) merged_extensions['dss']['previousSources'] = [ conflict.existing.source, conflict.incoming.source ] return DesignToken( name=conflict.incoming.name, value=conflict.incoming.value, type=conflict.incoming.type or conflict.existing.type, description=conflict.incoming.description or conflict.existing.description, source=conflict.incoming.source, source_file=conflict.incoming.source_file, source_line=conflict.incoming.source_line, original_name=conflict.incoming.original_name, original_value=conflict.incoming.original_value, category=conflict.incoming.category or conflict.existing.category, tags=merged_tags, deprecated=conflict.incoming.deprecated or conflict.existing.deprecated, deprecated_message=conflict.incoming.deprecated_message or conflict.existing.deprecated_message, version=conflict.incoming.version, updated_at=datetime.now(), extensions=merged_extensions, ) class TokenDiff: """ Compare two token collections and find differences. """ @staticmethod def diff( source: TokenCollection, target: TokenCollection ) -> Dict[str, List]: """ Compare two token collections. Returns: Dict with 'added', 'removed', 'changed', 'unchanged' lists """ source_by_name = {t.normalize_name(): t for t in source.tokens} target_by_name = {t.normalize_name(): t for t in target.tokens} source_names = set(source_by_name.keys()) target_names = set(target_by_name.keys()) result = { 'added': [], # In target but not source 'removed': [], # In source but not target 'changed': [], # In both but different value 'unchanged': [], # In both with same value } # Find added (in target, not in source) for name in target_names - source_names: result['added'].append(target_by_name[name]) # Find removed (in source, not in target) for name in source_names - target_names: result['removed'].append(source_by_name[name]) # Find changed/unchanged (in both) for name in source_names & target_names: source_token = source_by_name[name] target_token = target_by_name[name] if str(source_token.value) != str(target_token.value): result['changed'].append({ 'name': name, 'old_value': source_token.value, 'new_value': target_token.value, 'source_token': source_token, 'target_token': target_token, }) else: result['unchanged'].append(source_token) return result @staticmethod def summary(diff_result: Dict[str, List]) -> str: """Generate human-readable diff summary.""" lines = ["Token Diff Summary:", "=" * 40] if diff_result['added']: lines.append(f"\n+ Added ({len(diff_result['added'])}):") for token in diff_result['added'][:10]: lines.append(f" + {token.name}: {token.value}") if len(diff_result['added']) > 10: lines.append(f" ... and {len(diff_result['added']) - 10} more") if diff_result['removed']: lines.append(f"\n- Removed ({len(diff_result['removed'])}):") for token in diff_result['removed'][:10]: lines.append(f" - {token.name}: {token.value}") if len(diff_result['removed']) > 10: lines.append(f" ... and {len(diff_result['removed']) - 10} more") if diff_result['changed']: lines.append(f"\n~ Changed ({len(diff_result['changed'])}):") for change in diff_result['changed'][:10]: lines.append( f" ~ {change['name']}: {change['old_value']} → {change['new_value']}" ) if len(diff_result['changed']) > 10: lines.append(f" ... and {len(diff_result['changed']) - 10} more") lines.append(f"\n Unchanged: {len(diff_result['unchanged'])}") return "\n".join(lines)