342 lines
12 KiB
Python
342 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import errno
|
|
import hashlib
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
import threading
|
|
from dataclasses import dataclass
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from typing import AsyncIterable, Iterable
|
|
from uuid import uuid4
|
|
|
|
from fontTools.ttLib import TTCollection, TTFont, TTLibError
|
|
|
|
from app.display.text_renderer import DEFAULT_FONT_CANDIDATES, FontFace
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ACCEPTED_FONT_EXTENSIONS = (".ttf", ".otf", ".ttc", ".otc")
|
|
MAX_FONT_UPLOAD_BYTES = 32 * 1024 * 1024
|
|
_FONTCONFIG_TIMEOUT_SECONDS = 8
|
|
_USER_FONT_NAME = "{digest}.font"
|
|
|
|
|
|
class FontCatalogError(RuntimeError):
|
|
"""Base error for font catalog operations."""
|
|
|
|
|
|
class FontTooLargeError(FontCatalogError):
|
|
pass
|
|
|
|
|
|
class FontTypeError(FontCatalogError):
|
|
pass
|
|
|
|
|
|
class FontValidationError(FontCatalogError):
|
|
pass
|
|
|
|
|
|
class FontStorageFullError(FontCatalogError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _DiscoveredFace:
|
|
face: FontFace
|
|
family: str
|
|
style: str
|
|
|
|
|
|
def _friendly_name(font: TTFont, fallback: str) -> tuple[str, str]:
|
|
names = font.get("name")
|
|
if names is None:
|
|
return fallback, "Regular"
|
|
family = names.getBestFamilyName() or names.getDebugName(1) or fallback
|
|
style = names.getBestSubFamilyName() or names.getDebugName(2) or "Regular"
|
|
return str(family).strip() or fallback, str(style).strip() or "Regular"
|
|
|
|
|
|
def _has_unicode_cmap(font: TTFont) -> bool:
|
|
cmap = font.get("cmap")
|
|
return bool(cmap and any(table.isUnicode() and table.cmap for table in cmap.tables))
|
|
|
|
|
|
def _inspect_font_file(path: Path, *, require_unicode: bool) -> tuple[_DiscoveredFace, ...]:
|
|
path = Path(path)
|
|
fallback = path.stem or "Unnamed font"
|
|
collection: TTCollection | None = None
|
|
fonts: list[TTFont] = []
|
|
try:
|
|
content = path.read_bytes()
|
|
try:
|
|
collection = TTCollection(BytesIO(content), lazy=False)
|
|
fonts = list(collection.fonts)
|
|
except TTLibError:
|
|
fonts = [TTFont(BytesIO(content), lazy=False)]
|
|
discovered: list[_DiscoveredFace] = []
|
|
for index, font in enumerate(fonts):
|
|
if require_unicode and not _has_unicode_cmap(font):
|
|
continue
|
|
family, style = _friendly_name(font, fallback)
|
|
discovered.append(
|
|
_DiscoveredFace(FontFace(str(path.resolve()), index), family, style)
|
|
)
|
|
if not discovered:
|
|
raise FontValidationError("font file does not contain a usable Unicode font face")
|
|
return tuple(discovered)
|
|
except FontValidationError:
|
|
raise
|
|
except (OSError, TTLibError, KeyError, ValueError) as exc:
|
|
raise FontValidationError("font file is invalid or unsupported") from exc
|
|
finally:
|
|
if collection is not None:
|
|
collection.close()
|
|
else:
|
|
for font in fonts:
|
|
font.close()
|
|
|
|
|
|
def _system_font_id(face: FontFace) -> str:
|
|
raw = f"{Path(face.path).resolve()}\0{face.index}".encode("utf-8")
|
|
return f"system:{hashlib.sha256(raw).hexdigest()}"
|
|
|
|
|
|
def _user_font_id(digest: str, index: int) -> str:
|
|
return f"user:{digest}:{index}"
|
|
|
|
|
|
def _entry(identifier: str, family: str, style: str, source: str) -> dict[str, str]:
|
|
family = family.strip() or "未命名字体"
|
|
style = style.strip() or "Regular"
|
|
label = family if style.casefold() in {"regular", "normal", "book"} else f"{family} — {style}"
|
|
return {
|
|
"id": identifier,
|
|
"label": label,
|
|
"family": family,
|
|
"style": style,
|
|
"source": source,
|
|
}
|
|
|
|
|
|
def _fontconfig_faces() -> tuple[_DiscoveredFace, ...]:
|
|
try:
|
|
result = subprocess.run(
|
|
["fc-list", "--format=%{file}\t%{index}\t%{family[0]}\t%{style[0]}\n"],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
timeout=_FONTCONFIG_TIMEOUT_SECONDS,
|
|
)
|
|
except (FileNotFoundError, OSError, subprocess.SubprocessError):
|
|
return ()
|
|
|
|
discovered: list[_DiscoveredFace] = []
|
|
for line in result.stdout.splitlines():
|
|
fields = line.split("\t", 3)
|
|
if len(fields) != 4:
|
|
continue
|
|
raw_path, raw_index, family, style = fields
|
|
path = Path(raw_path.strip())
|
|
if path.suffix.casefold() not in ACCEPTED_FONT_EXTENSIONS or not path.is_file():
|
|
continue
|
|
try:
|
|
index = int(raw_index.strip() or "0")
|
|
except ValueError:
|
|
continue
|
|
discovered.append(
|
|
_DiscoveredFace(
|
|
FontFace(str(path.resolve()), index),
|
|
family.strip() or path.stem,
|
|
style.strip() or "Regular",
|
|
)
|
|
)
|
|
return tuple(discovered)
|
|
|
|
|
|
def _configured_paths() -> Iterable[Path]:
|
|
raw = os.environ.get("MATRIX_FONT_PATHS", "")
|
|
for value in raw.split(os.pathsep):
|
|
if value.strip():
|
|
yield Path(value.strip()).expanduser()
|
|
|
|
|
|
def _sync_directory(path: Path) -> None:
|
|
if os.name == "nt":
|
|
return
|
|
flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
|
|
descriptor = os.open(path, flags)
|
|
try:
|
|
os.fsync(descriptor)
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
|
|
class FontCatalog:
|
|
def __init__(self, data_dir: Path) -> None:
|
|
self.root = Path(data_dir) / "fonts"
|
|
self._lock = threading.RLock()
|
|
self._items: tuple[dict[str, str], ...] = ()
|
|
self._faces: dict[str, tuple[FontFace, ...]] = {}
|
|
self._warnings: tuple[str, ...] = ()
|
|
self.root.mkdir(parents=True, exist_ok=True)
|
|
if os.name != "nt":
|
|
self.root.chmod(0o750)
|
|
self._cleanup_upload_temporaries()
|
|
self._system_catalog = self._system_faces()
|
|
self.refresh()
|
|
|
|
def _cleanup_upload_temporaries(self) -> None:
|
|
for path in self.root.glob(".upload-*.tmp"):
|
|
if path.is_file() and path.parent == self.root:
|
|
path.unlink(missing_ok=True)
|
|
|
|
def _system_faces(self) -> tuple[_DiscoveredFace, ...]:
|
|
discovered = list(_fontconfig_faces())
|
|
known = {(str(Path(item.face.path).resolve()), item.face.index) for item in discovered}
|
|
known_paths = {path for path, _index in known}
|
|
for candidate in (*_configured_paths(), *(Path(value) for value in DEFAULT_FONT_CANDIDATES)):
|
|
if candidate.suffix.casefold() not in ACCEPTED_FONT_EXTENSIONS or not candidate.is_file():
|
|
continue
|
|
resolved_candidate = str(candidate.resolve())
|
|
if resolved_candidate in known_paths:
|
|
continue
|
|
try:
|
|
inspected = _inspect_font_file(candidate, require_unicode=True)
|
|
except FontValidationError:
|
|
continue
|
|
for item in inspected:
|
|
key = (str(Path(item.face.path).resolve()), item.face.index)
|
|
if key not in known:
|
|
known.add(key)
|
|
known_paths.add(key[0])
|
|
discovered.append(item)
|
|
return tuple(discovered)
|
|
|
|
def refresh(self) -> None:
|
|
faces: dict[str, tuple[FontFace, ...]] = {}
|
|
system_items: list[dict[str, str]] = []
|
|
imported_items: list[dict[str, str]] = []
|
|
warnings: list[str] = []
|
|
|
|
seen_system: set[tuple[str, int]] = set()
|
|
for item in self._system_catalog:
|
|
key = (str(Path(item.face.path).resolve()), item.face.index)
|
|
if key in seen_system:
|
|
continue
|
|
seen_system.add(key)
|
|
identifier = _system_font_id(item.face)
|
|
faces[identifier] = (item.face,)
|
|
system_items.append(_entry(identifier, item.family, item.style, "system"))
|
|
|
|
for path in sorted(self.root.glob("*.font"), key=lambda item: item.name):
|
|
digest = path.stem.casefold()
|
|
if len(digest) != 64 or any(character not in "0123456789abcdef" for character in digest):
|
|
warnings.append(f"已忽略未登记字体文件：{path.name}")
|
|
continue
|
|
try:
|
|
inspected = _inspect_font_file(path, require_unicode=True)
|
|
except FontValidationError:
|
|
warnings.append(f"已保留但忽略损坏字体：{path.name}")
|
|
logger.warning("Ignoring invalid persisted font %s", path)
|
|
continue
|
|
for item in inspected:
|
|
identifier = _user_font_id(digest, item.face.index)
|
|
faces[identifier] = (item.face,)
|
|
imported_items.append(_entry(identifier, item.family, item.style, "imported"))
|
|
|
|
sort_key = lambda value: (value["family"].casefold(), value["style"].casefold(), value["id"])
|
|
automatic = {
|
|
"id": "default",
|
|
"label": "自动多语言字体",
|
|
"family": "自动多语言字体",
|
|
"style": "自动回退",
|
|
"source": "automatic",
|
|
}
|
|
with self._lock:
|
|
self._items = (
|
|
automatic,
|
|
*sorted(imported_items, key=sort_key),
|
|
*sorted(system_items, key=sort_key),
|
|
)
|
|
self._faces = faces
|
|
self._warnings = tuple(warnings)
|
|
|
|
def document(self) -> dict:
|
|
with self._lock:
|
|
return {
|
|
"items": [dict(item) for item in self._items],
|
|
"accepted_extensions": list(ACCEPTED_FONT_EXTENSIONS),
|
|
"max_upload_bytes": MAX_FONT_UPLOAD_BYTES,
|
|
"warnings": list(self._warnings),
|
|
}
|
|
|
|
def resolve(self, identifier: str) -> tuple[FontFace, ...]:
|
|
with self._lock:
|
|
return self._faces.get(identifier, ())
|
|
|
|
async def import_font(
|
|
self,
|
|
filename: str,
|
|
chunks: AsyncIterable[bytes],
|
|
) -> tuple[bool, list[dict[str, str]]]:
|
|
if (
|
|
not filename
|
|
or len(filename) > 255
|
|
or "\0" in filename
|
|
or "/" in filename
|
|
or "\\" in filename
|
|
):
|
|
raise FontTypeError("font filename is missing or invalid")
|
|
if Path(filename).suffix.casefold() not in ACCEPTED_FONT_EXTENSIONS:
|
|
raise FontTypeError("font type must be TTF, OTF, TTC, or OTC")
|
|
|
|
temporary = self.root / f".upload-{uuid4().hex}.tmp"
|
|
digest = hashlib.sha256()
|
|
total = 0
|
|
try:
|
|
with temporary.open("xb") as handle:
|
|
if os.name != "nt":
|
|
os.chmod(temporary, 0o640)
|
|
async for chunk in chunks:
|
|
if not chunk:
|
|
continue
|
|
total += len(chunk)
|
|
if total > MAX_FONT_UPLOAD_BYTES:
|
|
raise FontTooLargeError("font file exceeds the 32 MiB limit")
|
|
digest.update(chunk)
|
|
handle.write(chunk)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
if total == 0:
|
|
raise FontValidationError("font file is empty")
|
|
inspected = _inspect_font_file(temporary, require_unicode=True)
|
|
file_digest = digest.hexdigest()
|
|
target = self.root / _USER_FONT_NAME.format(digest=file_digest)
|
|
with self._lock:
|
|
created = not target.exists()
|
|
if created:
|
|
os.replace(temporary, target)
|
|
_sync_directory(self.root)
|
|
else:
|
|
temporary.unlink(missing_ok=True)
|
|
self.refresh()
|
|
identifiers = {_user_font_id(file_digest, item.face.index) for item in inspected}
|
|
imported = [dict(item) for item in self._items if item["id"] in identifiers]
|
|
return created, imported
|
|
except FontCatalogError:
|
|
temporary.unlink(missing_ok=True)
|
|
raise
|
|
except OSError as exc:
|
|
temporary.unlink(missing_ok=True)
|
|
if exc.errno in {errno.ENOSPC, errno.EDQUOT}:
|
|
raise FontStorageFullError("not enough disk space to import font") from exc
|
|
raise FontCatalogError(str(exc)) from exc
|
|
except BaseException:
|
|
temporary.unlink(missing_ok=True)
|
|
raise
|