feat(ebook-search): implement async file path resolution and EPUB discovery
This commit is contained in:
@@ -11,6 +11,7 @@ from pathlib import Path
|
|||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
import tiktoken
|
import tiktoken
|
||||||
|
from anyio import Path as AsyncPath
|
||||||
from sqlalchemy import or_, select
|
from sqlalchemy import or_, select
|
||||||
|
|
||||||
from python.ebook_search.epub_parse import parse_epub
|
from python.ebook_search.epub_parse import parse_epub
|
||||||
@@ -74,18 +75,18 @@ def chunk_text(
|
|||||||
return [chunk for chunk in chunks if chunk.text]
|
return [chunk for chunk in chunks if chunk.text]
|
||||||
|
|
||||||
|
|
||||||
def find_library_epubs(library_path: str) -> tuple[Path, list[Path] | None]:
|
async def find_library_epubs(library_path: str) -> tuple[AsyncPath, list[AsyncPath] | None]:
|
||||||
"""Resolve one configured library path and collect its EPUB files (blocking filesystem walk).
|
"""Resolve one configured library path and collect its EPUB files asynchronously.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
tuple[Path, list[Path] | None]: The expanded path and its EPUB files, or ``None`` when
|
tuple[Path, list[Path] | None]: The expanded path and its EPUB files, or ``None`` when
|
||||||
the path is neither an EPUB file nor a directory.
|
the path is neither an EPUB file nor a directory.
|
||||||
"""
|
"""
|
||||||
path = Path(library_path).expanduser()
|
path = await AsyncPath(library_path).expanduser()
|
||||||
if path.is_file() and path.suffix.lower() == ".epub":
|
if await path.is_file() and path.suffix.lower() == ".epub":
|
||||||
return path, [path]
|
return path, [path]
|
||||||
if path.is_dir():
|
if await path.is_dir():
|
||||||
return path, sorted(path.rglob("*.epub"))
|
return path, sorted([epub_path async for epub_path in path.rglob("*.epub")])
|
||||||
return path, None
|
return path, None
|
||||||
|
|
||||||
|
|
||||||
@@ -93,7 +94,7 @@ async def ingest_configured_paths(session: AsyncSession, config: EbookSearchConf
|
|||||||
"""Ingest every EPUB found under configured library paths."""
|
"""Ingest every EPUB found under configured library paths."""
|
||||||
count = 0
|
count = 0
|
||||||
for library_path in config.library_paths:
|
for library_path in config.library_paths:
|
||||||
path, epub_paths = await asyncio.to_thread(find_library_epubs, library_path)
|
path, epub_paths = await find_library_epubs(library_path)
|
||||||
logger.info(f"ebook_ingest_path_start {path=}")
|
logger.info(f"ebook_ingest_path_start {path=}")
|
||||||
if epub_paths is None:
|
if epub_paths is None:
|
||||||
logger.warning(f"ebook_ingest_path_missing {path=}")
|
logger.warning(f"ebook_ingest_path_missing {path=}")
|
||||||
@@ -104,20 +105,21 @@ async def ingest_configured_paths(session: AsyncSession, config: EbookSearchConf
|
|||||||
return count
|
return count
|
||||||
|
|
||||||
|
|
||||||
def resolve_ingest_path(path: Path) -> Path:
|
async def resolve_ingest_path(path: Path | AsyncPath) -> AsyncPath:
|
||||||
"""Expand and resolve an ingest path (blocking filesystem call)."""
|
"""Expand and resolve an ingest path without blocking the event loop."""
|
||||||
return path.expanduser().resolve()
|
expanded_path = await AsyncPath(path).expanduser()
|
||||||
|
return await expanded_path.resolve()
|
||||||
|
|
||||||
|
|
||||||
async def ingest_file(session: AsyncSession, path: Path, config: EbookSearchConfig) -> bool:
|
async def ingest_file(session: AsyncSession, path: Path | AsyncPath, config: EbookSearchConfig) -> bool:
|
||||||
"""Ingest one EPUB file. Return True when the database changed."""
|
"""Ingest one EPUB file. Return True when the database changed."""
|
||||||
try:
|
try:
|
||||||
resolved_path = await asyncio.to_thread(resolve_ingest_path, path)
|
resolved_path = await resolve_ingest_path(path)
|
||||||
logger.info(f"ebook_ingest_file_start {resolved_path=}")
|
logger.info(f"ebook_ingest_file_start {resolved_path=}")
|
||||||
file_hash = await asyncio.to_thread(sha256_file, resolved_path)
|
file_hash = await sha256_file(resolved_path)
|
||||||
existing = await find_existing_source(session, resolved_path, file_hash)
|
existing = await find_existing_source(session, resolved_path, file_hash)
|
||||||
if existing is not None and existing.file_sha256 == file_hash:
|
if existing is not None and existing.file_sha256 == file_hash:
|
||||||
stat = resolved_path.stat()
|
stat = await resolved_path.stat()
|
||||||
existing.file_path = str(resolved_path)
|
existing.file_path = str(resolved_path)
|
||||||
existing.file_mtime = datetime.fromtimestamp(stat.st_mtime, tz=UTC)
|
existing.file_mtime = datetime.fromtimestamp(stat.st_mtime, tz=UTC)
|
||||||
existing.file_size = stat.st_size
|
existing.file_size = stat.st_size
|
||||||
@@ -129,8 +131,8 @@ async def ingest_file(session: AsyncSession, path: Path, config: EbookSearchConf
|
|||||||
await session.delete(existing)
|
await session.delete(existing)
|
||||||
await session.flush()
|
await session.flush()
|
||||||
|
|
||||||
stat = resolved_path.stat()
|
stat = await resolved_path.stat()
|
||||||
parsed = await asyncio.to_thread(parse_epub, resolved_path)
|
parsed = await asyncio.to_thread(parse_epub, Path(resolved_path))
|
||||||
source = EbookSource(
|
source = EbookSource(
|
||||||
title=parsed.title,
|
title=parsed.title,
|
||||||
author=parsed.author,
|
author=parsed.author,
|
||||||
@@ -170,7 +172,7 @@ async def ingest_file(session: AsyncSession, path: Path, config: EbookSearchConf
|
|||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
async def find_existing_source(session: AsyncSession, path: Path, file_hash: str) -> EbookSource | None:
|
async def find_existing_source(session: AsyncSession, path: Path | AsyncPath, file_hash: str) -> EbookSource | None:
|
||||||
"""Find an existing source by canonical path or file hash."""
|
"""Find an existing source by canonical path or file hash."""
|
||||||
return await session.scalar(
|
return await session.scalar(
|
||||||
select(EbookSource).where(or_(EbookSource.file_path == str(path), EbookSource.file_sha256 == file_hash))
|
select(EbookSource).where(or_(EbookSource.file_path == str(path), EbookSource.file_sha256 == file_hash))
|
||||||
@@ -209,10 +211,10 @@ def add_chapter_chunks(
|
|||||||
return chunk_index
|
return chunk_index
|
||||||
|
|
||||||
|
|
||||||
def sha256_file(path: Path) -> str:
|
async def sha256_file(path: AsyncPath) -> str:
|
||||||
"""Calculate the SHA-256 digest for a file."""
|
"""Calculate the SHA-256 digest for a file without blocking the event loop."""
|
||||||
digest = hashlib.sha256()
|
digest = hashlib.sha256()
|
||||||
with path.open("rb") as file:
|
async with await path.open("rb") as file:
|
||||||
for block in iter(lambda: file.read(1024 * 1024), b""):
|
while block := await file.read(1024 * 1024):
|
||||||
digest.update(block)
|
digest.update(block)
|
||||||
return digest.hexdigest()
|
return digest.hexdigest()
|
||||||
|
|||||||
@@ -27,7 +27,13 @@ from python.ebook_search.bm25_corpus import (
|
|||||||
)
|
)
|
||||||
from python.ebook_search.config import EbookSearchConfig, RerankConfig, load_config
|
from python.ebook_search.config import EbookSearchConfig, RerankConfig, load_config
|
||||||
from python.ebook_search.embeddings import MODEL_DIMENSIONS, ensure_embedding_models
|
from python.ebook_search.embeddings import MODEL_DIMENSIONS, ensure_embedding_models
|
||||||
from python.ebook_search.ingest import chunk_text, find_existing_source
|
from python.ebook_search.ingest import (
|
||||||
|
chunk_text,
|
||||||
|
find_existing_source,
|
||||||
|
find_library_epubs,
|
||||||
|
resolve_ingest_path,
|
||||||
|
sha256_file,
|
||||||
|
)
|
||||||
from python.ebook_search.search import (
|
from python.ebook_search.search import (
|
||||||
SearchResponse,
|
SearchResponse,
|
||||||
SearchResult,
|
SearchResult,
|
||||||
@@ -106,6 +112,33 @@ async def test_find_existing_source_matches_path_or_hash() -> None:
|
|||||||
assert await find_existing_source(session, Path("/new/book.epub"), "a" * 64) == source
|
assert await find_existing_source(session, Path("/new/book.epub"), "a" * 64) == source
|
||||||
|
|
||||||
|
|
||||||
|
async def test_async_ingest_path_resolution_and_hashing(tmp_path: Path) -> None:
|
||||||
|
"""Ingest file metadata uses AnyIO's asynchronous path operations."""
|
||||||
|
path = tmp_path / "book.epub"
|
||||||
|
path.write_bytes(b"epub content")
|
||||||
|
|
||||||
|
resolved_path = await resolve_ingest_path(path)
|
||||||
|
|
||||||
|
assert resolved_path == await resolved_path.resolve()
|
||||||
|
assert await sha256_file(resolved_path) == "4b71fc6fc452ae1ff7832bfe374ae56ab4d0649acf6af8ab4d068d858ea60448"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_find_library_epubs_uses_async_path_walk(tmp_path: Path) -> None:
|
||||||
|
"""Library discovery finds EPUB files through AnyIO's async path API."""
|
||||||
|
(tmp_path / "nested").mkdir()
|
||||||
|
first_epub = tmp_path / "first.epub"
|
||||||
|
second_epub = tmp_path / "nested" / "second.epub"
|
||||||
|
first_epub.write_bytes(b"first")
|
||||||
|
second_epub.write_bytes(b"second")
|
||||||
|
(tmp_path / "not-a-book.txt").write_text("ignore", encoding="utf-8")
|
||||||
|
|
||||||
|
path, epub_paths = await find_library_epubs(str(tmp_path))
|
||||||
|
|
||||||
|
assert str(path) == str(tmp_path)
|
||||||
|
assert epub_paths is not None
|
||||||
|
assert [str(epub_path) for epub_path in epub_paths] == [str(first_epub), str(second_epub)]
|
||||||
|
|
||||||
|
|
||||||
async def test_bm25_corpus_uses_existing_search_text_without_duplicate_metadata() -> None:
|
async def test_bm25_corpus_uses_existing_search_text_without_duplicate_metadata() -> None:
|
||||||
engine = await build_async_engine()
|
engine = await build_async_engine()
|
||||||
async with AsyncSession(engine, expire_on_commit=False) as session:
|
async with AsyncSession(engine, expire_on_commit=False) as session:
|
||||||
|
|||||||
Reference in New Issue
Block a user