feat: add duplicate code detection

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)

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__":

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": [],
     }
 

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")

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 == []