Archived
feat: uv config other feats
- add uv configuration for the backend - update frontend to make auth work - add new auth endpoints - add bookmars feat - add reader feat
This commit is contained in:
@@ -0,0 +1,78 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from django.core.files.base import ContentFile
|
||||
|
||||
from apps.books.models import EBook
|
||||
from apps.books.services.openlibrary import (
|
||||
HIGH_CONFIDENCE,
|
||||
download_cover,
|
||||
fetch_metadata,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def enrich_ebook_metadata(ebook: EBook) -> EBook:
|
||||
"""Fetch Open Library metadata and update the ebook. Never raises to callers."""
|
||||
user_title = ebook.title
|
||||
user_author = ebook.author or ""
|
||||
|
||||
try:
|
||||
result = fetch_metadata(user_title, user_author)
|
||||
except Exception:
|
||||
logger.exception("Open Library metadata fetch failed for ebook %s", ebook.pk)
|
||||
return ebook
|
||||
|
||||
if result is None:
|
||||
return ebook
|
||||
|
||||
now = datetime.now(timezone.utc).isoformat()
|
||||
metadata = {
|
||||
"source": "openlibrary",
|
||||
"matched_at": now,
|
||||
"match_language": result.match_language,
|
||||
"match_score": result.match_score,
|
||||
"match_status": result.match_status,
|
||||
"user_input": {"title": user_title, "author": user_author},
|
||||
"openlibrary": result.openlibrary,
|
||||
}
|
||||
ebook.metadata_json = metadata
|
||||
|
||||
update_fields = ["metadata_json", "updated_at"]
|
||||
|
||||
if result.match_status == "not_found":
|
||||
ebook.save(update_fields=update_fields)
|
||||
return ebook
|
||||
|
||||
ol = result.openlibrary
|
||||
cover_id = ol.get("cover_id")
|
||||
|
||||
cover_missing = (
|
||||
not ebook.cover_image
|
||||
or not ebook.cover_image.name
|
||||
or not ebook.cover_image.storage.exists(ebook.cover_image.name)
|
||||
)
|
||||
if cover_id and cover_missing:
|
||||
if ebook.cover_image:
|
||||
ebook.cover_image.delete(save=False)
|
||||
cover_bytes = download_cover(int(cover_id))
|
||||
if cover_bytes:
|
||||
filename = f"ol_cover_{ebook.pk}_{cover_id}.jpg"
|
||||
ebook.cover_image.save(filename, ContentFile(cover_bytes), save=False)
|
||||
update_fields.append("cover_image")
|
||||
|
||||
if result.match_score >= HIGH_CONFIDENCE:
|
||||
ol_title = ol.get("title")
|
||||
ol_authors = ol.get("authors") or []
|
||||
if ol_title:
|
||||
ebook.title = ol_title[:512]
|
||||
update_fields.append("title")
|
||||
if ol_authors:
|
||||
ebook.author = ol_authors[0][:256]
|
||||
update_fields.append("author")
|
||||
|
||||
ebook.save(update_fields=list(dict.fromkeys(update_fields)))
|
||||
return ebook
|
||||
@@ -0,0 +1,304 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
import unicodedata
|
||||
from dataclasses import dataclass
|
||||
from difflib import SequenceMatcher
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
from config.settings import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SEARCH_URL = "https://openlibrary.org/search.json"
|
||||
COVERS_URL = "https://covers.openlibrary.org/b/id/{cover_id}-{size}.jpg"
|
||||
SEARCH_FIELDS = (
|
||||
"key,title,author_name,cover_i,first_publish_year,subject,language,"
|
||||
"edition_key,number_of_pages_median,publisher"
|
||||
)
|
||||
|
||||
HIGH_CONFIDENCE = 0.8
|
||||
LOW_CONFIDENCE = 0.6
|
||||
|
||||
# EPUB release noise often copied from filenames, e.g. "Title [6494] (r2.3)"
|
||||
_TITLE_NOISE_PATTERNS = (
|
||||
re.compile(r"\s*\[\d+\]"), # [6494]
|
||||
re.compile(r"\s*\([rv][\d.]+\)", re.IGNORECASE), # (r2.3), (v1.0)
|
||||
re.compile(r"\s*\(rev[\d.]*\)", re.IGNORECASE), # (rev2)
|
||||
)
|
||||
|
||||
|
||||
def _sanitize_search_title(title: str) -> str:
|
||||
cleaned = title.strip()
|
||||
for pattern in _TITLE_NOISE_PATTERNS:
|
||||
cleaned = pattern.sub("", cleaned)
|
||||
return re.sub(r"\s+", " ", cleaned).strip()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class OpenLibraryHit:
|
||||
work_key: str
|
||||
title: str
|
||||
authors: list[str]
|
||||
cover_id: int | None
|
||||
first_publish_year: int | None
|
||||
subjects: list[str]
|
||||
languages: list[str]
|
||||
publishers: list[str]
|
||||
edition_key: str | None
|
||||
number_of_pages_median: int | None
|
||||
match_language: str
|
||||
score: float
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class OpenLibraryMetadata:
|
||||
match_language: str
|
||||
match_score: float
|
||||
match_status: str
|
||||
openlibrary: dict[str, Any]
|
||||
|
||||
|
||||
def _normalize(text: str) -> str:
|
||||
cleaned = re.sub(r"[^\w\s]", " ", _strip_accents(text).lower())
|
||||
return re.sub(r"\s+", " ", cleaned).strip()
|
||||
|
||||
|
||||
def _strip_accents(text: str) -> str:
|
||||
normalized = unicodedata.normalize("NFKD", text)
|
||||
return "".join(ch for ch in normalized if not unicodedata.combining(ch))
|
||||
|
||||
|
||||
def _build_search_params(
|
||||
title: str,
|
||||
author: str,
|
||||
*,
|
||||
lang: str | None,
|
||||
search_mode: str,
|
||||
) -> dict[str, str | int]:
|
||||
title = title.strip()
|
||||
author = author.strip()
|
||||
params: dict[str, str | int] = {"limit": 10, "fields": SEARCH_FIELDS}
|
||||
|
||||
if search_mode == "spanish_q":
|
||||
q_parts = ["language:spa", title]
|
||||
if author:
|
||||
q_parts.append(author)
|
||||
params["q"] = " ".join(q_parts)
|
||||
elif search_mode == "q":
|
||||
params["q"] = " ".join(part for part in (title, author) if part)
|
||||
elif search_mode == "q_unaccent":
|
||||
params["q"] = " ".join(
|
||||
part for part in (_strip_accents(title), _strip_accents(author) if author else "") if part
|
||||
)
|
||||
else:
|
||||
params["title"] = title
|
||||
if author:
|
||||
params["author"] = author
|
||||
|
||||
if lang:
|
||||
params["lang"] = lang
|
||||
return params
|
||||
|
||||
|
||||
def _title_similarity(a: str, b: str) -> float:
|
||||
na, nb = _normalize(a), _normalize(b)
|
||||
if not na or not nb:
|
||||
return 0.0
|
||||
if na in nb or nb in na:
|
||||
return 1.0
|
||||
return SequenceMatcher(None, na, nb).ratio()
|
||||
|
||||
|
||||
def _author_overlap(user_author: str, ol_authors: list[str]) -> float:
|
||||
if not user_author.strip():
|
||||
return 0.5 if ol_authors else 0.0
|
||||
user_tokens = set(_normalize(user_author).split())
|
||||
if not user_tokens:
|
||||
return 0.0
|
||||
best = 0.0
|
||||
for name in ol_authors:
|
||||
name_tokens = set(_normalize(name).split())
|
||||
if not name_tokens:
|
||||
continue
|
||||
overlap = len(user_tokens & name_tokens) / len(user_tokens)
|
||||
best = max(best, overlap)
|
||||
if user_tokens <= name_tokens or name_tokens <= user_tokens:
|
||||
best = max(best, 0.95)
|
||||
return best
|
||||
|
||||
|
||||
def _score_hit(title: str, author: str, doc: dict[str, Any], *, lang: str) -> float:
|
||||
ol_title = doc.get("title") or ""
|
||||
ol_authors = doc.get("author_name") or []
|
||||
title_score = _title_similarity(title, ol_title)
|
||||
author_score = _author_overlap(author, ol_authors)
|
||||
combined = (title_score * 0.6) + (author_score * 0.4)
|
||||
if doc.get("cover_i"):
|
||||
combined += 0.05
|
||||
return min(combined, 1.0)
|
||||
|
||||
|
||||
def _parse_hit(doc: dict[str, Any], *, lang: str, score: float) -> OpenLibraryHit:
|
||||
edition_keys = doc.get("edition_key") or []
|
||||
edition_key = edition_keys[0] if edition_keys else None
|
||||
cover_id = doc.get("cover_i")
|
||||
return OpenLibraryHit(
|
||||
work_key=doc.get("key") or "",
|
||||
title=doc.get("title") or "",
|
||||
authors=list(doc.get("author_name") or []),
|
||||
cover_id=int(cover_id) if cover_id else None,
|
||||
first_publish_year=doc.get("first_publish_year"),
|
||||
subjects=list(doc.get("subject") or [])[:10],
|
||||
languages=list(doc.get("language") or []),
|
||||
publishers=list(doc.get("publisher") or [])[:5],
|
||||
edition_key=edition_key,
|
||||
number_of_pages_median=doc.get("number_of_pages_median"),
|
||||
match_language=lang,
|
||||
score=score,
|
||||
)
|
||||
|
||||
|
||||
def build_cover_url(cover_id: int, size: str = "L") -> str:
|
||||
return COVERS_URL.format(cover_id=cover_id, size=size)
|
||||
|
||||
|
||||
def _client() -> httpx.Client:
|
||||
read_timeout = settings.OPENLIBRARY_TIMEOUT_SECONDS
|
||||
connect_timeout = settings.OPENLIBRARY_CONNECT_TIMEOUT_SECONDS
|
||||
return httpx.Client(
|
||||
timeout=httpx.Timeout(
|
||||
connect=connect_timeout,
|
||||
read=read_timeout,
|
||||
write=read_timeout,
|
||||
pool=connect_timeout,
|
||||
),
|
||||
headers={"User-Agent": settings.OPENLIBRARY_USER_AGENT},
|
||||
follow_redirects=True,
|
||||
)
|
||||
|
||||
|
||||
def search_works(
|
||||
title: str,
|
||||
author: str,
|
||||
*,
|
||||
lang: str | None = None,
|
||||
search_mode: str = "title",
|
||||
client: httpx.Client | None = None,
|
||||
) -> list[OpenLibraryHit]:
|
||||
if not title.strip():
|
||||
return []
|
||||
|
||||
params = _build_search_params(title, author, lang=lang, search_mode=search_mode)
|
||||
|
||||
try:
|
||||
if client is not None:
|
||||
response = client.get(SEARCH_URL, params=params)
|
||||
response.raise_for_status()
|
||||
docs = response.json().get("docs") or []
|
||||
else:
|
||||
with _client() as owned_client:
|
||||
response = owned_client.get(SEARCH_URL, params=params)
|
||||
response.raise_for_status()
|
||||
docs = response.json().get("docs") or []
|
||||
except (httpx.HTTPError, ValueError) as exc:
|
||||
logger.warning("Open Library search failed: %s", exc)
|
||||
return []
|
||||
|
||||
hits: list[OpenLibraryHit] = []
|
||||
for doc in docs:
|
||||
score = _score_hit(title, author, doc, lang=lang or "")
|
||||
if score < LOW_CONFIDENCE:
|
||||
continue
|
||||
hits.append(_parse_hit(doc, lang=lang or "", score=score))
|
||||
hits.sort(key=lambda h: (h.score, h.cover_id is not None), reverse=True)
|
||||
return hits
|
||||
|
||||
|
||||
def pick_best_match(title: str, author: str, hits: list[OpenLibraryHit]) -> OpenLibraryHit | None:
|
||||
return hits[0] if hits else None
|
||||
|
||||
|
||||
def download_cover(cover_id: int) -> bytes | None:
|
||||
url = build_cover_url(cover_id, size="L")
|
||||
try:
|
||||
with _client() as client:
|
||||
response = client.get(url)
|
||||
if response.status_code == 404:
|
||||
return None
|
||||
response.raise_for_status()
|
||||
content_type = response.headers.get("content-type", "")
|
||||
if not content_type.startswith("image/"):
|
||||
return None
|
||||
return response.content
|
||||
except httpx.HTTPError as exc:
|
||||
logger.warning("Open Library cover download failed for %s: %s", cover_id, exc)
|
||||
return None
|
||||
|
||||
|
||||
def _hit_to_openlibrary_dict(hit: OpenLibraryHit) -> dict[str, Any]:
|
||||
return {
|
||||
"work_key": hit.work_key,
|
||||
"edition_key": hit.edition_key,
|
||||
"title": hit.title,
|
||||
"authors": hit.authors,
|
||||
"cover_id": hit.cover_id,
|
||||
"cover_url": build_cover_url(hit.cover_id) if hit.cover_id else None,
|
||||
"first_publish_year": hit.first_publish_year,
|
||||
"subjects": hit.subjects,
|
||||
"languages": hit.languages,
|
||||
"publishers": hit.publishers,
|
||||
"number_of_pages_median": hit.number_of_pages_median,
|
||||
}
|
||||
|
||||
|
||||
def fetch_metadata(title: str, author: str) -> OpenLibraryMetadata | None:
|
||||
if not settings.OPENLIBRARY_ENABLED:
|
||||
return None
|
||||
|
||||
preferred = settings.OPENLIBRARY_PREFERRED_LANG
|
||||
fallback = settings.OPENLIBRARY_FALLBACK_LANG
|
||||
search_title = _sanitize_search_title(title)
|
||||
|
||||
if not search_title:
|
||||
return OpenLibraryMetadata(
|
||||
match_language=preferred,
|
||||
match_score=0.0,
|
||||
match_status="not_found",
|
||||
openlibrary={},
|
||||
)
|
||||
|
||||
search_plan: list[tuple[str, str | None]] = [
|
||||
("spanish_q", preferred),
|
||||
("title", preferred),
|
||||
("q", None),
|
||||
("q_unaccent", None),
|
||||
]
|
||||
if fallback != preferred:
|
||||
search_plan.extend([("title", fallback), ("q", fallback)])
|
||||
|
||||
hits: list[OpenLibraryHit] = []
|
||||
with _client() as client:
|
||||
for search_mode, lang in search_plan:
|
||||
hits = search_works(search_title, author, lang=lang, search_mode=search_mode, client=client)
|
||||
if hits:
|
||||
break
|
||||
|
||||
hit = pick_best_match(title, author, hits)
|
||||
if not hit:
|
||||
return OpenLibraryMetadata(
|
||||
match_language=preferred,
|
||||
match_score=0.0,
|
||||
match_status="not_found",
|
||||
openlibrary={},
|
||||
)
|
||||
|
||||
status = "matched" if hit.score >= HIGH_CONFIDENCE else "partial"
|
||||
return OpenLibraryMetadata(
|
||||
match_language=hit.match_language,
|
||||
match_score=hit.score,
|
||||
match_status=status,
|
||||
openlibrary=_hit_to_openlibrary_dict(hit),
|
||||
)
|
||||
@@ -0,0 +1,177 @@
|
||||
"""Extract TOC, metadata, and page count from uploaded EPUB/PDF files."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from apps.books.models import BookChapter, EBook
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def process_ebook(file_path: str, original_filename: str | None = None) -> dict[str, Any]:
|
||||
"""Parse an e-book file and return format, metadata, TOC, and page count."""
|
||||
ext = Path(original_filename or file_path).suffix.lower()
|
||||
if ext == ".epub" or file_path.lower().endswith(".epub"):
|
||||
return _process_epub(file_path)
|
||||
if ext == ".pdf" or file_path.lower().endswith(".pdf"):
|
||||
return _process_pdf(file_path)
|
||||
return {"format": ext.lstrip(".") or "unknown", "page_count": 0, "metadata": {}, "toc": []}
|
||||
|
||||
|
||||
def store_chapters(ebook: EBook, toc: list[dict[str, Any]], parent_index: int = 0) -> None:
|
||||
"""Recursively store TOC entries as BookChapter records."""
|
||||
for idx, entry in enumerate(toc):
|
||||
BookChapter.objects.create(
|
||||
ebook=ebook,
|
||||
title=entry.get("title", "Untitled"),
|
||||
index=parent_index + idx,
|
||||
href=entry.get("href", ""),
|
||||
children=entry.get("children", []),
|
||||
)
|
||||
children = entry.get("children", [])
|
||||
if children:
|
||||
store_chapters(ebook, children, parent_index + idx + 1)
|
||||
|
||||
|
||||
def apply_processing_to_ebook(ebook: EBook) -> dict[str, Any]:
|
||||
"""Run processing on an EBook instance and persist chapters + metadata."""
|
||||
if not ebook.file:
|
||||
raise ValueError("No file found for this e-book.")
|
||||
|
||||
file_path = ebook.file.path
|
||||
result = process_ebook(file_path, original_filename=ebook.filename())
|
||||
|
||||
ebook.format = result.get("format", ebook.format)
|
||||
ebook.page_count = result.get("page_count", 0)
|
||||
file_metadata = result.get("metadata") or {}
|
||||
if file_metadata:
|
||||
merged = dict(ebook.metadata_json or {})
|
||||
merged["file"] = file_metadata
|
||||
ebook.metadata_json = merged
|
||||
ebook.save(update_fields=["format", "page_count", "metadata_json", "updated_at"])
|
||||
|
||||
raw_toc: list[dict[str, Any]] = result.get("toc", [])
|
||||
BookChapter.objects.filter(ebook=ebook).delete()
|
||||
store_chapters(ebook, raw_toc)
|
||||
|
||||
return {
|
||||
"format": ebook.format,
|
||||
"page_count": ebook.page_count,
|
||||
"metadata": ebook.metadata_json,
|
||||
"toc_count": len(raw_toc),
|
||||
"status": "processed",
|
||||
}
|
||||
|
||||
|
||||
def _process_epub(file_path: str) -> dict[str, Any]:
|
||||
from ebooklib import epub, ITEM_DOCUMENT
|
||||
|
||||
book = epub.read_epub(file_path)
|
||||
metadata = _extract_epub_metadata(book)
|
||||
toc = _extract_epub_toc(book)
|
||||
|
||||
if not toc:
|
||||
toc = _fallback_toc_from_spine(book, ITEM_DOCUMENT)
|
||||
|
||||
flat_count = _count_toc_entries(toc)
|
||||
page_count = flat_count or len(book.spine)
|
||||
|
||||
return {
|
||||
"format": "epub",
|
||||
"page_count": page_count,
|
||||
"metadata": metadata,
|
||||
"toc": toc,
|
||||
}
|
||||
|
||||
|
||||
def _process_pdf(file_path: str) -> dict[str, Any]:
|
||||
return {"format": "pdf", "page_count": 0, "metadata": {}, "toc": []}
|
||||
|
||||
|
||||
def _extract_epub_metadata(book) -> dict[str, Any]:
|
||||
metadata: dict[str, Any] = {}
|
||||
|
||||
def first(namespace: str, name: str) -> str:
|
||||
values = book.get_metadata(namespace, name)
|
||||
if values:
|
||||
return str(values[0][0])
|
||||
return ""
|
||||
|
||||
title = first("DC", "title")
|
||||
if title:
|
||||
metadata["title"] = title
|
||||
creator = first("DC", "creator")
|
||||
if creator:
|
||||
metadata["author"] = creator
|
||||
language = first("DC", "language")
|
||||
if language:
|
||||
metadata["language"] = language
|
||||
identifier = first("DC", "identifier")
|
||||
if identifier:
|
||||
metadata["identifier"] = identifier
|
||||
return metadata
|
||||
|
||||
|
||||
def _parse_toc_item(item) -> dict[str, Any]:
|
||||
from ebooklib import epub
|
||||
|
||||
if isinstance(item, epub.Link):
|
||||
return {
|
||||
"title": item.title or "Untitled",
|
||||
"href": item.href or "",
|
||||
"children": [],
|
||||
}
|
||||
if isinstance(item, tuple):
|
||||
section, children = item
|
||||
entry = {
|
||||
"title": getattr(section, "title", None) or "Untitled",
|
||||
"href": getattr(section, "href", None) or "",
|
||||
"children": [],
|
||||
}
|
||||
for child in children:
|
||||
entry["children"].append(_parse_toc_item(child))
|
||||
return entry
|
||||
if hasattr(item, "title"):
|
||||
return {
|
||||
"title": item.title or "Untitled",
|
||||
"href": getattr(item, "href", "") or "",
|
||||
"children": [],
|
||||
}
|
||||
return {"title": "Untitled", "href": "", "children": []}
|
||||
|
||||
|
||||
def _extract_epub_toc(book) -> list[dict[str, Any]]:
|
||||
return [_parse_toc_item(item) for item in book.toc]
|
||||
|
||||
|
||||
def _fallback_toc_from_spine(book, item_document_type) -> list[dict[str, Any]]:
|
||||
toc: list[dict[str, Any]] = []
|
||||
seen: set[str] = set()
|
||||
chapter_num = 0
|
||||
|
||||
for spine_entry in book.spine:
|
||||
item_id = spine_entry[0] if isinstance(spine_entry, tuple) else spine_entry
|
||||
item = book.get_item_with_id(item_id)
|
||||
if not item or item.get_type() != item_document_type:
|
||||
continue
|
||||
href = item.get_name() or ""
|
||||
if not href or href in seen:
|
||||
continue
|
||||
seen.add(href)
|
||||
chapter_num += 1
|
||||
title = os.path.splitext(os.path.basename(href))[0] or f"Chapter {chapter_num}"
|
||||
toc.append({"title": title.replace("_", " ").replace("-", " "), "href": href, "children": []})
|
||||
|
||||
return toc
|
||||
|
||||
|
||||
def _count_toc_entries(toc: list[dict[str, Any]]) -> int:
|
||||
count = 0
|
||||
for entry in toc:
|
||||
count += 1
|
||||
count += _count_toc_entries(entry.get("children", []))
|
||||
return count
|
||||
Reference in New Issue
Block a user