#!/usr/bin/env python3
"""Главный Flask-сервер для Домашней библиотеки"""
from flask import Flask, request, Response, render_template_string
import os
from xml.etree.ElementTree import Element, SubElement, tostring
from xml.dom.minidom import parseString
from db_utils import get_cursor, get_stats
from author_utils import (
get_authors_by_prefix,
get_author_prefixes,
get_author_by_hash,
get_author_series,
get_author_books,
)
from book_utils import (
find_book_archive,
get_parsed_fb2,
get_annotation,
get_cover_cached,
get_table_of_contents,
get_section_html,
)
from series_utils import (
get_series_by_letter,
get_series_by_hash,
get_series_books,
)
app = Flask(__name__)
# ============ HTML ШАБЛОН ============
HTML_TEMPLATE = '''
Домашняя библиотека
'''
# ============ ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ ============
def get_stats_text():
"""Возвращает текст статистики"""
stats = get_stats()
return (f"{stats['books']:,} книг · "
f"{stats['authors']:,} авторов · "
f"{stats['series']:,} серий").replace(',', ' ')
def make_xml(feed):
"""Форматирует XML для OPDS"""
xml_str = parseString(tostring(feed)).toprettyxml()
return Response(xml_str, mimetype='application/atom+xml')
# ============ ВЕБ-МАРШРУТЫ ============
@app.route('/')
def index():
stats_text = get_stats_text()
content = '''
💡 Подсказка
Используйте поиск для быстрого нахождения книг. Или выберите раздел для навигации.
OPDS-адрес для читалок:
http://192.168.30.2:8085/opds
'''
return render_template_string(HTML_TEMPLATE, query='', content=content, stats=stats_text)
@app.route('/search')
def web_search():
q = request.args.get('q', '')
stats_text = get_stats_text()
if not q:
return render_template_string(
HTML_TEMPLATE, query='',
content='Введите поисковый запрос
',
stats=stats_text
)
page = int(request.args.get('page', 1))
offset = (page - 1) * 20
conn, cur = get_cursor()
cur.execute("""
SELECT b.id, b.title, b.language, b.added_date,
string_agg(DISTINCT a.full_name, ', ' ORDER BY ba.position) as authors,
(SELECT COUNT(*) FROM book_authors ba2 WHERE ba2.book_id = b.id) as authors_count,
string_agg(DISTINCT g.title_ru, ', ' ORDER BY g.title_ru) as genres
FROM books b
LEFT JOIN book_authors ba ON ba.book_id = b.id
LEFT JOIN authors a ON a.id = ba.author_id
LEFT JOIN book_genres bg ON bg.book_id = b.id
LEFT JOIN genres g ON g.id = bg.genre_id
WHERE b.fts @@ plainto_tsquery('russian', %s)
OR a.full_name ILIKE %s
GROUP BY b.id, b.title, b.language, b.added_date
ORDER BY ts_rank(b.fts, plainto_tsquery('russian', %s)) DESC
LIMIT 20 OFFSET %s
""", (q, f'%{q}%', q, offset))
books = cur.fetchall()
conn.close()
content = '🔍 Результаты поиска: {}
'.format(q)
if books:
content += ''
for book in books:
genres_html = ''
if book['genres']:
for genre in book['genres'].split(', ')[:3]:
genres_html += '
{}'.format(genre)
coauthors_html = ''
if book['authors_count'] > 1:
if book['authors_count'] <= 3:
coauthors_html = '
👥 соавторы: {}'.format(book['authors'])
else:
coauthors_html = '
👥 {} авторов'.format(book['authors_count'])
title_display = book['title']
if book.get('language') and book['language'] != 'ru':
title_display += f" ({book['language']})"
content += '''
'''.format(book['id'], title_display, book['authors'] or 'Автор неизвестен',
genres_html, coauthors_html, book['id'], book['id'])
content += '
'
if page > 1:
content += '
← Назад'.format(q, page-1)
content += '
Вперёд →'.format(q, page+1)
content += '
Ничего не найдено
'
return render_template_string(HTML_TEMPLATE, query=q, content=content, stats=stats_text)
@app.route('/authors')
def web_authors():
prefix = request.args.get('prefix', '')
stats_text = get_stats_text()
if prefix:
conn, cur = get_cursor()
cur.execute("SELECT COUNT(*) as cnt FROM authors WHERE last_name ILIKE %s", (prefix + '%',))
cnt = cur.fetchone()['cnt']
conn.close()
if cnt > 50:
groups = get_author_prefixes(prefix)
content = 'Авторы'
for i in range(len(prefix)):
content += '
→{}'.format(
prefix[:i+1], prefix[:i+1].upper())
content += '
'
content += 'Авторы: {}
'.format(prefix.upper())
content += ''
for g in groups:
next_prefix = prefix + g['next_letter']
content += '''
'''.format(next_prefix, next_prefix.upper(), g['cnt'], g['first_author'], g['last_author'])
content += '
Авторы'
for i in range(len(prefix)):
content += '
→{}'.format(
prefix[:i+1], prefix[:i+1].upper())
content += '
'
content += 'Авторы: {}
'.format(prefix.upper())
content += ''
for a in authors:
content += '''
'''.format(a['name_hash'], a['full_name'], a['books_count'])
content += '
Выберите букву:
'
content += ''
for l in letters:
content += '
{}'.format(
l['letter'], l['letter'].upper())
content += '
Автор не найден
',
stats=stats_text
)
author_series = get_author_series(author['id'])
author_books = get_author_books(author['id'])
content = '''
'''.format(
author['full_name'], author['full_name'],
len(author_books) + sum(s['books_count'] for s in author_series)
)
if author_series:
content += '📚 Серии
'
content += ''
for s in author_series:
content += '''
'''.format(s['name_hash'], s['name'], s['books_count'])
content += '
📖 Книги и сборники
'
if author_books:
content += ''
for book in author_books:
coauthors_html = ''
if book['authors_count'] > 1:
if book['authors_count'] <= 3:
coauthors_list = [
a.strip() for a in book['coauthors'].split(', ')
if a.strip() != author['full_name']
]
coauthors_html = '
👥 соавторы: {}'.format(
', '.join(coauthors_list[:2]))
else:
coauthors_html = '
👥 {} авторов'.format(
book['authors_count'])
series_html = ''
if book['series_name']:
series_html = '
📚 {}'.format(book['series_name'])
date_html = book['added_date'].strftime('%Y') if book['added_date'] else ''
title_display = book['title']
if book.get('language') and book['language'] != 'ru':
title_display += f" ({book['language']})"
content += '''
'''.format(book['id'], title_display, date_html, series_html, coauthors_html,
book['id'], book['id'])
content += '
Нет книг
'
return render_template_string(HTML_TEMPLATE, query='', content=content, stats=stats_text)
@app.route('/series/')
def web_series_page(series_hash):
stats_text = get_stats_text()
series = get_series_by_hash(series_hash)
if not series:
return render_template_string(
HTML_TEMPLATE, query='',
content='Серия не найдена
',
stats=stats_text
)
books = get_series_books(series['id'])
content = '''
'''.format(series['name'], series['name'], len(books))
if books:
content += ''
for book in books:
number_html = ''
if book['series_number'] and book['series_number'] > 0:
number_html = '
{:.0f}.'.format(
book['series_number'])
title_display = book['title']
if book.get('language') and book['language'] != 'ru':
title_display += f" ({book['language']})"
content += '''
'''.format(number_html, title_display, book['authors'] or '', book['id'], book['id'])
content += '
Нет книг
'
return render_template_string(HTML_TEMPLATE, query='', content=content, stats=stats_text)
@app.route('/genres')
def web_genres():
stats_text = get_stats_text()
conn, cur = get_cursor()
cur.execute("""
SELECT g.id, g.code, g.title_ru,
COUNT(DISTINCT bg.book_id) as books_count
FROM genres g
LEFT JOIN book_genres bg ON bg.genre_id = g.id
WHERE g.is_root = TRUE
GROUP BY g.id, g.code, g.title_ru
ORDER BY books_count DESC
""")
genres = cur.fetchall()
conn.close()
content = 'Жанры
'
content += ''
for g in genres:
content += '''
'''.format(g['code'], g['title_ru'] or g['code'], g['books_count'])
content += '
Книга не найдена
',
stats=stats_text
)
cur.execute("""
SELECT a.id, a.full_name, a.name_hash, ba.position
FROM book_authors ba
JOIN authors a ON a.id = ba.author_id
WHERE ba.book_id = %s
ORDER BY ba.position
""", (book_id,))
authors = cur.fetchall()
cur.execute("""
SELECT g.id, g.code, g.title_ru
FROM book_genres bg
JOIN genres g ON g.id = bg.genre_id
WHERE bg.book_id = %s
ORDER BY g.code
""", (book_id,))
genres = cur.fetchall()
conn.close()
internal_id = book['internal_id2'] or book['internal_id']
title_display = book['title']
if book['language'] and book['language'] != 'ru':
title_display += ' (' + book['language'] + ')'
content = ''
content += ''
content += '
'
content += '
 + ')
