| |
| """ |
| Python Script Engine β Write, Validate, Run Code Safely |
| ======================================================== |
| Provides: |
| 1. Code generation with GLM-guided structure |
| 2. AST parsing for structural validation |
| 3. Sandboxed execution with timeout/memory limits |
| 4. NRCI scoring for code coherence |
| 5. CRG grounding for concept verification |
| 6. Three Column Thinking: Code + Logic + Verification |
| |
| This replaces the screenplay engine with real script writing. |
| """ |
| import ast, sys, os, re, time, hashlib, traceback, signal, io |
| from typing import List, Dict, Tuple, Optional, Any, Set |
| from dataclasses import dataclass, field |
| from contextlib import contextmanager |
| from fractions import Fraction |
|
|
|
|
| |
| |
| |
|
|
| @dataclass |
| class CodeAnalysis: |
| """Structural analysis of Python code.""" |
| valid_syntax: bool |
| error_message: str = "" |
| |
| |
| n_functions: int = 0 |
| n_classes: int = 0 |
| n_imports: int = 0 |
| n_variables: int = 0 |
| n_loops: int = 0 |
| n_conditionals: int = 0 |
| n_try_except: int = 0 |
| n_list_comps: int = 0 |
| n_decorators: int = 0 |
| |
| |
| cyclomatic_complexity: int = 0 |
| max_depth: int = 0 |
| total_lines: int = 0 |
| blank_lines: int = 0 |
| comment_lines: int = 0 |
| |
| |
| function_names: List[str] = field(default_factory=list) |
| class_names: List[str] = field(default_factory=list) |
| import_names: List[str] = field(default_factory=list) |
| variable_names: List[str] = field(default_factory=list) |
| |
| |
| has_docstrings: bool = False |
| has_type_hints: bool = False |
| has_error_handling: bool = False |
| has_main_guard: bool = False |
| has_tests: bool = False |
| |
| |
| nrci_score: float = 0.0 |
| quality_verdict: str = "" |
|
|
|
|
| class CodeParser: |
| """Parses Python code into structural analysis.""" |
| |
| def analyze(self, code: str) -> CodeAnalysis: |
| """Full structural analysis of Python code.""" |
| result = CodeAnalysis(valid_syntax=True) |
| result.total_lines = len(code.split('\n')) |
| result.blank_lines = sum(1 for l in code.split('\n') if not l.strip()) |
| result.comment_lines = sum(1 for l in code.split('\n') if l.strip().startswith('#')) |
| |
| |
| result.has_main_guard = 'if __name__' in code or "if __name__=='__main__'" in code |
| |
| |
| result.has_tests = bool(re.search(r'def test_|class Test|import unittest|import pytest', code)) |
| |
| try: |
| tree = ast.parse(code) |
| except SyntaxError as e: |
| result.valid_syntax = False |
| result.error_message = f"SyntaxError at line {e.lineno}: {e.msg}" |
| return result |
| |
| |
| self._walk(tree, result, depth=0) |
| |
| |
| result.cyclomatic_complexity = 1 + result.n_conditionals + result.n_loops + result.n_try_except |
| |
| |
| result.nrci_score = self._compute_nrci(result) |
| |
| |
| result.quality_verdict = self._verdict(result) |
| |
| return result |
| |
| def _walk(self, node, result: CodeAnalysis, depth: int): |
| """Recursively walk AST.""" |
| result.max_depth = max(result.max_depth, depth) |
| |
| |
| if isinstance(node, ast.FunctionDef) or isinstance(node, ast.AsyncFunctionDef): |
| result.n_functions += 1 |
| result.function_names.append(node.name) |
| |
| if (node.body and isinstance(node.body[0], ast.Expr) and |
| isinstance(node.body[0].value, (ast.Constant, ast.Str))): |
| result.has_docstrings = True |
| |
| if node.returns: |
| result.has_type_hints = True |
| |
| if node.name.startswith('test_'): |
| result.has_tests = True |
| |
| |
| elif isinstance(node, ast.ClassDef): |
| result.n_classes += 1 |
| result.class_names.append(node.name) |
| |
| |
| elif isinstance(node, (ast.Import, ast.ImportFrom)): |
| result.n_imports += 1 |
| if isinstance(node, ast.Import): |
| for alias in node.names: |
| result.import_names.append(alias.name) |
| else: |
| result.import_names.append(node.module or "") |
| |
| |
| elif isinstance(node, ast.Assign): |
| result.n_variables += len(node.targets) |
| for target in node.targets: |
| if isinstance(target, ast.Name): |
| result.variable_names.append(target.id) |
| |
| |
| elif isinstance(node, (ast.For, ast.While, ast.AsyncFor)): |
| result.n_loops += 1 |
| |
| |
| elif isinstance(node, (ast.If, ast.IfExp)): |
| result.n_conditionals += 1 |
| |
| |
| elif isinstance(node, ast.Try): |
| result.n_try_except += 1 |
| result.has_error_handling = True |
| |
| |
| elif isinstance(node, (ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp)): |
| result.n_list_comps += 1 |
| |
| |
| if hasattr(node, 'decorator_list'): |
| result.n_decorators += len(node.decorator_list) |
| |
| |
| for child in ast.iter_child_nodes(node): |
| self._walk(child, result, depth + 1) |
| |
| def _compute_nrci(self, analysis: CodeAnalysis) -> float: |
| """Compute NRCI-like coherence score for code. |
| |
| Based on structural balance: |
| - Functions should have docstrings |
| - Error handling should be present |
| - Complexity should be moderate (not too simple, not too complex) |
| - Code should have tests |
| """ |
| score = 0.5 |
| |
| |
| if analysis.has_docstrings: |
| score += 0.1 |
| |
| |
| if analysis.has_type_hints: |
| score += 0.05 |
| |
| |
| if analysis.has_error_handling: |
| score += 0.1 |
| |
| |
| if analysis.has_main_guard: |
| score += 0.05 |
| |
| |
| if analysis.has_tests: |
| score += 0.1 |
| |
| |
| if analysis.cyclomatic_complexity <= 1: |
| score -= 0.05 |
| elif analysis.cyclomatic_complexity > 20: |
| score -= 0.15 |
| elif analysis.cyclomatic_complexity > 10: |
| score -= 0.05 |
| |
| |
| if analysis.total_lines > 0: |
| comment_ratio = analysis.comment_lines / analysis.total_lines |
| if 0.05 <= comment_ratio <= 0.3: |
| score += 0.05 |
| elif comment_ratio > 0.5: |
| score -= 0.05 |
| |
| return max(0.0, min(1.0, score)) |
| |
| def _verdict(self, analysis: CodeAnalysis) -> str: |
| """Quality verdict based on analysis.""" |
| score = analysis.nrci_score |
| |
| issues = [] |
| if not analysis.has_docstrings: |
| issues.append("no docstrings") |
| if not analysis.has_error_handling and analysis.n_functions > 0: |
| issues.append("no error handling") |
| if analysis.cyclomatic_complexity > 15: |
| issues.append(f"high complexity ({analysis.cyclomatic_complexity})") |
| if not analysis.has_tests and analysis.n_functions > 2: |
| issues.append("no tests") |
| if analysis.max_depth > 6: |
| issues.append(f"deep nesting ({analysis.max_depth})") |
| |
| if score >= 0.75: |
| return "EXCELLENT" + (f" ({', '.join(issues)})" if issues else "") |
| elif score >= 0.60: |
| return "GOOD" + (f" β {', '.join(issues)}" if issues else "") |
| elif score >= 0.45: |
| return "FAIR" + (f" β {', '.join(issues)}" if issues else "") |
| else: |
| return "NEEDS_WORK" + (f" β {', '.join(issues)}" if issues else "") |
|
|
|
|
| |
| |
| |
|
|
| @dataclass |
| class ExecutionResult: |
| """Result of sandboxed code execution.""" |
| success: bool |
| stdout: str = "" |
| stderr: str = "" |
| return_value: Any = None |
| exception: str = "" |
| execution_time_ms: float = 0.0 |
| timeout: bool = False |
| memory_exceeded: bool = False |
|
|
|
|
| class SandboxExecutor: |
| """Executes Python code in a restricted sandbox. |
| |
| Restrictions: |
| - No file system access (except /tmp) |
| - No network access |
| - No subprocess/os.system |
| - No import of dangerous modules |
| - Timeout enforcement |
| - Output capture |
| """ |
| |
| |
| SAFE_MODULES = { |
| 'math', 'json', 're', 'hashlib', 'random', 'itertools', |
| 'collections', 'functools', 'fractions', 'decimal', |
| 'string', 'textwrap', 'typing', 'dataclasses', |
| 'copy', 'operator', 'bisect', 'heapq', 'array', |
| 'datetime', 'time', 'calendar', |
| 'statistics', 'numbers', 'cmath', |
| } |
| |
| |
| BLOCKED_MODULES = { |
| 'os', 'sys', 'subprocess', 'shutil', 'socket', 'http', 'urllib', |
| 'requests', 'ftplib', 'smtplib', 'telnetlib', 'xmlrpc', |
| 'ctypes', 'importlib', 'compileall', 'py_compile', |
| 'multiprocessing', 'threading', 'signal', |
| 'shelve', 'pickle', 'marshal', 'dbm', |
| 'webbrowser', 'cgi', 'wsgiref', |
| } |
| |
| |
| BLOCKED_BUILTINS = { |
| 'exec', 'eval', 'compile', |
| 'breakpoint', 'exit', 'quit', |
| } |
| |
| def __init__(self, timeout_seconds: float = 5.0, |
| allow_io: bool = False, |
| allow_imports: Optional[Set[str]] = None): |
| self.timeout = timeout_seconds |
| self.allow_io = allow_io |
| self.allow_imports = allow_imports or set() |
| |
| def execute(self, code: str, context: Optional[Dict] = None) -> ExecutionResult: |
| """Execute code in sandbox. |
| |
| Args: |
| code: Python code to execute |
| context: Optional variables to inject into scope |
| |
| Returns: ExecutionResult with stdout, stderr, return value, etc. |
| """ |
| t0 = time.perf_counter() |
| |
| |
| block_reason = self._check_safety(code) |
| if block_reason: |
| return ExecutionResult( |
| success=False, |
| exception=f"Blocked: {block_reason}", |
| execution_time_ms=(time.perf_counter() - t0) * 1000, |
| ) |
| |
| |
| old_stdout = sys.stdout |
| old_stderr = sys.stderr |
| captured_stdout = io.StringIO() |
| captured_stderr = io.StringIO() |
| |
| |
| safe_globals = self._build_safe_globals(context) |
| |
| timeout_hit = False |
| |
| try: |
| sys.stdout = captured_stdout |
| sys.stderr = captured_stderr |
| |
| |
| compiled = compile(code, '<sandbox>', 'exec') |
| |
| |
| def timeout_handler(signum, frame): |
| nonlocal timeout_hit |
| timeout_hit = True |
| raise TimeoutError(f"Execution exceeded {self.timeout}s") |
| |
| |
| if hasattr(signal, 'SIGALRM'): |
| old_handler = signal.signal(signal.SIGALRM, timeout_handler) |
| signal.setitimer(signal.ITIMER_REAL, self.timeout) |
| |
| try: |
| exec(compiled, safe_globals) |
| finally: |
| if hasattr(signal, 'SIGALRM'): |
| signal.setitimer(signal.ITIMER_REAL, 0) |
| signal.signal(signal.SIGALRM, old_handler) |
| |
| |
| return_val = safe_globals.get('_result', None) |
| |
| return ExecutionResult( |
| success=True, |
| stdout=captured_stdout.getvalue(), |
| stderr=captured_stderr.getvalue(), |
| return_value=return_val, |
| execution_time_ms=(time.perf_counter() - t0) * 1000, |
| ) |
| |
| except TimeoutError: |
| return ExecutionResult( |
| success=False, |
| stdout=captured_stdout.getvalue(), |
| stderr=captured_stderr.getvalue(), |
| exception="TimeoutError: execution exceeded time limit", |
| timeout=True, |
| execution_time_ms=(time.perf_counter() - t0) * 1000, |
| ) |
| except Exception as e: |
| return ExecutionResult( |
| success=False, |
| stdout=captured_stdout.getvalue(), |
| stderr=captured_stderr.getvalue(), |
| exception=f"{type(e).__name__}: {e}", |
| execution_time_ms=(time.perf_counter() - t0) * 1000, |
| ) |
| finally: |
| sys.stdout = old_stdout |
| sys.stderr = old_stderr |
| |
| def _check_safety(self, code: str) -> Optional[str]: |
| """Pre-execution safety check. Returns reason if blocked, None if safe.""" |
| code_lower = code.lower() |
| |
| |
| for mod in self.BLOCKED_MODULES: |
| if mod in self.allow_imports: |
| continue |
| |
| patterns = [ |
| f'import {mod}', |
| f'from {mod}', |
| f'import{mod}', |
| ] |
| for pattern in patterns: |
| if pattern in code_lower: |
| return f"blocked module: {mod}" |
| |
| |
| |
| |
| |
| |
| for builtin in self.BLOCKED_BUILTINS: |
| if re.search(rf'\b{builtin}\b', code) and builtin not in ('open',): |
| return f"blocked builtin: {builtin}" |
| |
| |
| if not self.allow_io: |
| fs_patterns = [ |
| r'open\s*\(', r'\.read\(', r'\.write\(', r'\.delete\(', |
| r'os\.', r'pathlib', r'shutil', r'glob', |
| ] |
| for pattern in fs_patterns: |
| if re.search(pattern, code): |
| return f"blocked: file system access ({pattern})" |
| |
| |
| net_patterns = [ |
| r'requests\.', r'urllib', r'http\.', r'socket\.', |
| r'aiohttp', r'httpx', |
| ] |
| for pattern in net_patterns: |
| if re.search(pattern, code): |
| return f"blocked: network access ({pattern})" |
| |
| |
| if re.search(r'\bexec\s*\(', code) or re.search(r'\beval\s*\(', code): |
| return "blocked: exec/eval" |
| |
| return None |
| |
| def _build_safe_globals(self, context: Optional[Dict] = None) -> Dict: |
| """Build restricted global namespace.""" |
| import math, json, re, hashlib, random, itertools, collections, functools |
| import fractions, decimal, string, textwrap, typing, dataclasses |
| from fractions import Fraction |
| from decimal import Decimal |
| from collections import Counter, defaultdict, OrderedDict |
| from typing import List, Dict, Tuple, Optional, Set, Any |
| |
| safe_builtins = { |
| 'abs': abs, 'all': all, 'any': any, 'bool': bool, |
| 'chr': chr, 'dict': dict, 'dir': dir, 'divmod': divmod, |
| 'enumerate': enumerate, 'filter': filter, 'float': float, |
| 'format': format, 'frozenset': frozenset, 'getattr': getattr, |
| 'hasattr': hasattr, 'hash': hash, 'hex': hex, 'id': id, |
| 'int': int, 'isinstance': isinstance, 'issubclass': issubclass, |
| 'iter': iter, 'len': len, 'list': list, 'map': map, |
| 'max': max, 'min': min, 'next': next, 'object': object, |
| 'oct': oct, 'ord': ord, 'pow': pow, 'print': print, |
| 'property': property, 'range': range, 'repr': repr, |
| 'reversed': reversed, 'round': round, 'set': set, |
| 'setattr': setattr, 'slice': slice, 'sorted': sorted, |
| 'str': str, 'sum': sum, 'super': super, 'tuple': tuple, |
| 'type': type, 'vars': vars, 'zip': zip, |
| 'True': True, 'False': False, 'None': None, |
| |
| 'Exception': Exception, 'ValueError': ValueError, |
| 'TypeError': TypeError, 'KeyError': KeyError, |
| 'IndexError': IndexError, 'AttributeError': AttributeError, |
| 'ZeroDivisionError': ZeroDivisionError, 'StopIteration': StopIteration, |
| 'RuntimeError': RuntimeError, 'NotImplementedError': NotImplementedError, |
| 'OverflowError': OverflowError, 'ImportError': ImportError, |
| 'NameError': NameError, 'SyntaxError': SyntaxError, |
| 'AssertionError': AssertionError, 'OSError': OSError, |
| 'FileNotFoundError': FileNotFoundError, 'PermissionError': PermissionError, |
| 'TimeoutError': TimeoutError, 'MemoryError': MemoryError, |
| 'ArithmeticError': ArithmeticError, 'LookupError': LookupError, |
| 'UnicodeError': UnicodeError, 'DeprecationWarning': DeprecationWarning, |
| 'Warning': Warning, 'UserWarning': UserWarning, |
| |
| 'map': map, 'filter': filter, 'zip': zip, |
| 'enumerate': enumerate, 'reversed': reversed, 'sorted': sorted, |
| 'any': any, 'all': all, 'min': min, 'max': max, 'sum': sum, |
| 'abs': abs, 'round': round, 'pow': pow, 'divmod': divmod, |
| 'isinstance': isinstance, 'issubclass': issubclass, |
| 'getattr': getattr, 'setattr': setattr, 'hasattr': hasattr, |
| 'dir': dir, 'vars': vars, 'type': type, 'id': id, 'hash': hash, |
| 'repr': repr, 'format': format, 'chr': chr, 'ord': ord, |
| 'hex': hex, 'oct': oct, 'bool': bool, 'int': int, 'float': float, |
| 'str': str, 'list': list, 'dict': dict, 'set': set, 'tuple': tuple, |
| 'frozenset': frozenset, 'slice': slice, 'property': property, |
| 'object': object, 'super': super, |
| } |
| |
| safe_modules = { |
| 'math': math, 'json': json, 're': re, 'hashlib': hashlib, |
| 'random': random, 'itertools': itertools, |
| 'collections': collections, 'functools': functools, |
| 'fractions': fractions, 'decimal': decimal, |
| 'string': string, 'textwrap': textwrap, |
| 'typing': typing, 'dataclasses': dataclasses, |
| 'Fraction': Fraction, 'Decimal': Decimal, |
| 'Counter': Counter, 'defaultdict': defaultdict, |
| 'OrderedDict': OrderedDict, |
| 'List': List, 'Dict': Dict, 'Tuple': Tuple, |
| 'Optional': Optional, 'Set': Set, 'Any': Any, |
| } |
| |
| |
| safe_imports = set(self.SAFE_MODULES) | set(self.allow_imports) |
| |
| def safe_import(name, *args, **kwargs): |
| base_name = name.split('.')[0] |
| if base_name in safe_imports: |
| return __import__(name, *args, **kwargs) |
| raise ImportError(f"Import of '{name}' is not allowed in sandbox") |
| |
| safe_builtins['__import__'] = safe_import |
| |
| safe_globals = { |
| '__builtins__': safe_builtins, |
| '__name__': '__main__', |
| '__doc__': None, |
| '__file__': '<sandbox>', |
| **safe_modules, |
| } |
| |
| |
| if context: |
| safe_globals.update(context) |
| |
| return safe_globals |
|
|
|
|
| |
| |
| |
|
|
| class CodeComposer: |
| """Generates code analysis using Three Column Thinking. |
| |
| Column 1: CODE β The actual Python implementation |
| Column 2: LOGIC β Formal description of what the code does |
| Column 3: VERIFICATION β Tests and assertions to prove correctness |
| """ |
| |
| def __init__(self): |
| self.parser = CodeParser() |
| self.sandbox = SandboxExecutor(timeout_seconds=5.0) |
| |
| def analyze_and_verify(self, code: str, |
| run_tests: bool = True) -> Dict[str, Any]: |
| """Full analysis: parse β verify structure β run β test. |
| |
| Returns comprehensive report with Three Column Thinking. |
| """ |
| result = { |
| "code": code, |
| "analysis": None, |
| "execution": None, |
| "test_results": None, |
| "three_column": None, |
| "overall_verdict": "", |
| } |
| |
| |
| analysis = self.parser.analyze(code) |
| result["analysis"] = analysis |
| |
| if not analysis.valid_syntax: |
| result["overall_verdict"] = f"SYNTAX_ERROR: {analysis.error_message}" |
| return result |
| |
| |
| exec_result = self.sandbox.execute(code) |
| result["execution"] = exec_result |
| |
| |
| if run_tests and analysis.has_tests: |
| test_code = self._extract_tests(code) |
| if test_code: |
| test_result = self.sandbox.execute(test_code) |
| result["test_results"] = test_result |
| |
| |
| result["three_column"] = self._three_column(code, analysis, exec_result) |
| |
| |
| result["overall_verdict"] = self._overall_verdict(analysis, exec_result, result.get("test_results")) |
| |
| return result |
| |
| def _extract_tests(self, code: str) -> Optional[str]: |
| """Extract test functions from code.""" |
| try: |
| tree = ast.parse(code) |
| except SyntaxError: |
| return None |
| |
| test_lines = [] |
| for node in ast.walk(tree): |
| if isinstance(node, ast.FunctionDef) and node.name.startswith('test_'): |
| |
| start = node.lineno - 1 |
| end = node.end_lineno if hasattr(node, 'end_lineno') else start + 10 |
| lines = code.split('\n')[start:end] |
| test_lines.extend(lines) |
| test_lines.append(f" {node.name}()") |
| test_lines.append("") |
| |
| if not test_lines: |
| return None |
| |
| |
| test_code = code + "\n\n# Auto-extracted tests\n" |
| test_code += "\n".join(test_lines) |
| return test_code |
| |
| def _three_column(self, code: str, analysis: CodeAnalysis, |
| exec_result: ExecutionResult) -> Dict[str, str]: |
| """Generate Three Column Thinking analysis.""" |
| |
| |
| code_col = f"Python code: {analysis.n_functions} functions, {analysis.n_classes} classes, " |
| code_col += f"{analysis.n_imports} imports. " |
| code_col += f"Complexity: {analysis.cyclomatic_complexity}. " |
| if analysis.has_main_guard: |
| code_col += "Has __main__ guard. " |
| if analysis.has_tests: |
| code_col += "Has tests. " |
| |
| |
| logic_col = "" |
| if analysis.function_names: |
| logic_col += f"Functions: {', '.join(analysis.function_names[:5])}. " |
| if analysis.class_names: |
| logic_col += f"Classes: {', '.join(analysis.class_names[:3])}. " |
| if analysis.has_error_handling: |
| logic_col += "Includes error handling. " |
| if analysis.has_docstrings: |
| logic_col += "Documented with docstrings. " |
| logic_col += f"NRCI coherence: {analysis.nrci_score:.2f}. " |
| |
| |
| ver_col = "" |
| if exec_result.success: |
| ver_col += "Execution: PASS. " |
| if exec_result.stdout: |
| ver_col += f"Output: {exec_result.stdout[:100]}. " |
| else: |
| ver_col += f"Execution: FAIL ({exec_result.exception}). " |
| if analysis.has_tests: |
| ver_col += "Tests present. " |
| ver_col += f"Verdict: {analysis.quality_verdict}. " |
| |
| return { |
| "code": code_col, |
| "logic": logic_col, |
| "verification": ver_col, |
| } |
| |
| def _overall_verdict(self, analysis: CodeAnalysis, |
| exec_result: ExecutionResult, |
| test_result: Optional[ExecutionResult]) -> str: |
| """Determine overall verdict.""" |
| if not analysis.valid_syntax: |
| return "SYNTAX_ERROR" |
| |
| if not exec_result.success: |
| return f"RUNTIME_ERROR: {exec_result.exception}" |
| |
| if test_result and not test_result.success: |
| return f"TEST_FAILURE: {test_result.exception}" |
| |
| score = analysis.nrci_score |
| if score >= 0.75: |
| return "EXCELLENT" |
| elif score >= 0.60: |
| return "GOOD" |
| elif score >= 0.45: |
| return "FAIR" |
| else: |
| return "NEEDS_WORK" |
|
|
|
|
| |
| |
| |
|
|
| class CodeValidator: |
| """Validates Python code for correctness and safety. |
| |
| Use this to check code before execution or to verify |
| generated code meets quality standards. |
| """ |
| |
| def __init__(self): |
| self.parser = CodeParser() |
| self.sandbox = SandboxExecutor(timeout_seconds=2.0) |
| |
| def validate(self, code: str) -> Dict[str, Any]: |
| """Validate code. Returns detailed report.""" |
| analysis = self.parser.analyze(code) |
| |
| |
| sandbox = SandboxExecutor() |
| safety_issue = sandbox._check_safety(code) |
| |
| |
| try: |
| compile(code, '<validate>', 'exec') |
| compiles = True |
| compile_error = "" |
| except SyntaxError as e: |
| compiles = False |
| compile_error = f"Line {e.lineno}: {e.msg}" |
| |
| return { |
| "valid_syntax": analysis.valid_syntax, |
| "compiles": compiles, |
| "compile_error": compile_error, |
| "safe": safety_issue is None, |
| "safety_issue": safety_issue, |
| "analysis": analysis, |
| "nrci_score": analysis.nrci_score, |
| "verdict": analysis.quality_verdict, |
| "recommendations": self._recommendations(analysis, safety_issue), |
| } |
| |
| def _recommendations(self, analysis: CodeAnalysis, |
| safety_issue: Optional[str]) -> List[str]: |
| """Generate improvement recommendations.""" |
| recs = [] |
| |
| if not analysis.has_docstrings: |
| recs.append("Add docstrings to functions and classes") |
| if not analysis.has_type_hints: |
| recs.append("Add type hints for better code clarity") |
| if not analysis.has_error_handling and analysis.n_functions > 0: |
| recs.append("Add try/except for error handling") |
| if not analysis.has_tests and analysis.n_functions > 2: |
| recs.append("Add test functions (def test_*)") |
| if analysis.cyclomatic_complexity > 15: |
| recs.append(f"Reduce complexity (currently {analysis.cyclomatic_complexity})") |
| if analysis.max_depth > 5: |
| recs.append(f"Reduce nesting depth (currently {analysis.max_depth})") |
| if safety_issue: |
| recs.append(f"Safety issue: {safety_issue}") |
| |
| return recs |
|
|
|
|
| |
| |
| |
|
|
| def analyze_code(code: str) -> CodeAnalysis: |
| """Quick structural analysis.""" |
| return CodeParser().analyze(code) |
|
|
| def run_code(code: str, timeout: float = 5.0) -> ExecutionResult: |
| """Quick sandboxed execution.""" |
| return SandboxExecutor(timeout_seconds=timeout).execute(code) |
|
|
| def validate_code(code: str) -> Dict[str, Any]: |
| """Quick validation.""" |
| return CodeValidator().validate(code) |
|
|
| def verify_code(code: str) -> Dict[str, Any]: |
| """Full Three Column Thinking verification.""" |
| return CodeComposer().analyze_and_verify(code) |
|
|