Merge remote-tracking branch 'add-mobi-metadata-extraction/codex/add-mobi-metadata-extraction'

cps/mobi.py

@@ -0,0 +1,326 @@
+# -*- coding: utf-8 -*-
+
+#  This file is part of the Calibre-Web (https://github.com/janeczku/calibre-web)
+#    Copyright (C) 2026 Martin Grzeslowski
+#
+#  This program is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU 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 General Public License for more details.
+#
+#  You should have received a copy of the GNU General Public License
+#  along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+import os
+import re
+import struct
+
+from . import cover, isoLanguages, logger
+from .constants import BookMeta
+from .string_helper import strip_whitespaces
+
+log = logger.create()
+
+PDB_HEADER_LEN = 78
+PDB_RECORD_INFO_LEN = 8
+PALMDOC_HEADER_LEN = 16
+MOBI_HEADER_START = PALMDOC_HEADER_LEN
+MOBI_FULL_NAME_OFFSET = 0x44
+MOBI_FULL_NAME_LENGTH = 0x48
+MOBI_FIRST_IMAGE_INDEX = 0x5C
+UNKNOWN_INDEX = 0xFFFFFFFF
+
+EXTH_AUTHOR = 100
+EXTH_PUBLISHER = 101
+EXTH_DESCRIPTION = 103
+EXTH_ISBN = 104
+EXTH_SUBJECT = 105
+EXTH_PUBLISH_DATE = 106
+EXTH_COVER_OFFSET = 201
+EXTH_THUMBNAIL_OFFSET = 202
+EXTH_TITLE = 503
+EXTH_LANGUAGE = 524
+
+
+def get_mobi_info(tmp_file_path, original_file_name, original_file_extension, no_cover_processing):
+    data = _read_file(tmp_file_path)
+    record_offsets = _read_record_offsets(data)
+    record0 = _record(data, record_offsets, 0)
+    _ensure_mobi_record(record0)
+
+    encoding = _text_encoding(record0)
+    exth_records = _read_exth_records(record0)
+
+    title = _book_title(record0, encoding) or _first_text(exth_records, EXTH_TITLE, encoding) or original_file_name
+    author = _author_string(_texts(exth_records, EXTH_AUTHOR, encoding))
+    publisher = _first_text(exth_records, EXTH_PUBLISHER, encoding)
+    description = _first_text(exth_records, EXTH_DESCRIPTION, encoding)
+    tags = ', '.join(_unique(_texts(exth_records, EXTH_SUBJECT, encoding)))
+    pubdate = _first_text(exth_records, EXTH_PUBLISH_DATE, encoding)[:10]
+    languages = _language_code(_first_text(exth_records, EXTH_LANGUAGE, encoding))
+
+    identifiers = []
+    isbn = _first_text(exth_records, EXTH_ISBN, encoding)
+    if isbn:
+        identifiers.append(['isbn', isbn])
+
+    cover_file = None
+    if not no_cover_processing:
+        cover_file = _extract_cover(data, record_offsets, record0, exth_records, tmp_file_path)
+
+    return BookMeta(
+        file_path=tmp_file_path,
+        extension=original_file_extension,
+        title=title,
+        author=author or 'Unknown',
+        cover=cover_file,
+        description=description,
+        tags=tags,
+        series="",
+        series_id="",
+        languages=languages,
+        publisher=publisher,
+        pubdate=pubdate,
+        identifiers=identifiers)
+
+
+def _read_file(file_path):
+    with open(file_path, 'rb') as mobi_file:
+        return mobi_file.read()
+
+
+def _u16(data, offset):
+    return struct.unpack('>H', data[offset:offset + 2])[0]
+
+
+def _u32(data, offset):
+    return struct.unpack('>I', data[offset:offset + 4])[0]
+
+
+def _read_record_offsets(data):
+    if len(data) < PDB_HEADER_LEN:
+        raise ValueError('MOBI file is shorter than the Palm database header')
+
+    record_count = _u16(data, 76)
+    table_end = PDB_HEADER_LEN + record_count * PDB_RECORD_INFO_LEN
+    if record_count < 1 or table_end > len(data):
+        raise ValueError('Invalid Palm database record table')
+
+    offsets = [_u32(data, PDB_HEADER_LEN + index * PDB_RECORD_INFO_LEN) for index in range(record_count)]
+    offsets.append(len(data))
+
+    previous = table_end
+    for offset in offsets[:-1]:
+        if offset < previous or offset > len(data):
+            raise ValueError('Invalid Palm database record offset')
+        previous = offset
+    return offsets
+
+
+def _record(data, record_offsets, index):
+    if index < 0 or index >= len(record_offsets) - 1:
+        return b''
+    return data[record_offsets[index]:record_offsets[index + 1]]
+
+
+def _ensure_mobi_record(record0):
+    if len(record0) < MOBI_HEADER_START + 8 or record0[MOBI_HEADER_START:MOBI_HEADER_START + 4] != b'MOBI':
+        raise ValueError('File does not contain a MOBI header')
+
+
+def _mobi_header_len(record0):
+    return _u32(record0, MOBI_HEADER_START + 4)
+
+
+def _text_encoding(record0):
+    encoding = _u32(record0, MOBI_HEADER_START + 12)
+    if encoding == 1252:
+        return 'cp1252'
+    if encoding == 65001:
+        return 'utf-8'
+    return 'utf-8'
+
+
+def _read_exth_records(record0):
+    header_len = _mobi_header_len(record0)
+    exth_start = MOBI_HEADER_START + header_len
+
+    if record0[exth_start:exth_start + 4] != b'EXTH':
+        exth_start = record0.find(b'EXTH', MOBI_HEADER_START)
+    if exth_start < 0:
+        return {}
+
+    exth_len = _u32(record0, exth_start + 4)
+    record_count = _u32(record0, exth_start + 8)
+    exth_end = min(len(record0), exth_start + exth_len)
+    position = exth_start + 12
+    records = {}
+
+    for __ in range(record_count):
+        if position + 8 > exth_end:
+            break
+        record_type = _u32(record0, position)
+        record_len = _u32(record0, position + 4)
+        if record_len < 8 or position + record_len > exth_end:
+            break
+        records.setdefault(record_type, []).append(record0[position + 8:position + record_len])
+        position += record_len
+
+    return records
+
+
+def _decode_text(value, encoding):
+    value = value.rstrip(b'\0')
+    for candidate in (encoding, 'utf-8', 'cp1252', 'latin-1'):
+        try:
+            return strip_whitespaces(value.decode(candidate))
+        except UnicodeDecodeError:
+            pass
+    return ''
+
+
+def _texts(exth_records, record_type, encoding):
+    return [text for text in (_decode_text(value, encoding) for value in exth_records.get(record_type, [])) if text]
+
+
+def _first_text(exth_records, record_type, encoding):
+    values = _texts(exth_records, record_type, encoding)
+    return values[0] if values else ''
+
+
+def _book_title(record0, encoding):
+    header_len = _mobi_header_len(record0)
+    if header_len < MOBI_FULL_NAME_LENGTH + 4:
+        return ''
+
+    name_offset = _u32(record0, MOBI_HEADER_START + MOBI_FULL_NAME_OFFSET)
+    name_length = _u32(record0, MOBI_HEADER_START + MOBI_FULL_NAME_LENGTH)
+    if name_length and name_offset + name_length <= len(record0):
+        return _decode_text(record0[name_offset:name_offset + name_length], encoding)
+
+    alternate_offset = MOBI_HEADER_START + name_offset
+    if name_length and alternate_offset + name_length <= len(record0):
+        return _decode_text(record0[alternate_offset:alternate_offset + name_length], encoding)
+    return ''
+
+
+def _author_string(authors):
+    normalized = []
+    for value in authors:
+        normalized.extend(_split_author_value(value))
+
+    return ' & '.join(_unique(normalized))
+
+
+def _split_author_value(value):
+    authors = []
+    for part in re.split('[&;]', value):
+        part = strip_whitespaces(part)
+        if not part:
+            continue
+
+        if part.count(',') == 1:
+            left, right = [strip_whitespaces(item) for item in part.split(',', 1)]
+            if ' ' in left and ' ' in right:
+                authors.extend([left, right])
+            else:
+                authors.append(strip_whitespaces('%s %s' % (right, left)))
+        elif part.count(',') > 1:
+            authors.extend([strip_whitespaces(item) for item in part.split(',') if strip_whitespaces(item)])
+        else:
+            authors.append(part)
+    return authors
+
+
+def _unique(values):
+    seen = set()
+    result = []
+    for value in values:
+        if value not in seen:
+            seen.add(value)
+            result.append(value)
+    return result
+
+
+def _language_code(language):
+    if not language:
+        return ''
+    language = language.split('-', 1)[0].lower()
+    return isoLanguages.get_lang3(language)
+
+
+def _first_image_index(record0):
+    header_len = _mobi_header_len(record0)
+    if header_len < MOBI_FIRST_IMAGE_INDEX + 4:
+        return None
+    index = _u32(record0, MOBI_HEADER_START + MOBI_FIRST_IMAGE_INDEX)
+    return None if index == UNKNOWN_INDEX else index
+
+
+def _exth_int(exth_records, record_type):
+    values = exth_records.get(record_type, [])
+    if not values or len(values[0]) < 4:
+        return None
+    return _u32(values[0], 0)
+
+
+def _extract_cover(data, record_offsets, record0, exth_records, tmp_file_path):
+    first_image_index = _first_image_index(record0)
+    candidates = []
+    if first_image_index is not None:
+        for record_type in (EXTH_COVER_OFFSET, EXTH_THUMBNAIL_OFFSET):
+            image_offset = _exth_int(exth_records, record_type)
+            if image_offset is not None:
+                candidates.append(first_image_index + image_offset)
+
+    for index in candidates:
+        cover_file = _cover_from_record(data, record_offsets, index, tmp_file_path)
+        if cover_file:
+            return cover_file
+
+    return _largest_image_cover(data, record_offsets, first_image_index, tmp_file_path)
+
+
+def _cover_from_record(data, record_offsets, index, tmp_file_path):
+    blob = _record(data, record_offsets, index)
+    extension = _image_extension(blob)
+    if not extension:
+        return None
+    try:
+        return cover.cover_processing(tmp_file_path, blob, extension)
+    except Exception as ex:
+        log.warning('Cannot extract MOBI cover image, using default: %s', ex)
+        return None
+
+
+def _largest_image_cover(data, record_offsets, first_image_index, tmp_file_path):
+    start = first_image_index if first_image_index is not None else 1
+    images = []
+    for index in range(start, len(record_offsets) - 1):
+        blob = _record(data, record_offsets, index)
+        extension = _image_extension(blob)
+        if extension:
+            images.append((len(blob), index))
+
+    for __, index in sorted(images, reverse=True):
+        cover_file = _cover_from_record(data, record_offsets, index, tmp_file_path)
+        if cover_file:
+            return cover_file
+    return None
+
+
+def _image_extension(blob):
+    if blob.startswith(b'\xff\xd8\xff'):
+        return '.jpg'
+    if blob.startswith(b'\x89PNG\r\n\x1a\n'):
+        return '.png'
+    if blob.startswith(b'RIFF') and blob[8:12] == b'WEBP':
+        return '.webp'
+    if blob.startswith(b'BM'):
+        return '.bmp'
+    return None

