Create utils.py

utils.py

@@ -0,0 +1,484 @@
+"""Utility functions for Salesforce B2B Commerce Migration Assistant"""
+
+import re
+import json
+import logging
+from typing import Dict, Tuple, List, Optional
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Apex syntax patterns for validation
+APEX_PATTERNS = {
+    "class_declaration": r"(?:public|private|global|protected)\s+(?:virtual|abstract|with sharing|without sharing|inherited sharing)?\s*class\s+\w+",
+    "trigger_declaration": r"trigger\s+\w+\s+on\s+\w+\s*\([^)]+\)",
+    "method_declaration": r"(?:public|private|global|protected)\s+(?:static)?\s*(?:void|\w+)\s+\w+\s*\([^)]*\)",
+    "soql_query": r"(?:\[|Database\.query\s*\()\s*SELECT\s+.*?\s+FROM\s+\w+.*?(?:\]|\))",
+    "dml_operation": r"(?:insert|update|delete|undelete|upsert|merge)\s+\w+",
+    "bulkification_issue": r"for\s*\([^)]+\)\s*{[^}]*(?:insert|update|delete|undelete)\s+",
+    "hardcoded_id": r"(?:\'[a-zA-Z0-9]{15}\'|\'[a-zA-Z0-9]{18}\')",
+    "missing_null_check": r"(\w+)\.(\w+)(?!\s*(?:!=|==)\s*null)",
+    "governor_limit_risk": r"(?:for\s*\([^)]+\)\s*{[^}]*\[SELECT|Database\.query)",
+}
+
+# Common Apex errors and their fixes
+APEX_ERRORS = {
+    "missing_semicolon": {
+        "pattern": r"[^{};]\s*\n\s*(?:public|private|global|protected|if|for|while|try)",
+        "message": "Missing semicolon at end of statement",
+        "severity": "error"
+    },
+    "unclosed_bracket": {
+        "pattern": r"(?:\{(?:[^{}]|(?:\{[^{}]*\}))*$)|(?:^[^{}]*\})",
+        "message": "Unclosed or extra bracket detected",
+        "severity": "error"
+    },
+    "invalid_soql": {
+        "pattern": r"\[\s*SELECT\s+FROM\s+\w+",
+        "message": "Invalid SOQL: Missing field selection",
+        "severity": "error"
+    },
+    "missing_try_catch_dml": {
+        "pattern": r"(?<!try\s{[^}]*)(insert|update|delete|upsert)\s+(?!.*catch)",
+        "message": "DML operation without try-catch block",
+        "severity": "warning"
+    }
+}
+
+# B2B Commerce specific patterns
+B2B_COMMERCE_PATTERNS = {
+    "cloudcraze_reference": r"(?:ccrz__|E_\w+|CC_\w+)",
+    "b2b_lex_object": r"(?:OrderSummary|CartItem|WebCart|ProductCatalog|BuyerGroup|CommerceEntitlementPolicy)",
+    "deprecated_method": r"(?:ccrz\.cc_CallContext|ccrz\.ccAPI|cc_bean_\w+)",
+    "migration_required": r"(?:E_Product__|E_Cart__|E_Order__|CC_Promotions__|CC_Tax__)"
+}
+
+VALIDATION_SCHEMA = {
+    "quality_rating": "int (1–10)",
+    "accuracy": "float (0.0–1.0)",
+    "completeness": "float (0.0–1.0)",
+    "best_practices_alignment": "float (0.0–1.0)",
+    "syntax_validity": "float (0.0–1.0)",
+    "security_score": "float (0.0–1.0)",
+    "performance_score": "float (0.0–1.0)",
+    "explanations": {
+        "quality_rating": "string",
+        "accuracy": "string",
+        "completeness": "string",
+        "best_practices_alignment": "string",
+        "syntax_validity": "string",
+        "security_score": "string",
+        "performance_score": "string"
+    },
+    "errors": ["list of syntax errors"],
+    "warnings": ["list of potential issues"],
+    "suggestions": ["list of improvement suggestions"]
+}
+
+def validate_apex_syntax(code: str) -> Tuple[bool, List[Dict[str, str]]]:
+    """Validate Apex syntax and return errors/warnings."""
+    issues = []
+    
+    # Check for basic syntax errors
+    for error_type, error_info in APEX_ERRORS.items():
+        matches = re.finditer(error_info["pattern"], code, re.MULTILINE | re.DOTALL)
+        for match in matches:
+            issues.append({
+                "type": error_info["severity"],
+                "message": error_info["message"],
+                "line": code[:match.start()].count('\n') + 1,
+                "position": match.start()
+            })
+    
+    # Check for Apex-specific patterns
+    if not re.search(APEX_PATTERNS["class_declaration"], code) and \
+       not re.search(APEX_PATTERNS["trigger_declaration"], code):
+        issues.append({
+            "type": "error",
+            "message": "No valid Apex class or trigger declaration found",
+            "line": 1,
+            "position": 0
+        })
+    
+    # Check for bulkification issues
+    bulk_issues = re.finditer(APEX_PATTERNS["bulkification_issue"], code, re.DOTALL)
+    for match in bulk_issues:
+        issues.append({
+            "type": "error",
+            "message": "DML operation inside loop - violates bulkification best practices",
+            "line": code[:match.start()].count('\n') + 1,
+            "position": match.start()
+        })
+    
+    # Check for hardcoded IDs
+    hardcoded_ids = re.finditer(APEX_PATTERNS["hardcoded_id"], code)
+    for match in hardcoded_ids:
+        issues.append({
+            "type": "warning",
+            "message": "Hardcoded Salesforce ID detected - use Custom Settings or Custom Metadata",
+            "line": code[:match.start()].count('\n') + 1,
+            "position": match.start()
+        })
+    
+    # Check for governor limit risks
+    gov_limit_risks = re.finditer(APEX_PATTERNS["governor_limit_risk"], code, re.DOTALL)
+    for match in gov_limit_risks:
+        issues.append({
+            "type": "warning",
+            "message": "SOQL query inside loop - potential governor limit issue",
+            "line": code[:match.start()].count('\n') + 1,
+            "position": match.start()
+        })
+    
+    has_errors = any(issue["type"] == "error" for issue in issues)
+    return not has_errors, issues
+
+def perform_skeptical_evaluation(code: str, context: str = "trigger") -> Dict[str, any]:
+    """Perform skeptical evaluation of code looking for common issues."""
+    evaluation = {
+        "syntax_issues": [],
+        "security_concerns": [],
+        "performance_issues": [],
+        "best_practice_violations": [],
+        "b2b_commerce_issues": []
+    }
+    
+    # Syntax validation
+    is_valid, syntax_issues = validate_apex_syntax(code)
+    evaluation["syntax_issues"] = syntax_issues
+    
+    # Security checks
+    if re.search(r"without\s+sharing", code, re.IGNORECASE):
+        evaluation["security_concerns"].append({
+            "type": "warning",
+            "message": "Class declared 'without sharing' - ensure this is intentional"
+        })
+    
+    if not re.search(r"\.stripInaccessible\(", code) and re.search(r"(insert|update)\s+", code):
+        evaluation["security_concerns"].append({
+            "type": "warning",
+            "message": "DML operations without stripInaccessible - potential FLS violation"
+        })
+    
+    # Performance checks
+    nested_loops = re.findall(r"for\s*\([^)]+\)\s*\{[^}]*for\s*\([^)]+\)", code, re.DOTALL)
+    if nested_loops:
+        evaluation["performance_issues"].append({
+            "type": "warning",
+            "message": f"Nested loops detected ({len(nested_loops)} occurrences) - review for O(n²) complexity"
+        })
+    
+    # Check for missing test assertions (if it's a test class)
+    if re.search(r"@isTest|testMethod", code, re.IGNORECASE):
+        if not re.search(r"System\.assert|Assert\.", code):
+            evaluation["best_practice_violations"].append({
+                "type": "error",
+                "message": "Test class without assertions - tests must verify behavior"
+            })
+    
+    # B2B Commerce specific checks
+    cloudcraze_refs = re.findall(B2B_COMMERCE_PATTERNS["cloudcraze_reference"], code)
+    if cloudcraze_refs:
+        evaluation["b2b_commerce_issues"].append({
+            "type": "error",
+            "message": f"CloudCraze references found ({len(set(cloudcraze_refs))} unique) - must be migrated to B2B LEX"
+        })
+    
+    deprecated_methods = re.findall(B2B_COMMERCE_PATTERNS["deprecated_method"], code)
+    if deprecated_methods:
+        evaluation["b2b_commerce_issues"].append({
+            "type": "error",
+            "message": f"Deprecated CloudCraze methods found: {', '.join(set(deprecated_methods))}"
+        })
+    
+    return evaluation
+
+def extract_code_blocks(text: str) -> str:
+    """Enhanced code extraction with multiple strategies."""
+    # Strategy 1: Standard code blocks with language markers
+    pattern = r"```(?:apex|java|Apex|Java|APEX|JAVA)?\s*(.*?)```"
+    matches = re.findall(pattern, text, re.DOTALL | re.IGNORECASE)
+    
+    code_blocks = []
+    for block in matches:
+        cleaned_block = block.strip()
+        if cleaned_block:
+            code_blocks.append(cleaned_block)
+    
+    # Strategy 2: Improved fallback detection for Apex-specific patterns
+    if not code_blocks:
+        apex_patterns = [
+            # Class declarations (including inner classes)
+            r"((?:public|private|global|protected)\s+(?:virtual|abstract|with sharing|without sharing|inherited sharing)?\s*class\s+\w+(?:\s+extends\s+\w+)?(?:\s+implements\s+[\w\s,]+)?\s*\{(?:[^{}]|\{[^{}]*\})*\})",
+            # Trigger declarations
+            r"(trigger\s+\w+\s+on\s+\w+\s*\([^)]+\)\s*\{(?:[^{}]|\{[^{}]*\})*\})",
+            # Interface declarations
+            r"((?:public|private|global)\s+interface\s+\w+(?:\s+extends\s+[\w\s,]+)?\s*\{(?:[^{}]|\{[^{}]*\})*\})",
+            # Enum declarations
+            r"((?:public|private|global)\s+enum\s+\w+\s*\{[^}]+\})",
+            # Annotated methods or classes
+            r"(@\w+(?:\([^)]*\))?\s*(?:public|private|global|protected).*?(?:\{(?:[^{}]|\{[^{}]*\})*\}|;))"
+        ]
+        
+        for pattern in apex_patterns:
+            found = re.findall(pattern, text, re.DOTALL | re.MULTILINE)
+            code_blocks.extend(found)
+    
+    # Strategy 3: Look for code between specific markers
+    if not code_blocks:
+        # Look for code after phrases like "corrected code:", "here's the code:", etc.
+        marker_patterns = [
+            r"(?:corrected|fixed|updated|converted|modified)\s+code\s*:\s*\n((?:(?:public|private|global|trigger).*?)(?=\n\n|\Z))",
+            r"(?:here'?s?|below is)\s+(?:the|your)\s+(?:corrected|fixed|updated)\s+\w+\s*:\s*\n((?:(?:public|private|global|trigger).*?)(?=\n\n|\Z))"
+        ]
+        
+        for pattern in marker_patterns:
+            found = re.findall(pattern, text, re.DOTALL | re.IGNORECASE)
+            code_blocks.extend(found)
+    
+    return '\n\n'.join(filter(None, code_blocks))
+
+def format_structured_explanation(response: str, code_output: str) -> str:
+    """Format the explanation in a structured, brief manner."""
+    # Extract key sections using regex
+    sections = {
+        "key_changes": "",
+        "critical_issues": "",
+        "warnings": ""
+    }
+    
+    # Extract KEY CHANGES section
+    key_match = re.search(r"##\s*KEY CHANGES.*?\n((?:[-•]\s*.*?\n)+)", response, re.IGNORECASE | re.DOTALL)
+    if key_match:
+        sections["key_changes"] = key_match.group(1).strip()
+    
+    # Extract CRITICAL ISSUES section
+    critical_match = re.search(r"##\s*CRITICAL ISSUES.*?\n((?:\d+\..*?\n)+)", response, re.IGNORECASE | re.DOTALL)
+    if critical_match:
+        sections["critical_issues"] = critical_match.group(1).strip()
+    
+    # Extract WARNINGS section
+    warning_match = re.search(r"##\s*REMAINING WARNINGS.*?\n((?:[-•]\s*.*?\n)*)", response, re.IGNORECASE | re.DOTALL)
+    if warning_match:
+        sections["warnings"] = warning_match.group(1).strip()
+    
+    # Build formatted explanation
+    formatted = "### Summary of Changes\n\n"
+    
+    if sections["key_changes"]:
+        formatted += "**Key Changes:**\n" + sections["key_changes"] + "\n\n"
+    
+    if sections["critical_issues"]:
+        formatted += "**Critical Issues Fixed:**\n" + sections["critical_issues"] + "\n\n"
+    
+    if sections["warnings"]:
+        formatted += "**⚠️ Remaining Warnings:**\n" + sections["warnings"]
+    
+    # If structured extraction failed, provide a brief summary
+    if not any(sections.values()):
+        # Fall back to a simple extraction
+        formatted = "### Code Correction Summary\n\n"
+        formatted += "The code has been corrected and optimized. "
+        formatted += "Check the code output for inline comments explaining specific changes.\n\n"
+        formatted += "For detailed analysis, see the Full Model Response."
+    
+    return formatted.strip()
+
+def format_object_conversion_explanation(response: str, code_output: str) -> str:
+    """Format the object conversion explanation in a structured manner."""
+    sections = {
+        "mapping": "",
+        "field_table": "",
+        "steps": "",
+        "warnings": ""
+    }
+    
+    # Extract object mapping section
+    mapping_match = re.search(r"##\s*B2B LEX OBJECT MAPPING.*?\n((?:[-•]\s*.*?\n)+)", response, re.IGNORECASE | re.DOTALL)
+    if mapping_match:
+        sections["mapping"] = mapping_match.group(1).strip()
+    
+    # Extract field mappings table
+    table_match = re.search(r"##\s*FIELD MAPPINGS.*?\n((?:\|.*?\|.*?\n)+)", response, re.IGNORECASE | re.DOTALL)
+    if table_match:
+        sections["field_table"] = table_match.group(1).strip()
+    
+    # Extract migration steps
+    steps_match = re.search(r"##\s*MIGRATION STEPS.*?\n((?:\d+\..*?\n)+)", response, re.IGNORECASE | re.DOTALL)
+    if steps_match:
+        sections["steps"] = steps_match.group(1).strip()
+    
+    # Extract warnings
+    warning_match = re.search(r"##\s*WARNINGS.*?\n((?:[-•]\s*.*?\n)*)", response, re.IGNORECASE | re.DOTALL)
+    if warning_match:
+        sections["warnings"] = warning_match.group(1).strip()
+    
+    # Build formatted explanation
+    formatted = "### Conversion Summary\n\n"
+    
+    if sections["mapping"]:
+        formatted += "**Object Mapping:**\n" + sections["mapping"] + "\n\n"
+    
+    if sections["field_table"]:
+        formatted += "**Field Mappings:**\n" + sections["field_table"] + "\n\n"
+    
+    if sections["steps"]:
+        formatted += "**Migration Steps:**\n" + sections["steps"] + "\n\n"
+    
+    if sections["warnings"]:
+        formatted += "**⚠️ Important Notes:**\n" + sections["warnings"]
+    
+    # Fallback if structured extraction failed
+    if not any(sections.values()):
+        formatted = "### Conversion Summary\n\n"
+        formatted += "The CloudCraze object has been converted to B2B Lightning Experience format. "
+        formatted += "Check the code output for the complete implementation.\n\n"
+        formatted += "For detailed field mappings and migration steps, see the Full Model Response."
+    
+    return formatted.strip()
+
+def extract_validation_metrics(validation_text: str) -> Optional[Dict[str, float]]:
+    """Enhanced JSON extraction for validation metrics."""
+    try:
+        # Strategy 1: Look for JSON after specific markers
+        json_patterns = [
+            r'(?:json|JSON|assessment|Assessment)[\s:]*({[^{}]*(?:{[^{}]*}[^{}]*)*})',
+            r'```json\s*({[^`]+})\s*```',
+            r'({[^{}]*"quality_rating"[^{}]*(?:{[^{}]*}[^{}]*)*})'
+        ]
+        
+        for pattern in json_patterns:
+            matches = re.findall(pattern, validation_text, re.DOTALL)
+            for match in matches:
+                try:
+                    data = json.loads(match)
+                    if "quality_rating" in data:
+                        return normalize_metrics(data)
+                except json.JSONDecodeError:
+                    continue
+        
+        # Strategy 2: Extract individual metrics if JSON parsing fails
+        metrics = {}
+        metric_patterns = {
+            "quality_rating": r"quality_rating[\"']?\s*:\s*(\d+(?:\.\d+)?)",
+            "accuracy": r"accuracy[\"']?\s*:\s*(\d+(?:\.\d+)?)",
+            "completeness": r"completeness[\"']?\s*:\s*(\d+(?:\.\d+)?)",
+            "best_practices_alignment": r"best_practices_alignment[\"']?\s*:\s*(\d+(?:\.\d+)?)",
+            "syntax_validity": r"syntax_validity[\"']?\s*:\s*(\d+(?:\.\d+)?)",
+            "security_score": r"security_score[\"']?\s*:\s*(\d+(?:\.\d+)?)",
+            "performance_score": r"performance_score[\"']?\s*:\s*(\d+(?:\.\d+)?)"
+        }
+        
+        for metric, pattern in metric_patterns.items():
+            match = re.search(pattern, validation_text, re.IGNORECASE)
+            if match:
+                metrics[metric] = float(match.group(1))
+        
+        if metrics:
+            return normalize_metrics(metrics)
+        
+        return None
+        
+    except Exception as e:
+        logger.error(f"Error extracting metrics: {e}")
+        return None
+
+def normalize_metrics(data: Dict) -> Dict[str, float]:
+    """Ensure metrics are in the correct format and range."""
+    normalized = {
+        "quality_rating": min(10, max(0, float(data.get("quality_rating", 0)))),
+        "accuracy": min(1.0, max(0.0, float(data.get("accuracy", 0.0)))),
+        "completeness": min(1.0, max(0.0, float(data.get("completeness", 0.0)))),
+        "best_practices_alignment": min(1.0, max(0.0, float(data.get("best_practices_alignment", 0.0)))),
+        "syntax_validity": min(1.0, max(0.0, float(data.get("syntax_validity", 0.0)))),
+        "security_score": min(1.0, max(0.0, float(data.get("security_score", 0.0)))),
+        "performance_score": min(1.0, max(0.0, float(data.get("performance_score", 0.0))))
+    }
+    return normalized
+
+def generate_test_cases(code_type: str, code: str) -> str:
+    """Generate test cases for the given code."""
+    if code_type == "trigger":
+        return f"""
+// Test class for the trigger
+@isTest
+private class Test_MigratedTrigger {{
+    @TestSetup
+    static void setup() {{
+        // Create test data
+        // TODO: Add specific test data setup
+    }}
+    
+    @isTest
+    static void testBulkInsert() {{
+        // Test bulk insert scenario
+        List<SObject> testRecords = new List<SObject>();
+        for(Integer i = 0; i < 200; i++) {{
+            // TODO: Create test records
+        }}
+        
+        Test.startTest();
+        insert testRecords;
+        Test.stopTest();
+        
+        // TODO: Add assertions
+        System.assert(true, 'Bulk insert test needs implementation');
+    }}
+    
+    @isTest
+    static void testBulkUpdate() {{
+        // Test bulk update scenario
+        // TODO: Implement bulk update test
+    }}
+    
+    @isTest
+    static void testErrorHandling() {{
+        // Test error scenarios
+        // TODO: Test validation rules, required fields, etc.
+    }}
+    
+    @isTest
+    static void testGovernorLimits() {{
+        // Test near governor limits
+        // TODO: Test with large data volumes
+    }}
+}}
+"""
+    else:  # object conversion
+        return f"""
+// Test data creation for migrated object
+@isTest
+public class Test_MigratedObjectData {{
+    public static SObject createTestRecord() {{
+        // TODO: Create and return test instance
+        return null;
+    }}
+    
+    public static List<SObject> createBulkTestRecords(Integer count) {{
+        List<SObject> records = new List<SObject>();
+        for(Integer i = 0; i < count) {{
+            // TODO: Create test records
+        }}
+        return records;
+    }}
+    
+    public static void validateMigrationMapping() {{
+        // Validate that all fields are properly mapped
+        // TODO: Add field mapping validation
+    }}
+}}
+"""
+
+def handle_api_error(status_code: int, response_text: str) -> str:
+    """Handle API errors with appropriate user-friendly messages."""
+    if status_code == 401:
+        return "Authentication failed. Please check API configuration."
+    elif status_code == 429:
+        return "Rate limit exceeded. Please try again later."
+    elif status_code == 403:
+        return "Access forbidden. Please check your permissions."
+    elif status_code >= 500:
+        return "Service temporarily unavailable. Please try again."
+    else:
+        return f"Request failed with status {status_code}"