Работающий функционал
This commit is contained in:
astakhovp committed
1 parent
3b1df81c41
commit
04138d36b1
14 files changed
+2879
-933
No files matched your search
@@ -0,0 +1,247 @@
|
||||
#!/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()
|
||||
Reference in new issue
Block a user