import csv import json import re import os from datetime import datetime import html class CSVGenerator: def __init__(self): pass def generate_csv_from_mapping(self, mapping_content, filename_prefix="mapping", json_data=None): """Generate CSV file from mapping document content""" try: # Generate filename timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") csv_filename = f"{filename_prefix}_{timestamp}.csv" csv_path = os.path.join("temp", csv_filename) # Ensure temp directory exists os.makedirs("temp", exist_ok=True) # Simple pipe-to-comma conversion csv_content = self._convert_table_to_csv(mapping_content) if not csv_content: return None, "No table data found to convert to CSV" # Write CSV file with open(csv_path, 'w', newline='', encoding='utf-8') as f: f.write(csv_content) return csv_path, csv_filename except Exception as e: return None, f"Error generating CSV: {str(e)}" def _convert_table_to_csv(self, content): """Convert pipe-separated table to CSV format""" lines = content.split('\n') csv_lines = [] for line in lines: line = line.strip() # Skip empty lines, headers, and separators if not line or line.startswith('#') or '---' in line or line.startswith('='): continue # Process table rows with pipes if '|' in line: # Split by pipe and clean up cells = [cell.strip() for cell in line.split('|')] # Remove empty cells at start/end while cells and not cells[0]: cells.pop(0) while cells and not cells[-1]: cells.pop() if len(cells) >= 2: # Valid row # Escape commas in cell content escaped_cells = [] for cell in cells: if ',' in cell or '"' in cell: cell = '"' + cell.replace('"', '""') + '"' escaped_cells.append(cell) csv_lines.append(','.join(escaped_cells)) return '\n'.join(csv_lines) def _clean_html_entities(self, content): """Clean HTML entities from content""" # Multiple passes to handle nested encoding for _ in range(3): content = html.unescape(content) content = content.replace('&', '&') content = content.replace('>', '>') content = content.replace('<', '<') content = content.replace('"', '"') content = content.replace(''', "'") return content def _extract_from_text(self, content): """Extract mapping data from unstructured text""" mappings = [] # Enhanced patterns for different content formats endpoint_pattern = r'(GET|POST|PUT|DELETE|PATCH)\s+([/\w\-{}]+)' field_pattern = r'(\w+)\s*:\s*(\w+)(?:\s*\((.*?)\))?' # Pattern for structured field descriptions structured_field_pattern = r'[-*]\s*(\w+)\s*[:-]\s*(\w+)(?:\s*[,;]?\s*(.+?))?$' current_endpoint = None current_method = None lines = content.split('\n') for i, line in enumerate(lines): line = line.strip() # Skip empty lines and headers if not line or line.startswith('#') or line.startswith('='): continue # Check for endpoint endpoint_match = re.search(endpoint_pattern, line) if endpoint_match: current_method = endpoint_match.group(1) current_endpoint = endpoint_match.group(2) continue # Check for structured field descriptions (bullet points) structured_match = re.search(structured_field_pattern, line) if structured_match: field_name = structured_match.group(1) data_type = structured_match.group(2) description = structured_match.group(3) if structured_match.group(3) else '' mapping = { 'endpoint': current_endpoint or 'N/A', 'method': current_method or 'N/A', 'field_name': field_name, 'data_type': data_type, 'description': description.strip(), 'field_path': f"{current_endpoint or 'root'}.{field_name}", 'required': 'Yes' if 'required' in line.lower() or 'mandatory' in line.lower() else 'No', 'example': '' } mappings.append(mapping) continue # Check for standard field patterns field_match = re.search(field_pattern, line) if field_match: mapping = { 'endpoint': current_endpoint or 'N/A', 'method': current_method or 'N/A', 'field_name': field_match.group(1), 'data_type': field_match.group(2), 'description': field_match.group(3) if field_match.group(3) else '', 'field_path': f"{current_endpoint or 'root'}.{field_match.group(1)}", 'required': 'Yes' if 'required' in line.lower() or 'mandatory' in line.lower() else 'No', 'example': '' } mappings.append(mapping) # If still no mappings found, try to extract any field-like content if not mappings: mappings = self._extract_fallback(content) return mappings def _extract_fallback(self, content): """Fallback extraction for any field-like content""" mappings = [] lines = content.split('\n') current_endpoint = None current_method = None for line in lines: line = line.strip() if not line or len(line) < 3: continue # Skip headers and separators if line.startswith('#') or line.startswith('|') or line.startswith('-') or line.startswith('='): continue # Check for endpoint endpoint_match = re.search(r'(GET|POST|PUT|DELETE|PATCH)\s+([/\w\-{}]+)', line) if endpoint_match: current_method = endpoint_match.group(1) current_endpoint = endpoint_match.group(2) continue # Pattern 1: field_name: type - description pattern1 = r'^(\w+)\s*:\s*(\w+)\s*[-–]\s*(.+)$' match1 = re.search(pattern1, line) if match1: field_name = match1.group(1) data_type = match1.group(2) description = match1.group(3).strip() mapping = { 'endpoint': current_endpoint or 'N/A', 'method': current_method or 'N/A', 'field_name': field_name, 'data_type': data_type, 'description': description, 'field_path': f"{current_endpoint or 'root'}.{field_name}", 'required': 'Yes' if 'required' in description.lower() or 'mandatory' in description.lower() else 'No', 'example': '' } mappings.append(mapping) continue # Pattern 2: just field names (current case) if re.match(r'^\w+$', line) and len(line) > 2: field_name = line # Infer data type from field name data_type = 'string' if any(indicator in field_name.lower() for indicator in ['id', 'uuid']): data_type = 'string' elif any(indicator in field_name.lower() for indicator in ['date', 'time']): data_type = 'datetime' elif any(indicator in field_name.lower() for indicator in ['email']): data_type = 'email' elif any(indicator in field_name.lower() for indicator in ['code', 'postal']): data_type = 'string' # Generate description from field name description = f"Field for {field_name.replace('_', ' ').replace('Name', ' name').replace('Line', ' line')}" mapping = { 'endpoint': current_endpoint or 'N/A', 'method': current_method or 'N/A', 'field_name': field_name, 'data_type': data_type, 'description': description, 'field_path': f"{current_endpoint or 'root'}.{field_name}", 'required': 'No', 'example': '' } mappings.append(mapping) return mappings