"""
Utilities for writing PDF files.
Contains code from the PyPDF2 project; see :ref:`here <pypdf2-license>`
for the original license.
"""

import logging
import os
import typing
from typing import Dict, Iterable, List, Optional, Set, Tuple, Union, cast

from asn1crypto import x509
from pyhanko.pdf_utils import generic
from pyhanko.pdf_utils.crypt import (
    PubKeySecurityHandler,
    SecurityHandler,
    StandardSecurityHandler,
)
from pyhanko.pdf_utils.extensions import (
    DeveloperExtension,
    DevExtensionMultivalued,
)
from pyhanko.pdf_utils.generic import pdf_name
from pyhanko.pdf_utils.metadata.info import (
    update_info_dict,
    view_from_info_dict,
)
from pyhanko.pdf_utils.metadata.model import DocumentMetadata
from pyhanko.pdf_utils.misc import (
    IndirectObjectExpected,
    PdfError,
    PdfReadError,
    PdfWriteError,
    instance_test,
)
from pyhanko.pdf_utils.rw_common import PdfHandler
from pyhanko.pdf_utils.xref import (
    OBJSTREAM_FORBIDDEN,
    ObjectStream,
    PositionDict,
    XRefStream,
    write_xref_table,
)

if typing.TYPE_CHECKING:
    from .font.api import FontSubsetCollection

__all__ = [
    'BasePdfFileWriter',
    'PageObject',
    'PdfFileWriter',
    'init_xobject_dictionary',
    'copy_into_new_writer',
]

logger = logging.getLogger(__name__)


# TODO move this to content.py?
def init_xobject_dictionary(
    command_stream: bytes,
    box_width,
    box_height,
    resources: Optional[generic.DictionaryObject] = None,
) -> generic.StreamObject:
    """
    Helper function to initialise form XObject dictionaries.

    .. note::
        For utilities to handle image XObjects, see :mod:`.images`.

    :param command_stream:
        The XObject's raw appearance stream.
    :param box_width:
        The width of the XObject's bounding box.
    :param box_height:
        The height of the XObject's bounding box.
    :param resources:
        A resource dictionary to include with the form object.
    :return:
        A :class:`~.generic.StreamObject` representation of the form XObject.
    """
    resources = resources or generic.DictionaryObject()
    return generic.StreamObject(
        {
            pdf_name('/BBox'): generic.ArrayObject(
                list(
                    map(generic.FloatObject, (0.0, box_height, box_width, 0.0))
                )
            ),
            pdf_name('/Resources'): resources,
            pdf_name('/Type'): pdf_name('/XObject'),
            pdf_name('/Subtype'): pdf_name('/Form'),
        },
        stream_data=command_stream,
    )


