#!/usr/bin/env python3 """Модуль для работы с книгами (извлечение из fb2, конвертация)""" import zipfile import os import base64 import io from lxml import etree BOOKS_PATH = os.environ.get( 'BOOKS_PATH', '/mnt/hdd/share/Torrents/Downloaded/Libruks' ) COVERS_PATH = os.environ.get('COVERS_PATH', '/mnt/hdd/opds/covers') # Пространства имён fb2 FB2_NS = 'http://www.gribuser.ru/xml/fictionbook/2.0' XLINK_NS = 'http://www.w3.org/1999/xlink' NS = {'fb': FB2_NS, 'xl': XLINK_NS} # Кэш для распарсенных fb2 файлов (в памяти) _fb2_cache = {} def find_book_archive(internal_id): """Находит архив с книгой по internal_id""" archives_dir = os.path.join(BOOKS_PATH, 'Архивы Либрусек') if not os.path.exists(archives_dir): return None for archive_name in sorted(os.listdir(archives_dir)): if not archive_name.endswith('.zip'): continue try: parts = archive_name.replace('.zip', '').split('-') start_id = int(parts[1]) end_id = int(parts[2]) except: continue if start_id <= internal_id <= end_id: return os.path.join(archives_dir, archive_name) return None def get_book_fb2(internal_id): """Извлекает fb2 данные из архива, обрабатывая вложенные zip""" archive_path = find_book_archive(internal_id) if not archive_path: return None try: with zipfile.ZipFile(archive_path, 'r') as zf: for filename in zf.namelist(): if str(internal_id) in filename: data = zf.read(filename) # Проверяем, не zip ли это внутри if data[:4] == b'PK\x03\x04': inner_zip = zipfile.ZipFile(io.BytesIO(data)) # Ищем fb2 внутри for inner_name in inner_zip.namelist(): if inner_name.lower().endswith('.fb2'): return inner_zip.read(inner_name) # Если fb2 не найден, берём первый файл if inner_zip.namelist(): return inner_zip.read(inner_zip.namelist()[0]) return data except: pass return None def get_parsed_fb2(internal_id): """Возвращает распарсенное fb2 дерево (с кэшированием)""" if internal_id in _fb2_cache: return _fb2_cache[internal_id] fb2_data = get_book_fb2(internal_id) if not fb2_data: return None try: parser = etree.XMLParser(recover=True) tree = etree.fromstring(fb2_data, parser) _fb2_cache[internal_id] = tree return tree except: return None def get_annotation(tree): """Извлекает аннотацию из fb2""" if tree is None: return None annotation = tree.find(f'.//{{{FB2_NS}}}annotation') if annotation is not None: texts = [] for p in annotation.findall(f'.//{{{FB2_NS}}}p'): text = ''.join(p.itertext()).strip() if text: texts.append(text) if texts: return '\n\n'.join(texts) text = ''.join(annotation.itertext()).strip() if text: return text return None def get_cover_image(tree): """Извлекает обложку из fb2. Возвращает (data, mime_type) или (None, None)""" if tree is None: return None, None coverpage = tree.find(f'.//{{{FB2_NS}}}coverpage') if coverpage is None: return None, None image = coverpage.find(f'.//{{{FB2_NS}}}image') if image is None: return None, None href = image.get(f'{{{XLINK_NS}}}href') if not href: return None, None binary_id = href.lstrip('#') binary = tree.find(f".//{{{FB2_NS}}}binary[@id='{binary_id}']") if binary is None: return None, None mime_type = binary.get('content-type', 'image/jpeg') data = binary.text if not data: return None, None try: return base64.b64decode(data), mime_type except: return None, None def get_cover_cached(internal_id): """Возвращает обложку из кэша или извлекает из fb2 и кэширует""" os.makedirs(COVERS_PATH, exist_ok=True) cover_path = os.path.join(COVERS_PATH, f"{internal_id}.jpg") if os.path.exists(cover_path): with open(cover_path, 'rb') as f: return f.read(), 'image/jpeg' tree = get_parsed_fb2(internal_id) if tree is None: return None, None data, mime_type = get_cover_image(tree) if data is None: return None, None try: with open(cover_path, 'wb') as f: f.write(data) except: pass return data, mime_type def get_table_of_contents(tree): """Извлекает оглавление из fb2""" if tree is None: return [] sections = [] body = tree.find(f'.//{{{FB2_NS}}}body') if body is not None: for i, section in enumerate(body.findall(f'.//{{{FB2_NS}}}section')): title_elem = section.find(f'{{{FB2_NS}}}title') if title_elem is not None: title = ''.join(title_elem.itertext()).strip() if title: sections.append({ 'index': i, 'title': title }) return sections def get_section_html(tree, section_index): """Конвертирует секцию fb2 в HTML""" if tree is None: return None body = tree.find(f'.//{{{FB2_NS}}}body') if body is None: return None sections = body.findall(f'.//{{{FB2_NS}}}section') if section_index >= len(sections): return None section = sections[section_index] html_parts = [] for elem in section.iter(): tag = etree.QName(elem).localname if tag == 'title': html_parts.append(f'
{text}
') elif tag == 'image': href = elem.get(f'{{{XLINK_NS}}}href') if href: binary_id = href.lstrip('#') binary = tree.find(f".//{{{FB2_NS}}}binary[@id='{binary_id}']") if binary is not None: mime_type = binary.get('content-type', 'image/jpeg') data = binary.text if data: html_parts.append( f'