From 7c2b6f34e9b2a03fa752070e75da21fe30250d71 Mon Sep 17 00:00:00 2001 From: Azaucifer <47237500+Azaucifer@users.noreply.github.com> Date: Mon, 7 Sep 2026 15:49:37 +0530 Subject: [PATCH] feat: add duplicate code detection --- analysis/duplicate_detection.py | 73 ++++++++++++++++++ analyzer.py | 7 +- reporting/json_report.py | 19 ++++- reporting/terminal.py | 19 ++++- test_analyzer.py | 126 +++++++++++++++++++++++++++++++- 5 files changed, 236 insertions(+), 8 deletions(-) create mode 100644 analysis/duplicate_detection.py diff --git a/analysis/duplicate_detection.py b/analysis/duplicate_detection.py new file mode 100644 index 0000000..4258929 --- /dev/null +++ b/analysis/duplicate_detection.py @@ -0,0 +1,73 @@ +import ast +import copy + +from analysis.file_analysis import parse_python_file + + +def extract_functions(tree): + functions = [] + + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + functions.append(node) + + return functions + + +def normalize_function(function): + function = copy.deepcopy(function) + function = ast.fix_missing_locations(function) + + for node in ast.walk(function): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + node.name = "FUNCTION" + + elif isinstance(node, ast.arg): + node.arg = "ARGUMENT" + + elif isinstance(node, ast.Name): + node.id = "VARIABLE" + + return ast.dump(function) + + +def find_duplicates(functions): + normalized = {} + + for file, function in functions: + key = normalize_function(function) + + if key not in normalized: + normalized[key] = [] + + normalized[key].append( + { + "file": file, + "name": function.name, + "start_line": function.lineno, + } + ) + + return [ + group + for group in normalized.values() + if len(group) > 1 + ] + + +def detect_duplicates(py_files): + functions = [] + + for file in py_files: + with file.open(encoding="utf-8") as f: + source = f.read() + + tree = parse_python_file(source, file) + + if tree is None: + continue + + for function in extract_functions(tree): + functions.append((file, function)) + + return find_duplicates(functions) diff --git a/analyzer.py b/analyzer.py index 890c32d..5390795 100644 --- a/analyzer.py +++ b/analyzer.py @@ -3,6 +3,7 @@ from cli.arguments import create_parser from analysis.file_analysis import analyze_file +from analysis.duplicate_detection import detect_duplicates from reporting.terminal import ( generate_codebase_summary, @@ -38,6 +39,8 @@ def analyze_codebase( output_file="codebase_report.json", ): results = [] + duplicates = detect_duplicates(py_files) + for file in py_files: metrics = analyze_file(file) @@ -51,11 +54,11 @@ def analyze_codebase( generate_report(metrics) print() - generate_codebase_summary(results) + generate_codebase_summary(results, duplicates) print() if generate_json: - generate_json_report(results, output_file) + generate_json_report(results, output_file, duplicates) if __name__ == "__main__": diff --git a/reporting/json_report.py b/reporting/json_report.py index 8a8fb6f..d5280d4 100644 --- a/reporting/json_report.py +++ b/reporting/json_report.py @@ -3,7 +3,11 @@ from analysis.quality import get_health_rating -def generate_json_report(results, output_file="codebase_report.json"): +def generate_json_report( + results, + output_file="codebase_report.json", + duplicates=None, +): if not results: print("No valid Python files found.") return @@ -30,6 +34,19 @@ def generate_json_report(results, output_file="codebase_report.json"): "average_health_score": round(average_health_score, 1), "rating": get_health_rating(average_health_score), }, + "duplicates": [ + { + "functions": [ + { + "file": function["file"].name, + "name": function["name"], + "start_line": function["start_line"], + } + for function in group + ] + } + for group in (duplicates or []) + ], "files": [], } diff --git a/reporting/terminal.py b/reporting/terminal.py index cc3effe..0d3a6f5 100644 --- a/reporting/terminal.py +++ b/reporting/terminal.py @@ -1,7 +1,7 @@ from analysis.quality import get_health_rating -def generate_codebase_summary(results): +def generate_codebase_summary(results, duplicates=None): if not results: print("No valid Python files found.") return @@ -51,6 +51,23 @@ def generate_codebase_summary(results): print(f" - {issue}") print() + print("\nDuplicate Code") + print("-" * 20) + + if not duplicates: + print("No duplicate functions detected.") + else: + print(f"Duplicate groups: {len(duplicates)}") + + for index, group in enumerate(duplicates, start=1): + print(f"\nGroup {index}") + + for function in group: + print( + f" {function['name']}() - " + f"{function['file'].name}:{function['start_line']}" + ) + def display_line_metrics(metrics): print("\nLines") diff --git a/test_analyzer.py b/test_analyzer.py index 5f00162..fa190f3 100644 --- a/test_analyzer.py +++ b/test_analyzer.py @@ -1,9 +1,10 @@ -import pytest -from pathlib import Path -import sys import ast -import os import json +import os +import sys +from pathlib import Path + +import pytest from analyzer import ( main, @@ -33,6 +34,10 @@ analyze_operations, ) +from analysis.duplicate_detection import ( + detect_duplicates, +) + from analysis.complexity import ( calculate_complexity, count_function_arguments, @@ -874,6 +879,56 @@ def test_main_json_output(tmp_path): os.chdir(original_cwd) +def test_main_json_duplicate_output(tmp_path): + """Test that duplicate functions are included in JSON output""" + first = tmp_path / "first.py" + second = tmp_path / "second.py" + + first.write_text( + "def add(a, b):\n" + " return a + b\n" + ) + + second.write_text( + "def calculate(x, y):\n" + " return x + y\n" + ) + + output_file = tmp_path / "codebase_report.json" + + original_argv = sys.argv.copy() + original_cwd = Path.cwd() + sys.argv = ["analyzer.py", str(tmp_path), "--json"] + + try: + os.chdir(tmp_path) + + main() + + assert output_file.exists() + + with open(output_file, encoding="utf-8") as f: + report = json.load(f) + + assert "duplicates" in report + assert len(report["duplicates"]) == 1 + + duplicate_functions = report["duplicates"][0]["functions"] + + assert len(duplicate_functions) == 2 + assert duplicate_functions[0]["file"] == "first.py" + assert duplicate_functions[0]["name"] == "add" + assert duplicate_functions[0]["start_line"] == 1 + + assert duplicate_functions[1]["file"] == "second.py" + assert duplicate_functions[1]["name"] == "calculate" + assert duplicate_functions[1]["start_line"] == 1 + + finally: + sys.argv = original_argv + os.chdir(original_cwd) + + def test_main_json_custom_output(tmp_path): """Test main with a custom JSON output filename""" test_file = tmp_path / "test.py" @@ -1028,3 +1083,66 @@ def test(): assert result["functions"] == 2 assert result["function_details"][0]["name"] == "test" assert result["function_details"][1]["name"] == "test" + + +def test_detect_duplicates(tmp_path): + """Test detection of structurally identical functions""" + first = tmp_path / "first.py" + second = tmp_path / "second.py" + + first.write_text( + "def add(a, b):\n" + " return a + b\n" + ) + + second.write_text( + "def calculate(x, y):\n" + " return x + y\n" + ) + + result = detect_duplicates([first, second]) + + assert len(result) == 1 + assert len(result[0]) == 2 + assert result[0][0]["name"] == "add" + assert result[0][1]["name"] == "calculate" + + +def test_no_duplicates(tmp_path): + """Test that different functions are not detected as duplicates""" + first = tmp_path / "first.py" + second = tmp_path / "second.py" + + first.write_text( + "def add(a, b):\n" + " return a + b\n" + ) + + second.write_text( + "def multiply(x, y):\n" + " return x * y\n" + ) + + result = detect_duplicates([first, second]) + + assert result == [] + + +def test_detect_duplicates_skips_syntax_errors(tmp_path): + """Test that files with syntax errors are skipped""" + valid_file = tmp_path / "valid.py" + invalid_file = tmp_path / "invalid.py" + + valid_file.write_text( + "def add(a, b):\n" + " return a + b\n" + ) + + invalid_file.write_text( + "def broken(:\n" + " return 1\n" + ) + + result = detect_duplicates([valid_file, invalid_file]) + + assert result == []