#!/usr/bin/env python3 import sqlite3 from datetime import datetime from pathlib import Path import tempfile import os import json import sys, time docstring = """ This script reads metadata and application state from a Calibre library, merges the information, and generates an Org-mode file suitable for org-roam, categorising books by shelves and including read status. If the output file already exists, it preserves the existing org-roam header, and writes the updated book list below it. This script runs on a cron schedule at midnight every day and outputs a JSON summary of its execution, in the form of {"script": "calibre_org_sync", "status": "success", "duration_ms": 3, "output": "/home/zaine/master-folder/org_files/org_roam/20250724230557-books_org_agenda.org", "timestamp": "2026-01-14T00:00:02.077267"}. """ # ───────────────────────────────────────────────────────────── # Configuration # ───────────────────────────────────────────────────────────── METADATA_DB = Path("/home/zaine/master-folder/projects/calibre/library/metadata.db") APP_DB = Path("/home/zaine/master-folder/projects/calibre/data/app.db") CALIBRE_LIBRARY_ROOT = Path("/home/zaine/master-folder/projects/calibre/library") OUTPUT_ORG = Path("/home/zaine/master-folder/org_files/org_roam/20250724230557-books_org_agenda.org") UNSORTED_SHELF = "Unsorted" # ───────────────────────────────────────────────────────────── # Database readers # ───────────────────────────────────────────────────────────── def read_metadata(db_path): conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row cur = conn.cursor() query = """ SELECT b.id AS book_id, b.title AS title, b.path AS calibre_path, GROUP_CONCAT(a.name, ', ') AS authors FROM books b LEFT JOIN books_authors_link bal ON b.id = bal.book LEFT JOIN authors a ON bal.author = a.id GROUP BY b.id ORDER BY b.title COLLATE NOCASE; """ cur.execute(query) rows = cur.fetchall() conn.close() books = {} for row in rows: books[row["book_id"]] = { "id": row["book_id"], "title": row["title"], "authors": row["authors"] or "Unknown", "calibre_path": row["calibre_path"], } return books def read_app_state(db_path): conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row cur = conn.cursor() # --- Shelves per book --- shelf_query = """ SELECT bsl.book_id AS book_id, s.name AS shelf FROM book_shelf_link bsl JOIN shelf s ON bsl.shelf = s.id; """ cur.execute(shelf_query) shelf_rows = cur.fetchall() shelves_by_book = {} for row in shelf_rows: shelves_by_book.setdefault(row["book_id"], []).append(row["shelf"]) # --- Read status + completion time --- status_query = """ SELECT book_id, read_status, last_modified FROM book_read_link; """ cur.execute(status_query) status_rows = cur.fetchall() status_by_book = {} completed_at_by_book = {} for r in status_rows: if r["read_status"] == 1: status_by_book[r["book_id"]] = "Read" completed_at_by_book[r["book_id"]] = r["last_modified"] else: status_by_book[r["book_id"]] = "Unread" conn.close() return shelves_by_book, status_by_book, completed_at_by_book # ───────────────────────────────────────────────────────────── # Merge logic # ───────────────────────────────────────────────────────────── def merge_books(metadata, shelves, statuses, completed): library = {} for book_id, book in metadata.items(): book_shelves = shelves.get(book_id, [UNSORTED_SHELF]) status = statuses.get(book_id, "Unread") completed_at = completed.get(book_id) for shelf in book_shelves: library.setdefault(shelf, []).append({ **book, "status": status, "completed_at": completed_at, "abs_path": CALIBRE_LIBRARY_ROOT / book["calibre_path"], }) return library def read_org_roam_header(path): """ Returns the org-roam header (inclusive of first :END:) or None if file does not exist. """ if not path.exists(): return None lines = path.read_text(encoding="utf-8").splitlines() for i, line in enumerate(lines): if line.strip() == ":END:": return "\n".join(lines[: i + 1]).rstrip() raise RuntimeError("No org-roam PROPERTIES drawer found") # ───────────────────────────────────────────────────────────── # Org generation # ───────────────────────────────────────────────────────────── def emit_org_body(library): lines = [] lines.append("#+TITLE: Library") lines.append("#+AUTHOR: Auto-generated") lines.append("#+filetags: :books:org:index:") lines.append("#+DATE: 2025-05-19") lines.append("#+STARTUP: content") lines.append(f"#+PROPERTY: GENERATED_AT {datetime.now().isoformat()}") lines.append("") for shelf in sorted(library.keys(), key=str.lower): lines.append(f"* {shelf}") books = sorted( library[shelf], key=lambda b: b["title"].lower() ) for b in books: lines.append(f"** {b['title']} :{b['status']}:") lines.append(":PROPERTIES:") lines.append(f":AUTHOR: {b['authors']}") lines.append(f":STATUS: {b['status']}") lines.append(f":CALIBRE_ID: {b['id']}") if b.get("completed_at"): lines.append(f":COMPLETED_AT: {b['completed_at']}") lines.append(f":PATH: {b['abs_path']}") lines.append(":END:") lines.append("") return "\n".join(lines) def write_library_org(path, library): header = read_org_roam_header(path) if header is None: raise RuntimeError( "library.org must already exist and contain an org-roam ID" ) body = emit_org_body(library) content = header + "\n\n" + body atomic_write(path, content) # ───────────────────────────────────────────────────────────── # Atomic write # ───────────────────────────────────────────────────────────── def atomic_write(path, content): path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile( "w", delete=False, dir=str(path.parent), encoding="utf-8" ) as tmp: tmp.write(content) temp_name = tmp.name os.replace(temp_name, path) # ───────────────────────────────────────────────────────────── # Main # ───────────────────────────────────────────────────────────── def main(): metadata = read_metadata(METADATA_DB) shelves, statuses, completed = read_app_state(APP_DB) library = merge_books(metadata, shelves, statuses, completed) write_library_org(OUTPUT_ORG, library) if __name__ == "__main__": start = time.time() try: main() duration_ms = int((time.time() - start) * 1000) print(json.dumps({ "script": "calibre_org_sync", "status": "success", "duration_ms": duration_ms, "output": str(OUTPUT_ORG), "timestamp": datetime.now().isoformat() })) sys.exit(0) except Exception as e: duration_ms = int((time.time() - start) * 1000) print(json.dumps({ "script": "calibre_org_sync", "status": "failure", "duration_ms": duration_ms, "error": str(e) }), file=sys.stderr) sys.exit(1)