cps/uploader.py

@@ -65,6 +65,13 @@ except ImportError as e:
     log.debug('Cannot import epub, extracting epub metadata will not work: %s', e)
     use_epub_meta = False
 
+try:
+    from . import mobi
+    use_mobi_meta = True
+except ImportError as e:
+    log.debug('Cannot import mobi, extracting mobi metadata will not work: %s', e)
+    use_mobi_meta = False
+
 try:
     from . import fb2
     use_fb2_meta = True
@@ -88,6 +95,8 @@ def process(tmp_file_path, original_file_name, original_file_extension, rar_exec
             meta = pdf_meta(tmp_file_path, original_file_name, original_file_extension, no_cover)
         elif extension_upper in [".KEPUB", ".EPUB"] and use_epub_meta is True:
             meta = epub.get_epub_info(tmp_file_path, original_file_name, original_file_extension, no_cover)
+        elif extension_upper in [".MOBI", ".AZW", ".AZW3", ".PRC", ".POBI"] and use_mobi_meta is True:
+            meta = mobi.get_mobi_info(tmp_file_path, original_file_name, original_file_extension, no_cover)
         elif ".FB2" == extension_upper and use_fb2_meta is True:
             meta = fb2.get_fb2_info(tmp_file_path, original_file_extension)
         elif extension_upper in ['.CBZ', '.CBT', '.CBR', ".CB7"]: