""" Diff Service Handles parsing and applying SEARCH/REPLACE diff blocks from AI responses """ import re import logging from typing import List, Dict, Tuple, Optional from diff_match_patch import diff_match_patch logger = logging.getLogger(__name__) # Diff block delimiters SEARCH_START = "<<<<<<< SEARCH" DIVIDER = "=======" REPLACE_END = ">>>>>>> REPLACE" class DiffBlock: """Represents a single SEARCH/REPLACE diff block""" def __init__(self, original: str, updated: str): self.original = original self.updated = updated class DiffService: """Service for parsing and applying diff blocks""" @staticmethod def parse_diff_blocks(content: str) -> List[DiffBlock]: """ Parse AI response content for SEARCH/REPLACE blocks. Args: content: The AI response content Returns: List of DiffBlock objects """ blocks = [] lines = content.split('\n') i = 0 while i < len(lines): # Trim lines for comparison to handle potential trailing whitespace from AI if lines[i].strip() == SEARCH_START: original_lines = [] updated_lines = [] i += 1 # Move past SEARCH_START # Collect original lines until DIVIDER while i < len(lines) and lines[i].strip() != DIVIDER: original_lines.append(lines[i]) i += 1 if i >= len(lines) or lines[i].strip() != DIVIDER: logger.warning( f"Malformed diff block: Missing or misplaced '{DIVIDER}' after SEARCH block. " f"Block content: {chr(10).join(original_lines)}" ) # Skip to next potential block start while i < len(lines) and SEARCH_START not in lines[i]: i += 1 continue i += 1 # Move past DIVIDER # Collect updated lines until REPLACE_END while i < len(lines) and lines[i].strip() != REPLACE_END: updated_lines.append(lines[i]) i += 1 if i >= len(lines) or lines[i].strip() != REPLACE_END: logger.warning( f"Malformed diff block: Missing or misplaced '{REPLACE_END}' after REPLACE block. " f"Block content: {chr(10).join(updated_lines)}" ) # Skip to next potential block start while i < len(lines) and SEARCH_START not in lines[i]: i += 1 continue # Join lines back together original_text = '\n'.join(original_lines) updated_text = '\n'.join(updated_lines) blocks.append(DiffBlock(original_text, updated_text)) i += 1 return blocks @staticmethod def apply_single_diff(current_html: str, original_block: str, updated_block: str) -> Optional[str]: """ Apply a single diff block to the current HTML content using diff-match-patch for fuzzy matching. Args: current_html: The current HTML content original_block: The content from the SEARCH block updated_block: The content from the REPLACE block Returns: The updated HTML content, or None if application failed """ dmp = diff_match_patch() # Handle potential trailing newline inconsistencies (same as DeepSite) search_block = original_block if not original_block.endswith('\n') and (original_block + '\n') in current_html: search_block = original_block + '\n' # Match replacement block newline handling replace_block = updated_block if search_block.endswith('\n') and updated_block and not updated_block.endswith('\n'): replace_block = updated_block + '\n' # For deletion (empty updated block) if search_block.endswith('\n') and not updated_block: replace_block = '' # First try: Exact string match if search_block in current_html: logger.info("Found exact match, using direct string replacement.") index = current_html.index(search_block) return current_html[:index] + replace_block + current_html[index + len(search_block):] # Second try: Normalize whitespace and try again normalized_search = ' '.join(search_block.split()) normalized_html = ' '.join(current_html.split()) if normalized_search in normalized_html: logger.info("Found normalized match, using direct string replacement.") # Find the original position in the non-normalized HTML search_lines = search_block.split('\n') if len(search_lines) > 0: first_line = search_lines[0].strip() if first_line in current_html: start_index = current_html.find(first_line) # Find the end of the search block end_index = start_index for line in search_lines[1:]: next_line_start = current_html.find(line.strip(), end_index) if next_line_start != -1: end_index = next_line_start + len(line.strip()) else: break else: # Found all lines, replace the block return current_html[:start_index] + replace_block + current_html[end_index:] # Third try: Use diff-match-patch with more lenient settings patches = dmp.patch_make(search_block, replace_block) # More lenient thresholds for better matching dmp.Match_Threshold = 0.5 dmp.Patch_DeleteThreshold = 0.5 dmp.Match_Distance = 1000 # Increase search distance patched_html, results = dmp.patch_apply(patches, current_html) # Check if the patch applied successfully if all(results): logger.info("Successfully applied patch using diff-match-patch.") return patched_html else: logger.warning(f"Patch application failed using diff-match-patch. Results: {results}") logger.error(f"Search block not found in HTML. Search block:\n{repr(search_block)}") return None @staticmethod def apply_diffs(original_html: str, ai_response_content: str) -> Tuple[str, bool, str]: """ Apply all parsed diff blocks sequentially to the original HTML. Args: original_html: The initial HTML content ai_response_content: The full response from the AI containing diff blocks Returns: Tuple of (final_html, success, error_message) """ diff_blocks = DiffService.parse_diff_blocks(ai_response_content) if not diff_blocks: logger.warning("AI response did not contain valid SEARCH/REPLACE blocks.") # Check if AI tried to use the format but failed if any(marker in ai_response_content for marker in [SEARCH_START, DIVIDER, REPLACE_END]): error_msg = "AI response contained malformed or unparseable diff blocks. Could not apply changes." return original_html, False, error_msg # Check if AI gave full HTML instead trimmed_response = ai_response_content.strip().lower() if trimmed_response.startswith('