class BasePdfFileWriter(PdfHandler):
    """Base class for PDF writers."""

    output_version = (1, 7)
    """Output version to be declared in the output file."""

    stream_xrefs: bool
    """
    Boolean controlling whether or not the output file will contain 
    its cross-references in stream format, or as a classical XRef table.
    
    The default for new files is ``True``. For incremental updates,
    the writer adapts to the system used in the previous iteration of the
    document (as mandated by the standard).
    """

    def __init__(
        self,
        root: Union[generic.IndirectObject, generic.DictionaryObject],
        info: Union[generic.IndirectObject, generic.DictionaryObject, None],
        document_id: generic.ArrayObject,
        obj_id_start: int = 0,
        stream_xrefs: bool = True,
    ):
        self.objects: Dict[Tuple[int, int], generic.PdfObject] = {}
        self.object_streams: List[ObjectStream] = list()
        self.objs_in_streams: Dict[int, generic.PdfObject] = {}
        self._lastobj_id = obj_id_start
        self._resolves_objs_from: Iterable[PdfHandler] = (self,)
        self._allocated_placeholders: Set[int] = set()

        if isinstance(root, generic.IndirectObject):
            self._root = root
        else:
            self._root = self.add_object(root)

        self.security_handler: Optional[SecurityHandler] = None
        self._encrypt: Optional[generic.IndirectObject] = None
        self._encrypt_key: Optional[bytes] = None
        self._document_id = document_id
        self.stream_xrefs = stream_xrefs
        info_ref = None
        if info is not None:
            if not isinstance(info, generic.IndirectObject):
                info_ref = self.add_object(info)
            else:
                info_ref = generic.IndirectObject(
                    info.idnum, info.generation, self
                )
        self._info = info_ref
        self._meta: DocumentMetadata = DocumentMetadata()

        self._font_resources: Dict[str, 'FontSubsetCollection'] = {}
        self.digest_aware_write = False
        self._mac_handler_cls = None

    def get_subset_collection(self, base_postscript_name: str):
        from .font.api import FontSubsetCollection

        try:
            fsc = self._font_resources[base_postscript_name]
        except KeyError:
            fsc = FontSubsetCollection(base_postscript_name)
            self._font_resources[base_postscript_name] = fsc

        return fsc

    @property
    def document_meta(self) -> DocumentMetadata:
        return self._meta

    @property
    def document_meta_view(self) -> DocumentMetadata:
        # we need the view_over (not just a copy) because
        # e.g. copy_into_new_writer will populate the info dict with
        # base values that can then be "cleanly" overridden in the
        # high-level API without destroying any unsupported entries
        # in the original info dict
        if self._info:
            base = view_from_info_dict(self._info.get_object())
        else:
            base = DocumentMetadata()
        return self._meta.view_over(base)

    def ensure_output_version(self, version):
        if self.output_version < version:
            self.output_version = version

    def set_info(
        self,
        info: Union[generic.IndirectObject, generic.DictionaryObject, None],
    ) -> Optional[generic.IndirectObject]:
        """
        Set the ``/Info`` entry of the document trailer.

        :param info:
            The new ``/Info`` dictionary, as an indirect reference.
        """
        new_info: Optional[generic.IndirectObject]
        if info is not None and not isinstance(info, generic.IndirectObject):
            self._info = new_info = self.add_object(info)
        else:
            self._info = new_info = info
        return new_info

    def set_custom_trailer_entry(
        self, key: generic.NameObject, value: generic.PdfObject
    ):
        """
        Set a custom, unmanaged entry in the document trailer or cross-reference
        stream dictionary.

        .. warning::
            Calling this method to set an entry that is managed by pyHanko
            internally (info dictionary, document catalog, etc.) has undefined
            results.

        :param key:
            Dictionary key to use in the trailer.
        :param value:
            Value to set
        """
        raise NotImplementedError

    @property
    def document_id(self) -> Tuple[bytes, bytes]:
        id_arr = self._document_id
        return id_arr[0].original_bytes, id_arr[1].original_bytes

    def mark_update(
        self, obj_ref: Union[generic.Reference, generic.IndirectObject]
    ):
        """
        Mark an object reference to be updated.
        This is only relevant for incremental updates, but is included
        as a no-op by default for interoperability reasons.

        :param obj_ref:
            An indirect object instance or a reference.
        """
        pass

    def update_container(self, obj: generic.PdfObject):
        """
        Mark the container of an object (as indicated by the
        :attr:`~.generic.PdfObject.container_ref` attribute on
        :class:`~.generic.PdfObject`) for an update.

        As with :meth:`mark_update`, this only applies to incremental updates,
        but defaults to a no-op.

        :param obj:
            The object whose top-level container needs to be rewritten.
        """
        pass

    @property
    def root_ref(self) -> generic.Reference:
        """
        :return:
            A reference to the document catalog.
        """
        return self._root.reference

    def update_root(self):
        """
        Signal that the document catalog should be written to the output.
        Equivalent to calling :meth:`mark_update` with :attr:`root_ref`.
        """
        pass

    def register_extension(self, ext: DeveloperExtension):
        try:
            extensions = self.root['/Extensions']
        except KeyError:
            self.root['/Extensions'] = extensions = generic.DictionaryObject()

        # check if the extension is already registered,
        try:
            cur_ext_value = extensions[ext.prefix_name]
        except KeyError:
            cur_ext_value = None
        extension_dicts: Iterable[generic.PdfObject] = ()
        old_ext_multivalued = False
        if isinstance(cur_ext_value, generic.DictionaryObject):
            extension_dicts = (cur_ext_value,)
        elif isinstance(cur_ext_value, generic.ArrayObject):
            extension_dicts = tuple(cur_ext_value.iterate())
            old_ext_multivalued = True
        elif cur_ext_value is not None:
            cls_name = type(cur_ext_value).__name__
            raise PdfReadError(
                f"PDF extension value is of type {cls_name}, "
                f"expected (direct) PDF dictionary or array."
            )

        # investigate the relationship between the new extension's extension
        # level and that of any existing extension dictionaries with the same
        # prefix
        for ix, ext_dict in enumerate(extension_dicts):
            if not isinstance(ext_dict, generic.DictionaryObject):
                cls_name = type(ext_dict).__name__
                raise PdfReadError(
                    f"PDF extension array entry is of type "
                    f"{cls_name}, expected (direct) PDF dictionary"
                )
            try:
                lvl = int(ext_dict.raw_get('/ExtensionLevel'))
            except (TypeError, ValueError, KeyError):
                raise PdfReadError("Could not read developer extension level")
            # TODO is this still appropriate with the new ExtensionRevision
            #  values?
            if lvl == ext.extension_level:
                return  # nothing to do
            elif ext.compare_by_level:
                old_ext_applies = lvl >= ext.extension_level
                replace_old = not old_ext_applies
            else:
                # this covers the case where there's no obvious comparison
                old_ext_applies = lvl in ext.subsumed_by
                replace_old = lvl in ext.subsumes
            if old_ext_applies:
                return  # nothing to do, old extension doesn't need replacing
            elif replace_old:
                # New extension takes priority,
                # replace the old extension
                if old_ext_multivalued:
                    # it was an array, so replace the value at the proper
                    # index
                    cur_ext_value[ix] = ext.as_pdf_object()
                elif ext.multivalued == DevExtensionMultivalued.ALWAYS:
                    # extension was previously not multivalued, but we now
                    # make it so
                    extensions[ext.prefix_name] = generic.ArrayObject(
                        [ext.as_pdf_object()]
                    )
                else:
                    extensions[ext.prefix_name] = ext.as_pdf_object()
                self.update_container(extensions)
                return  # we're done here
            elif ext.multivalued == DevExtensionMultivalued.NEVER:
                # there is a matching extension in the file, but we can't
                # replace it -> error
                raise PdfWriteError(
                    f"Could not register extension with prefix "
                    f"{ext.prefix_name} and level {ext.extension_level}; "
                    f"file contains extension with same prefix and "
                    f"extension level {lvl}. If this extension level is "
                    f"safe to override, mark it as subsumed."
                )
        # if we got here, we're going to add the new extension without
        # replacing any existing ones
        ext_dict = ext.as_pdf_object()
        if cur_ext_value is None:
            # nothing there yet, just write the extension
            if ext.multivalued == DevExtensionMultivalued.ALWAYS:
                extensions[ext.prefix_name] = generic.ArrayObject([ext_dict])
            else:
                extensions[ext.prefix_name] = ext_dict
        else:
            if old_ext_multivalued:
                # there was already an array, so just tack on our new value
                cur_ext_value.append(ext_dict)
            else:
                # turn the extension dict into an array
                extensions[ext.prefix_name] = generic.ArrayObject(
                    [cur_ext_value, ext_dict]
                )
        self.update_container(extensions)

    def get_object(self, ido, as_metadata_stream: bool = False):
        if ido.pdf not in self._resolves_objs_from:
            raise PdfError(
                f'Reference {ido} has no relation to this PDF writer.'
            )
        idnum = ido.idnum
        generation = ido.generation
        try:
            return self.objects[(generation, idnum)]
        except KeyError:
            if generation == 0:
                if idnum in self._allocated_placeholders:
                    return generic.NullObject()
                try:
                    return self.objs_in_streams[idnum]
                except KeyError:
                    pass
            raise KeyError(ido)

    def allocate_placeholder(self) -> generic.IndirectObject:
        """
        Allocate an object reference to populate later.
        Calls to :meth:`get_object` for this reference will
        return :class:`~.generic.NullObject` until it is populated using
        :meth:`add_object`.

        This method is only relevant in certain advanced contexts where
        an object ID needs to be known before the object it refers
        to can be built; chances are you'll never need it.

        :return:
            A :class:`~.generic.IndirectObject` instance referring to
            the object just allocated.
        """

        idnum = self._lastobj_id + 1
        self._allocated_placeholders.add(idnum)
        self._lastobj_id += 1
        return generic.IndirectObject(idnum, 0, self)

    def add_object(
        self, obj, obj_stream: Optional[ObjectStream] = None, idnum=None
    ) -> generic.IndirectObject:
        """
        Add a new object to this writer.

        :param obj:
            The object to add.
        :param obj_stream:
            An object stream to add the object to.
        :param idnum:
            Manually specify the object ID of the object to be added.
            This is only allowed for object IDs that have previously been
            allocated using :meth:`allocate_placeholder`.
        :return:
            A :class:`~.generic.IndirectObject` instance referring to
            the object just added.
        """

        if idnum is not None:
            if idnum not in self._allocated_placeholders:
                raise PdfWriteError(
                    "Manually specifying idnum is only allowed for "
                    "references previously allocated using "
                    "allocate_placeholder()."
                )
            preallocated = True
        else:
            preallocated = False
            idnum = self._lastobj_id + 1

        if obj_stream is None:
            self.objects[(0, idnum)] = obj
        elif obj_stream in self.object_streams:
            obj_stream.add_object(idnum, obj)
            self.objs_in_streams[idnum] = obj
        else:
            raise PdfWriteError(
                f'Stream {repr(obj_stream)} is unknown to this PDF writer.'
            )

        if preallocated:
            self._allocated_placeholders.remove(idnum)
        else:
            self._lastobj_id += 1
        return generic.IndirectObject(idnum, 0, self)

    def prepare_object_stream(self, compress=True):
        """Prepare and return a new :class:`.ObjectStream` object.

        :param compress:
            Indicates whether the resulting object stream should be compressed.
        :return:
            An :class:`.ObjectStream` object.
        """
        if not self.stream_xrefs:
            raise PdfWriteError(
                'Object streams require Xref streams to be enabled.'
            )
        stream = ObjectStream(compress=compress)
        self.object_streams.append(stream)
        return stream

    def _write_header(self, stream):
        raise NotImplementedError

    def _assign_security_handler(self, sh: SecurityHandler):
        self.security_handler = sh
        self._encrypt = self.add_object(sh.as_pdf_object())
        min_pdf_version = sh.get_min_pdf_version()
        if min_pdf_version is not None:
            self.ensure_output_version(min_pdf_version)
        for extension in self.security_handler.get_extensions():
            self.register_extension(extension)

    def _flush_obj_stream(self, obj_stm: ObjectStream):
        """
        Internal method to flush an object stream as part of the file
        writing process.
        """
        stream_ref = obj_stm.ref
        if obj_stm and not stream_ref:
            # first, register the object stream object
            #  (will get written later)
            obj_stm.ref = stream_ref = self.add_object(obj_stm.as_pdf_object())
        # loop over all objects in the stream, and prepare
        # the data to put in the XRef table
        for ix, (idnum, obj) in enumerate(obj_stm):
            yield idnum, (stream_ref.idnum, ix)

    def _write_objects(self, stream, object_position_dict: PositionDict):
        # deal with objects in object streams first
        for obj_stream in self.object_streams:
            for idnum, pos_record in self._flush_obj_stream(obj_stream):
                object_position_dict[(0, idnum)] = pos_record

        for ix in sorted(self.objects.keys()):
            generation, idnum = ix
            obj = self.objects[ix]
            object_position_dict[ix] = stream.tell()
            stream.write(('%d %d obj\n' % (idnum, generation)).encode('ascii'))
            handler: Optional[SecurityHandler] = None
            if self.security_handler is not None:
                assert self._encrypt is not None
                if idnum != self._encrypt.idnum:
                    handler = self.security_handler
            container_ref = generic.Reference(idnum, generation, self)
            obj.write_to_stream(stream, handler, container_ref)
            stream.write(b'\nendobj\n')

    def _prep_dom_for_writing(self):
        # ensure that all font resources are flushed
        for fsc in self._font_resources.values():
            for engine in fsc.subsets.values():
                engine.prepare_write()

        self._update_meta()

    def _update_meta(self):
        xmp_xml = None
        try:
            # delayed import since the namespace registration operation
            # is global (thank you ElementTree...)
            # also, we want the XML dep(s) to be optional
            # noinspection PyUnresolvedReferences
            from pyhanko.pdf_utils.metadata import xmp_xml

            need_xmp = (
                self._meta.xmp_unmanaged
                or bool(self._meta.xmp_extra)
                or self.output_version >= (2, 0)
                or '/Metadata' in self.root
            )
        except ImportError:  # pragma: nocover
            need_xmp = False

        self._meta.last_modified = 'now'
        if self._info is not None:
            mod = update_info_dict(
                self._meta,
                self._info.get_object(),
                # if we write XMP, we only update existing entries
                only_update_existing=need_xmp,
            )
            if mod:
                self.mark_update(self._info)
        elif not need_xmp:
            info_dict = generic.DictionaryObject()
            update_info_dict(self._meta, info_dict)
            self._info = self.add_object(info_dict)
            # if there's no info dict, and we're going to write XMP anyhow,
            # don't bother creating one

        if need_xmp:
            meta_stm = None
            if '/Metadata' in self.root:
                meta_obj = self.root['/Metadata']
                if isinstance(meta_obj, xmp_xml.MetadataStream):
                    meta_stm = meta_obj
                    meta_stm.update_xmp_with_meta(self._meta)
                    self.update_container(meta_stm)
            if meta_stm is None:
                meta_stm = xmp_xml.MetadataStream.from_xmp(
                    xmp_xml.update_xmp_with_meta(self._meta)
                )
                sh = self.security_handler
                meta_stm._handler = sh
                if (
                    sh is not None
                    and not self.security_handler.encrypt_metadata
                ):
                    # note: this will add the /Identity crypt filter, hence
                    # metadata encryption will be omitted
                    meta_stm.add_crypt_filter()
                self.root['/Metadata'] = self.add_object(meta_stm)
                self.update_root()
            self.update_root()

    def _populate_trailer(self, trailer):
        # prepare trailer dictionary entries
        trailer[pdf_name('/Root')] = self._root
        if self._info is not None:
            trailer[pdf_name('/Info')] = self._info
        if self._encrypt is not None:
            trailer[pdf_name('/Encrypt')] = self._encrypt
        # before doing anything else, we attempt to load the crypto-relevant
        # data, so that we can bail early if something's not right
        trailer[pdf_name('/ID')] = self._document_id

    @property
    def trailer_view(self) -> generic.DictionaryObject:
        trailer = generic.DictionaryObject()
        self._populate_trailer(trailer)
        return trailer

    @property
    def _need_mac_on_write(self):
        return (
            not self.digest_aware_write
            and self.security_handler is not None
            and self.security_handler.pdf_mac_enabled
        )

    def write(self, stream):
        """
        Write the contents of this PDF writer to a stream.

        :param stream:
            A writable output stream.
        """
        self._prep_dom_for_writing()
        if self._need_mac_on_write:
            from pyhanko.pdf_utils.crypt import pdfmac

            self.register_extension(pdfmac.ISO32004)
            pdfmac.add_standalone_mac(self, output=stream)
        else:
            self._write(stream)

    def _init_mac_handler(self, md_algorithm: str = 'sha256'):
        """
        MAC-adding method that can easily be manipulated in tests.
        """
        from pyhanko.pdf_utils.crypt import pdfmac

        cls = self._mac_handler_cls or pdfmac.PdfMacTokenHandler
        sh = self.security_handler
        assert sh is not None
        return cls.from_key_mat(
            file_encryption_key=sh.get_file_encryption_key(),
            kdf_salt=sh.get_kdf_salt(),
            md_algorithm=md_algorithm,
        )

    def _write(self, stream, skip_header: bool = False):
        object_positions: PositionDict = {}

        trailer: generic.DictionaryObject
        if self.stream_xrefs:
            xmp_trailer = XRefStream(object_positions)
            xmp_trailer.compress()
            trailer = xmp_trailer
        else:
            trailer = generic.DictionaryObject()

        if not skip_header:
            self._write_header(stream)
        self._populate_trailer(trailer)
        self._write_objects(stream, object_positions)

        if self.stream_xrefs:
            xref_location = stream.tell()
            xrefs_id = self._lastobj_id + 1
            # add position of XRef stream to the XRef stream
            object_positions[(0, xrefs_id)] = xref_location
            trailer[pdf_name('/Size')] = generic.NumberObject(xrefs_id + 1)
            # write XRef stream
            stream.write(('%d %d obj\n' % (xrefs_id, 0)).encode('ascii'))
            trailer.write_to_stream(stream, None)
            stream.write(b'\nendobj\n')
        else:
            # classical xref table
            xref_location = write_xref_table(
                stream, cast(Dict[Tuple[int, int], int], object_positions)
            )
            trailer[pdf_name('/Size')] = generic.NumberObject(
                self._lastobj_id + 1
            )
            # write trailer
            stream.write(b'trailer\n')
            trailer.write_to_stream(stream, None)

        # write xref table pointer and EOF
        xref_pointer_string = '\nstartxref\n%s\n' % xref_location
        stream.write(xref_pointer_string.encode('ascii') + b'%%EOF\n')

    def register_annotation(self, page_ref, annot_ref):
        """
        Register an annotation to be added to a page.
        This convenience function takes care of calling :meth:`mark_update`
        where necessary.

        :param page_ref:
            Reference to the page object involved.
        :param annot_ref:
            Reference to the annotation object to be added.
        """
        page_obj = page_ref.get_object()
        try:
            annot_arr_ref = page_obj.raw_get('/Annots')
            if isinstance(annot_arr_ref, generic.IndirectObject):
                annots = annot_arr_ref.get_object()
                self.mark_update(annot_arr_ref)
            else:
                # we need to update the entire page object if the annots array
                # is a direct object
                annots = annot_arr_ref
                self.mark_update(page_ref)
        except KeyError:
            annots = generic.ArrayObject()
            self.mark_update(page_ref)
            page_obj[pdf_name('/Annots')] = annots

        annots.append(annot_ref)

    def insert_page(self, new_page, after=None):
        """
        Insert a page object into the tree.

        :param new_page:
            Page object to insert.
        :param after:
            Page number (zero-indexed) after which to insert the page.
        :return:
            A reference to the newly inserted page.
        """
        if new_page['/Type'] != pdf_name('/Page'):
            raise PdfWriteError('Not a page object')
        if '/Parent' in new_page:
            raise PdfWriteError('/Parent must not be set.')

        page_tree_root_ref = self.root.raw_get('/Pages')
        if after is None:
            page_count = page_tree_root_ref.get_object()['/Count']
            after = page_count - 1

        if after == -1:
            # there are no pages yet, this will be the first
            pages_obj_ref = page_tree_root_ref
            kid_ix = -1
        else:
            pages_obj_ref, kid_ix, _ = self.find_page_container(after)

        pages_obj = pages_obj_ref.get_object()
        try:
            kids = pages_obj['/Kids']
        except KeyError:  # pragma: nocover
            raise PdfError('/Pages must have /Kids')

        # increase page count for all parents
        parent = pages_obj
        while parent is not None:
            # can't use += 1 because of the way PyPDF2's generic types work
            count = parent['/Count']
            parent[pdf_name('/Count')] = generic.NumberObject(count + 1)
            parent = parent.get('/Parent')
        new_page_ref = self.add_object(new_page)
        kids.insert(kid_ix + 1, new_page_ref)
        new_page[pdf_name('/Parent')] = pages_obj_ref
        self.update_container(pages_obj)
        self.update_container(kids)

        return new_page_ref

    def import_object(
        self, obj: generic.PdfObject, obj_stream: Optional[ObjectStream] = None
    ) -> generic.PdfObject:
        """
        Deep-copy an object into this writer, dealing with resolving indirect
        references in the process.

        .. danger::
            The table mapping indirect references in the input to indirect
            references in the writer is not preserved between calls.
            Concretely, this means that invoking :meth:`import_object` twice
            on the same input reader may cause object duplication.

        :param obj:
            The object to import.
        :param obj_stream:
            The object stream to import objects into.

            .. note::
                Stream objects and bare references will not be put into
                the object stream; the standard forbids this.
        :return:
            The object as associated with this writer.
            If the input object was an indirect reference, a dictionary
            (incl. streams) or an array, the returned value will always be
            a new instance.
        """

        importer = _ObjectImporter(
            source=obj.get_container_ref().get_pdf_handler(),
            target=self,
            obj_stream=obj_stream,
            reference_map={},
        )
        return importer.import_object(obj)

    def import_page_as_xobject(
        self, other: PdfHandler, page_ix=0, inherit_filters=True
    ):
        """
        Import a page content stream from some other
        :class:`~.rw_common.PdfHandler` into the current one as a form XObject.

        :param other:
            A :class:`~.rw_common.PdfHandler`
        :param page_ix:
            Index of the page to copy (default: 0)
        :param inherit_filters:
            Inherit the content stream's filters, if present.
        :return:
            An :class:`~.generic.IndirectObject` referring to the page object
            as added to the current reader.
        """
        page_ref, resources = other.find_page_for_modification(page_ix)
        page_obj = page_ref.get_object()

        # find the page's /MediaBox by going up the tree until we encounter it
        pagetree_obj = page_obj
        while True:
            try:
                mb = pagetree_obj['/MediaBox']
                break
            except KeyError:
                try:
                    pagetree_obj = pagetree_obj['/Parent']
                except KeyError:  # pragma: nocover
                    raise PdfReadError(
                        f'Page {page_ix} does not have a /MediaBox'
                    )

        stream_dict = {
            pdf_name('/BBox'): mb,
            pdf_name('/Resources'): self.import_object(resources),
            pdf_name('/Type'): pdf_name('/XObject'),
            pdf_name('/Subtype'): pdf_name('/Form'),
        }
        command_stream = page_obj['/Contents']
        # if the page /Contents is an array, retrieve the content stream
        # with the appropriate index
        if isinstance(command_stream, generic.ArrayObject):
            if len(command_stream) == 1 and inherit_filters:
                command_stream = command_stream[0].get_object()
            else:
                # There are multiple streams, so inheriting filters is not
                # a well-defined operations.
                # Decode and concatenate, then put in a flate filter, and return
                result = generic.StreamObject(
                    stream_dict,
                    stream_data=b''.join(
                        partial_stream.get_object().data
                        for partial_stream in command_stream
                    ),
                )
                result.compress()
                return self.add_object(result)
        assert isinstance(command_stream, generic.StreamObject)
        filters = None
        if inherit_filters:
            try:
                # try to inherit filters from the original command stream
                filters = command_stream['/Filter']
            except KeyError:
                pass

        if filters is not None:
            stream_dict[pdf_name('/Filter')] = self.import_object(filters)
            result = generic.StreamObject(
                stream_dict, encoded_data=command_stream.encoded_data
            )
        else:
            result = generic.StreamObject(
                stream_dict, stream_data=command_stream.data
            )

        return self.add_object(result)

    # TODO these can be simplified considerably using the new update_container
    def add_stream_to_page(
        self, page_ix, stream_ref, resources=None, prepend=False
    ):
        """Append an indirect stream object to a page in a PDF as a content
        stream.

        :param page_ix:
            Index of the page to modify.
            The first page has index `0`.
        :param stream_ref:
            :class:`~.generic.IndirectObject` reference to the stream
            object to add.
        :param resources:
            Resource dictionary containing resources to add to the page's
            existing resource dictionary.
        :param prepend:
            Prepend the content stream to the list of content streams, as
            opposed to appending it to the end.
            This has the effect of causing the stream to be rendered
            underneath the already existing content on the page.
        :return:
            An :class:`~.generic.IndirectObject` reference to the page object
            that was modified.
        """

        page_obj_ref, res_ref = self.find_page_for_modification(page_ix)

        page_obj = page_obj_ref.get_object()

        contents_ref = page_obj.raw_get('/Contents')

        if isinstance(contents_ref, generic.IndirectObject):
            contents = contents_ref.get_object()
            if isinstance(contents, generic.ArrayObject):
                # This is the easy case. It suffices to mark
                # this array for an update, and append our stream to it.
                self.mark_update(contents_ref)
                if prepend:
                    contents.insert(0, stream_ref)
                else:
                    contents.append(stream_ref)
            elif isinstance(contents, generic.StreamObject):
                # replace the old stream with an array containing
                # a reference to the original stream, and our own stream.
                new = (
                    [stream_ref, contents_ref]
                    if prepend
                    else [contents_ref, stream_ref]
                )
                contents = generic.ArrayObject(new)
                page_obj[pdf_name('/Contents')] = self.add_object(contents)
                # mark the page to be updated as well
                self.mark_update(page_obj_ref)
            else:
                raise PdfError('Unexpected type for page /Contents')
        elif isinstance(contents_ref, generic.ArrayObject):
            # make /Contents an indirect array, and append our stream
            contents = contents_ref
            if prepend:
                contents.insert(0, stream_ref)
            else:
                contents.append(stream_ref)
            page_obj[pdf_name('/Contents')] = self.add_object(contents)
            self.mark_update(page_obj_ref)
        else:
            raise PdfError('Unexpected type for page /Contents')

        if resources is None:
            return

        if isinstance(res_ref, generic.IndirectObject):
            # we can get away with only updating this reference
            orig_resource_dict = res_ref.get_object()
            assert isinstance(orig_resource_dict, generic.DictionaryObject)
            if self.merge_resources(orig_resource_dict, resources):
                self.mark_update(res_ref)
        else:
            # don't bother trying to update the resource object, just
            # clone it and add it to the current page object.
            orig_resource_dict = generic.DictionaryObject(res_ref)
            page_obj[pdf_name('/Resources')] = self.add_object(
                orig_resource_dict
            )
            self.merge_resources(orig_resource_dict, resources)

        return page_obj_ref

    # TODO this doesn't really belong here
    def merge_resources(
        self,
        orig_dict: generic.DictionaryObject,
        new_dict: generic.DictionaryObject,
    ) -> bool:
        """
        Update an existing resource dictionary object with data from another
        one. Returns ``True`` if the original dict object was modified directly.

        The caller is responsible for avoiding name conflicts with existing
        resources.
        """

        update_needed = False
        for key, value in new_dict.items():
            try:
                orig_value_ref = orig_dict.raw_get(key)
            except KeyError:
                update_needed = True
                orig_dict[key] = value
                continue

            if isinstance(orig_value_ref, generic.IndirectObject):
                orig_value = orig_value_ref.get_object()
                self.mark_update(orig_value_ref)
            else:
                orig_value = orig_value_ref
                update_needed = True

            if (
                isinstance(orig_value, generic.ArrayObject)
                and key == '/ProcSet'
            ):
                orig_value.extend(
                    x
                    for x in value.get_object()
                    if isinstance(x, generic.NameObject) and x not in orig_value
                )
            elif isinstance(orig_value, generic.DictionaryObject):
                for key_, value_ in value.items():
                    if key_ in orig_value:
                        raise PdfError(
                            'Naming conflict in resource of type %s: '
                            'key %s occurs in both.' % (key, key_)
                        )
                    orig_value[key_] = value_

        return update_needed


