Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -3,12 +3,12 @@ name = "lenslogic"
version = "1.0.2"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.13"
requires-python = ">=3.8"
dependencies = []

[tool.ruff]
line-length = 120
target-version = "py311"
target-version = "py38"

[tool.ruff.lint]
select = [
Expand All @@ -21,6 +21,10 @@ select = [
]
ignore = [
"E501", # Line too long - let ruff-format handle it
"UP006", # Use `dict` instead of `Dict` - not compatible with Python 3.8
"UP007", # Use `X | Y` for type annotations - not compatible with Python 3.8
"UP035", # `typing.Dict` is deprecated - not in Python 3.8
"UP045", # Use `X | None` instead of `Optional[X]` - not compatible with Python 3.8
]

[tool.ruff.lint.per-file-ignores]
Expand Down
18 changes: 9 additions & 9 deletions src/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import logging
import sys
from pathlib import Path
from typing import Any
from typing import Any, Dict, List, Optional

from modules.backup_manager import BackupManager
from modules.config_wizard import ConfigurationWizard
Expand All @@ -24,7 +24,7 @@


class LensLogic:
def __init__(self, config_path: str | None = None, args: dict[str, Any] | None = None):
def __init__(self, config_path: Optional[str] = None, args: Optional[Dict[str, Any]] = None):
self.config_manager = ConfigManager(config_path)

# Store custom destination separately - don't save to config
Expand Down Expand Up @@ -68,7 +68,7 @@ def setup_logging(self):
],
)

def organize_photos(self, dry_run: bool = False, custom_destination: str | None = None):
def organize_photos(self, dry_run: bool = False, custom_destination: Optional[str] = None):
source_dir = Path(self.config.get("general", {}).get("source_directory", "."))

# Use custom destination if provided, or class custom_destination, otherwise use config destination
Expand Down Expand Up @@ -200,7 +200,7 @@ def organize_photos(self, dry_run: bool = False, custom_destination: str | None

return success_count > 0

def _collect_files(self, source_dir: Path) -> list[Path]:
def _collect_files(self, source_dir: Path) -> List[Path]:
files = []

all_extensions = set()
Expand Down Expand Up @@ -230,7 +230,7 @@ def analyze_library(self):
for ext, count in sorted(stats["file_types"].items(), key=lambda x: x[1], reverse=True)[:10]:
self.progress_tracker.console.print(f" {ext}: {count} files")

def export_gps_locations(self, output_path: str | None = None):
def export_gps_locations(self, output_path: Optional[str] = None):
if not output_path:
output_path = "photo_locations.kml"

Expand Down Expand Up @@ -348,7 +348,7 @@ def run_interactive(self):
self.progress_tracker.print_info("Goodbye!")
break

def generate_advanced_statistics(self, output_dir: str | None = None):
def generate_advanced_statistics(self, output_dir: Optional[str] = None):
"""Generate comprehensive statistics with charts"""
source_dir = self.config.get("general", {}).get("source_directory", ".")
files = self._collect_files(Path(source_dir))
Expand Down Expand Up @@ -462,7 +462,7 @@ def detect_sessions(self, organize_by_sessions: bool = False):
f"Organized {result['files_organized']} files into {result['sessions_processed']} session folders"
)

