Spaces:
Sleeping
Sleeping
| 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 |