Merge pull request #131 from openzim/generate_library

Generate library

Author: benoit74

backend/pyproject.toml

@@ -216,8 +216,8 @@ extend-immutable-calls = ["fastapi.Depends", "fastapi.Query"]
 ban-relative-imports = "all"
 
 [tool.ruff.lint.per-file-ignores]
-# Tests can use magic values, assertions, and relative imports
-"tests/**/*" = ["PLR2004", "S101", "TID252"]
+# Tests can use magic values, assertions, relative imports and unsafe XML
+"tests/**/*" = ["PLR2004", "S101", "TID252", "S314"]
 # Ignore invalid module names for scripts
 "src/cms_backend/periodic-tasks.py" = ["N999"]
 "src/cms_backend/periodic-scheduler.py" = ["N999"]

backend/src/cms_backend/api/main.py

@@ -12,6 +12,7 @@
 from cms_backend.api.routes.books import router as books_router
 from cms_backend.api.routes.healthcheck import router as healthcheck_router
 from cms_backend.api.routes.http_errors import BadRequestError
+from cms_backend.api.routes.library import router as library_router
 from cms_backend.api.routes.titles import router as titles_router
 from cms_backend.api.routes.warehouse_paths import router as warehouse_paths_router
 from cms_backend.api.routes.zimfarm_notifications import (
@@ -58,6 +59,7 @@ def create_app(*, debug: bool = True):
     main_router.include_router(router=titles_router)
     main_router.include_router(router=books_router)
     main_router.include_router(router=warehouse_paths_router)
+    main_router.include_router(router=library_router)
 
     app.include_router(router=main_router)
 

backend/src/cms_backend/api/routes/library.py

@@ -0,0 +1,103 @@
+from http import HTTPStatus
+from typing import Annotated
+from uuid import UUID
+from xml.etree import ElementTree as ET
+
+from fastapi import APIRouter, Depends, Path
+from fastapi.responses import Response
+from sqlalchemy.orm import Session as OrmSession
+
+from cms_backend.db import gen_dbsession
+from cms_backend.db.exceptions import RecordDoesNotExistError
+from cms_backend.db.library import (
+    get_latest_books_for_library,
+    get_library,
+    get_library_by_name_or_none,
+)
+from cms_backend.db.models import Book
+
+router = APIRouter(prefix="/libraries", tags=["libraries"])
+
+
+def _build_library_xml(books: list[Book]) -> str:
+    """Build XML library catalog from books."""
+    library_elem = ET.Element("library")
+    library_elem.set("version", "20110515")
+
+    for book in books:
+        if not book.zim_metadata:
+            continue
+
+        book_elem = ET.SubElement(library_elem, "book")
+
+        # Required attributes
+        book_elem.set("id", str(book.id))
+        book_elem.set("size", str(book.size))
+        book_elem.set("mediaCount", str(book.media_count))
+        book_elem.set("articleCount", str(book.article_count))
+
+        # Metadata from zim_metadata dict
+        zim_meta = book.zim_metadata
+        book_elem.set("title", zim_meta.get("Title", ""))
+        book_elem.set("description", zim_meta.get("Description", ""))
+        book_elem.set("language", zim_meta.get("Language", ""))
+        book_elem.set("creator", zim_meta.get("Creator", ""))
+        book_elem.set("publisher", zim_meta.get("Publisher", ""))
+        book_elem.set("name", zim_meta.get("Name", ""))
+        book_elem.set("date", zim_meta.get("Date", ""))
+
+        # Optional tags - combine with underscores if present
+        tags = zim_meta.get("Tags", "")
+        if tags:
+            book_elem.set("tags", tags)
+
+        # Favicon and faviconMimeType - these are typically extracted from the ZIM
+        # but for now we'll use empty strings as the data structure doesn't contain them
+        favicon = zim_meta.get("Illustration_48x48@1", "")
+        if favicon:
+            book_elem.set("favicon", favicon)
+            book_elem.set("faviconMimeType", "image/png")
+
+        # URL - would need to be constructed from warehouse config
+        # For now, leaving empty as warehouse config mapping is not implemented
+        url = zim_meta.get("URL", "")
+        if url:
+            book_elem.set("url", url)
+
+    return ET.tostring(library_elem, encoding="unicode")
+
+
+@router.get("/{library_id_or_name}/catalog.xml")
+async def get_library_catalog_xml(
+    library_id_or_name: Annotated[str, Path()],
+    session: Annotated[OrmSession, Depends(gen_dbsession)],
+):
+    """Get library catalog as XML. Library can be specified by ID (UUID) or name."""
+    # Try to parse as UUID first, otherwise treat as name
+    library = None
+    try:
+        library_id = UUID(library_id_or_name)
+        try:
+            library = get_library(session, library_id)
+        except RecordDoesNotExistError:
+            pass
+    except ValueError:
+        # Not a valid UUID, try as name
+        library = get_library_by_name_or_none(session, library_id_or_name)
+
+    if library is None:
+        return Response(
+            content='<?xml version="1.0" encoding="UTF-8"?>'
+            '<library version="20110515"></library>',
+            status_code=HTTPStatus.NOT_FOUND,
+            media_type="application/xml",
+        )
+
+    books = get_latest_books_for_library(session, library.id)
+    xml_content = _build_library_xml(books)
+
+    return Response(
+        content=xml_content,
+        status_code=HTTPStatus.OK,
+        media_type="application/xml",
+    )

backend/src/cms_backend/db/library.py

@@ -0,0 +1,78 @@
+from uuid import UUID
+
+from sqlalchemy import and_, select
+from sqlalchemy.orm import Session as OrmSession
+
+from cms_backend.db.exceptions import RecordDoesNotExistError
+from cms_backend.db.models import Book, BookLocation, Library, LibraryWarehousePath
+
+
+def get_library_or_none(session: OrmSession, library_id: UUID) -> Library | None:
+    """Get a library by ID if possible else None"""
+    return session.scalars(
+        select(Library).where(Library.id == library_id)
+    ).one_or_none()
+
+
+def get_library(session: OrmSession, library_id: UUID) -> Library:
+    """Get a library by ID if possible else raise an exception"""
+    if (library := get_library_or_none(session, library_id=library_id)) is None:
+        raise RecordDoesNotExistError(f"Library with ID {library_id} does not exist")
+    return library
+
+
+def get_library_by_name_or_none(
+    session: OrmSession, library_name: str
+) -> Library | None:
+    """Get a library by name if possible else None"""
+    return session.scalars(
+        select(Library).where(Library.name == library_name)
+    ).one_or_none()
+
+
+def get_latest_books_for_library(session: OrmSession, library_id: UUID) -> list[Book]:
+    """
+    Get the latest book for each name+flavour combination in a library.
+
+    A library contains multiple warehouse paths. For each unique name+flavour
+    combination found in the library's warehouse paths, return only the most
+    recent published book (by created_at).
+
+    Args:
+        session: ORM session
+        library_id: ID of the library
+
+    Returns:
+        List of Book objects, one per name+flavour combination
+    """
+    # Get all books in the library's warehouse paths that are published
+    # and currently located there
+    stmt = (
+        select(Book)
+        .join(BookLocation)
+        .join(
+            LibraryWarehousePath,
+            BookLocation.warehouse_path_id == LibraryWarehousePath.warehouse_path_id,
+        )
+        .where(
+            and_(
+                LibraryWarehousePath.library_id == library_id,
+                BookLocation.status == "current",
+                Book.status == "published",
+            )
+        )
+        .order_by(Book.name, Book.flavour, Book.created_at.desc())
+    )
+
+    books = session.scalars(stmt).all()
+
+    # Filter to keep only the latest book per name+flavour combination
+    seen: set[tuple[str | None, str | None]] = set()
+    latest_books: list[Book] = []
+    for book in books:
+        key = (book.name, book.flavour)
+        if key not in seen:
+            seen.add(key)
+            latest_books.append(book)
+
+    return latest_books

backend/src/cms_backend/db/models.py

@@ -246,3 +246,33 @@ def full_str(self) -> str:
             f"{self.warehouse_path.warehouse.name}:"
             f"{self.warehouse_path.folder_name}/{self.filename}"
         )
