yakamo/o-phone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ec56e7cf3839ce6b8229a21e31e71e02a8cb0206
o-phone/o-phone.py
T

1833 lines
37 KiB
Python
Raw Normal View History

Upload files to "/"
2026-08-14 21:06:25 +01:00
#!/usr/bin/env python3
import argparse
import hashlib
import os
import random
import shutil
import sqlite3
import subprocess
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
# ============================================================================
# CONFIGURATION
# ============================================================================
DATABASE_NAME = ".photo_organiser.db"
IMAGE_EXTENSIONS = {
".jpg", ".jpeg", ".jpe",
".png",
".gif",
".bmp",
".tif", ".tiff",
".webp",
".heic", ".heif",
".avif",
# RAW
".dng",
".raw",
".arw",
".cr2", ".cr3",
".nef", ".nrw",
".orf",
".rw2",
".raf",
".pef",
".srw",
".x3f",
}
VIDEO_EXTENSIONS = {
".mp4",
".mov",
".m4v",
".avi",
".mkv",
".wmv",
".webm",
".3gp",
".3g2",
".mts",
".m2ts",
".ts",
".mpg",
".mpeg",
".mxf",
}
MONTH_NAMES = [
"January",
"February",
"March",
"April",
"May",
"June",
"July",
"August",
"September",
"October",
"November",
"December",
]
# ============================================================================
# DATABASE
# ============================================================================
def database_path(destination: Path) -> Path:
return destination / DATABASE_NAME
def connect_database(destination: Path) -> sqlite3.Connection:
db_path = database_path(destination)
connection = sqlite3.connect(db_path)
# Good settings for a database on an external drive.
connection.execute("PRAGMA journal_mode=WAL")
connection.execute("PRAGMA synchronous=FULL")
connection.execute("PRAGMA foreign_keys=ON")
return connection
def create_database(connection: sqlite3.Connection):
connection.execute(
"""
CREATE TABLE IF NOT EXISTS files (
id INTEGER PRIMARY KEY,
sha256 TEXT NOT NULL UNIQUE,
size INTEGER NOT NULL,
original_name TEXT NOT NULL,
stored_name TEXT NOT NULL,
relative_path TEXT NOT NULL UNIQUE,
date_taken TEXT NOT NULL,
file_type TEXT NOT NULL,
device_make TEXT,
device_model TEXT,
imported_at TEXT NOT NULL
)
"""
)
connection.execute(
"""
CREATE INDEX IF NOT EXISTS idx_files_sha256
ON files(sha256)
"""
)
connection.execute(
"""
CREATE INDEX IF NOT EXISTS idx_files_date_taken
ON files(date_taken)
"""
)
connection.execute(
"""
CREATE INDEX IF NOT EXISTS idx_files_device
ON files(device_make, device_model)
"""
)
connection.commit()
def hash_exists(
connection: sqlite3.Connection,
file_hash: str,
):
cursor = connection.execute(
"""
SELECT id, relative_path
FROM files
WHERE sha256 = ?
""",
(file_hash,),
)
return cursor.fetchone()
# ============================================================================
# FILE HASHING
# ============================================================================
def calculate_sha256(
path: Path,
chunk_size: int = 8 * 1024 * 1024,
) -> str:
"""
Calculate SHA-256 without loading the entire file into memory.
"""
sha256 = hashlib.sha256()
with path.open("rb") as file:
while True:
chunk = file.read(chunk_size)
if not chunk:
break
sha256.update(chunk)
return sha256.hexdigest()
# ============================================================================
# METADATA
# ============================================================================
def parse_metadata_date(value: str):
if not value:
return None
value = value.strip()
# Remove fractional seconds.
if "." in value:
value = value.split(".", 1)[0]
formats = [
"%Y:%m:%d %H:%M:%S",
"%Y-%m-%d %H:%M:%S",
"%Y-%m-%dT%H:%M:%S",
"%Y-%m-%d %H:%M",
"%Y:%m:%d %H:%M",
"%Y-%m-%d",
"%Y:%m:%d",
]
for fmt in formats:
try:
return datetime.strptime(value, fmt)
except ValueError:
continue
return None
def run_exiftool(path: Path, tag: str):
"""
Get a single metadata tag from ExifTool.
Returns None if ExifTool isn't installed or the tag isn't available.
"""
try:
result = subprocess.run(
[
"exiftool",
"-s3",
f"-{tag}",
str(path),
],
capture_output=True,
text=True,
timeout=15,
)
except (
FileNotFoundError,
subprocess.SubprocessError,
OSError,
):
return None
if result.returncode != 0:
return None
value = result.stdout.strip()
return value or None
def get_exiftool_date(path: Path):
"""
Try the most useful capture/creation date fields.
"""
tags = [
"DateTimeOriginal",
"CreateDate",
"MediaCreateDate",
"CreationDate",
]
for tag in tags:
value = run_exiftool(path, tag)
if value:
parsed = parse_metadata_date(value)
if parsed:
return parsed
return None
def get_pillow_date(path: Path):
"""
Optional fallback for image EXIF.
Pillow is not required.
"""
try:
from PIL import Image
except ImportError:
return None
try:
with Image.open(path) as image:
exif = image.getexif()
if not exif:
return None
# 36867 = DateTimeOriginal
# 36868 = DateTimeDigitized
# 306 = DateTime
for tag in (36867, 36868, 306):
value = exif.get(tag)
if value:
parsed = parse_metadata_date(
str(value)
)
if parsed:
return parsed
except Exception:
pass
return None
def get_file_date(path: Path) -> datetime:
"""
Date priority:
1. ExifTool metadata
2. Pillow EXIF
3. Filesystem modification time
"""
date = get_exiftool_date(path)
if date:
return date
if path.suffix.lower() in IMAGE_EXTENSIONS:
date = get_pillow_date(path)
if date:
return date
return datetime.fromtimestamp(
path.stat().st_mtime
)
def get_device_metadata(path: Path):
"""
Return:
(make, model)
from ExifTool.
Examples:
("Motorola", "moto g13")
("Apple", "iPhone 15")
("GoPro", "HERO12 Black")
"""
make = run_exiftool(path, "Make")
model = run_exiftool(path, "Model")
if make:
make = make.strip()
if model:
model = model.strip()
return make, model
# ============================================================================
# FILE CLASSIFICATION
# ============================================================================
def classify_file(path: Path) -> str:
extension = path.suffix.lower()
# PNG is deliberately treated as screenshot.
if extension == ".png":
return "screenshot"
if extension in IMAGE_EXTENSIONS:
return "image"
if extension in VIDEO_EXTENSIONS:
return "video"
return "unknown"
# ============================================================================
# DESTINATION PATH
# ============================================================================
def unknown_folder(
destination: Path,
extension: str,
) -> Path:
if extension:
clean_extension = (
extension
.lower()
.lstrip(".")
)
return destination / f"u-{clean_extension}"
return destination / "u-no-extension"
def get_destination_directory(
destination: Path,
source_file: Path,
file_type: str,
date: datetime,
) -> Path:
if file_type == "screenshot":
root = destination / "screen shots"
elif file_type in ("image", "video"):
root = destination
else:
return unknown_folder(
destination,
source_file.suffix,
)
year = date.strftime("%Y")
month = MONTH_NAMES[
date.month - 1
]
day = date.strftime(
"%d-%m-%Y"
)
return (
root
/ year
/ month
/ day
)
# ============================================================================
# FILE NAMING
# ============================================================================
def random_six_digits() -> str:
return (
f"{random.SystemRandom().randint(0, 999999):06d}"
)
def create_destination_filename(
directory: Path,
date: datetime,
extension: str,
) -> Path:
directory.mkdir(
parents=True,
exist_ok=True,
)
date_string = date.strftime(
"%d-%m-%Y"
)
for _ in range(10000):
random_number = random_six_digits()
filename = (
f"{date_string}-"
f"{random_number}"
f"{extension.lower()}"
)
candidate = directory / filename
if not candidate.exists():
return candidate
raise RuntimeError(
"Could not generate a unique filename."
)
# ============================================================================
# SOURCE FILE DISCOVERY
# ============================================================================
def find_source_files(source: Path):
for path in source.rglob("*"):
if not path.is_file():
continue
# macOS resource fork.
if path.name.startswith("._"):
continue
# macOS directory metadata.
if path.name == ".DS_Store":
continue
yield path
# ============================================================================
# DESTINATION SAFETY
# ============================================================================
def destination_contains_library_files(
destination: Path,
) -> bool:
database = database_path(destination)
for path in destination.rglob("*"):
if not path.is_file():
continue
if path == database:
continue
if path.name.startswith(
database.name
):
continue
if path.name.startswith("._"):
continue
if path.name == ".DS_Store":
continue
return True
return False
# ============================================================================
# SAFE COPY
# ============================================================================
def safe_copy_and_verify(
source: Path,
destination: Path,
source_hash: str,
):
"""
Copy to a temporary file.
Hash the temporary copy.
Only rename to the final filename after the hash matches.
This protects against:
- incomplete copies
- disconnected drives
- corrupted transfers
"""
destination.parent.mkdir(
parents=True,
exist_ok=True,
)
temporary_path = None
try:
fd, temp_name = tempfile.mkstemp(
prefix=".photo-import-",
suffix=".tmp",
dir=str(destination.parent),
)
os.close(fd)
temporary_path = Path(temp_name)
shutil.copy2(
source,
temporary_path,
)
copied_hash = calculate_sha256(
temporary_path
)
if copied_hash != source_hash:
raise RuntimeError(
"Hash mismatch after copy.\n"
f"Source: {source_hash}\n"
f"Copy: {copied_hash}"
)
# Same filesystem = atomic rename.
temporary_path.replace(
destination
)
temporary_path = None
finally:
if temporary_path is not None:
try:
temporary_path.unlink()
except OSError:
pass
# ============================================================================
# IMPORT ONE FILE
# ============================================================================
def process_file(
source_file: Path,
source_root: Path,
destination_root: Path,
connection: sqlite3.Connection,
dry_run: bool,
move: bool,
):
relative_source = (
source_file.relative_to(
source_root
)
)
print()
print(
f"Processing: {relative_source}"
)
# ------------------------------------------------------------------------
# Hash source
# ------------------------------------------------------------------------
try:
source_hash = calculate_sha256(
source_file
)
except Exception as exc:
print(
f" ERROR hashing: {exc}"
)
return "error"
# ------------------------------------------------------------------------
# Duplicate check
# ------------------------------------------------------------------------
existing = hash_exists(
connection,
source_hash,
)
if existing:
_, existing_path = existing
print(" DUPLICATE")
print(
f" Existing file: {existing_path}"
)
return "duplicate"
# ------------------------------------------------------------------------
# Metadata
# ------------------------------------------------------------------------
try:
date_taken = get_file_date(
source_file
)
except Exception as exc:
print(
f" ERROR getting date: {exc}"
)
return "error"
file_type = classify_file(
source_file
)
device_make, device_model = (
get_device_metadata(
source_file
)
)
# ------------------------------------------------------------------------
# Destination
# ------------------------------------------------------------------------
destination_directory = (
get_destination_directory(
destination_root,
source_file,
file_type,
date_taken,
)
)
try:
destination_file = (
create_destination_filename(
destination_directory,
date_taken,
source_file.suffix,
)
)
except Exception as exc:
print(
f" ERROR creating filename: {exc}"
)
return "error"
relative_destination = (
destination_file.relative_to(
destination_root
)
)
print(
f" Type: {file_type}"
)
print(
f" Date: "
f"{date_taken.strftime('%d-%m-%Y')}"
)
if device_make or device_model:
device = " ".join(
value
for value in (
device_make,
device_model,
)
if value
)
print(
f" Device: {device}"
)
else:
print(
" Device: unknown"
)
print(
f" SHA-256: {source_hash}"
)
print(
f" Destination: {relative_destination}"
)
# ------------------------------------------------------------------------
# Dry run
# ------------------------------------------------------------------------
if dry_run:
print(
" DRY RUN - nothing copied"
)
return "dry-run"
# ------------------------------------------------------------------------
# Copy + verify + database transaction
# ------------------------------------------------------------------------
try:
# FIRST:
# Copy and verify the physical file.
safe_copy_and_verify(
source_file,
destination_file,
source_hash,
)
# SECOND:
# Only after successful verification do we write to SQLite.
imported_at = datetime.now(
timezone.utc
).isoformat()
connection.execute(
"BEGIN"
)
try:
connection.execute(
"""
INSERT INTO files (
sha256,
size,
original_name,
stored_name,
relative_path,
date_taken,
file_type,
device_make,
device_model,
imported_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
source_hash,
source_file.stat().st_size,
source_file.name,
destination_file.name,
str(relative_destination),
date_taken.isoformat(),
file_type,
device_make,
device_model,
imported_at,
),
)
connection.commit()
except Exception:
connection.rollback()
# The database transaction failed.
# Remove the copied file so filesystem and database
# remain consistent.
try:
destination_file.unlink()
except OSError:
pass
raise
# --------------------------------------------------------------------
# Optional source deletion
# --------------------------------------------------------------------
if move:
source_file.unlink()
print(" IMPORTED")
return (
"moved"
if move
else "imported"
)
except Exception as exc:
print(
f" ERROR: {exc}"
)
return "error"
# ============================================================================
# NORMAL IMPORT
# ============================================================================
def import_files(
source: Path,
destination: Path,
dry_run: bool,
move: bool,
):
destination.mkdir(
parents=True,
exist_ok=True,
)
db_path = database_path(
destination
)
# ------------------------------------------------------------------------
# Safety check
#
# If files already exist in the destination but there is no DB,
# refuse to continue.
#
# This prevents a user accidentally importing thousands of duplicates
# into an existing library.
# ------------------------------------------------------------------------
if (
not db_path.exists()
and destination_contains_library_files(
destination
)
):
print(
"ERROR: The destination already contains files,"
)
print(
"but there is no photo organiser database."
)
print()
print(
"Before importing, build the database with:"
)
print()
print(
f'python3 organise_phone.py '
f'--build-db "{destination}"'
)
print()
print(
"This is a safety feature to prevent accidental"
)
print(
"duplicate imports."
)
sys.exit(1)
connection = connect_database(
destination
)
try:
create_database(
connection
)
files = list(
find_source_files(
source
)
)
print("=" * 70)
print("PHOTO ORGANISER")
print("=" * 70)
print(
f"Source: {source}"
)
print(
f"Destination: {destination}"
)
print(
f"Database: {db_path}"
)
print(
f"Operation: "
f"{'MOVE' if move else 'COPY'}"
)
print(
f"Mode: "
f"{'DRY RUN' if dry_run else 'LIVE'}"
)
print()
print(
f"Found {len(files)} files."
)
print("=" * 70)
imported = 0
duplicates = 0
errors = 0
dry_runs = 0
moved = 0
for source_file in files:
result = process_file(
source_file=source_file,
source_root=source,
destination_root=destination,
connection=connection,
dry_run=dry_run,
move=move,
)
if result == "imported":
imported += 1
elif result == "moved":
moved += 1
elif result == "duplicate":
duplicates += 1
elif result == "error":
errors += 1
elif result == "dry-run":
dry_runs += 1
print()
print("=" * 70)
print("IMPORT COMPLETE")
print("=" * 70)
if dry_run:
print(
f"Would import: {dry_runs}"
)
else:
print(
f"Imported: {imported}"
)
if moved:
print(
f"Moved: {moved}"
)
print(
f"Duplicates skipped: {duplicates}"
)
print(
f"Errors: {errors}"
)
print("=" * 70)
finally:
connection.close()
# ============================================================================
# BUILD DATABASE
# ============================================================================
def build_database(
destination: Path,
):
"""
Build a database for an existing library.
This does not move, rename or delete anything.
"""
db_path = database_path(
destination
)
connection = connect_database(
destination
)
try:
create_database(
connection
)
files = list(
find_source_files(
destination
)
)
print("=" * 70)
print("BUILDING PHOTO DATABASE")
print("=" * 70)
print(
f"Library: {destination}"
)
print(
f"Database: {db_path}"
)
print(
f"Files found: {len(files)}"
)
print()
added = 0
already_indexed = 0
duplicates = 0
errors = 0
for path in files:
# Never index the database itself.
if path == db_path:
continue
# SQLite WAL / SHM files.
if path.name in (
f"{DATABASE_NAME}-wal",
f"{DATABASE_NAME}-shm",
):
continue
try:
relative_path = (
path.relative_to(
destination
)
)
# Check whether this exact path is already in DB.
path_exists = connection.execute(
"""
SELECT id
FROM files
WHERE relative_path = ?
""",
(str(relative_path),),
).fetchone()
if path_exists:
already_indexed += 1
continue
print(
f"Hashing: {relative_path}"
)
file_hash = (
calculate_sha256(
path
)
)
existing = hash_exists(
connection,
file_hash,
)
if existing:
print(
" DUPLICATE HASH:"
)
print(
f" {existing[1]}"
)
duplicates += 1
continue
date_taken = (
get_file_date(
path
)
)
device_make, device_model = (
get_device_metadata(
path
)
)
connection.execute(
"""
INSERT INTO files (
sha256,
size,
original_name,
stored_name,
relative_path,
date_taken,
file_type,
device_make,
device_model,
imported_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
file_hash,
path.stat().st_size,
path.name,
path.name,
str(relative_path),
date_taken.isoformat(),
classify_file(path),
device_make,
device_model,
datetime.now(
timezone.utc
).isoformat(),
),
)
connection.commit()
added += 1
except Exception as exc:
print(
f" ERROR: {exc}"
)
errors += 1
print()
print("=" * 70)
print("DATABASE BUILD COMPLETE")
print("=" * 70)
print(
f"Added: {added}"
)
print(
f"Already indexed: {already_indexed}"
)
print(
f"Duplicate hashes: {duplicates}"
)
print(
f"Errors: {errors}"
)
print("=" * 70)
finally:
connection.close()
# ============================================================================
# VERIFY DATABASE
# ============================================================================
def verify_database(
destination: Path,
deep: bool,
):
"""
Verify database ↔ filesystem consistency.
Normal mode checks:
- DB files exist
- sizes match
- every library file is indexed
Deep mode additionally:
- hashes every file
- compares against DB SHA-256
"""
db_path = database_path(
destination
)
if not db_path.exists():
print(
"ERROR: Database does not exist:"
)
print(
db_path
)
print()
print(
"Build it with:"
)
print()
print(
f'python3 organise_phone.py '
f'--build-db "{destination}"'
)
sys.exit(1)
connection = connect_database(
destination
)
try:
create_database(
connection
)
print("=" * 70)
print("VERIFYING PHOTO LIBRARY")
print("=" * 70)
print(
f"Library: {destination}"
)
print(
f"Database: {db_path}"
)
print(
f"Mode: "
f"{'DEEP - hash every file' if deep else 'NORMAL'}"
)
print()
rows = connection.execute(
"""
SELECT
id,
sha256,
size,
relative_path
FROM files
ORDER BY id
"""
).fetchall()
database_paths = set()
missing = 0
size_mismatches = 0
hash_mismatches = 0
errors = 0
# --------------------------------------------------------------------
# Check every database entry
# --------------------------------------------------------------------
for (
row_id,
expected_hash,
expected_size,
relative_path,
) in rows:
database_paths.add(
relative_path
)
actual_path = (
destination
/ relative_path
)
if not actual_path.exists():
print(
f"MISSING: {relative_path}"
)
missing += 1
continue
if not actual_path.is_file():
print(
f"NOT A FILE: {relative_path}"
)
errors += 1
continue
try:
actual_size = (
actual_path.stat().st_size
)
if actual_size != expected_size:
print(
f"SIZE MISMATCH: "
f"{relative_path}"
)
size_mismatches += 1
continue
if deep:
print(
f"Hashing: "
f"{relative_path}"
)
actual_hash = (
calculate_sha256(
actual_path
)
)
if (
actual_hash
!= expected_hash
):
print(
f"HASH MISMATCH: "
f"{relative_path}"
)
hash_mismatches += 1
except Exception as exc:
print(
f"ERROR: "
f"{relative_path}: "
f"{exc}"
)
errors += 1
# --------------------------------------------------------------------
# Find files on disk that aren't in DB
# --------------------------------------------------------------------
unindexed = []
for path in find_source_files(
destination
):
if path == db_path:
continue
if path.name in (
f"{DATABASE_NAME}-wal",
f"{DATABASE_NAME}-shm",
):
continue
try:
relative_path = str(
path.relative_to(
destination
)
)
except ValueError:
continue
if (
relative_path
not in database_paths
):
unindexed.append(
relative_path
)
# --------------------------------------------------------------------
# Results
# --------------------------------------------------------------------
print()
print("=" * 70)
print("VERIFICATION COMPLETE")
print("=" * 70)
print(
f"Database records: {len(rows)}"
)
print(
f"Missing files: {missing}"
)
print(
f"Size mismatches: {size_mismatches}"
)
if deep:
print(
f"Hash mismatches: {hash_mismatches}"
)
print(
f"Unindexed files: {len(unindexed)}"
)
print(
f"Other errors: {errors}"
)
if unindexed:
print()
print(
"UNINDEXED FILES:"
)
for path in unindexed:
print(
f" {path}"
)
print()
if (
missing == 0
and size_mismatches == 0
and hash_mismatches == 0
and len(unindexed) == 0
and errors == 0
):
print(
"RESULT: LIBRARY IS CONSISTENT"
)
else:
print(
"RESULT: PROBLEMS FOUND"
)
print("=" * 70)
finally:
connection.close()
# ============================================================================
# MAIN
# ============================================================================
def main():
parser = argparse.ArgumentParser(
description=(
"Organise photos and videos into a date-based "
"library with SHA-256 duplicate detection."
)
)
parser.add_argument(
"source",
type=Path,
nargs="?",
help="Source folder",
)
parser.add_argument(
"destination",
type=Path,
nargs="?",
help="Destination folder / external drive",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Show what would happen without copying",
)
parser.add_argument(
"--move",
action="store_true",
help="Move files instead of copying them",
)
parser.add_argument(
"--build-db",
metavar="DESTINATION",
type=Path,
help="Build database for an existing library",
)
parser.add_argument(
"--verify",
metavar="DESTINATION",
type=Path,
help="Verify database against library",
)
parser.add_argument(
"--deep",
action="store_true",
help=(
"With --verify, calculate SHA-256 for every "
"file and verify its contents"
),
)
args = parser.parse_args()
# ------------------------------------------------------------------------
# BUILD DATABASE
# ------------------------------------------------------------------------
if args.build_db:
destination = (
args.build_db
.expanduser()
.resolve()
)
if not destination.exists():
print(
f"ERROR: Folder does not exist:\n"
f"{destination}",
file=sys.stderr,
)
sys.exit(1)
if not destination.is_dir():
print(
f"ERROR: Not a folder:\n"
f"{destination}",
file=sys.stderr,
)
sys.exit(1)
build_database(
destination
)
return
# ------------------------------------------------------------------------
# VERIFY
# ------------------------------------------------------------------------
if args.verify:
destination = (
args.verify
.expanduser()
.resolve()
)
if not destination.exists():
print(
f"ERROR: Folder does not exist:\n"
f"{destination}",
file=sys.stderr,
)
sys.exit(1)
if not destination.is_dir():
print(
f"ERROR: Not a folder:\n"
f"{destination}",
file=sys.stderr,
)
sys.exit(1)
verify_database(
destination,
args.deep,
)
return
# ------------------------------------------------------------------------
# NORMAL IMPORT
# ------------------------------------------------------------------------
if (
not args.source
or not args.destination
):
parser.print_help()
sys.exit(1)
source = (
args.source
.expanduser()
.resolve()
)
destination = (
args.destination
.expanduser()
.resolve()
)
if not source.exists():
print(
f"ERROR: Source folder does not exist:\n"
f"{source}",
file=sys.stderr,
)
sys.exit(1)
if not source.is_dir():
print(
f"ERROR: Source is not a folder:\n"
f"{source}",
file=sys.stderr,
)
sys.exit(1)
# Destination cannot be inside source.
try:
destination.relative_to(
source
)
print(
"ERROR: Destination cannot be "
"inside the source folder.",
file=sys.stderr,
)
sys.exit(1)
except ValueError:
pass
import_files(
source=source,
destination=destination,
dry_run=args.dry_run,
move=args.move,
)
if __name__ == "__main__":
main()
Reference in New Issue Copy Permalink