dedup

#!/usr/bin/env python3

# Copyright (c) Thulio Ferraz Assis 2025
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
File Deduplication Tool
Finds duplicate files in a directory tree and replaces them with symlinks.
"""

import os
import sys
import hashlib
import argparse
from pathlib import Path
from collections import defaultdict
from typing import Dict, List, Tuple


def calculate_file_hash(filepath: Path, chunk_size: int = 8192) -> str:
    """Calculate SHA256 hash of a file."""
    sha256 = hashlib.sha256()
    try:
        with open(filepath, 'rb') as f:
            while chunk := f.read(chunk_size):
                sha256.update(chunk)
        return sha256.hexdigest()
    except (IOError, OSError) as e:
        print(f"Error reading file {filepath}: {e}")
        return None


def scan_directory(root_dir: Path, follow_symlinks: bool = False) -> Dict[str, List[Path]]:
    """
    Scan directory tree and group files by their hash.
    
    Returns:
        Dictionary mapping hash -> list of file paths with that hash
    """
    hash_to_files = defaultdict(list)
    total_files = 0
    skipped_symlinks = 0
    
    for dirpath, dirnames, filenames in os.walk(root_dir, followlinks=follow_symlinks):
        for filename in filenames:
            filepath = Path(dirpath) / filename
            
            if filepath.is_symlink():
                skipped_symlinks += 1
                continue
            
            if not filepath.is_file():
                continue
            
            file_hash = calculate_file_hash(filepath)
            if file_hash:
                hash_to_files[file_hash].append(filepath)
                total_files += 1
                
                if total_files % 100 == 0:
                    print(f"Processed {total_files} files...", end='\r')
    
    print(f"\nScanned {total_files} files, skipped {skipped_symlinks} symlinks")
    return hash_to_files


def find_duplicates(hash_to_files: Dict[str, List[Path]]) -> List[Tuple[Path, List[Path]]]:
    """
    Find groups of duplicate files.
    
    Returns:
        List of tuples (original_file, [duplicate_files])
    """
    duplicates = []
    
    for file_hash, file_list in hash_to_files.items():
        if len(file_list) > 1:
            # Sort by path length (shorter paths first) then alphabetically.
            # This helps keep the "original" in a more logical location.
            file_list.sort(key=lambda p: (len(str(p)), str(p)))
            
            original = file_list[0]
            copies = file_list[1:]
            duplicates.append((original, copies))
    
    return duplicates


def replace_with_symlink(original: Path, duplicate: Path, dry_run: bool = False) -> bool:
    """Replace duplicate file with a symlink to the original."""
    try:
        relative_path = os.path.relpath(original, duplicate.parent)
        
        if dry_run:
            print(f"Would create symlink: {duplicate} -> {relative_path}")
            return True
        
        duplicate.unlink()
        
        duplicate.symlink_to(relative_path)
        
        return True
    except Exception as e:
        print(f"Error replacing {duplicate} with symlink: {e}")
        return False


def format_size(size_bytes: int) -> str:
    """Format byte size in human-readable format."""
    for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
        if size_bytes < 1024.0:
            return f"{size_bytes:.2f} {unit}"
        size_bytes /= 1024.0
    return f"{size_bytes:.2f} PB"


def main():
    parser = argparse.ArgumentParser(
        description="Find duplicate files and replace them with symlinks"
    )
    parser.add_argument(
        "directory",
        type=Path,
        help="Directory to scan for duplicates"
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Show what would be done without making changes"
    )
    parser.add_argument(
        "--min-size",
        type=int,
        default=1,
        help="Minimum file size in bytes to consider (default: 1)"
    )
    parser.add_argument(
        "--interactive",
        action="store_true",
        help="Ask for confirmation before creating each symlink"
    )
    parser.add_argument(
        "--absolute-links",
        action="store_true",
        help="Create absolute symlinks instead of relative ones"
    )
    
    args = parser.parse_args()
    
    if not args.directory.exists():
        print(f"Error: Directory '{args.directory}' does not exist")
        sys.exit(1)
    
    if not args.directory.is_dir():
        print(f"Error: '{args.directory}' is not a directory")
        sys.exit(1)
    
    print(f"Scanning directory: {args.directory.absolute()}")
    
    hash_to_files = scan_directory(args.directory)
    duplicates = find_duplicates(hash_to_files)
    
    if not duplicates:
        print("No duplicate files found!")
        return
    
    total_duplicates = sum(len(copies) for _, copies in duplicates)
    total_size_saved = 0
    
    print(f"\nFound {len(duplicates)} groups of duplicate files")
    print(f"Total duplicate files: {total_duplicates}")
    
    replaced_count = 0
    
    for original, copies in duplicates:
        print(f"\n{'='*60}")
        print(f"Original: {original}")
        
        try:
            file_size = original.stat().st_size
            
            if file_size < args.min_size:
                print(f"Skipping (size {file_size} < {args.min_size} bytes)")
                continue
            
            print(f"Size: {format_size(file_size)}")
            print(f"Duplicates ({len(copies)}):")
            
            for copy in copies:
                print(f"  - {copy}")
            
            potential_savings = file_size * len(copies)
            print(f"Potential space savings: {format_size(potential_savings)}")
            
            for copy in copies:
                if args.interactive and not args.dry_run:
                    response = input(f"Replace {copy} with symlink? [y/N] ").lower()
                    if response != 'y':
                        continue
                
                if args.absolute_links:
                    # For absolute links, we need to modify the function.
                    try:
                        if args.dry_run:
                            print(f"Would create absolute symlink: {copy} -> {original.absolute()}")
                            success = True
                        else:
                            copy.unlink()
                            copy.symlink_to(original.absolute())
                            success = True
                    except Exception as e:
                        print(f"Error creating absolute symlink: {e}")
                        success = False
                else:
                    success = replace_with_symlink(original, copy, args.dry_run)
                
                if success:
                    replaced_count += 1
                    total_size_saved += file_size
                    
        except Exception as e:
            print(f"Error processing group: {e}")
            continue
    
    print(f"\n{'='*60}")
    print("Summary:")
    print(f"  Files replaced with symlinks: {replaced_count}")
    print(f"  Space saved: {format_size(total_size_saved)}")
    
    if args.dry_run:
        print("\nThis was a dry run. No changes were made.")


if __name__ == "__main__":
    main()