def optimize_for_social_media(self, platform: str, format_type: str = "post", output_dir: str | None = None):
def optimize_for_social_media(self, platform: str, format_type: str = "post", output_dir: Optional[str] = None):
"""Optimize photos for social media platforms"""
source_dir = self.config.get("general", {}).get("source_directory", ".")
files = self._collect_files(Path(source_dir))
Expand Down Expand Up @@ -502,7 +502,7 @@ def optimize_for_social_media(self, platform: str, format_type: str = "post", ou
output_path = Path(results[0]["output_path"]).parent
self.progress_tracker.print_info(f"Optimized images saved to: {output_path}")

def backup_photos(self, destinations: list[str] | None = None, verify: bool = True):
def backup_photos(self, destinations: Optional[List[str]] = None, verify: bool = True):
"""Backup organized photos to specified destinations"""
# Backup the organized photos (destination directory), not the source directory
source_dir = self.config.get("general", {}).get("destination_directory", "./organized")
Expand Down Expand Up @@ -567,7 +567,7 @@ def run_config_wizard(self, quick: bool = False):
else:
self.progress_tracker.print_info("Configuration wizard cancelled or failed")

def analyze_xmp_library(self, library_path: str | None = None, output_dir: str | None = None):
def analyze_xmp_library(self, library_path: Optional[str] = None, output_dir: Optional[str] = None):
"""Analyze photo library using XMP sidecar files"""
if not library_path:
library_path = self.config.get("general", {}).get("source_directory", ".")
Expand Down
32 changes: 16 additions & 16 deletions src/modules/backup_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,15 +5,15 @@
import time
from datetime import datetime
from pathlib import Path
from typing import Any
from typing import Any, Dict, List, Optional

from send2trash import send2trash

logger = logging.getLogger(__name__)


class BackupManager:
def __init__(self, config: dict[str, Any]):
def __init__(self, config: Dict[str, Any]):
self.config = config
self.backup_config = config.get("backup", {})
self.verification_enabled = self.backup_config.get("enable_verification", True)
Expand All @@ -24,7 +24,7 @@ def __init__(self, config: dict[str, Any]):

self.checksum_cache = self._load_checksum_cache()

def _load_checksum_cache(self) -> dict[str, dict[str, Any]]:
def _load_checksum_cache(self) -> Dict[str, Dict[str, Any]]:
"""Load checksum cache from file"""
try:
if Path(self.checksum_cache_file).exists():
Expand All @@ -43,7 +43,7 @@ def _save_checksum_cache(self):
except Exception as e:
logger.warning(f"Could not save checksum cache: {e}")

def calculate_file_checksum(self, file_path: str, algorithm: str = "sha256") -> str | None:
def calculate_file_checksum(self, file_path: str, algorithm: str = "sha256") -> Optional[str]:
"""Calculate checksum for a file"""
file_path_obj = Path(file_path)

Expand Down Expand Up @@ -84,7 +84,7 @@ def calculate_file_checksum(self, file_path: str, algorithm: str = "sha256") ->
logger.error(f"Error calculating checksum for {file_path_obj}: {e}")
return None

def verify_backup(self, source_dir: str, backup_dir: str, quick_mode: bool = False) -> dict[str, Any]:
def verify_backup(self, source_dir: str, backup_dir: str, quick_mode: bool = False) -> Dict[str, Any]:
"""Verify backup integrity against source"""
source_path = Path(source_dir)
backup_path = Path(backup_dir)
Expand Down Expand Up @@ -158,7 +158,7 @@ def verify_backup(self, source_dir: str, backup_dir: str, quick_mode: bool = Fal

return result

def _get_file_list(self, directory: Path) -> list[Path]:
def _get_file_list(self, directory: Path) -> List[Path]:
"""Get list of all files in directory, excluding patterns"""
files = []

Expand Down Expand Up @@ -205,7 +205,7 @@ def _full_file_compare(self, file1: Path, file2: Path) -> bool:
except Exception:
return False

def incremental_sync(self, source_dir: str, destination_dirs: list[str], dry_run: bool = False) -> dict[str, Any]:
def incremental_sync(self, source_dir: str, destination_dirs: List[str], dry_run: bool = False) -> Dict[str, Any]:
"""Perform incremental sync to multiple destinations"""
source_path = Path(source_dir)

Expand Down Expand Up @@ -280,8 +280,8 @@ def incremental_sync(self, source_dir: str, destination_dirs: list[str], dry_run
return result

def _sync_to_destination(
self, source_path: Path, source_index: dict, dest_dir: str, dry_run: bool
) -> dict[str, Any]:
self, source_path: Path, source_index: Dict, dest_dir: str, dry_run: bool
) -> Dict[str, Any]:
"""Sync source to a single destination"""
dest_path = Path(dest_dir)

Expand Down Expand Up @@ -360,7 +360,7 @@ def _sync_to_destination(

return result

def _needs_update(self, source_info: dict, dest_info: dict) -> bool:
def _needs_update(self, source_info: Dict, dest_info: Dict) -> bool:
"""Check if destination file needs updating"""
# Check size first (quick)
if source_info["size"] != dest_info["size"]:
Expand All @@ -384,7 +384,7 @@ def _copy_file(self, source_path: Path, dest_path: Path) -> bool:
logger.error(f"Error copying {source_path} to {dest_path}: {e}")
return False

def get_backup_status(self, source_dir: str, backup_dirs: list[str]) -> dict[str, Any]:
def get_backup_status(self, source_dir: str, backup_dirs: List[str]) -> Dict[str, Any]:
"""Get status of all configured backups"""
status = {
"source_directory": source_dir,
Expand Down Expand Up @@ -454,7 +454,7 @@ def get_backup_status(self, source_dir: str, backup_dirs: list[str]) -> dict[str

return status

def cleanup_old_backups(self, backup_dir: str, keep_days: int = 30, dry_run: bool = False) -> dict[str, Any]:
def cleanup_old_backups(self, backup_dir: str, keep_days: int = 30, dry_run: bool = False) -> Dict[str, Any]:
"""Clean up old backup files"""
backup_path = Path(backup_dir)

Expand Down Expand Up @@ -501,11 +501,11 @@ def restore_from_backup(
self,
backup_dir: str,
restore_dir: str,
file_patterns: list[str] | None = None,
file_patterns: Optional[List[str]] = None,
preserve_structure: bool = True,
overwrite_newer: bool = True,
dry_run: bool = False,
) -> dict[str, Any]:
) -> Dict[str, Any]:
"""Restore files from backup to specified directory"""
backup_path = Path(backup_dir)
restore_path = Path(restore_dir)
Expand Down Expand Up @@ -590,7 +590,7 @@ def restore_from_backup(
result["restore_time"] = time.time() - start_time
return result

def list_backup_contents(self, backup_dir: str, show_details: bool = False) -> dict[str, Any]:
def list_backup_contents(self, backup_dir: str, show_details: bool = False) -> Dict[str, Any]:
"""List contents of a backup directory with optional details"""
backup_path = Path(backup_dir)

Expand Down Expand Up @@ -648,7 +648,7 @@ def list_backup_contents(self, backup_dir: str, show_details: bool = False) -> d

return result

def get_restore_candidates(self, backup_dirs: list[str]) -> dict[str, Any]:
def get_restore_candidates(self, backup_dirs: List[str]) -> Dict[str, Any]:
"""Get information about available restore sources"""
candidates = {
"available_backups": [],
Expand Down
19 changes: 10 additions & 9 deletions src/modules/duplicate_detector.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import hashlib
import logging
from pathlib import Path
from typing import Dict, List, Optional, Tuple

import numpy as np
from PIL import Image
Expand All @@ -9,15 +10,15 @@


class DuplicateDetector:
def __init__(self, config: dict):
def __init__(self, config: Dict):
self.config = config.get("duplicate_detection", {})
self.method = self.config.get("method", "hash")
self.threshold = self.config.get("threshold", 0.95)
self.action = self.config.get("action", "skip")
self.duplicate_folder = self.config.get("duplicate_folder", "DUPLICATES")
self.hash_cache = {}

def find_duplicates(self, file_paths: list[str]) -> dict[str, list[str]]:
def find_duplicates(self, file_paths: List[str]) -> Dict[str, List[str]]:
duplicates = {}

if self.method == "hash":
Expand All @@ -29,7 +30,7 @@ def find_duplicates(self, file_paths: list[str]) -> dict[str, list[str]]:

return duplicates

def _find_by_hash(self, file_paths: list[str]) -> dict[str, list[str]]:
def _find_by_hash(self, file_paths: List[str]) -> Dict[str, List[str]]:
hash_groups = {}

for file_path in file_paths:
Expand All @@ -44,7 +45,7 @@ def _find_by_hash(self, file_paths: list[str]) -> dict[str, list[str]]:

return duplicates

def _calculate_file_hash(self, file_path: str) -> str | None:
def _calculate_file_hash(self, file_path: str) -> Optional[str]:
if file_path in self.hash_cache:
return self.hash_cache[file_path]

Expand All @@ -62,7 +63,7 @@ def _calculate_file_hash(self, file_path: str) -> str | None:
logger.error(f"Error calculating hash for {file_path}: {e}")
return None

def _find_by_pixels(self, file_paths: list[str]) -> dict[str, list[str]]:
def _find_by_pixels(self, file_paths: List[str]) -> Dict[str, List[str]]:
duplicates = {}
processed = set()
image_data_cache = {}
Expand Down Expand Up @@ -96,7 +97,7 @@ def _find_by_pixels(self, file_paths: list[str]) -> dict[str, list[str]]:

return duplicates

def _get_image_array(self, file_path: str, cache: dict) -> np.ndarray | None:
def _get_image_array(self, file_path: str, cache: Dict) -> Optional[np.ndarray]:
if file_path in cache:
return cache[file_path]

Expand Down Expand Up @@ -127,7 +128,7 @@ def _compare_images(self, img1: np.ndarray, img2: np.ndarray) -> bool:
logger.debug(f"Error comparing images: {e}")
return False

def _find_by_histogram(self, file_paths: list[str]) -> dict[str, list[str]]:
def _find_by_histogram(self, file_paths: List[str]) -> Dict[str, List[str]]:
duplicates = {}
processed = set()
histogram_cache = {}
Expand Down Expand Up @@ -161,7 +162,7 @@ def _find_by_histogram(self, file_paths: list[str]) -> dict[str, list[str]]:

return duplicates

def _calculate_histogram(self, file_path: str, cache: dict) -> np.ndarray | None:
def _calculate_histogram(self, file_path: str, cache: Dict) -> Optional[np.ndarray]:
if file_path in cache:
return cache[file_path]

Expand Down Expand Up @@ -213,7 +214,7 @@ def is_duplicate(self, file1: str, file2: str) -> bool:

return False

def handle_duplicate(self, original: str, duplicate: str, destination_base: str) -> tuple[str, str]:
def handle_duplicate(self, original: str, duplicate: str, destination_base: str) -> Tuple[str, str]:
if self.action == "skip":
return "skip", f"Skipping duplicate: {duplicate}"

Expand Down
Loading
Loading