# SPDX-FileCopyrightText: 2022 James R. Barlow
# SPDX-License-Identifier: MPL-2.0

"""PdfMetadata - facade for XMP and DocumentInfo metadata."""

from __future__ import annotations

import logging
from collections.abc import Iterator, MutableMapping
from datetime import datetime, timezone
from typing import TYPE_CHECKING, Any
from warnings import warn

from lxml.etree import QName

from pikepdf._version import __version__ as pikepdf_version
from pikepdf.models.metadata._constants import (
    XMP_NS_PDF,
    XMP_NS_PDFA_ID,
    XMP_NS_PDFX_ID,
    XMP_NS_XMP,
    clean,
)
from pikepdf.models.metadata._converters import DOCINFO_MAPPING, DocinfoMapping
from pikepdf.models.metadata._docinfo import DocinfoStore
from pikepdf.models.metadata._xmp import XmpDocument
from pikepdf.objects import Name, Stream

if TYPE_CHECKING:  # pragma: no cover
    from pikepdf import Pdf

log = logging.getLogger(__name__)


class PdfMetadata(MutableMapping):
    """Read and edit the metadata associated with a PDF.

    The PDF specification contain two types of metadata, the newer XMP
    (Extensible Metadata Platform, XML-based) and older DocumentInformation
    dictionary. The PDF 2.0 specification removes the DocumentInformation
    dictionary.

    This primarily works with XMP metadata, but includes methods to generate
    XMP from DocumentInformation and will also coordinate updates to
    DocumentInformation so that the two are kept consistent.

    XMP metadata fields may be accessed using the full XML namespace URI or
    the short name. For example ``metadata['dc:description']``
    and ``metadata['{http://purl.org/dc/elements/1.1/}description']``
    both refer to the same field. Several common XML namespaces are registered
    automatically.

    See the XMP specification for details of allowable fields.

    To update metadata, use a with block.

    Example:
        >>> with pdf.open_metadata() as records:
        ...     records['dc:title'] = 'New Title'

    See Also:
        :meth:`pikepdf.Pdf.open_metadata`
    """

    # Keep DOCINFO_MAPPING at class level for backward compatibility
    DOCINFO_MAPPING: list[DocinfoMapping] = DOCINFO_MAPPING

    # Delegate namespace dicts to XmpDocument for backward compatibility
    NS: dict[str, str] = XmpDocument.NS
    REVERSE_NS: dict[str, str] = XmpDocument.REVERSE_NS

    def __init__(
        self,
        pdf: Pdf,
        pikepdf_mark: bool = True,
        sync_docinfo: bool = True,
        overwrite_invalid_xml: bool = True,
    ):
        """Construct PdfMetadata. Use Pdf.open_metadata() instead."""
        self._pdf = pdf
        self.mark = pikepdf_mark
        self.sync_docinfo = sync_docinfo
        self._updating = False
        self._overwrite_invalid_xml = overwrite_invalid_xml

        # Initialize XmpDocument with PDF's XMP data
        self._xmp_doc = self._load_xmp()

        # Initialize DocinfoStore
        self._docinfo = DocinfoStore(pdf)

    def _load_xmp(self) -> XmpDocument:
        """Load XMP from PDF or create empty XmpDocument."""
        try:
            data = self._pdf.Root.Metadata.read_bytes()
        except AttributeError:
            data = b''

        return XmpDocument(
            data, overwrite_invalid_xml=self._overwrite_invalid_xml
        )

    def load_from_docinfo(
        self, docinfo, delete_missing: bool = False, raise_failure: bool = False
    ) -> None:
        """Populate the XMP metadata object with DocumentInfo.

        Arguments:
            docinfo: a DocumentInfo, e.g pdf.docinfo
            delete_missing: if the entry is not DocumentInfo, delete the equivalent
                from XMP
            raise_failure: if True, raise any failure to convert docinfo;
                otherwise warn and continue

        A few entries in the deprecated DocumentInfo dictionary are considered
        approximately equivalent to certain XMP records. This method copies
        those entries into the XMP metadata.
        """

        def warn_or_raise(msg, e=None):
            if raise_failure:
                raise ValueError(msg) from e
            warn(msg)

        for uri, shortkey, docinfo_name, converter in self.DOCINFO_MAPPING:
            qname = QName(uri, shortkey)
            # docinfo might be a dict or pikepdf.Dictionary, so lookup keys
            # by str(Name)
            val = docinfo.get(str(docinfo_name))
            if val is None:
                if delete_missing and qname in self:
                    del self[qname]
                continue
            try:
                val = str(val)
                if converter:
                    val = converter.xmp_from_docinfo(val)
                if not val:
                    continue
                self._setitem(qname, val, True)
            except (ValueError, AttributeError, NotImplementedError) as e:
                warn_or_raise(
                    f"The metadata field {docinfo_name} could not be copied to XMP", e
                )
        valid_docinfo_names = {
            str(docinfo_name) for _, _, docinfo_name, _ in self.DOCINFO_MAPPING
        }
        extra_docinfo_names = {str(k) for k in docinfo.keys()} - valid_docinfo_names
        for extra in extra_docinfo_names:
            warn_or_raise(
                f"The metadata field {extra} with value '{repr(docinfo.get(extra))}' "
                "has no XMP equivalent, so it was discarded",
            )

    def __enter__(self):
        """Open metadata for editing."""
        self._updating = True
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Close metadata and apply changes."""
        try:
            if exc_type is not None:
                return
            self._apply_changes()
        finally:
            self._updating = False

    def _update_docinfo(self):
        """Update the PDF's DocumentInfo dictionary to match XMP metadata.

        The standard mapping is described here:
            https://www.pdfa.org/pdfa-metadata-xmp-rdf-dublin-core/
        """
        # Touch object to ensure it exists
        self._pdf.docinfo  # pylint: disable=pointless-statement
        for uri, element, docinfo_name, converter in self.DOCINFO_MAPPING:
            qname = QName(uri, element)
            try:
                value = self[qname]
            except KeyError:
                if docinfo_name in self._pdf.docinfo:
                    del self._pdf.docinfo[docinfo_name]
                continue
            if converter:
                try:
                    value = converter.docinfo_from_xmp(value)
                except ValueError:
                    warn(
                        f"The DocumentInfo field {docinfo_name} could not be "
                        "updated from XMP"
                    )
                    value = None
                except Exception as e:
                    raise ValueError(
                        "An error occurred while updating DocumentInfo field "
                        f"{docinfo_name} from XMP {qname} with value {value}"
                    ) from e
            if value is None:
                if docinfo_name in self._pdf.docinfo:
                    del self._pdf.docinfo[docinfo_name]
                continue
            self._docinfo.set(docinfo_name, clean(value))

    def _apply_changes(self):
        """Serialize our changes back to the PDF in memory.

        Depending how we are initialized, leave our metadata mark and producer.
        """
        if self.mark:
            # We were asked to mark the file as being edited by pikepdf
            self._setitem(
                QName(XMP_NS_XMP, 'MetadataDate'),
                datetime.now(timezone.utc).isoformat(),
                applying_mark=True,
            )
            self._setitem(
                QName(XMP_NS_PDF, 'Producer'),
                'pikepdf ' + pikepdf_version,
                applying_mark=True,
            )
        xml = self._xmp_doc.to_bytes()
        self._pdf.Root.Metadata = Stream(self._pdf, xml)
        self._pdf.Root.Metadata[Name.Type] = Name.Metadata
        self._pdf.Root.Metadata[Name.Subtype] = Name.XML
        if self.sync_docinfo:
            self._update_docinfo()

    @classmethod
    def _qname(cls, name: QName | str) -> str:
        """Convert name to an XML QName.

        e.g. pdf:Producer -> {http://ns.adobe.com/pdf/1.3/}Producer
        """
        return XmpDocument.qname(name)

    @classmethod
    def register_xml_namespace(cls, uri: str, prefix: str) -> None:
        """Register a new XML/XMP namespace.

        Arguments:
            uri: The long form of the namespace.
            prefix: The alias to use when interpreting XMP.
        """
        XmpDocument.register_xml_namespace(uri, prefix)

    def _prefix_from_uri(self, uriname: str) -> str:
        """Given a fully qualified XML name, find a prefix.

        e.g. {http://ns.adobe.com/pdf/1.3/}Producer -> pdf:Producer
        """
        return self._xmp_doc.prefix_from_uri(uriname)

    def __contains__(self, key: object) -> bool:  # type: ignore[override]
        """Test if XMP key is in metadata."""
        if not isinstance(key, (str, QName)):
            raise TypeError(f"{key!r} must be str or QName")
        return key in self._xmp_doc

    def __getitem__(self, key: str | QName) -> Any:
        """Retrieve XMP metadata for key."""
        return self._xmp_doc[key]

    def __iter__(self) -> Iterator[str]:
        """Iterate through XMP metadata attributes and nodes."""
        return iter(self._xmp_doc)

    def __len__(self) -> int:
        """Return number of items in metadata."""
        return len(self._xmp_doc)

    def _setitem(
        self,
        key: str | QName,
        val: set[str] | list[str] | str,
        applying_mark: bool = False,
    ) -> None:
        if not self._updating:
            raise RuntimeError("Metadata not opened for editing, use with block")

        qkey = self._qname(key)
        self._setitem_check_args(key, val, applying_mark, qkey)
        self._xmp_doc.set_value(key, val)

    def _setitem_check_args(
        self, key: str | QName, val: Any, applying_mark: bool, qkey: str
    ) -> None:
        if (
            self.mark
            and not applying_mark
            and qkey
            in (
                self._qname('xmp:MetadataDate'),
                self._qname('pdf:Producer'),
            )
        ):
            # Complain if user writes self[pdf:Producer] = ... and because it will
            # be overwritten on save, unless self._updating_mark, in which case
            # the action was initiated internally
            log.warning(
                f"Update to {key} will be overwritten because metadata was opened "
                "with set_pikepdf_as_editor=True"
            )
        if isinstance(val, str) and qkey in (self._qname('dc:creator')):
            log.error(f"{key} should be set to a list of strings")

    def __setitem__(self, key: str | QName, val: set[str] | list[str] | str) -> None:
        """Set XMP metadata key to value."""
        return self._setitem(key, val, False)

    def __delitem__(self, key: str | QName) -> None:
        """Delete item from XMP metadata."""
        if not self._updating:
            raise RuntimeError("Metadata not opened for editing, use with block")
        del self._xmp_doc[key]

    @property
    def pdfa_status(self) -> str:
        """Return the PDF/A conformance level claimed by this PDF, or False.

        A PDF may claim to PDF/A compliant without this being true. Use an
        independent verifier such as veraPDF to test if a PDF is truly
        conformant.

        Returns:
            The conformance level of the PDF/A, or an empty string if the
            PDF does not claim PDF/A conformance. Possible valid values
            are: 1A, 1B, 2A, 2B, 2U, 3A, 3B, 3U. Note that ISO standard
            typically refers to PDF/A-1b for example, using lower case;
            this function returns the value as it appears in the PDF, which
            is uppercase.
        """
        key_part = QName(XMP_NS_PDFA_ID, 'part')
        key_conformance = QName(XMP_NS_PDFA_ID, 'conformance')
        try:
            return self[key_part] + self[key_conformance]
        except KeyError:
            return ''

    @property
    def pdfx_status(self) -> str:
        """Return the PDF/X conformance level claimed by this PDF, or False.

        A PDF may claim to PDF/X compliant without this being true. Use an
        independent verifier such as veraPDF to test if a PDF is truly
        conformant.

        Returns:
            The conformance level of the PDF/X, or an empty string if the
            PDF does not claim PDF/X conformance.
        """
        pdfx_version = QName(XMP_NS_PDFX_ID, 'GTS_PDFXVersion')
        try:
            return self[pdfx_version]
        except KeyError:
            return ''

    def __str__(self) -> str:
        """Convert XMP metadata to XML string."""
        return str(self._xmp_doc)

    # Backward compatibility methods for internal API access
    def _load(self) -> None:
        """No-op for backward compatibility.

        Previously this triggered lazy loading of XMP. Now XMP is loaded
        immediately in __init__.
        """
        pass

    def _get_rdf_root(self):
        """Get the rdf:RDF root element.

        Provided for backward compatibility with code that accesses
        internal XMP structure.
        """
        return self._xmp_doc._get_rdf_root()

    def _get_xml_bytes(self, xpacket: bool = True) -> bytes:
        """Serialize XMP to XML bytes.

        Provided for backward compatibility.
        """
        return self._xmp_doc.to_bytes(xpacket=xpacket)
