Handle books location, and update them when adding book to title

backend/src/cms_backend/db/book.py

@@ -3,7 +3,7 @@
 
 from sqlalchemy.orm import Session as OrmSession
 
-from cms_backend.db.models import Book, ZimfarmNotification
+from cms_backend.db.models import Book, BookLocation, WarehousePath, ZimfarmNotification
 from cms_backend.utils.datetime import getnow
 
 
@@ -53,3 +53,47 @@ def create_book(
     )
 
     return book
+
+
+def create_book_location(
+    session: OrmSession,
+    *,
+    book: Book,
+    warehouse_path_id: UUID,
+    filename: str,
+    status: str = "current",
+) -> BookLocation:
+    """Create a new book location.
+
+    Args:
+        session: SQLAlchemy session
+        book: Book instance
+        warehouse_path_id: ID of the warehouse path
+        filename: Filename in warehouse
+        status: Location status ('current' or 'target'), defaults to 'current'
+
+    Returns:
+        Created BookLocation instance
+    """
+    # Get warehouse path info for event message
+    warehouse_path = session.get(WarehousePath, warehouse_path_id)
+    if not warehouse_path:
+        raise ValueError(f"WarehousePath with id {warehouse_path_id} not found")
+
+    warehouse_name = warehouse_path.warehouse.name
+    folder_name = warehouse_path.folder_name
+
+    location = BookLocation(
+        book_id=book.id,
+        status=status,
+        filename=filename,
+    )
+    location.warehouse_path_id = warehouse_path_id
+    session.add(location)
+    book.locations.append(location)
+    book.events.append(
+        f"{getnow()}: added {status} location: {filename} in {warehouse_name}: "
+        f"{folder_name} ({warehouse_path_id})"
+    )
+
+    return location

backend/src/cms_backend/db/models.py

@@ -112,6 +112,12 @@ class Book(Base):
         back_populates="book"
     )
 
