scripts/org-books-calibre.py

#!/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)