Kai
35af26c794
The curator app (Python backend, tests, systemd units, config, scripts) now lives in this repo under scenarios/curator/backend, exported from the standalone checkout's tracked tree (.pi mirror, venv and caches excluded). 149 unit tests pass from the new location; _SHARED_LIB and eval GOLDEN_DIR resolve unchanged. History not preserved per decision. verify-no-secrets: the ASSIGN heuristic now requires the value to carry entropy (a digit or uppercase letter), so vendored Python kwargs like token=extraction_token no longer false-positive while real base64/hex/random secrets still trip it.
105 lines
4.4 KiB
Python
105 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import unicodedata
|
|
from typing import Any
|
|
|
|
from .config import Settings
|
|
from .db import Database
|
|
from .book_pages import BookPageProvider
|
|
|
|
|
|
class BookReviewProvider:
|
|
"""Fetch attributed book metadata and ratings without making them catalog facts."""
|
|
|
|
def __init__(self, settings: Settings, database: Database):
|
|
self.settings = settings
|
|
self.database = database
|
|
|
|
@staticmethod
|
|
def _cache_key(plan: dict[str, Any]) -> str:
|
|
value = json.dumps(
|
|
{
|
|
"title": plan.get("title") or "",
|
|
"creator": plan.get("creator") or plan.get("author") or "",
|
|
"isbn": plan.get("isbn") or "",
|
|
"aliases": plan.get("aliases") or [],
|
|
},
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
)
|
|
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
|
|
|
def _public_pages(self, title: str, author: str, isbn: str) -> list[dict[str, Any]]:
|
|
return BookPageProvider().search(title, author, isbn, 8)
|
|
|
|
@staticmethod
|
|
def _text_key(value: str) -> str:
|
|
folded = unicodedata.normalize("NFKC", value).casefold()
|
|
return re.sub(r"[^\w]+", "", folded, flags=re.UNICODE)
|
|
|
|
@classmethod
|
|
def _matched_results(
|
|
cls,
|
|
results: list[dict[str, Any]],
|
|
title: str,
|
|
aliases: list[str],
|
|
author: str,
|
|
isbn: str = "",
|
|
) -> list[dict[str, Any]]:
|
|
title_keys = {cls._text_key(value) for value in (title, *aliases) if cls._text_key(value)}
|
|
author_key = cls._text_key(author)
|
|
isbn_key = re.sub(r"[^0-9X]", "", isbn.upper())
|
|
matched: list[dict[str, Any]] = []
|
|
for raw in results:
|
|
item = dict(raw)
|
|
result_key = cls._text_key(str(item.get("title") or ""))
|
|
if not result_key:
|
|
continue
|
|
result_isbns = {
|
|
re.sub(r"[^0-9X]", "", str(value).upper())
|
|
for value in item.get("isbns") or []
|
|
}
|
|
isbn_match = bool(isbn_key and isbn_key in result_isbns)
|
|
exact = result_key in title_keys
|
|
partial = any(len(key) >= 6 and (key in result_key or result_key in key) for key in title_keys)
|
|
if not isbn_match and not exact and not partial:
|
|
continue
|
|
author_keys = [cls._text_key(str(value)) for value in item.get("authors") or []]
|
|
author_match = bool(author_key and any(author_key in value or value in author_key for value in author_keys if value))
|
|
item["match"] = {
|
|
"title": "exact" if exact else "partial" if partial else "isbn",
|
|
"author": author_match,
|
|
"isbn": isbn_match,
|
|
"confidence": "high" if isbn_match or (exact and (author_match or not author_key)) else "medium",
|
|
}
|
|
matched.append(item)
|
|
return matched[:5]
|
|
|
|
def lookup(self, plan: dict[str, Any]) -> dict[str, Any]:
|
|
title = str(plan.get("title") or "").strip()
|
|
author = str(plan.get("creator") or plan.get("author") or "").strip()
|
|
isbn = str(plan.get("isbn") or "").replace("-", "").strip()
|
|
aliases = [str(value).strip() for value in plan.get("aliases") or [] if str(value).strip()]
|
|
result: dict[str, Any] = {"results": [], "errors": [], "providers_checked": []}
|
|
if not title and not isbn:
|
|
result["errors"].append("book-reviews: 缺少书名或 ISBN")
|
|
return result
|
|
cache_key = self._cache_key(plan)
|
|
cached = None if plan.get("refresh") else self.database.cache_get("book-reviews", cache_key)
|
|
if cached is not None:
|
|
cached["cache"] = {"hit": True, "ttl_seconds": self.settings.review_cache_ttl_seconds}
|
|
return cached
|
|
for name, loader in (("douban-goodreads-pages", self._public_pages),):
|
|
result["providers_checked"].append(name)
|
|
try:
|
|
result["results"].extend(self._matched_results(loader(title, author, isbn), title, aliases, author, isbn))
|
|
except Exception as exc:
|
|
result["errors"].append(f"{name}: {exc}")
|
|
if result["results"]:
|
|
self.database.cache_put("book-reviews", cache_key, result, self.settings.review_cache_ttl_seconds)
|
|
result["cache"] = {"hit": False, "ttl_seconds": self.settings.review_cache_ttl_seconds}
|
|
return result
|