+
+
+class Library(Base):
+    __tablename__ = "library"
+    id: Mapped[UUID] = mapped_column(
+        init=False, primary_key=True, server_default=text("uuid_generate_v4()")
+    )
+    name: Mapped[str] = mapped_column(unique=True, index=True)
+
+    # Warehouse paths via junction table
+    warehouse_paths: Mapped[list["LibraryWarehousePath"]] = relationship(
+        back_populates="library",
+        cascade="all, delete-orphan",
+        init=False,
+    )
+
+
+class LibraryWarehousePath(Base):
+    __tablename__ = "library_warehouse_path"
+    library_id: Mapped[UUID] = mapped_column(
+        ForeignKey("library.id"), primary_key=True, init=False
+    )
+    warehouse_path_id: Mapped[UUID] = mapped_column(
+        ForeignKey("warehouse_path.id"), primary_key=True, init=False
+    )
+
+    library: Mapped["Library"] = relationship(
+        back_populates="warehouse_paths", init=False
+    )
+    warehouse_path: Mapped["WarehousePath"] = relationship(init=False)

backend/src/cms_backend/migrations/versions/add_library_tables.py

@@ -0,0 +1,59 @@
+"""Add library and library_warehouse_path tables
+
+Revision ID: add_library_tables
+Revises: add_pending_move_index
+Create Date: 2025-11-18 00:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects.postgresql import UUID
+
+# revision identifiers, used by Alembic.
+revision = "add_library_tables"
+down_revision = "add_pending_move_index"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # Create library table
+    op.create_table(
+        "library",
+        sa.Column(
+            "id", UUID(), server_default=sa.text("uuid_generate_v4()"), nullable=False
+        ),
+        sa.Column("name", sa.String(), nullable=False),
+        sa.PrimaryKeyConstraint("id", name="pk_library"),
+        sa.UniqueConstraint("name", name="uq_library_name"),
+    )
+
+    # Create unique index on library name for fast lookups
+    op.create_index("ix_library_name", "library", ["name"], unique=True)
+
+    # Create library_warehouse_path junction table
+    op.create_table(
+        "library_warehouse_path",
+        sa.Column("library_id", UUID(), nullable=False),
+        sa.Column("warehouse_path_id", UUID(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["library_id"],
+            ["library.id"],
+            name="fk_library_warehouse_path_library_id_library",
+        ),
+        sa.ForeignKeyConstraint(
+            ["warehouse_path_id"],
+            ["warehouse_path.id"],
+            name="fk_library_warehouse_path_warehouse_path_id_warehouse_path",
+        ),
+        sa.PrimaryKeyConstraint(
+            "library_id", "warehouse_path_id", name="pk_library_warehouse_path"
+        ),
+    )
+
+
+def downgrade():
+    # Drop tables in reverse order due to foreign key constraints
+    op.drop_table("library_warehouse_path")
+    op.drop_table("library")