import ast import re import difflib import subprocess import tempfile import os import json import traceback from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline import torch from config import config, get_complexity_score, get_language_config, AI_PATTERNS class CodingAnalyzer: def __init__(self): self.config = config self.setup_ai_model() def setup_ai_model(self): """Setup Qwen2.5-1.5B-Instruct model for code analysis (CPU-friendly)""" try: model_config = self.config.model print(f"šŸ”„ Loading {model_config.model_name} model...") use_fp16 = torch.cuda.is_available() dtype = torch.float16 if use_fp16 else torch.float32 self.model = AutoModelForCausalLM.from_pretrained( model_config.model_name, torch_dtype=dtype, trust_remote_code=model_config.trust_remote_code, low_cpu_mem_usage=True ) device = "cuda" if torch.cuda.is_available() else "cpu" self.model = self.model.to(device) print(f"šŸ“ Running on: {device.upper()}") self.tokenizer = AutoTokenizer.from_pretrained( model_config.model_name, trust_remote_code=model_config.trust_remote_code ) self.pipe = pipeline( "text-generation", model=self.model, tokenizer=self.tokenizer, max_new_tokens=model_config.max_new_tokens, temperature=model_config.temperature, do_sample=model_config.do_sample, top_p=model_config.top_p, repetition_penalty=model_config.repetition_penalty ) param_count = sum(p.numel() for p in self.model.parameters()) / 1e9 print(f"āœ… {model_config.model_name} loaded! ({param_count:.2f}B params)") except Exception as e: print(f"āŒ Model load failed: {e}") print(f"šŸ“‹ Traceback:\n{traceback.format_exc()}") print("šŸ”„ Falling back to rule-based analysis...") self.pipe = None def _call_ai(self, messages): """ Centralized helper to call the Qwen chat pipeline. Supports chat template format. Returns generated text string or raises on failure. """ if self.pipe is None: raise RuntimeError("AI pipeline not available") output = self.pipe(messages, return_full_text=False) if isinstance(output, list) and len(output) > 0: first = output[0] if isinstance(first, dict) and 'generated_text' in first: return first['generated_text'] if isinstance(first, list) and len(first) > 0 and 'generated_text' in first[0]: return first[0]['generated_text'] raise ValueError(f"Unexpected pipeline output format: {type(output)}") def analyze_solution(self, question, user_code, correct_solution, language="python", difficulty="medium", test_cases=None): """Main analysis function""" results = { 'status': 'completed', 'overall_score': 0, 'max_score': 100, 'analysis': { 'correctness': {}, 'code_quality': {}, 'efficiency': {}, 'ai_detection': {}, 'similarity': {}, 'test_results': {} }, 'feedback': '', 'recommendations': [] } try: # 1. Correctness Analysis (40 points) correctness = self.analyze_correctness(user_code, correct_solution, test_cases, language) results['analysis']['correctness'] = correctness results['overall_score'] += correctness['score'] # 2. Code Quality Analysis (25 points) quality = self.analyze_code_quality(user_code, language) results['analysis']['code_quality'] = quality results['overall_score'] += quality['score'] # 3. Efficiency Analysis (20 points) efficiency = self.analyze_efficiency(user_code, correct_solution, language) results['analysis']['efficiency'] = efficiency results['overall_score'] += efficiency['score'] # 4. AI Detection (10 points penalty) ai_detection = self.detect_ai_generated(user_code, language) results['analysis']['ai_detection'] = ai_detection results['overall_score'] -= ai_detection['penalty'] # 5. Similarity Analysis (5 points penalty) similarity = self.analyze_similarity(user_code, correct_solution) results['analysis']['similarity'] = similarity results['overall_score'] -= similarity['penalty'] # Generate feedback and recommendations results['feedback'] = self.generate_feedback(results['analysis'], question, difficulty) results['recommendations'] = self.generate_recommendations(results['analysis']) results['overall_score'] = max(0, min(100, results['overall_score'])) except Exception as e: results['status'] = 'error' results['error'] = str(e) print(f"āŒ analyze_solution failed: {e}\n{traceback.format_exc()}") return results # ───────────────────────────────────────────── # 1. CORRECTNESS # ───────────────────────────────────────────── def analyze_correctness(self, user_code, correct_solution, test_cases, language): """Correctness analysis: syntax + real test execution + AI review""" result = { 'score': 0, 'max_score': 40, 'syntax_valid': False, 'logic_correct': False, 'test_cases_passed': 0, 'total_test_cases': 0, 'execution_errors': [], 'ai_analysis': '' } try: # --- Syntax check (Python only via AST) --- if language == 'python': try: ast.parse(user_code) result['syntax_valid'] = True except SyntaxError as e: result['execution_errors'].append(f"Syntax Error: {str(e)}") result['score'] = 0 return result else: # For other languages assume syntax valid (no AST available) result['syntax_valid'] = True # --- Real test execution for Python --- if language == 'python' and test_cases: passed, total, errors = self._run_python_tests(user_code, test_cases) result['test_cases_passed'] = passed result['total_test_cases'] = total result['execution_errors'].extend(errors) if total > 0: pass_ratio = passed / total result['score'] = int(10 + 30 * pass_ratio) # 10 syntax + up to 30 logic result['logic_correct'] = pass_ratio >= 0.8 else: result['score'] = 10 # only syntax points if no tests # --- AI-powered review --- if self.pipe: correctness_prompt = f"""Analyze this code solution for correctness: User Code: {user_code} Reference Solution: {correct_solution} Evaluate: 1. Syntax validity (0-10 points) 2. Logic correctness (0-30 points) 3. Edge case handling 4. Algorithm accuracy Respond ONLY with valid JSON, no extra text: {{"syntax_score": 8, "logic_score": 25, "analysis": "your explanation here"}}""" messages = [ {"role": "system", "content": "You are an expert code reviewer. Respond with valid JSON only, no markdown, no extra text."}, {"role": "user", "content": correctness_prompt} ] try: ai_response = self._call_ai(messages) json_match = re.search(r'\{.*?\}', ai_response, re.DOTALL) if json_match: ai_eval = json.loads(json_match.group()) syntax_score = min(10, max(0, int(ai_eval.get('syntax_score', 0)))) logic_score = min(30, max(0, int(ai_eval.get('logic_score', 0)))) # If no real tests ran, use AI score if result['total_test_cases'] == 0: result['score'] = syntax_score + logic_score result['logic_correct'] = logic_score >= 25 result['ai_analysis'] = ai_eval.get('analysis', '') except Exception as e: print(f"āš ļø Correctness AI analysis failed: {e}") # Fallback: syntax valid = 10 + 30 base if result['total_test_cases'] == 0: result['score'] = 40 if result['syntax_valid'] else 0 result['logic_correct'] = result['syntax_valid'] else: # No AI: give full marks if syntax valid and no tests failed if result['total_test_cases'] == 0 and result['syntax_valid']: result['score'] = 40 result['logic_correct'] = True except Exception as e: result['execution_errors'].append(str(e)) print(f"āŒ analyze_correctness failed: {e}") return result def _run_python_tests(self, user_code, test_cases): """Actually execute Python code against test cases using subprocess""" passed = 0 total = len(test_cases) errors = [] for tc in test_cases: input_data = tc.get('input', '') expected = str(tc.get('expected_output', '')).strip() input_repr = repr(input_data) # Build test script without backslashes inside f-strings (Python 3.10 compat) lines = [ user_code, "", "import sys", "import ast as _ast", "result = None", "try:", " src = '''", user_code, " '''", " tree = _ast.parse(src)", " func_names = [node.name for node in _ast.walk(tree) if isinstance(node, _ast.FunctionDef)]", " if func_names:", " fn = globals()[func_names[0]]", " args = " + input_repr, " if isinstance(args, (list, tuple)):", " result = fn(*args)", " else:", " result = fn(args)", " print(result)", "except Exception as e:", " print('ERROR: ' + str(e), file=sys.stderr)", " sys.exit(1)", ] test_script = "\n".join(lines) try: proc = subprocess.run( ['python3', '-c', test_script], capture_output=True, text=True, timeout=self.config.analysis.test_timeout ) actual = proc.stdout.strip() if proc.returncode == 0 and actual == expected: passed += 1 else: if proc.stderr: errors.append(proc.stderr.strip()[:200]) except subprocess.TimeoutExpired: errors.append("Test timed out (>" + str(self.config.analysis.test_timeout) + "s)") except Exception as e: errors.append(str(e)) return passed, total, errors # ───────────────────────────────────────────── # 2. CODE QUALITY # ───────────────────────────────────────────── def analyze_code_quality(self, user_code, language): """Code quality analysis: AI-powered with rule-based fallback""" result = { 'score': 0, 'max_score': 25, 'readability': 0, 'structure': 0, 'naming': 0, 'comments': 0, 'best_practices': 0, 'ai_analysis': '' } try: if self.pipe: quality_prompt = f"""Analyze this {language} code for quality: Code: {user_code} Score each (0-5 points): 1. readability: formatting, line length, clarity 2. structure: functions/classes, organization 3. naming: variable and function names 4. comments: documentation quality 5. best_practices: language conventions Respond ONLY with valid JSON: {{"readability": 4, "structure": 4, "naming": 3, "comments": 1, "best_practices": 3, "analysis": "explanation"}}""" messages = [ {"role": "system", "content": "You are a code quality expert. Respond with valid JSON only, no markdown."}, {"role": "user", "content": quality_prompt} ] try: ai_response = self._call_ai(messages) json_match = re.search(r'\{.*?\}', ai_response, re.DOTALL) if json_match: ai_eval = json.loads(json_match.group()) result['readability'] = min(5, max(0, int(ai_eval.get('readability', 0)))) result['structure'] = min(5, max(0, int(ai_eval.get('structure', 0)))) result['naming'] = min(5, max(0, int(ai_eval.get('naming', 0)))) result['comments'] = min(5, max(0, int(ai_eval.get('comments', 0)))) result['best_practices'] = min(5, max(0, int(ai_eval.get('best_practices', 0)))) result['ai_analysis'] = ai_eval.get('analysis', '') result['score'] = sum([result['readability'], result['structure'], result['naming'], result['comments'], result['best_practices']]) return result else: raise ValueError("No JSON in AI response") except Exception as e: print(f"āš ļø Code quality AI failed: {e}") self._basic_quality_analysis(user_code, language, result) else: self._basic_quality_analysis(user_code, language, result) except Exception as e: print(f"āŒ analyze_code_quality failed: {e}") self._basic_quality_analysis(user_code, language, result) return result def _basic_quality_analysis(self, user_code, language, result): """Rule-based quality analysis fallback""" lines = user_code.split('\n') avg_line_length = sum(len(l) for l in lines) / len(lines) if lines else 0 comment_lines = sum(1 for l in lines if l.strip().startswith('#')) result['readability'] = 5 if avg_line_length < 80 else (3 if avg_line_length < 120 else 1) result['structure'] = 4 if 'def ' in user_code else 2 result['naming'] = 3 # neutral default result['comments'] = min(5, comment_lines) result['best_practices'] = 3 # neutral default result['score'] = sum([result['readability'], result['structure'], result['naming'], result['comments'], result['best_practices']]) # ───────────────────────────────────────────── # 3. EFFICIENCY # ───────────────────────────────────────────── def analyze_efficiency(self, user_code, correct_solution, language): """Efficiency analysis: AI complexity detection with regex fallback""" result = { 'score': 0, 'max_score': 20, 'time_complexity': 'Unknown', 'space_complexity': 'Unknown', 'optimization_score': 0, 'ai_analysis': '' } try: if self.pipe: efficiency_prompt = f"""Analyze algorithm efficiency: User Code: {user_code} Reference Solution: {correct_solution} Evaluate: 1. Time Complexity in Big O 2. Space Complexity in Big O 3. Efficiency score (0-20 points) 4. Optimization opportunities Respond ONLY with valid JSON: {{"time_complexity": "O(n)", "space_complexity": "O(1)", "efficiency_score": 15, "analysis": "explanation"}}""" messages = [ {"role": "system", "content": "You are an algorithm complexity expert. Respond with valid JSON only, no markdown."}, {"role": "user", "content": efficiency_prompt} ] try: ai_response = self._call_ai(messages) json_match = re.search(r'\{.*?\}', ai_response, re.DOTALL) if json_match: ai_eval = json.loads(json_match.group()) result['time_complexity'] = ai_eval.get('time_complexity', 'Unknown') result['space_complexity'] = ai_eval.get('space_complexity', 'Unknown') result['score'] = min(20, max(0, int(ai_eval.get('efficiency_score', 0)))) result['optimization_score'] = result['score'] result['ai_analysis'] = ai_eval.get('analysis', '') return result else: raise ValueError("No JSON in AI response") except Exception as e: print(f"āš ļø Efficiency AI failed: {e}") self._basic_efficiency_analysis(user_code, result) else: self._basic_efficiency_analysis(user_code, result) except Exception as e: print(f"āŒ analyze_efficiency failed: {e}") self._basic_efficiency_analysis(user_code, result) return result def _basic_efficiency_analysis(self, user_code, result): """Rule-based complexity detection""" nested_loops = len(re.findall(r'for\s.+\n.*for\s', user_code)) while_nested = len(re.findall(r'while\s.+\n.*while\s', user_code)) total_nested = nested_loops + while_nested if total_nested == 0: complexity_score = 15 result['time_complexity'] = 'O(n) or better' elif total_nested == 1: complexity_score = 10 result['time_complexity'] = 'O(n²)' else: complexity_score = 5 result['time_complexity'] = 'O(n³) or worse' # Bonus for efficient data structures if 'dict' in user_code or '{' in user_code or 'set(' in user_code: complexity_score = min(20, complexity_score + 3) result['space_complexity'] = 'O(n)' if ('list(' in user_code or '[]' in user_code) else 'O(1)' result['optimization_score'] = complexity_score result['score'] = min(20, complexity_score) # ───────────────────────────────────────────── # 4. AI DETECTION # ───────────────────────────────────────────── def detect_ai_generated(self, user_code, language): """Detect AI-generated code patterns""" result = { 'penalty': 0, 'max_penalty': 10, 'ai_probability': 0, 'indicators': [], 'ai_analysis': '' } try: if self.pipe: ai_detection_prompt = f"""Analyze if this code is AI-generated: Code: {user_code} Look for: 1. Overly verbose comments 2. Perfect documentation style 3. Generic variable names (result, output, temp) 4. Unnatural comment patterns 5. Typical ChatGPT/AI structure Respond ONLY with valid JSON: {{"ai_probability": 20, "penalty_score": 2, "indicators": ["reason1"], "analysis": "explanation"}} ai_probability: 0-100, penalty_score: 0-10""" messages = [ {"role": "system", "content": "You are an expert at detecting AI-generated code. Respond with valid JSON only, no markdown."}, {"role": "user", "content": ai_detection_prompt} ] try: ai_response = self._call_ai(messages) json_match = re.search(r'\{.*?\}', ai_response, re.DOTALL) if json_match: ai_eval = json.loads(json_match.group()) result['ai_probability'] = min(100, max(0, int(ai_eval.get('ai_probability', 0)))) result['penalty'] = min(10, max(0, int(ai_eval.get('penalty_score', 0)))) result['indicators'] = ai_eval.get('indicators', []) result['ai_analysis'] = ai_eval.get('analysis', '') return result else: raise ValueError("No JSON in AI response") except Exception as e: print(f"āš ļø AI detection failed: {e}") self._basic_ai_detection(user_code, language, result) else: self._basic_ai_detection(user_code, language, result) except Exception as e: print(f"āŒ detect_ai_generated failed: {e}") self._basic_ai_detection(user_code, language, result) return result def _basic_ai_detection(self, user_code, language, result): """Rule-based AI detection fallback""" ai_score = 0 lines = user_code.split('\n') comment_ratio = sum(1 for l in lines if l.strip().startswith('#')) / len(lines) if lines else 0 if comment_ratio > 0.3: ai_score += 3 result['indicators'].append('Excessive comments (>30% of lines)') for pattern in AI_PATTERNS: if re.search(pattern, user_code, re.IGNORECASE): ai_score += 2 result['indicators'].append(f'Generic AI comment pattern: {pattern}') break # Generic variable names if re.search(r'\bresult\b|\boutput\b|\btemp\b|\bans\b', user_code): ai_score += 1 result['indicators'].append('Generic variable names (result/output/temp)') result['ai_probability'] = min(100, ai_score * 15) result['penalty'] = min(10, ai_score) # ───────────────────────────────────────────── # 5. SIMILARITY # ───────────────────────────────────────────── def analyze_similarity(self, user_code, correct_solution): """Similarity / plagiarism check""" result = { 'penalty': 0, 'max_penalty': 5, 'similarity_ratio': 0, 'ai_analysis': '' } try: # Normalize both codes user_clean = re.sub(r'#.*', '', user_code).replace(' ', '').replace('\n', '') correct_clean = re.sub(r'#.*', '', correct_solution).replace(' ', '').replace('\n', '') similarity = difflib.SequenceMatcher(None, user_clean, correct_clean).ratio() result['similarity_ratio'] = round(similarity, 2) if self.pipe: similarity_prompt = f"""Analyze code similarity for plagiarism: User Code: {user_code} Reference Solution: {correct_solution} Similarity Ratio: {similarity:.2f} Evaluate: 1. Structural similarity 2. Algorithm approach 3. Variable naming patterns 4. Is it copying or just same logic? 5. Penalty (0-5 points) Respond ONLY with valid JSON: {{"penalty_score": 2, "is_plagiarism": false, "analysis": "explanation"}}""" messages = [ {"role": "system", "content": "You are a plagiarism detection expert. Respond with valid JSON only, no markdown."}, {"role": "user", "content": similarity_prompt} ] try: ai_response = self._call_ai(messages) json_match = re.search(r'\{.*?\}', ai_response, re.DOTALL) if json_match: ai_eval = json.loads(json_match.group()) result['penalty'] = min(5, max(0, int(ai_eval.get('penalty_score', 0)))) result['ai_analysis'] = ai_eval.get('analysis', '') return result else: raise ValueError("No JSON in AI response") except Exception as e: print(f"āš ļø Similarity AI failed: {e}") self._basic_similarity_penalty(similarity, result) else: self._basic_similarity_penalty(similarity, result) except Exception as e: print(f"āŒ analyze_similarity failed: {e}") return result def _basic_similarity_penalty(self, similarity, result): """Rule-based similarity penalty""" if similarity > 0.95: result['penalty'] = 5 elif similarity > 0.85: result['penalty'] = 3 elif similarity > 0.75: result['penalty'] = 1 else: result['penalty'] = 0 # ───────────────────────────────────────────── # 6. FEEDBACK & RECOMMENDATIONS # ───────────────────────────────────────────── def generate_feedback(self, analysis, question, difficulty): """Generate comprehensive AI-powered feedback""" correctness = analysis['correctness'] quality = analysis['code_quality'] efficiency = analysis['efficiency'] ai_detection = analysis['ai_detection'] similarity = analysis['similarity'] if not self.pipe: total = (correctness['score'] + quality['score'] + efficiency['score'] - ai_detection['penalty'] - similarity['penalty']) return ( f"Rule-based analysis completed. " f"Correctness: {correctness['score']}/40, " f"Quality: {quality['score']}/25, " f"Efficiency: {efficiency['score']}/20 | " f"Time complexity: {efficiency.get('time_complexity', 'Unknown')}. " f"Overall: {total}/100." ) try: feedback_prompt = f"""Give constructive feedback for this coding solution: Problem: {question} Difficulty: {difficulty} Scores: - Correctness: {correctness['score']}/40 - Code Quality: {quality['score']}/25 - Efficiency: {efficiency['score']}/20 - AI Penalty: {ai_detection['penalty']}/10 - Similarity Penalty: {similarity['penalty']}/5 Details: - Syntax Valid: {correctness['syntax_valid']} - Tests Passed: {correctness.get('test_cases_passed', 0)}/{correctness.get('total_test_cases', 0)} - Time Complexity: {efficiency['time_complexity']} - Space Complexity: {efficiency['space_complexity']} Write 3-5 sentences covering strengths, weaknesses, and specific improvements.""" messages = [ {"role": "system", "content": "You are an expert coding instructor. Give clear, constructive feedback in 3-5 sentences."}, {"role": "user", "content": feedback_prompt} ] return self._call_ai(messages) except Exception as e: print(f"āš ļø Feedback generation failed: {e}") total = (correctness['score'] + quality['score'] + efficiency['score'] - ai_detection['penalty'] - similarity['penalty']) return f"Analysis completed. Score: {total}/100. Time complexity: {efficiency.get('time_complexity', 'Unknown')}." def generate_recommendations(self, analysis): """Generate actionable recommendations""" if not self.pipe: return self._basic_recommendations(analysis) try: correctness = analysis['correctness'] quality = analysis['code_quality'] efficiency = analysis['efficiency'] ai_detection = analysis['ai_detection'] recommendations_prompt = f"""Generate 3-5 specific actionable recommendations: Scores: - Correctness: {correctness['score']}/40 - Quality: {quality['score']}/25 - Efficiency: {efficiency['score']}/20 - AI Detection Penalty: {ai_detection['penalty']} Issues: - Syntax Valid: {correctness['syntax_valid']} - Time Complexity: {efficiency['time_complexity']} - AI Indicators: {ai_detection['indicators']} Respond ONLY with valid JSON: {{"recommendations": ["action 1", "action 2", "action 3"]}}""" messages = [ {"role": "system", "content": "You are a coding mentor. Respond with valid JSON only, no markdown."}, {"role": "user", "content": recommendations_prompt} ] try: ai_response = self._call_ai(messages) json_match = re.search(r'\{.*?\}', ai_response, re.DOTALL) if json_match: ai_eval = json.loads(json_match.group()) recs = ai_eval.get('recommendations', []) if recs: return recs[:5] except Exception as e: print(f"āš ļø Recommendations AI failed: {e}") except Exception as e: print(f"āŒ generate_recommendations failed: {e}") return self._basic_recommendations(analysis) def _basic_recommendations(self, analysis): """Rule-based recommendations fallback""" recommendations = [] correctness = analysis['correctness'] quality = analysis['code_quality'] efficiency = analysis['efficiency'] ai_detection = analysis['ai_detection'] if not correctness['syntax_valid']: recommendations.append("Fix syntax errors before submission") if correctness.get('test_cases_passed', 0) < correctness.get('total_test_cases', 1): recommendations.append("Debug logic to pass all test cases") if quality['score'] < 15: recommendations.append("Improve code readability and add comments") if 'O(n²)' in efficiency.get('time_complexity', ''): recommendations.append("Optimize algorithm — consider using a hash map to reduce to O(n)") if 'O(n³)' in efficiency.get('time_complexity', ''): recommendations.append("Critical: algorithm is O(n³) or worse — needs major optimization") if ai_detection['penalty'] > 5: recommendations.append("Ensure code originality — avoid AI-generated patterns") if quality.get('comments', 0) == 0: recommendations.append("Add docstrings and inline comments for better documentation") return recommendations[:5] def execute_test_case(self, code, test_case, language): """Kept for API compatibility — use _run_python_tests for actual execution""" if language == 'python' and test_case: passed, total, _ = self._run_python_tests(code, [test_case]) return passed == total return True