from __future__ import annotations

import errno
import hashlib
import json
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
from app.persistence import atomic_write_bytes

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._favorite_path = self.root / "favorites.json"
        self._favorite_ids = self._load_favorites()
        self._cleanup_upload_temporaries()
        self._builtin = self._install_builtins()
        self._system_catalog = self._system_faces()
        self.refresh()

    def _load_favorites(self) -> list[str]:
        if not self._favorite_path.exists():
            atomic_write_bytes(self._favorite_path, b'{"schema_version":1,"favorite_ids":[]}\n')
            return []
        try:
            record = json.loads(self._favorite_path.read_text(encoding="utf-8"))
            if (not isinstance(record, dict) or set(record) != {"schema_version", "favorite_ids"}
                    or type(record["schema_version"]) is not int or record["schema_version"] != 1):
                raise ValueError("invalid favorites schema")
            identifiers = record["favorite_ids"]
            if (not isinstance(identifiers, list)
                    or any(not isinstance(value, str) or not value or len(value) > 512 for value in identifiers)
                    or len(set(identifiers)) != len(identifiers)):
                raise ValueError("invalid favorites identifiers")
            return identifiers
        except (ValueError, UnicodeError) as exc:
            raise FontValidationError("invalid font favorites; original file preserved") from exc

    def set_favorite(self, identifier: str, favorite: bool) -> list[str]:
        with self._lock:
            if favorite and not any(item["id"] == identifier for item in self._items):
                raise KeyError(identifier)
            updated = [value for value in self._favorite_ids if value != identifier]
            if favorite:
                updated.append(identifier)
            if set(updated) == set(self._favorite_ids):
                return list(self._favorite_ids)
            content = json.dumps({"schema_version": 1, "favorite_ids": updated}, ensure_ascii=False).encode("utf-8") + b"\n"
            atomic_write_bytes(self._favorite_path, content)
            self._favorite_ids = updated
            return list(updated)

    def _install_builtins(self) -> dict[str, tuple[dict, FontFace]]:
        bundle = Path(__file__).resolve().parent / "bundled"
        manifest = json.loads((bundle / "manifest.json").read_text(encoding="utf-8"))
        expected = {"builtin:walnut-bus-8x16": "SimilarBus8x16.ttf", "builtin:walnut-bus-16x32": "SimilarBus16x32.ttf"}
        entries = manifest.get("fonts", [])
        if manifest.get("version") != 2 or len(entries) != 2 or {item["id"]: item["file"] for item in entries} != expected:
            raise FontValidationError("invalid bundled font registration")
        self._manifest = manifest
        # Validate the entire bundle before modifying persisted resources.
        for item in entries:
            path = bundle / item["file"]
            if hashlib.sha256(path.read_bytes()).hexdigest() != item["sha256"]:
                raise FontValidationError("bundled font checksum mismatch")
            _inspect_font_file(path, require_unicode=True)
        installed = {item["id"]: (item, self._install_face(bundle, item)) for item in entries}
        # Only retire the exact previous built-in asset; never glob user fonts.
        for digest in manifest.get("obsolete", []):
            if len(digest) != 64 or any(ch not in "0123456789abcdef" for ch in digest):
                raise FontValidationError("invalid obsolete font registration")
            old = self.root / _USER_FONT_NAME.format(digest=digest)
            if old.exists():
                if old.is_symlink() or hashlib.sha256(old.read_bytes()).hexdigest() != digest:
                    raise FontValidationError("obsolete font checksum mismatch; file preserved")
                old.unlink()
                _sync_directory(self.root)
        return installed

    def _install_face(self, bundle: Path, manifest: dict) -> FontFace:
        payload = (bundle / manifest["file"]).read_bytes()
        digest = hashlib.sha256(payload).hexdigest()
        if digest != manifest["sha256"]:
            raise FontValidationError("bundled font checksum mismatch")
        target = self.root / _USER_FONT_NAME.format(digest=digest)
        if target.exists():
            if hashlib.sha256(target.read_bytes()).hexdigest() != digest:
                raise FontValidationError("persisted bundled font checksum mismatch; file preserved")
        else:
            temporary = self.root / f".upload-{uuid4().hex}.tmp"
            try:
                with temporary.open("xb") as stream:
                    if os.name != "nt":
                        os.chmod(temporary, 0o640)
                    stream.write(payload)
                    stream.flush()
                    os.fsync(stream.fileno())
                _inspect_font_file(temporary, require_unicode=True)
                # Concurrent initialization must never replace an existing file.
                try:
                    os.link(temporary, target)
                except FileExistsError:
                    if hashlib.sha256(target.read_bytes()).hexdigest() != digest:
                        raise FontValidationError("persisted bundled font checksum mismatch; file preserved")
                _sync_directory(self.root)
            finally:
                temporary.unlink(missing_ok=True)
        return FontFace(str(target.resolve()), 0)

    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] = []
        builtin_digests = {item["sha256"] for item, face in self._builtin.values()}
        for manifest, builtin_face in self._builtin.values():
            faces[manifest["id"]] = (builtin_face,)
            builtin_item = _entry(manifest["id"], manifest["family"], manifest["style"], "system")
            builtin_item["label"] = manifest["label"]
            system_items.append(builtin_item)

        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,)
                if digest not in builtin_digests and digest not in self._manifest.get("obsolete", []):
                    imported_items.append(_entry(identifier, item.family, item.style, "imported"))

        for alias, identifier in self._manifest.get("aliases", {}).items():
            faces[alias] = faces[identifier]

        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],
                "favorite_ids": list(self._favorite_ids),
                "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, ())

    def render_version(self, identifier: str) -> str | None:
        identifier = self._manifest.get("aliases", {}).get(identifier, identifier)
        item = self._builtin.get(identifier)
        if item:
            return item[0]["sha256"]
        for manifest, face in self._builtin.values():
            if identifier == _user_font_id(manifest["sha256"], 0):
                return manifest["sha256"]
        return None

    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()
            old_alias = _user_font_id(file_digest, 0)
            if file_digest in self._manifest.get("obsolete", []):
                temporary.unlink(missing_ok=True)
                destination = self._manifest["aliases"][old_alias]
                return False, [dict(item) for item in self._items if item["id"] == destination]
            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}
                for manifest, face in self._builtin.values():
                    if file_digest == manifest["sha256"]:
                        identifiers.add(manifest["id"])
                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