Kai
88b06d782f
The curator Python backend (package, tests, systemd units, config templates, scripts) now lives under scenarios/curator/backend and is the single source of truth; the live checkout at the workspace path is a runtime copy. Exported from the app repo's tracked tree via git archive (no history, .pi/venv/caches excluded). 149 unit tests pass from the new location. profile.toml backend is now repo-relative (scenarios/curator/backend); verify-generated.sh resolves a relative backend against REPO_ROOT. verify-no-secrets ASSIGN heuristic now requires value entropy so vendored kwargs like token=extraction_token no longer false-positive. README documents the backend/ layout and the operator-owned app rollout step.
199 lines
8.4 KiB
Python
199 lines
8.4 KiB
Python
from __future__ import annotations
|
|
|
|
import mimetypes
|
|
import posixpath
|
|
import re
|
|
import zipfile
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from xml.etree import ElementTree as ET
|
|
|
|
|
|
CONTAINER = "META-INF/container.xml"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EpubInfo:
|
|
title: str
|
|
author: str
|
|
language: str
|
|
identifier: str
|
|
publisher: str
|
|
package_path: str
|
|
spine: tuple[str, ...]
|
|
display_title: str = ""
|
|
title_aliases: tuple[str, ...] = ()
|
|
declared_language: str = "und"
|
|
identifiers: tuple[str, ...] = ()
|
|
source_identifiers: tuple[str, ...] = ()
|
|
isbn: str = ""
|
|
cover_path: str = ""
|
|
|
|
|
|
def _text(root: ET.Element, suffix: str) -> str:
|
|
for element in root.iter():
|
|
if element.tag.rsplit("}", 1)[-1] == suffix and element.text:
|
|
return element.text.strip()
|
|
return ""
|
|
|
|
|
|
def _values(root: ET.Element, suffix: str) -> list[ET.Element]:
|
|
return [element for element in root.iter() if element.tag.rsplit("}", 1)[-1] == suffix]
|
|
|
|
|
|
def _clean_isbn(value: str) -> str:
|
|
if value.casefold().startswith("urn:uuid:"):
|
|
return ""
|
|
candidate = re.sub(r"[^0-9Xx]", "", value.removeprefix("urn:isbn:"))
|
|
candidate = candidate.upper()
|
|
if len(candidate) == 13 and candidate.isdigit():
|
|
checksum = sum((1 if index % 2 == 0 else 3) * int(digit) for index, digit in enumerate(candidate))
|
|
return candidate if checksum % 10 == 0 else ""
|
|
if len(candidate) == 10 and candidate[:9].isdigit() and (candidate[9].isdigit() or candidate[9] == "X"):
|
|
checksum = sum((10 - index) * (10 if digit == "X" else int(digit)) for index, digit in enumerate(candidate))
|
|
return candidate if checksum % 11 == 0 else ""
|
|
return ""
|
|
|
|
|
|
def _visible_language(archive: zipfile.ZipFile, spine: tuple[str, ...], declared: str) -> str:
|
|
fragments: list[str] = []
|
|
for member in spine[:20]:
|
|
try:
|
|
root = ET.fromstring(archive.read(member))
|
|
except (KeyError, ET.ParseError):
|
|
continue
|
|
for element in root.iter():
|
|
if element.tag.rsplit("}", 1)[-1] not in {"style", "script"} and element.text:
|
|
fragments.append(element.text)
|
|
if sum(map(len, fragments)) >= 200_000:
|
|
break
|
|
text = " ".join(fragments)
|
|
cjk = len(re.findall(r"[\u3400-\u9fff]", text))
|
|
latin = len(re.findall(r"[A-Za-z]", text))
|
|
if cjk >= 200 and cjk / max(1, cjk + latin) >= 0.65:
|
|
return "zh-Hans"
|
|
parts = (declared or "und").replace("_", "-").split("-")
|
|
if parts[0].casefold() == "en":
|
|
return "-".join(["en", *[part.upper() if len(part) == 2 else part.title() for part in parts[1:]]])
|
|
if parts[0].casefold() == "zh":
|
|
return "-".join(["zh", *[part.title() for part in parts[1:]]])
|
|
return declared or "und"
|
|
|
|
|
|
def inspect_epub(path: Path) -> EpubInfo:
|
|
if not zipfile.is_zipfile(path):
|
|
raise ValueError("EPUB is not a ZIP container")
|
|
with zipfile.ZipFile(path) as archive:
|
|
try:
|
|
container = ET.fromstring(archive.read(CONTAINER))
|
|
except (KeyError, ET.ParseError) as exc:
|
|
raise ValueError("EPUB container.xml is missing or invalid") from exc
|
|
package_path = ""
|
|
for element in container.iter():
|
|
if element.tag.rsplit("}", 1)[-1] == "rootfile":
|
|
package_path = element.attrib.get("full-path", "")
|
|
break
|
|
if not package_path:
|
|
raise ValueError("EPUB package document is missing")
|
|
try:
|
|
package = ET.fromstring(archive.read(package_path))
|
|
except (KeyError, ET.ParseError) as exc:
|
|
raise ValueError("EPUB package document is invalid") from exc
|
|
|
|
manifest: dict[str, str] = {}
|
|
manifest_properties: dict[str, str] = {}
|
|
spine_ids: list[str] = []
|
|
for element in package.iter():
|
|
name = element.tag.rsplit("}", 1)[-1]
|
|
if name == "item":
|
|
item_id = element.attrib.get("id", "")
|
|
href = element.attrib.get("href", "")
|
|
if item_id and href:
|
|
manifest[item_id] = href
|
|
manifest_properties[item_id] = element.attrib.get("properties", "")
|
|
elif name == "itemref":
|
|
item_id = element.attrib.get("idref", "")
|
|
if item_id:
|
|
spine_ids.append(item_id)
|
|
|
|
package_dir = posixpath.dirname(package_path)
|
|
spine = tuple(
|
|
posixpath.normpath(posixpath.join(package_dir, manifest[item_id]))
|
|
for item_id in spine_ids
|
|
if item_id in manifest
|
|
)
|
|
if not spine:
|
|
raise ValueError("EPUB contains no readable spine")
|
|
legacy_cover_id = ""
|
|
for element in _values(package, "meta"):
|
|
if element.attrib.get("name") == "cover":
|
|
legacy_cover_id = element.attrib.get("content", "")
|
|
break
|
|
cover_id = next((item_id for item_id, properties in manifest_properties.items() if "cover-image" in properties.split()), "")
|
|
cover_id = cover_id or legacy_cover_id
|
|
cover_path = posixpath.normpath(posixpath.join(package_dir, manifest[cover_id])) if cover_id in manifest else ""
|
|
title_elements = _values(package, "title")
|
|
titles = [(element.attrib.get("id", ""), (element.text or "").strip()) for element in title_elements]
|
|
titles = [(element_id, value) for element_id, value in titles if value]
|
|
title_types: dict[str, str] = {}
|
|
file_as: dict[str, str] = {}
|
|
for element in _values(package, "meta"):
|
|
target = element.attrib.get("refines", "").removeprefix("#")
|
|
prop = element.attrib.get("property", "")
|
|
value = (element.text or "").strip()
|
|
if target and prop == "title-type":
|
|
title_types[target] = value
|
|
elif target and prop == "file-as":
|
|
file_as[target] = value
|
|
extended = next((value for element_id, value in titles if title_types.get(element_id) == "extended"), "")
|
|
display_title = next((value for element_id, value in titles if title_types.get(element_id) == "main"), "")
|
|
display_title = display_title or (titles[0][1] if titles else "")
|
|
title = extended or display_title
|
|
|
|
creator_elements = _values(package, "creator")
|
|
author = next(((element.text or "").strip() for element in creator_elements if (element.text or "").strip()), "")
|
|
if not author:
|
|
creator_ids = [element.attrib.get("id", "") for element in creator_elements]
|
|
author = next((file_as.get(element_id, "") for element_id in creator_ids if file_as.get(element_id)), "")
|
|
if "," in author:
|
|
family, given = [part.strip() for part in author.split(",", 1)]
|
|
author = f"{given} {family}".strip()
|
|
|
|
identifiers = tuple((element.text or "").strip() for element in _values(package, "identifier") if (element.text or "").strip())
|
|
source_identifiers = tuple((element.text or "").strip() for element in _values(package, "source") if (element.text or "").strip())
|
|
unique_id = package.attrib.get("unique-identifier", "")
|
|
identifier = next(
|
|
((element.text or "").strip() for element in _values(package, "identifier") if element.attrib.get("id") == unique_id),
|
|
identifiers[0] if identifiers else "",
|
|
)
|
|
isbn = _clean_isbn(identifier)
|
|
declared_language = _text(package, "language") or "und"
|
|
language = _visible_language(archive, spine, declared_language)
|
|
aliases = tuple(dict.fromkeys(value for _, value in titles if value != title))
|
|
return EpubInfo(
|
|
title=title,
|
|
author=author,
|
|
language=language,
|
|
identifier=identifier,
|
|
publisher=_text(package, "publisher"),
|
|
package_path=package_path,
|
|
spine=spine,
|
|
display_title=display_title,
|
|
title_aliases=aliases,
|
|
declared_language=declared_language,
|
|
identifiers=identifiers,
|
|
source_identifiers=source_identifiers,
|
|
isbn=isbn,
|
|
cover_path=cover_path,
|
|
)
|
|
|
|
|
|
def read_member(path: Path, member: str) -> tuple[bytes, str]:
|
|
member = posixpath.normpath(member.lstrip("/"))
|
|
if member == ".." or member.startswith("../"):
|
|
raise ValueError("invalid EPUB member path")
|
|
with zipfile.ZipFile(path) as archive:
|
|
data = archive.read(member)
|
|
mime = mimetypes.guess_type(member)[0] or "application/octet-stream"
|
|
return data, mime
|