class PageObject(generic.DictionaryObject):
    """Subclass of :class:`~.generic.DictionaryObject` that handles some of the
    initialisation boilerplate for page objects."""

    # TODO be more clever with inheritable required attributes,
    #  and enforce the requirements on insertion instead
    # (setting /MediaBox at the page tree root seems to make sense, for example)
    def __init__(self, contents, media_box, resources=None):
        resources = resources or generic.DictionaryObject()

        if isinstance(contents, list):
            if not all(map(instance_test(generic.IndirectObject), contents)):
                raise PdfWriteError(
                    'Contents array must consist of indirect references'
                )
            if not isinstance(contents, generic.ArrayObject):
                contents = generic.ArrayObject(contents)
        elif not isinstance(contents, generic.IndirectObject):
            raise PdfWriteError(
                'Contents must be either an indirect reference or an array'
            )

        if len(media_box) != 4:
            raise ValueError('Media box must consist of 4 coordinates.')
        super().__init__(
            {
                pdf_name('/Type'): pdf_name('/Page'),
                pdf_name('/MediaBox'): generic.ArrayObject(
                    map(generic.FloatObject, media_box)
                ),
                pdf_name('/Resources'): resources,
                pdf_name('/Contents'): contents,
            }
        )


class PdfFileWriter(BasePdfFileWriter):
    """Class to write new PDF files."""

    def __init__(self, stream_xrefs=True, init_page_tree=True, info=None):
        # root object
        root = generic.DictionaryObject(
            {
                pdf_name("/Type"): pdf_name("/Catalog"),
            }
        )

        id1 = generic.ByteStringObject(os.urandom(16))
        id2 = generic.ByteStringObject(os.urandom(16))
        id_obj = generic.ArrayObject([id1, id2])

        self._custom_trailer_entries = {}
        super().__init__(root, info, id_obj, stream_xrefs=stream_xrefs)

        if init_page_tree:
            pages = generic.DictionaryObject(
                {
                    pdf_name("/Type"): pdf_name("/Pages"),
                    pdf_name("/Count"): generic.NumberObject(0),
                    pdf_name("/Kids"): generic.ArrayObject(),
                }
            )

            root[pdf_name('/Pages')] = self.add_object(pages)

    def _write_header(self, stream):
        major, minor = self.output_version
        stream.write(f'%PDF-{major}.{minor}\n'.encode('ascii'))
        # write some binary characters to make sure the file is flagged
        # as binary (see § 7.5.2 in ISO 32000-1)
        stream.write(b'%\xc2\xa5\xc2\xb1\xc3\xab\n')

    def encrypt(self, owner_pass, user_pass=None, **kwargs):
        """
        Mark this document to be encrypted with PDF 2.0 encryption (AES-256).

        .. caution::
            While pyHanko supports legacy PDF encryption as well, the API
            to create new documents using outdated encryption is left
            largely undocumented on purpose to discourage its use.

            This caveat does *not* apply to incremental updates added to
            existing documents.

        .. danger::
            The PDF 2.0 standard mandates AES-256 in CBC mode, and also includes
            12 bytes of known plaintext by design. This implies that a
            sufficiently knowledgeable attacker can inject arbitrary content
            into your encrypted files without knowledge of the password.

            Adding a digital signature to the encrypted document is **not**
            a foolproof way to deal with this either, since most viewers will
            still allow the document to be opened before signatures are
            validated, and therefore end users are still exposed to potentially
            malicious content.

            Until the standard supports authenticated encryption schemes, you
            should **never** rely on its encryption provisions if tampering
            is a concern.


        :param owner_pass:
            The desired owner password.
        :param user_pass:
            The desired user password (defaults to the owner password
            if not specified)
        :param kwargs:
            Other keyword arguments to be passed to
            :meth:`.StandardSecurityHandler.build_from_pw`.
        """
        sh = StandardSecurityHandler.build_from_pw(
            owner_pass, user_pass, **kwargs
        )
        self._assign_security_handler(sh)

    def encrypt_pubkey(self, recipients: List[x509.Certificate], **kwargs):
        """
        Mark this document to be encrypted with PDF 2.0 public key encryption.
        The certificates passed in should be RSA certificates.

        PyHanko defaults to AES-256 to encrypt the actual file contents.
        The seed used to derive the file encryption key is also encrypted
        using AES-256 and bundled in a CMS EnvelopedData object.
        The envelope key is then encrypted separately for each recipient, using
        their respective public keys.

        .. caution::
            The caveats for :meth:`encrypt` also apply here.

        :param recipients:
            Certificates of the recipients that should be able to decrypt
            the document.
        :param kwargs:
            Other keyword arguments to be passed to
            :meth:`.PubKeySecurityHandler.build_from_certs`.
        """
        self.output_version = (2, 0)
        sh = PubKeySecurityHandler.build_from_certs(recipients, **kwargs)
        self._assign_security_handler(sh)

    def set_custom_trailer_entry(
        self, key: generic.NameObject, value: generic.PdfObject
    ):
        """
        Set a custom, unmanaged entry in the document trailer or cross-reference
        stream dictionary.

        .. warning::
            Calling this method to set an entry that is managed by pyHanko
            internally (info dictionary, document catalog, etc.) has undefined
            results.

        :param key:
            Dictionary key to use in the trailer.
        :param value:
            Value to set
        """
        self._custom_trailer_entries[key] = value

    def _populate_trailer(self, trailer):
        # allow the base implementation to override internally managed entries
        trailer.update(self._custom_trailer_entries)
        super()._populate_trailer(trailer)


