#!/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''' Photo Archive
Photo Archive
''' def generate_html(items: list[dict[str, Any]]) -> str: """Build final index.html with JSON data safely embedded.""" # Strip internal sort helper before serialising clean = [] for it in items: clean.append({ "id": it["id"], "name": it["name"], "original": it["original"], "path": it["path"], "type": it["type"], "date": it["date"], "displayDate": it["displayDate"], "displayDay": it["displayDay"], "time": it["time"], "year": it["year"], "month": it["month"], "monthName": it["monthName"], "day": it["day"], "device": it["device"], "size": it["size"], "sort": it["sort"], }) payload = {"items": clean} # Ensure safe embedding inside json_text = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) json_text = json_text.replace("<", "\\u003c").replace(">", "\\u003e") html_out = HTML_TEMPLATE.replace("__CATALOGUE_DATA__", json_text) return html_out # ============================================================================ # MAIN # ============================================================================ def main() -> None: parser = argparse.ArgumentParser( prog="catalogue.py", description=( "Generate a read-only offline HTML catalogue from an " "o-phone.py photo/video backup." ), formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python3 catalogue.py "/Volumes/all/phone-media-backup" python3 catalogue.py "/Volumes/all/phone-media-backup" --open python3 catalogue.py "/Volumes/all/phone-media-backup" --check python3 catalogue.py "/Volumes/all/phone-media-backup" --check --verbose """.strip(), ) parser.add_argument( "backup", type=Path, nargs="?", help="Root folder of the existing backup (contains .photo_organiser.db)", ) parser.add_argument( "--check", action="store_true", help="Check database/filesystem consistency without generating the catalogue", ) parser.add_argument( "--open", action="store_true", help="After generating, open index.html in the default browser", ) parser.add_argument( "--verbose", action="store_true", help="Show additional diagnostic information", ) args = parser.parse_args() if args.backup is None: parser.print_help() sys.exit(1) backup_root = args.backup.expanduser().resolve() if not backup_root.exists(): eprint(f"ERROR: Backup folder does not exist.\n{backup_root}") sys.exit(1) if not backup_root.is_dir(): eprint(f"ERROR: Not a folder.\n{backup_root}") sys.exit(1) db_path = backup_root / DATABASE_NAME print("Reading database...") conn = open_database_readonly(db_path) try: records = load_records(conn) finally: conn.close() print(f"Found {len(records):,} records.") if args.check: problems = run_check(backup_root, records, args.verbose) sys.exit(1 if problems else 0) # ---- Generate catalogue ---- print("Checking files...") items, present, missing, unknown = build_items( backup_root, records, args.verbose ) print(f"{present:,} files present.") if missing: print(f"WARNING: {missing:,} files referenced by the database are missing.") if unknown: print(f"Note: {unknown:,} items have unknown/unsupported type.") print("Generating catalogue...") html_content = generate_html(items) out_path = backup_root / "index.html" try: out_path.write_text(html_content, encoding="utf-8") except OSError as exc: eprint(f"ERROR: Could not write index.html.\n{exc}") sys.exit(1) print("Wrote:") print(f" {out_path}") if args.open: try: # Prefer OS open on macOS; fall back to webbrowser if sys.platform == "darwin": os.system(f'open "{out_path}"') # noqa: S605 else: webbrowser.open(out_path.as_uri()) except Exception as exc: eprint(f"WARNING: Could not open browser: {exc}") if __name__ == "__main__": main()