# SPDX-License-Identifier: AGPL-3.0-or-later
#
# Cata-Log - the central hub for digital flyers
# Copyright (C) 2026 David Aderbauer & The Cata-Log Contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import contextlib
import logging
import uuid
from datetime import datetime
from hashlib import sha256
from io import BytesIO
from pathlib import Path
from apscheduler.job import Job
from apscheduler.jobstores.base import JobLookupError
from apscheduler.triggers.cron import CronTrigger
from ebooklib import epub
from PIL import Image
from sqlalchemy import Enum as SQLEnum
from sqlalchemy import orm
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.schema import CheckConstraint, ForeignKey, MetaData, UniqueConstraint
from sqlalchemy.sql import select
from sqlalchemy.types import JSON, String, Text
from cata_log_hub.constants import StatusEnum
from cata_log_hub.exceptions import (
NetworkError,
ProviderUnknownClassWarning,
ProviderWarning,
)
from cata_log_hub.providers import Provider as ProviderType
from cata_log_hub.scheduler import scheduler
from cata_log_hub.settings import get_settings
from .mixins import TimestampMixin
from .types import PathType, UTCDatetime
logger = logging.getLogger(__name__)
[docs]
class ModelBase(orm.DeclarativeBase):
"""Base for all ORM models."""
metadata = MetaData(
naming_convention={
"ix": "ix_%(column_0_label)s",
"uq": "uq_%(table_name)s__%(column_0_N_name)s",
"ck": "ck_%(table_name)s__%(column_0_name)s_%(constraint_name)s ",
"fk": "fk_%(table_name)s__%(column_0_name)s__%(referred_table_name)s",
"pk": "pk_%(table_name)s",
}
)
[docs]
class Provider(ModelBase, TimestampMixin):
"""ORM model for a catalog provider."""
id: orm.Mapped[int] = orm.mapped_column(primary_key=True)
class_uid: orm.Mapped[str] = orm.mapped_column(String(127))
configuration: orm.Mapped[dict[str, str]] = orm.mapped_column(
MutableDict.as_mutable(JSON)
)
note: orm.Mapped[str] = orm.mapped_column(Text, default="")
job_id: orm.Mapped[str | None] = orm.mapped_column(
String(255), nullable=True, unique=True
)
catalogs: orm.Mapped[list[Catalog]] = orm.relationship(
back_populates="provider", cascade="all, delete-orphan"
)
status: orm.Mapped[StatusEnum] = orm.mapped_column(
SQLEnum(StatusEnum, name="provider_status_enum"),
default=StatusEnum.HEALTHY,
server_default=StatusEnum.HEALTHY,
)
__tablename__ = "providers"
[docs]
def get_provider_class(self) -> type[ProviderType]:
"""Get the class for this provider.
Returns:
The provider class.
Raises:
:exc:`cata_log_hub.exceptions.ProviderUnknownClassWarning`: If there is no class for this provider's class-uid.
"""
return ProviderType.get_class(self.class_uid)
[docs]
def get_provider_instance(self) -> ProviderType:
"""Get a class instance for this provider.
Returns:
The provider instance.
Raises:
:exc:`cata_log_hub.exceptions.ProviderUnknownClassWarning`: If there is no class for this provider's class-uid.
"""
return self.get_provider_class()(self.configuration)
[docs]
def fetch_catalog(self, db_session: orm.Session) -> None:
"""Fetch this provider's catalog and save it to storage and db.
Raises:
:exc:`cata_log_hub.exceptions.NetworkError`: Reraised
"""
try:
logger.debug(
"Fetching catalog of provider ...", extra={"provider_id": self.id}
)
with (
db_session.begin_nested(),
self.get_provider_instance() as provider_fetcher,
):
db_session.add(self)
new_catalog = Catalog(
provider_id=self.id,
valid_since=provider_fetcher.get_valid_since(),
valid_until=provider_fetcher.get_valid_until(),
)
db_session.add(new_catalog)
db_session.flush()
logger.debug(
"Success adding new catalog.",
extra={"provider_id": self.id, "catalog_id": new_catalog.id},
)
for (
page_number,
page_bytes,
) in provider_fetcher.iter_catalog_pages():
pagefile = PageFile.get_or_create(
db_session=db_session,
page_bytes=page_bytes,
)
logger.debug(
"Success saving page data to storage.",
extra={
"provider_id": self.id,
"catalog_id": new_catalog.id,
"page_nr": page_number,
"path": pagefile.path,
},
)
new_page = Page(
catalog_id=new_catalog.id,
file_id=pagefile.id,
number=page_number,
)
db_session.add(new_page)
except ProviderWarning as provider_warning:
logger.exception(
"Provider catalog is %s!",
provider_warning.provider_status.value,
extra={
"provider_id": self.id,
"provider_class_uid": self.class_uid,
},
)
self.status = provider_warning.provider_status
except NetworkError:
logger.warning(
"Network error while fetching provider catalog.",
extra={
"provider_id": self.id,
},
exc_info=True,
)
raise
else:
logger.debug(
"Success fetching catalog of provider.",
extra={"provider_id": self.id},
)
self.status = StatusEnum.HEALTHY
db_session.commit()
[docs]
def add_job(self) -> None:
"""Add the job for this provider."""
if not self.job_id:
try:
provider_class = self.get_provider_class()
except ProviderUnknownClassWarning:
logger.exception(
"No provider class found for provider instance!",
extra={
"provider_id": self.id,
"provider_class_uid": self.class_uid,
},
)
return
cron_trigger = CronTrigger.from_crontab(
provider_class.schedule, timezone=provider_class.region.timezone
)
cron_trigger.jitter = provider_class.jitter
job = scheduler.add_job(
name=f"{self.class_uid}-{self.configuration}",
func="cata_log_hub.jobs:fetch_provider",
args=[self.id],
trigger=cron_trigger,
)
self.job_id = job.id
[docs]
def remove_job(self) -> None:
"""Remove the job of this provider."""
if self.job_id:
with contextlib.suppress(JobLookupError):
scheduler.remove_job(self.job_id)
self.job_id = None
@property
def job(self) -> Job | None:
"""The job of this provider."""
return scheduler.get_job(self.job_id)
[docs]
class Catalog(ModelBase, TimestampMixin):
"""ORM model for a catalog."""
id: orm.Mapped[int] = orm.mapped_column(primary_key=True)
valid_since: orm.Mapped[datetime] = orm.mapped_column(UTCDatetime)
valid_until: orm.Mapped[datetime] = orm.mapped_column(UTCDatetime)
provider: orm.Mapped[Provider] = orm.relationship(back_populates="catalogs")
provider_id: orm.Mapped[int] = orm.mapped_column(
ForeignKey(Provider.__tablename__ + ".id", ondelete="CASCADE"), nullable=False
)
pages: orm.Mapped[list[Page]] = orm.relationship(
back_populates="catalog", cascade="all, delete-orphan"
)
__tablename__ = "catalogs"
[docs]
def as_pdf(self) -> bytes:
"""Compress the catalog to a pdf file.
Returns:
The pdf file bytes.
"""
pdf_bytes_io = BytesIO()
page_images = [Image.open(page.file.path) for page in self.pages]
page_images[0].save(
pdf_bytes_io,
format="PDF",
resolution=100.0,
save_all=True,
append_images=page_images[1:],
)
pdf_bytes_io.seek(0)
return pdf_bytes_io.read()
[docs]
def as_epub(self) -> bytes:
"""Convert the catalog to a epub file.
Returns:
The epub file bytes.
"""
provider_class = self.provider.get_provider_class()
book = epub.EpubBook()
book.set_title(
f"{self.provider.class_uid.title()}, {self.valid_since.date()} - {self.valid_until.date()}"
)
book.set_direction("rtl" if provider_class.region.is_rtl else "ltr")
book.set_language(provider_class.region.language_code)
book.set_cover(
f"cover_{self.pages[0].file.name}",
self.pages[0].file.path.read_bytes(),
)
book.add_author("Cata-Log")
book.add_metadata("DC", "description", provider_class.description)
book.add_metadata(
None,
"meta",
"",
{"valid_since": self.valid_since.isoformat(timespec="seconds")},
)
book.add_metadata(
None,
"meta",
"",
{"valid_until": self.valid_until.isoformat(timespec="seconds")},
)
for page in self.pages:
page_image = epub.EpubImage(
uid=str(page.id),
file_name=page.file.name,
content=page.file.path.read_bytes(),
)
book.add_item(page_image)
page_chapter = epub.EpubHtml(
title=f"Page {page.number + 1}",
file_name=f"chap_{page.number + 1}.xhtml",
content=f'<img src="{page.file.name}"/>',
)
book.add_item(page_chapter)
book.spine.append(page_chapter)
book.toc = book.spine
book.add_item(epub.EpubNcx())
book.add_item(epub.EpubNav())
book_bytes_io = BytesIO()
epub.write_epub(book_bytes_io, book)
book_bytes_io.seek(0)
return book_bytes_io.read()
[docs]
@classmethod
def cleanup(cls, db_session: orm.Session, deadline: datetime) -> None:
"""Cleanup catalogs created before a deadline.
Args:
db_session: A database session.
deadline: The deadline for catalog deletion.
"""
logger.info(
"Deleting outdated catalogs ...",
extra={"expiration_deadline": deadline},
)
for catalog in db_session.query(cls).filter(cls.created_at < deadline).all():
with db_session.begin_nested():
db_session.delete(catalog)
logger.debug(
"Deleted outdated catalog.",
extra={
"catalog_id": catalog.id,
"creation_date": catalog.created_at,
"expiration_deadline": deadline,
},
)
logger.info(
"Success deleting outdated catalogs.",
extra={"expiration_deadline": deadline},
)
[docs]
class PageFile(ModelBase, TimestampMixin):
"""ORM model for a catalog page file."""
id: orm.Mapped[int] = orm.mapped_column(primary_key=True)
path: orm.Mapped[Path] = orm.mapped_column(PathType, unique=True)
original_sha256: orm.Mapped[str] = orm.mapped_column(String(64))
sha256: orm.Mapped[str] = orm.mapped_column(String(64))
size: orm.Mapped[int] = orm.mapped_column(CheckConstraint("size > 0", name="gt_0"))
height: orm.Mapped[int] = orm.mapped_column(
CheckConstraint("height > 0", name="gt_0")
)
width: orm.Mapped[int] = orm.mapped_column(
CheckConstraint("width > 0", name="gt_0")
)
pages: orm.Mapped[list[Page]] = orm.relationship(
back_populates="file",
passive_deletes="all", # essential to make ondelete=RESTRICT work: https://stackoverflow.com/questions/55968951/sqlalchemy-fk-ondelete-does-not-restrict
)
__tablename__ = "pagefiles"
[docs]
@orm.validates("size", "width", "height")
def validate_nonnegative_number(self, key: str, value: int) -> int:
"""Ensure size values are positive."""
if value <= 0:
raise ValueError(f"Invalid {key} {value}")
return value
@property
def name(self) -> str:
"""The name of the stored file."""
return self.path.name
[docs]
@classmethod
def get_or_create(cls, db_session: orm.Session, page_bytes: bytes) -> PageFile:
"""Get or create a pagefile.
Args:
db_session: A database session to use.
page_bytes: The content of the pagefile.
path: The path for the pagefile.
"""
original_sha256 = sha256(page_bytes).hexdigest()
pagefile = (
db_session.query(cls).filter(cls.original_sha256 == original_sha256).first()
)
if not pagefile:
pagefile = cls(
original_sha256=original_sha256,
path=get_settings().storage_path / (str(uuid.uuid4()) + ".webp"),
)
with Image.open(BytesIO(page_bytes)) as image:
pagefile.width, pagefile.height = image.size
webp_image_io = BytesIO()
image.save(
webp_image_io,
format="WEBP",
lossless=False,
quality=80,
method=0,
save_all=True,
)
webp_image_io.seek(0)
webp_image_bytes = webp_image_io.read()
pagefile.sha256 = sha256(webp_image_bytes).hexdigest()
pagefile.size = len(webp_image_bytes)
db_session.add(pagefile)
db_session.flush()
pagefile.path.write_bytes(webp_image_bytes)
return pagefile
[docs]
@classmethod
def cleanup(cls, db_session: orm.Session) -> None:
"""Cleanup unused page files.
Args:
db_session: A database session.
"""
used_paths = set(db_session.execute(select(cls.path)).scalars().all())
for storage_filepath in get_settings().storage_path.iterdir():
if storage_filepath.is_dir() or storage_filepath.resolve() in used_paths:
continue
storage_filepath.unlink(missing_ok=True)
[docs]
class Page(ModelBase, TimestampMixin):
"""ORM model for a catalog page."""
id: orm.Mapped[int] = orm.mapped_column(primary_key=True)
number: orm.Mapped[int] = orm.mapped_column(
CheckConstraint("number >= 0", name="ge_0")
)
catalog_id: orm.Mapped[int] = orm.mapped_column(
ForeignKey(Catalog.__tablename__ + ".id", ondelete="CASCADE"), nullable=False
)
catalog: orm.Mapped[Catalog] = orm.relationship(back_populates="pages")
file_id: orm.Mapped[int] = orm.mapped_column(
ForeignKey(PageFile.__tablename__ + ".id", ondelete="RESTRICT"), nullable=False
)
file: orm.Mapped[PageFile] = orm.relationship(back_populates="pages")
__tablename__ = "pages"
__table_args__ = (UniqueConstraint("catalog_id", "number"),)
[docs]
@orm.validates("number")
def validate_nonnegative_number(self, key: str, number: int) -> int:
"""Ensure number is nonnegative."""
if number < 0:
raise ValueError(f"Invalid {key} {number}")
return number