class _ObjectImporter:

    def __init__(
        self,
        source: PdfHandler,
        target: BasePdfFileWriter,
        reference_map: Dict[generic.Reference, generic.IndirectObject],
        obj_stream: Optional[ObjectStream],
    ):
        self.source = source
        self.target = target
        self.obj_stream = obj_stream
        self.queued_references: List[
            Tuple[generic.Reference, generic.Reference]
        ] = []
        self.reference_map = reference_map

    def import_object(self, obj: generic.PdfObject) -> generic.PdfObject:
        result = self._ingest(obj)

        while self.queued_references:
            source_ref, target_ref = self.queued_references.pop()
            source_obj = source_ref.get_object()
            imported = self._ingest(source_obj)

            # if the imported object is a bare reference and/or a stream
            # object, we can't put it into an object stream.
            if isinstance(imported, OBJSTREAM_FORBIDDEN):
                obj_stream = None
            else:
                obj_stream = self.obj_stream

            # fill in the placeholder
            self.target.add_object(
                imported, obj_stream=obj_stream, idnum=target_ref.idnum
            )

        return result

    def _ingest(self, obj: generic.PdfObject):
        if isinstance(obj, generic.DecryptedObjectProxy):
            obj = obj.decrypted
        if isinstance(obj, generic.IndirectObject):
            return self.process_reference(obj.reference)
        elif isinstance(obj, generic.DictionaryObject):
            raw_dict = {
                k: self._ingest(v) for k, v in obj.items() if k != '/Metadata'
            }
            try:
                # make sure to import metadata streams as such
                meta_ref = obj.get_value_as_reference('/Metadata')
                # ensure a MetadataStream object ends up in the cache
                meta_ref.get_pdf_handler().get_object(
                    meta_ref, as_metadata_stream=True
                )
                # ...then import the reference
                raw_dict['/Metadata'] = self.process_reference(meta_ref)
            except (KeyError, IndirectObjectExpected):
                pass

            if isinstance(obj, generic.StreamObject):
                stm_cls = generic.StreamObject
                # again, make sure to import metadata streams as such
                try:
                    # noinspection PyUnresolvedReferences
                    from pyhanko.pdf_utils.metadata import xmp_xml

                    if isinstance(obj, xmp_xml.MetadataStream):
                        stm_cls = xmp_xml.MetadataStream
                except ImportError:  # pragma: nocover
                    pass
                # In the vast majority of use cases, I'd expect the content
                # to be available in encoded form by default.
                # By initialising the stream object in this way, we avoid
                # a potentially costly decoding operation.
                return stm_cls(raw_dict, encoded_data=obj.encoded_data)
            else:
                return generic.DictionaryObject(raw_dict)
        elif isinstance(obj, generic.ArrayObject):
            return generic.ArrayObject(self._ingest(v) for v in obj)
        else:
            return obj

    def process_reference(self, ref: generic.Reference) -> generic.PdfObject:
        try:
            return self.reference_map[ref]
        except KeyError:
            # Add a placeholder to reserve the reference value.
            new_ido = self.target.allocate_placeholder()
            self.reference_map[ref] = new_ido
            self.queued_references.append((ref, new_ido.reference))
            return new_ido

    def preprocess_signature_data(self):
        # Signature /Contents is never encrypted => ensure we respect that
        # (even though the import operation is guaranteed to break the signature
        # there are valid use cases for stripping the encryption on such files,
        # e.g. for downstream processing)
        from ..sign.fields import enumerate_sig_fields

        signature_dict_refs = [
            field_value.reference
            for fq_name, field_value, field_ref in enumerate_sig_fields(
                self.source, filled_status=True
            )
            # this is the case in all valid PDFs
            if isinstance(field_value, generic.IndirectObject)
        ]
        if signature_dict_refs:
            logger.warning(
                "Source document contains filled signature fields--the copy "
                "operation will invalidate them."
            )
        for ref in signature_dict_refs:
            sig_dict = ref.get_object()
            assert isinstance(sig_dict, generic.DictionaryObject)
            raw_dict = {
                k: self._ingest(v)
                for k, v in sig_dict.items()
                if k != '/Contents'
            }
            raw_dict['/Contents'] = generic.ByteStringObject(
                sig_dict.raw_get(
                    '/Contents', decrypt=generic.EncryptedObjAccess.RAW
                ).original_bytes
            )
            self.reference_map[ref] = self.target.add_object(
                generic.DictionaryObject(raw_dict),
                obj_stream=None,
            )


