mangareader/backend.py

from PIL import Image
from io import BytesIO
from mimetypes import types_map as mimetypes
from zipfile import ZipFile
import os.path
import sqlite3
import werkzeug.exceptions as exceptions

mimetypes['.webp'] = 'image/webp'

thumbnail_cache = {}


# https://docs.python.org/3.7/library/sqlite3.html#sqlite3.Connection.row_factory
def dict_factory(cursor, row):
    d = {}
    for idx, col in enumerate(cursor.description):
        d[col[0]] = row[idx]
    return d


def generate_thumbnail(filepath):
    if filepath in thumbnail_cache:
        thumbnail = BytesIO()
        thumbnail.write(thumbnail_cache[filepath])
    else:
        image = Image.open(filepath)
        image.thumbnail((512, 512))
        thumbnail = BytesIO()
        image.save(thumbnail, 'webp')
        thumbnail.seek(0)
        thumbnail_cache[filepath] = thumbnail.read()

    thumbnail.seek(0)

    return {
        'data': thumbnail,
        'mimetype': 'image/webp'
    }


class CalibreDB:
    def __init__(self, path='metadata.db'):
        self.database_path = f'file:{path}?mode=ro'

    def create_cursor(self):
        conn = sqlite3.connect(self.database_path, uri=True)
        conn.row_factory = dict_factory
        return conn.cursor()

    def get_series_list(self):
        cursor = self.create_cursor()
        series = cursor.execute('''
            select
                series.id as id,
                series.name as name,
                count(*) as volumes
            from
                books,
                books_series_link,
                data,
                series
            where
                books_series_link.series = series.id and
                books_series_link.book = books.id and
                books.id = data.book and
                data.format = \'CBZ\'
            group by series.name
            having min(books.series_index)
            ''')

        return series

    def get_series_cover(self, series_id):
        cursor = self.create_cursor()
        first_volume = cursor.execute('''
            select
                books.id
            from
                books,
                books_series_link,
                series
            where
                books_series_link.book = books.id and
                books_series_link.series = series.id and
                series.id = ?
            group by series.name
            having min(books.series_index)
        ''', (str(series_id),)).fetchone()

        if first_volume is None:
            raise exceptions.NotFound()

        return self.get_volume_cover(first_volume['id'])

    def get_series_cover_thumbnail(self, series_id):
        return generate_thumbnail(self.get_series_cover(series_id))

    def get_series_volumes(self, series_id):
        cursor = self.create_cursor()
        title = cursor.execute('''
            select
                series.name
            from
                series
            where
                series.id = ?
        ''', (str(series_id),)).fetchone()['name']
        volumes = cursor.execute('''
            select
                books.id,
                books.title,
                books.series_index as "index"
            from
                books,
                books_series_link,
                series
            where
                books_series_link.book = books.id and
                books_series_link.series = series.id and
                series.id = ?
            order by books.series_index
        ''', (str(series_id),)).fetchall()

        return {
            'title': title,
            'volumes': volumes
        }

    def get_volume_cover(self, volume_id):
        cursor = self.create_cursor()
        volume = cursor.execute('''
            select
                books.has_cover as has_cover,
                books.path as path
            from
                books
            where
                books.id = ?
            order by books.series_index
        ''', (str(volume_id),)).fetchone()

        if volume['has_cover']:
            return volume['path'] + '/cover.jpg'
        else:
            raise exceptions.NotFound()

    def get_volume_cover_thumbnail(self, volume_id):
        return generate_thumbnail(self.get_volume_cover(volume_id))

    def get_volume_filepath(self, volume_id):
        cursor = self.create_cursor()
        location = cursor.execute('''
            select
                books.path as path,
                data.name as filename,
                lower(data.format) as extension
            from
                books,
                data
            where
                data.book = books.id and
                books.id = ?
        ''', (str(volume_id),)).fetchone()

        if location is None:
            raise exceptions.NotFound()

        return location['path'] + '/' + location['filename'] + '.' + location['extension']

    def get_volume_info(self, volume_id):
        cursor = self.create_cursor()
        volume_info = cursor.execute('''
        select
            books.title,
            books_series_link.series
        from
            books,
            books_series_link
        where
            books_series_link.book = books.id and
            books.id = ?
        ''', (str(volume_id),)).fetchone()

        volume_info['pages'] = self.get_volume_page_number(volume_id)

        return volume_info

    def get_volume_page_number(self, volume_id):
        path = self.get_volume_filepath(volume_id)
        with ZipFile(path, 'r') as volume:
            return len(volume.filelist)

    def get_volume_page(self, volume_id, page_number):
        if page_number < 1:
            raise exceptions.NotFound()
        path = self.get_volume_filepath(volume_id)
        with ZipFile(path, 'r') as volume:
            try:
                zip_info = volume.filelist[page_number - 1]
            except IndexError:
                raise exceptions.NotFound()
                return None

            page_filename = zip_info.filename

            page_data = BytesIO()
            page_data.write(volume.read(page_filename))
            page_data.seek(0)

            return {
                'data': page_data,
                'mimetype': mimetypes[os.path.splitext(page_filename)[1]],
                'etag': str(zip_info.CRC)
            }