+    locations: Mapped[list["BookLocation"]] = relationship(
+        back_populates="book",
+        cascade="all, delete-orphan",
+        init=False,
+    )
+
 
 Index(
     "idx_book_status_pending_processing",
@@ -205,3 +211,17 @@ class WarehousePath(Base):
     warehouse: Mapped["Warehouse"] = relationship(
         back_populates="warehouse_paths", init=False
     )
+
+
+class BookLocation(Base):
+    __tablename__ = "book_location"
+    book_id: Mapped[UUID] = mapped_column(ForeignKey("book.id"), primary_key=True)
+    warehouse_path_id: Mapped[UUID] = mapped_column(
+        ForeignKey("warehouse_path.id"), primary_key=True, init=False
+    )
+    status: Mapped[str] = mapped_column(primary_key=True)  # 'current' or 'target'
+
+    filename: Mapped[str]
+
+    book: Mapped["Book"] = relationship(back_populates="locations", init=False)
+    warehouse_path: Mapped["WarehousePath"] = relationship(init=False)

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

@@ -0,0 +1,43 @@
+"""Add book_location table to store current and target file locations
+
+Revision ID: add_book_location_table
+Revises: title_warehouse_paths
+Create Date: 2025-11-11 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_book_location_table"
+down_revision = "title_warehouse_paths"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # Create the book_location table with composite primary key
+    op.create_table(
+        "book_location",
+        sa.Column("book_id", UUID(), nullable=False),
+        sa.Column("warehouse_path_id", UUID(), nullable=False),
+        sa.Column("status", sa.String(), nullable=False),
+        sa.Column("filename", sa.String(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["book_id"], ["book.id"], name="fk_book_location_book_id_book"
+        ),
+        sa.ForeignKeyConstraint(
+            ["warehouse_path_id"],
+            ["warehouse_path.id"],
+            name="fk_book_location_warehouse_path_id_warehouse_path",
+        ),
+        sa.PrimaryKeyConstraint(
+            "book_id", "warehouse_path_id", "status", name="pk_book_location"
+        ),
+    )
+
+
+def downgrade():
+    op.drop_table("book_location")

backend/src/cms_backend/processors/book.py

@@ -46,16 +46,15 @@ def check_book_qa(book: Book) -> bool:
 
 def get_matching_title(session: ORMSession, book: Book) -> Title | None:
     try:
-        name = str(book.zim_metadata.get("Name"))
-        if not name:
+        if not book.name:
             book.events.append(
                 f"{getnow()}: no title can be found because name is missing"
             )
             book.status = "qa_failed"
             return None
 
         title = get_title_by_name_and_producer_or_none(
-            session, name=name, producer_unique_id=book.producer_unique_id
+            session, name=book.name, producer_unique_id=book.producer_unique_id
         )
 
         if not title:

backend/src/cms_backend/processors/title.py

@@ -1,15 +1,24 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session as OrmSession
+
 from cms_backend import logger
-from cms_backend.db.models import Book, Title
+from cms_backend.db.book import create_book_location
+from cms_backend.db.models import Book, Title, TitleWarehousePath
 from cms_backend.utils.datetime import getnow
+from cms_backend.utils.filename import compute_target_filename
 
 
-def add_book_to_title(book: Book, title: Title):
+def add_book_to_title(session: OrmSession, book: Book, title: Title):
     try:
-        name = book.zim_metadata.get("Name")
-        if not name:
-            raise Exception("book Name is missing")
-        if not isinstance(name, str):
-            raise Exception(f"book Name is expected to be a string, found {type(name)}")
+        # Retrieve name from book.name directly
+        if not book.name:
+            raise Exception("book name is missing or invalid")
+
+        # Validate book.date is also present and valid
+        if not book.date:
+            raise Exception("book date is missing or invalid")
+
+        name = book.name
 
         title.books.append(book)
         book.events.append(f"{getnow()}: book added to title {title.id}")
@@ -35,6 +44,34 @@ def add_book_to_title(book: Book, title: Title):
             )
             title.producer_display_url = book.producer_display_url
 
+        # Compute target filename once for this book
+        target_filename = compute_target_filename(
+            session,
+            name=name,
+            flavour=book.flavour,
+            date=book.date,
+        )
+
+        # Determine which warehouse paths to use based on title.in_prod
+        path_type = "prod" if title.in_prod else "dev"
+
+        # Get all warehouse paths for this title and path_type
+        stmt = select(TitleWarehousePath).where(
+            TitleWarehousePath.title_id == title.id,
+            TitleWarehousePath.path_type == path_type,
+        )
+        target_warehouse_paths = session.scalars(stmt).all()
+
+        # Create target locations for each applicable warehouse path
+        for title_warehouse_path in target_warehouse_paths:
+            create_book_location(
+                session=session,
+                book=book,
+                warehouse_path_id=title_warehouse_path.warehouse_path_id,
+                filename=target_filename,
+                status="target",
+            )
+
     except Exception as exc:
         book.events.append(
             f"{getnow()}: error encountered while adding to title {title.id}\n{exc}"

backend/src/cms_backend/processors/zimfarm_notification.py

@@ -1,10 +1,11 @@
 from typing import cast
 
+from sqlalchemy import select
 from sqlalchemy.orm import Session as ORMSession
 
 from cms_backend import logger
-from cms_backend.db.book import create_book
-from cms_backend.db.models import ZimfarmNotification
+from cms_backend.db.book import create_book, create_book_location
+from cms_backend.db.models import Warehouse, WarehousePath, ZimfarmNotification
 from cms_backend.processors.book import check_book_qa, get_matching_title
 from cms_backend.processors.title import add_book_to_title
 from cms_backend.utils.datetime import getnow
@@ -27,7 +28,9 @@ def process_notification(session: ORMSession, notification: ZimfarmNotification)
                 "size",
                 "metadata",
                 "zimcheck",
-                "url",
+                "warehouse_name",
+                "folder_name",
+                "filename",
                 "producer",
             ]
             if key not in notification.content
@@ -67,6 +70,37 @@ def process_notification(session: ORMSession, notification: ZimfarmNotification)
             notification.status = "bad_notification"
             return
 
+        # Look up warehouse path by warehouse_name and folder_name
+        warehouse_name = notification.content.get("warehouse_name")
+        folder_name = notification.content.get("folder_name")
+        filename = notification.content.get("filename")
+
+        # Validate filename is a non-empty string
+        if not isinstance(filename, str) or not filename:
+            notification.events.append(
+                f"{getnow()}: filename must be a non-empty string, got "
+                f"{type(filename).__name__}: {filename}"
+            )
+            notification.status = "bad_notification"
+            return
+
+        stmt = (
+            select(WarehousePath)
+            .join(Warehouse)
+            .where(
+                Warehouse.name == warehouse_name,
+                WarehousePath.folder_name == folder_name,
+            )
+        )
+        warehouse_path = session.scalars(stmt).one_or_none()
+
+        if not warehouse_path:
+            notification.events.append(
+                f"{getnow()}: warehouse path not found: {warehouse_name}/{folder_name}"
+            )
+            notification.status = "bad_notification"
+            return
+
         book = create_book(
             session=session,
             book_id=notification.id,
@@ -80,6 +114,16 @@ def process_notification(session: ORMSession, notification: ZimfarmNotification)
             producer_display_url=producer["displayUrl"],
             producer_unique_id=producer["uniqueId"],
         )
+
+        # Create current book location
+        create_book_location(
+            session=session,
+            book=book,
+            warehouse_path_id=warehouse_path.id,
+            filename=filename,
+            status="current",
+        )
+
         notification.status = "processed"
 
         if not check_book_qa(book):
@@ -90,7 +134,7 @@ def process_notification(session: ORMSession, notification: ZimfarmNotification)
         if not title:
             return
 
-        add_book_to_title(book, title)
+        add_book_to_title(session, book, title)
 
     except Exception as exc:
         notification.events.append(

backend/src/cms_backend/routes/books.py

@@ -10,7 +10,12 @@
 from cms_backend.routes.fields import LimitFieldMax200, NotEmptyString, SkipField
 from cms_backend.routes.models import ListResponse, calculate_pagination_metadata
 from cms_backend.schemas import BaseModel
-from cms_backend.schemas.orms import BookFullSchema, BookLightSchema, ProducerSchema
+from cms_backend.schemas.orms import (
+    BookFullSchema,
+    BookLightSchema,
+    BookLocationSchema,
+    ProducerSchema,
+)
 
 router = APIRouter(prefix="/books", tags=["books"])
 
@@ -59,6 +64,31 @@ async def get_book(
 
     db_book = db_get_book(session=session, book_id=book_id)
 
+    # Separate current and target locations
+    current_locations = [
+        BookLocationSchema(
+            warehouse_path_id=location.warehouse_path_id,
+            warehouse_name=location.warehouse_path.warehouse.name,
+            folder_name=location.warehouse_path.folder_name,
+            filename=location.filename,
+            status=location.status,
+        )
+        for location in db_book.locations
+        if location.status == "current"
+    ]
+
+    target_locations = [
+        BookLocationSchema(
+            warehouse_path_id=location.warehouse_path_id,
+            warehouse_name=location.warehouse_path.warehouse.name,
+            folder_name=location.warehouse_path.folder_name,
+            filename=location.filename,
+            status=location.status,
+        )
+        for location in db_book.locations
+        if location.status == "target"
+    ]
+
     return BookFullSchema(
         id=db_book.id,
         title_id=db_book.title_id,
@@ -78,4 +108,6 @@ async def get_book(
             display_url=db_book.producer_display_url,
             unique_id=db_book.producer_unique_id,
         ),
+        current_locations=current_locations,
+        target_locations=target_locations,
     )

backend/src/cms_backend/schemas/orms.py

@@ -72,6 +72,18 @@ class ProducerSchema(BaseModel):
     unique_id: str
 
 
+class BookLocationSchema(BaseModel):
+    """
+    Schema for book location information
+    """
+
+    warehouse_path_id: UUID
+    warehouse_name: str
+    folder_name: str
+    filename: str
+    status: str  # 'current' or 'target'
+
+
 class BookLightSchema(BaseModel):
     """
     Schema for reading a book model with some fields
@@ -94,6 +106,8 @@ class BookFullSchema(BookLightSchema):
     zim_metadata: dict[str, Any]
     events: list[str]
     producer: ProducerSchema
+    current_locations: list[BookLocationSchema]
+    target_locations: list[BookLocationSchema]
 
 
 class WarehousePathSchema(BaseModel):