def copy_into_new_writer(
    input_handler: PdfHandler, writer_kwargs: Optional[dict] = None
) -> PdfFileWriter:
    """
    Copy all objects in a given PDF handler into a new :class:`.PdfFileWriter`.
    This operation will attempt to preserve the document catalog
    of the original ``input_handler``.

    Very roughly, calling this function and then immediately invoking
    :meth:`~.BasePdfFileWriter.write` on the resulting writer should result
    in an equivalent document as far as presentation is concerned.
    As a general rule, behaviour that is controlled from outside the document
    catalog (e.g. encryption) or that requires byte-for-byte equivalence with
    the original (e.g. digital signatures) will not survive this translation.


    :param input_handler:
        :class:`.PdfHandler` to source objects from.
    :param writer_kwargs:
        Keyword arguments to pass to the writer.
    :return:
        New :class:`.PdfFileWriter` containing all objects from the input
        handler.
    """

    writer_kwargs = writer_kwargs or {}
    writer_kwargs.setdefault("stream_xrefs", False)

    # TODO try to be more clever with object streams
    w = PdfFileWriter(init_page_tree=False, **writer_kwargs)
    input_root_ref = input_handler.root_ref
    output_root_ref = w.root_ref
    # call _ObjectImporter in such a way that we translate the input handler's
    # root to the new writer's root.
    # From a technical PoV this doesn't matter, but it makes the output file
    # somewhat "cleaner" (i.e. it doesn't leave an orphaned document catalog
    # cluttering up the file)
    importer = _ObjectImporter(
        source=input_handler,
        target=w,
        reference_map={
            input_root_ref: generic.IndirectObject(
                idnum=output_root_ref.idnum,
                generation=output_root_ref.generation,
                pdf=w,
            )
        },
        obj_stream=None,
    )
    importer.preprocess_signature_data()
    new_root_dict = importer.import_object(input_handler.root)
    # override the old root ref
    ix = (output_root_ref.generation, output_root_ref.idnum)
    w.objects[ix] = new_root_dict

    if "info" not in writer_kwargs:
        try:
            # migrate the info dict. We do this low-level to avoid issues
            # with the producer string handling, and to keep a nice separation
            # between user-supplied metadata values and values that were present
            # in the original doc.
            info_dict = input_handler.trailer_view['/Info']
        except KeyError:
            info_dict = None
        if info_dict is not None:
            importer = _ObjectImporter(
                source=input_handler,
                target=w,
                reference_map={},
                obj_stream=None,
            )
            imported_info = importer.import_object(info_dict)
            w._info = w.add_object(imported_info)

    return w
