#!/usr/bin/env python3
"""
catalogue.py — Read-only HTML catalogue generator for o-phone.py backups.
Does not import, modify, or depend on o-phone.py.
Reads the existing SQLite database and generates a single offline index.html.
"""
from __future__ import annotations
import argparse
import html
import json
import os
import sqlite3
import sys
import webbrowser
from collections import defaultdict
from datetime import datetime
from pathlib import Path
from typing import Any
# ============================================================================
# CONFIGURATION (must match o-phone.py)
# ============================================================================
DATABASE_NAME = ".photo_organiser.db"
# file_type values produced by o-phone.py classify_file()
TYPE_IMAGE = "image"
TYPE_VIDEO = "video"
TYPE_SCREENSHOT = "screenshot"
TYPE_UNKNOWN = "unknown"
MONTH_NAMES = [
"January", "February", "March", "April", "May", "June",
"July", "August", "September", "October", "November", "December",
]
# Extensions that browsers generally cannot display as
RAW_EXTENSIONS = {
".dng", ".raw", ".arw", ".cr2", ".cr3", ".nef", ".nrw",
".orf", ".rw2", ".raf", ".pef", ".srw", ".x3f",
}
# ============================================================================
# UTILITIES
# ============================================================================
def eprint(*args: Any, **kwargs: Any) -> None:
print(*args, file=sys.stderr, **kwargs)
def parse_date_taken(value: str) -> datetime | None:
"""Parse date_taken stored by o-phone.py (ISO-ish)."""
if not value:
return None
value = value.strip()
# Handle fractional seconds and timezone suffixes loosely
for fmt in (
"%Y-%m-%dT%H:%M:%S.%f%z",
"%Y-%m-%dT%H:%M:%S%z",
"%Y-%m-%dT%H:%M:%S.%f",
"%Y-%m-%dT%H:%M:%S",
"%Y-%m-%d %H:%M:%S",
"%Y-%m-%d",
"%Y:%m:%d %H:%M:%S",
"%Y:%m:%d",
):
try:
return datetime.strptime(value, fmt)
except ValueError:
continue
# Last resort: take first 19 chars as YYYY-MM-DDTHH:MM:SS
try:
return datetime.strptime(value[:19], "%Y-%m-%dT%H:%M:%S")
except ValueError:
return None
def format_display_date(dt: datetime) -> str:
return f"{dt.day} {MONTH_NAMES[dt.month - 1]} {dt.year}"
def format_display_day(dt: datetime) -> str:
return f"{dt.day} {MONTH_NAMES[dt.month - 1][:3]}"
def is_raw(path: str, original_name: str) -> bool:
ext = Path(path).suffix.lower() or Path(original_name).suffix.lower()
return ext in RAW_EXTENSIONS
def open_database_readonly(db_path: Path) -> sqlite3.Connection:
"""Open SQLite in read-only mode. Fail if the file does not exist."""
if not db_path.is_file():
eprint(f"ERROR: SQLite database not found:\n{db_path}")
sys.exit(1)
# URI mode forces read-only; also prevents accidental creation
uri = f"file:{db_path.resolve().as_posix()}?mode=ro"
try:
conn = sqlite3.connect(uri, uri=True)
conn.row_factory = sqlite3.Row
# Quick schema sanity check
cols = {
row[1]
for row in conn.execute("PRAGMA table_info(files)").fetchall()
}
required = {
"id", "sha256", "size", "original_name", "stored_name",
"relative_path", "date_taken", "file_type",
"device_make", "device_model", "imported_at",
}
if not required.issubset(cols):
eprint("ERROR: Database schema is not recognised.")
eprint(f"Expected columns including: {sorted(required)}")
eprint(f"Found: {sorted(cols)}")
sys.exit(1)
return conn
except sqlite3.Error as exc:
eprint(f"ERROR: Unable to open database.\n{exc}")
sys.exit(1)
def load_records(conn: sqlite3.Connection) -> list[dict[str, Any]]:
rows = conn.execute(
"""
SELECT
id,
sha256,
size,
original_name,
stored_name,
relative_path,
date_taken,
file_type,
device_make,
device_model,
imported_at
FROM files
ORDER BY date_taken DESC, id DESC
"""
).fetchall()
records: list[dict[str, Any]] = []
for row in rows:
dt = parse_date_taken(row["date_taken"])
records.append({
"id": row["id"],
"sha256": row["sha256"],
"size": row["size"],
"original_name": row["original_name"],
"stored_name": row["stored_name"],
"relative_path": row["relative_path"],
"date_taken": row["date_taken"],
"file_type": row["file_type"] or TYPE_UNKNOWN,
"device_make": row["device_make"] or "",
"device_model": row["device_model"] or "",
"imported_at": row["imported_at"],
"_dt": dt,
})
return records
# ============================================================================
# CHECK MODE
# ============================================================================
def run_check(
backup_root: Path,
records: list[dict[str, Any]],
verbose: bool,
) -> int:
"""
Report consistency issues. Returns number of problems found.
Does not modify anything.
"""
print("=" * 60)
print("CATALOGUE CHECK")
print("=" * 60)
print(f"Backup: {backup_root}")
print(f"Database: {backup_root / DATABASE_NAME}")
print(f"Records: {len(records):,}")
print()
missing: list[str] = []
path_escape: list[str] = []
present = 0
hash_counts: dict[str, int] = defaultdict(int)
for rec in records:
rel = rec["relative_path"]
hash_counts[rec["sha256"]] += 1
# Detect path traversal / escape
try:
full = (backup_root / rel).resolve()
if not str(full).startswith(str(backup_root.resolve())):
path_escape.append(rel)
continue
except Exception:
path_escape.append(rel)
continue
if full.is_file():
present += 1
else:
missing.append(rel)
duplicate_hashes = {
h: c for h, c in hash_counts.items() if c > 1
}
# Files on disk not in database (optional, can be slow on huge trees)
db_paths = {rec["relative_path"] for rec in records}
unindexed: list[str] = []
skip_names = {
DATABASE_NAME,
f"{DATABASE_NAME}-wal",
f"{DATABASE_NAME}-shm",
"index.html",
}
for path in backup_root.rglob("*"):
if not path.is_file():
continue
if path.name in skip_names or path.name.startswith("._"):
continue
if path.name == ".DS_Store":
continue
# Ignore anything under a .catalogue folder if present later
try:
rel = str(path.relative_to(backup_root))
except ValueError:
continue
if rel.startswith(".catalogue/") or rel == "index.html":
continue
if rel not in db_paths:
unindexed.append(rel)
print(f"Files present: {present:,}")
print(f"Missing files: {len(missing):,}")
print(f"Path escape attempts: {len(path_escape):,}")
print(f"Duplicate SHA-256: {len(duplicate_hashes):,}")
print(f"Unindexed on disk: {len(unindexed):,}")
problems = (
len(missing)
+ len(path_escape)
+ len(duplicate_hashes)
+ len(unindexed)
)
if missing:
print()
print("MISSING FILES:")
for p in missing[:50]:
print(f" {p}")
if len(missing) > 50:
print(f" ... and {len(missing) - 50} more")
if path_escape:
print()
print("PATHS THAT ESCAPE BACKUP ROOT:")
for p in path_escape:
print(f" {p}")
if duplicate_hashes and verbose:
print()
print("DUPLICATE HASHES (first few):")
for i, (h, c) in enumerate(duplicate_hashes.items()):
if i >= 10:
print(f" ... and {len(duplicate_hashes) - 10} more")
break
print(f" {h[:16]}… ×{c}")
if unindexed:
print()
print("UNINDEXED FILES ON DISK (sample):")
for p in unindexed[:30]:
print(f" {p}")
if len(unindexed) > 30:
print(f" ... and {len(unindexed) - 30} more")
print()
if problems == 0:
print("RESULT: CONSISTENT")
else:
print("RESULT: PROBLEMS FOUND")
print("=" * 60)
return problems
# ============================================================================
# BUILD CATALOGUE DATA
# ============================================================================
def build_items(
backup_root: Path,
records: list[dict[str, Any]],
verbose: bool,
) -> tuple[list[dict[str, Any]], int, int, int]:
"""
Convert DB records into catalogue items.
Returns (items, present_count, missing_count, unknown_count).
Only includes items whose files exist.
"""
items: list[dict[str, Any]] = []
missing = 0
unknown = 0
present = 0
for rec in records:
rel = rec["relative_path"]
full = backup_root / rel
if not full.is_file():
missing += 1
if verbose:
eprint(f" missing: {rel}")
continue
present += 1
dt = rec["_dt"]
ftype = rec["file_type"]
if ftype == TYPE_UNKNOWN:
unknown += 1
# Determine display type for the UI
if ftype == TYPE_SCREENSHOT:
ui_type = "screenshot"
elif ftype == TYPE_VIDEO:
ui_type = "video"
elif ftype == TYPE_IMAGE:
if is_raw(rel, rec["original_name"]):
ui_type = "raw"
else:
ui_type = "image"
else:
ui_type = "unknown"
# Relative path suitable for HTML (forward slashes, no leading ./)
web_path = rel.replace("\\", "/")
device = " ".join(
part for part in (rec["device_make"], rec["device_model"]) if part
).strip()
item = {
"id": rec["id"],
"name": rec["stored_name"],
"original": rec["original_name"],
"path": web_path,
"type": ui_type,
"date": dt.strftime("%Y-%m-%d") if dt else "",
"displayDate": format_display_date(dt) if dt else "",
"displayDay": format_display_day(dt) if dt else "",
"time": dt.strftime("%H:%M:%S") if dt and (dt.hour or dt.minute or dt.second) else "",
"year": dt.year if dt else 0,
"month": dt.month if dt else 0,
"monthName": MONTH_NAMES[dt.month - 1] if dt else "",
"day": dt.day if dt else 0,
"device": device,
"size": rec["size"],
# Sort key: newest first
"sort": dt.timestamp() if dt else 0,
}
items.append(item)
# Newest first overall
items.sort(key=lambda x: (-x["sort"], -x["id"]))
return items, present, missing, unknown
# ============================================================================
# HTML GENERATION
# ============================================================================
# Large HTML/JS template. Data is injected via a unique placeholder
# so we never use Python f-strings for the JS body (avoids { } conflicts).
HTML_TEMPLATE = r'''