'
content += '
'
content += '
'
content += '
'
content += '
' + title_display + '
'
if authors:
content += '
Автор: '
author_links = []
for a in authors:
author_links.append('' + a['full_name'] + '')
content += ', '.join(author_links) + '
'
if book['series_name']:
content += '
Серия: ' + book['series_name'] + ''
if book['series_number'] and book['series_number'] > 0:
content += ' (книга ' + str(int(book['series_number'])) + ')'
content += '
'
if genres:
content += '
Жанры: '
for g in genres:
content += '' + (g['title_ru'] or g['code']) + ' '
content += '
'
if book['added_date']:
content += '
Добавлена: ' + book['added_date'].strftime('%d.%m.%Y') + '
'
try:
tree = get_parsed_fb2(internal_id)
annotation = get_annotation(tree)
if annotation:
content += '
'
content += '
Аннотация:'
content += '
' + annotation + '
'
content += '
'
except:
pass
content += '
📋 Оглавление
'
content += ''
for i, section in enumerate(toc):
content += '
'
content += '
{{ content|safe }}
'''
@app.route('/read//')
def read_section(book_id, section_index):
"""Чтение главы"""
stats_text = get_stats_text()
conn, cur = get_cursor()
cur.execute("""
SELECT b.id, b.title, b.internal_id, b.internal_id2
FROM books b
WHERE b.id = %s
""", (book_id,))
book = cur.fetchone()
conn.close()
if not book:
return render_template_string(
HTML_TEMPLATE, query='',
content='Книга не найдена
',
stats=stats_text
)
internal_id = book['internal_id2'] or book['internal_id']
tree = get_parsed_fb2(internal_id)
if tree is None:
return render_template_string(
HTML_TEMPLATE, query='',
content='Не удалось загрузить книгу
',
stats=stats_text
)
toc = get_table_of_contents(tree)
if section_index < 0 or section_index >= len(toc):
return render_template_string(
HTML_TEMPLATE, query='',
content='Глава не найдена
',
stats=stats_text
)
section_html = get_section_html(tree, section_index)
if section_html is None:
return render_template_string(
HTML_TEMPLATE, query='',
content='Не удалось загрузить главу
',
stats=stats_text
)
prev_section = section_index - 1 if section_index > 0 else None
next_section = section_index + 1 if section_index < len(toc) - 1 else None
section_title = toc[section_index]['title'] if section_index < len(toc) else ''
return render_template_string(
READ_TEMPLATE,
title=book['title'],
book_id=book_id,
book_title=book['title'],
section_title=section_title,
prev_section=prev_section,
next_section=next_section,
content=section_html
)
# ============ СКАЧИВАНИЕ ============
@app.route('/download/')
def download_book(book_id):
"""Отдаёт fb2 файл из архива"""
import zipfile
conn, cur = get_cursor()
cur.execute("""
SELECT internal_id, internal_id2, file_size, format, title
FROM books
WHERE id = %s
""", (book_id,))
book = cur.fetchone()
conn.close()
if not book:
return "Книга не найдена", 404
book_internal_id = book['internal_id2'] or book['internal_id']
if not book_internal_id:
return "Не удалось определить ID книги", 404
archive_path = find_book_archive(book_internal_id)
if not archive_path:
return f"Архив для книги ID={book_internal_id} не найден", 404
try:
with zipfile.ZipFile(archive_path, 'r') as zf:
for filename in zf.namelist():
if str(book_internal_id) in filename:
file_data = zf.read(filename)
return Response(
file_data,
mimetype='application/fb2+xml',
headers={
'Content-Disposition': f'attachment; filename="{filename}"'
}
)
except Exception as e:
return f"Ошибка при чтении архива: {str(e)}", 500
return f"Файл для книги ID={book_internal_id} не найден в архиве", 404
# ============ OPDS ============
@app.route('/opds')
def opds_root():
feed = Element('feed', {
'xmlns': 'http://www.w3.org/2005/Atom',
'xmlns:opds': 'http://opds-spec.org/2010/catalog'
})
SubElement(feed, 'title').text = 'Домашняя библиотека'
SubElement(feed, 'updated').text = '2026-10-10T00:00:00Z'
SubElement(feed, 'id').text = 'urn:uuid:12345678-1234-1234-1234-123456789abc'
entries = [
('search', 'Поиск', '/opds/search?q={searchTerms}'),
('authors', 'Авторы', '/opds/authors'),
('genres', 'Жанры', '/opds/genres')
]
for id_, title, href in entries:
entry = SubElement(feed, 'entry')
SubElement(entry, 'title').text = title
SubElement(entry, 'id').text = f'urn:uuid:{id_}'
SubElement(entry, 'updated').text = '2026-10-10T00:00:00Z'
SubElement(entry, 'link', {
'href': href,
'type': 'application/atom+xml'
})
return make_xml(feed)
@app.route('/opds/search')
def opds_search():
q = request.args.get('q', '')
page = int(request.args.get('page', 1))
offset = (page - 1) * 50
if not q:
return opds_root()
conn, cur = get_cursor()
cur.execute("""
SELECT b.id, b.title,
string_agg(DISTINCT a.full_name, ', ' ORDER BY ba.position) as author,
string_agg(DISTINCT g.code, ':' ORDER BY g.code) as genre_codes
FROM books b
LEFT JOIN book_authors ba ON ba.book_id = b.id
LEFT JOIN authors a ON a.id = ba.author_id
LEFT JOIN book_genres bg ON bg.book_id = b.id
LEFT JOIN genres g ON g.id = bg.genre_id
WHERE b.fts @@ plainto_tsquery('russian', %s)
OR a.full_name ILIKE %s
GROUP BY b.id, b.title
ORDER BY ts_rank(b.fts, plainto_tsquery('russian', %s)) DESC
LIMIT 50 OFFSET %s
""", (q, f'%{q}%', q, offset))
books = cur.fetchall()
conn.close()
feed = Element('feed', {
'xmlns': 'http://www.w3.org/2005/Atom',
'xmlns:dc': 'http://purl.org/dc/terms/',
'xmlns:opds': 'http://opds-spec.org/2010/catalog'
})
SubElement(feed, 'title').text = f'Поиск: {q}'
SubElement(feed, 'updated').text = '2026-10-10T00:00:00Z'
SubElement(feed, 'id').text = 'urn:uuid:search'
for book in books:
entry = SubElement(feed, 'entry')
SubElement(entry, 'title').text = book['title']
SubElement(entry, 'id').text = f"urn:uuid:book:{book['id']}"
SubElement(entry, 'updated').text = '2026-10-10T00:00:00Z'
author = SubElement(entry, 'author')
SubElement(author, 'name').text = book['author']
SubElement(entry, 'link', {
'href': f"/download/{book['id']}",
'type': 'application/fb2+zip',
'rel': 'http://opds-spec.org/acquisition'
})
if book['genre_codes']:
for genre in book['genre_codes'].split(':'):
SubElement(entry, 'category', {'term': genre, 'label': genre})
return make_xml(feed)
@app.route('/opds/authors')
def opds_authors():
prefix = request.args.get('prefix', '')
feed = Element('feed', {
'xmlns': 'http://www.w3.org/2005/Atom',
'xmlns:opds': 'http://opds-spec.org/2010/catalog'
})
SubElement(feed, 'updated').text = '2026-10-10T00:00:00Z'
SubElement(feed, 'id').text = 'urn:uuid:authors'
if prefix:
conn, cur = get_cursor()
cur.execute("SELECT COUNT(*) as cnt FROM authors WHERE last_name ILIKE %s", (prefix + '%',))
cnt = cur.fetchone()['cnt']
conn.close()
if cnt > 50:
groups = get_author_prefixes(prefix)
SubElement(feed, 'title').text = f'Авторы: {prefix.upper()}'
for g in groups:
next_prefix = prefix + g['next_letter']
entry = SubElement(feed, 'entry')
SubElement(entry, 'title').text = f"{next_prefix.upper()} ({g['cnt']})"
SubElement(entry, 'id').text = f"urn:uuid:prefix:{next_prefix}"
SubElement(entry, 'updated').text = '2026-10-10T00:00:00Z'
SubElement(entry, 'link', {
'href': f"/opds/authors?prefix={next_prefix}",
'type': 'application/atom+xml'
})
else:
authors = get_authors_by_prefix(prefix)
SubElement(feed, 'title').text = f'Авторы: {prefix.upper()}'
for a in authors:
entry = SubElement(feed, 'entry')
SubElement(entry, 'title').text = a['full_name']
SubElement(entry, 'id').text = f"urn:uuid:author:{a['name_hash']}"
SubElement(entry, 'updated').text = '2026-10-10T00:00:00Z'
SubElement(entry, 'link', {
'href': f"/opds/author/{a['name_hash']}",
'type': 'application/atom+xml'
})
else:
conn, cur = get_cursor()
cur.execute("""
SELECT DISTINCT lower(substring(last_name, 1, 1)) as letter,
COUNT(*) as cnt
FROM authors
WHERE last_name ~ '^[а-яА-Яa-zA-Z]'
GROUP BY lower(substring(last_name, 1, 1))
ORDER BY letter
""")
letters = cur.fetchall()
conn.close()
SubElement(feed, 'title').text = 'Авторы по алфавиту'
for l in letters:
entry = SubElement(feed, 'entry')
SubElement(entry, 'title').text = f"{l['letter'].upper()} ({l['cnt']})"
SubElement(entry, 'id').text = f"urn:uuid:letter:{l['letter']}"
SubElement(entry, 'updated').text = '2026-10-10T00:00:00Z'
SubElement(entry, 'link', {
'href': f"/opds/authors?prefix={l['letter']}",
'type': 'application/atom+xml'
})
return make_xml(feed)
@app.route('/opds/author/')
def opds_author_books(author_hash):
author = get_author_by_hash(author_hash)
if not author:
return make_xml(Element('feed', {
'xmlns': 'http://www.w3.org/2005/Atom',
'xmlns:opds': 'http://opds-spec.org/2010/catalog'
}))
books = get_author_books(author['id'])
feed = Element('feed', {
'xmlns': 'http://www.w3.org/2005/Atom',
'xmlns:dc': 'http://purl.org/dc/terms/',
'xmlns:opds': 'http://opds-spec.org/2010/catalog'
})
SubElement(feed, 'title').text = f'Книги: {author["full_name"]}'
SubElement(feed, 'updated').text = '2026-10-10T00:00:00Z'
SubElement(feed, 'id').text = f'urn:uuid:author:{author_hash}'
for book in books:
entry = SubElement(feed, 'entry')
SubElement(entry, 'title').text = book['title']
SubElement(entry, 'id').text = f"urn:uuid:book:{book['id']}"
SubElement(entry, 'updated').text = '2026-10-10T00:00:00Z'
author_elem = SubElement(entry, 'author')
SubElement(author_elem, 'name').text = author['full_name']
SubElement(entry, 'link', {
'href': f"/download/{book['id']}",
'type': 'application/fb2+zip',
'rel': 'http://opds-spec.org/acquisition'
})
return make_xml(feed)
@app.route('/opds/genres')
def opds_genres():
conn, cur = get_cursor()
cur.execute("""
SELECT g.code, g.title_ru, COUNT(bg.book_id) as cnt
FROM genres g
LEFT JOIN book_genres bg ON bg.genre_id = g.id
WHERE g.is_root = TRUE
GROUP BY g.code, g.title_ru
ORDER BY cnt DESC
""")
genres = cur.fetchall()
conn.close()
feed = Element('feed', {
'xmlns': 'http://www.w3.org/2005/Atom',
'xmlns:opds': 'http://opds-spec.org/2010/catalog'
})
SubElement(feed, 'title').text = 'Жанры'
SubElement(feed, 'updated').text = '2026-10-10T00:00:00Z'
SubElement(feed, 'id').text = 'urn:uuid:genres'
for g in genres:
entry = SubElement(feed, 'entry')
SubElement(entry, 'title').text = f"{g['title_ru'] or g['code']} ({g['cnt']})"
SubElement(entry, 'id').text = f"urn:uuid:genre:{g['code']}"
SubElement(entry, 'updated').text = '2026-10-10T00:00:00Z'
SubElement(entry, 'link', {
'href': f"/opds/genres?parent={g['code']}",
'type': 'application/atom+xml'
})
return make_xml(feed)
# ============ ЗАПУСК ============
if __name__ == '__main__':
app.run(host='0.0.0.0', port=8085, debug=False)