247 lines
7.1 KiB
Python
247 lines
7.1 KiB
Python
#!/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'<h2>{"".join(elem.itertext())}</h2>')
|
|
elif tag == 'p':
|
|
text = ''.join(elem.itertext()).strip()
|
|
if text:
|
|
html_parts.append(f'<p>{text}</p>')
|
|
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'<img src="data:{mime_type};base64,{data}" '
|
|
f'style="max-width: 100%;"/>'
|
|
)
|
|
|
|
return '\n'.join(html_parts)
|
|
|
|
|
|
def clear_cache():
|
|
"""Очищает кэш распарсенных fb2 файлов"""
|
|
_fb2_cache.clear() |