image-match/python/py/Lib/site-packages/nbconvert/writers/files.py

"""Contains writer for writing nbconvert output to filesystem."""

# Copyright (c) IPython Development Team.
# Distributed under the terms of the Modified BSD License.

import errno
import glob
import os
from pathlib import Path

from traitlets import Unicode, observe

from nbconvert.utils.io import link_or_copy

from .base import WriterBase


class FilesWriter(WriterBase):
    """Consumes nbconvert output and produces files."""

    build_directory = Unicode(
        "",
        help="""Directory to write output(s) to. Defaults
                              to output to the directory of each notebook. To recover
                              previous default behaviour (outputting to the current
                              working directory) use . as the flag value.""",
    ).tag(config=True)

    relpath = Unicode(
        help="""When copying files that the notebook depends on, copy them in
        relation to this path, such that the destination filename will be
        os.path.relpath(filename, relpath). If FilesWriter is operating on a
        notebook that already exists elsewhere on disk, then the default will be
        the directory containing that notebook."""
    ).tag(config=True)

    # Make sure that the output directory exists.
    @observe("build_directory")
    def _build_directory_changed(self, change):
        new = change["new"]
        if new:
            self._makedir(new)

    def __init__(self, **kw):
        """Initialize the writer."""
        super().__init__(**kw)
        self._build_directory_changed({"new": self.build_directory})

    def _makedir(self, path, mode=0o755):
        """ensure that a directory exists

        If it doesn't exist, try to create it and protect against a race condition
        if another process is doing the same.

        The default permissions are 755, which differ from os.makedirs default of 777.
        """
        if not os.path.exists(path):
            self.log.info("Making directory %s", path)
            try:
                os.makedirs(path, mode=mode)
            except OSError as e:
                if e.errno != errno.EEXIST:
                    raise
        elif not os.path.isdir(path):
            raise OSError("%r exists but is not a directory" % path)

    def _write_items(self, items, build_dir):
        """Write a dict containing filename->binary data"""
        for filename, data in items:
            # Determine where to write the file to
            dest = os.path.join(build_dir, filename)
            path = os.path.dirname(dest)
            self._makedir(path)

            # Write file
            self.log.debug("Writing %i bytes to %s", len(data), dest)
            with open(dest, "wb") as f:
                f.write(data)

    def write(self, output, resources, notebook_name=None, **kw):
        """
        Consume and write Jinja output to the file system.  Output directory
        is set via the 'build_directory' variable of this instance (a
        configurable).

        See base for more...
        """

        # Verify that a notebook name is provided.
        if notebook_name is None:
            msg = "notebook_name"
            raise TypeError(msg)

        # Pull the extension and subdir from the resources dict.
        output_extension = resources.get("output_extension", None)

        # Get the relative path for copying files
        resource_path = resources.get("metadata", {}).get("path", "")
        relpath = self.relpath or resource_path
        build_directory = self.build_directory or resource_path

        # Write the extracted outputs to the destination directory.
        # NOTE: WE WRITE EVERYTHING AS-IF IT'S BINARY.  THE EXTRACT FIG
        # PREPROCESSOR SHOULD HANDLE UNIX/WINDOWS LINE ENDINGS...

        items = resources.get("outputs", {}).items()
        if items:
            self.log.info(
                "Support files will be in %s",
                os.path.join(resources.get("output_files_dir", ""), ""),
            )
            self._write_items(items, build_directory)

        # Write the extracted attachments
        # if ExtractAttachmentsOutput specified a separate directory
        attachments = resources.get("attachments", {}).items()
        if attachments:
            self.log.info(
                "Attachments will be in %s",
                os.path.join(resources.get("attachment_files_dir", ""), ""),
            )
            self._write_items(attachments, build_directory)

        # Copy referenced files to output directory
        if build_directory:
            for filename in self.files:
                # Copy files that match search pattern
                for matching_filename in glob.glob(filename):
                    # compute the relative path for the filename
                    if relpath != "":
                        dest_filename = os.path.relpath(matching_filename, relpath)
                    else:
                        dest_filename = matching_filename

                    # Make sure folder exists.
                    dest = os.path.join(build_directory, dest_filename)
                    path = os.path.dirname(dest)
                    self._makedir(path)

                    # Copy if destination is different.
                    if os.path.normpath(dest) != os.path.normpath(matching_filename):
                        self.log.info("Copying %s -> %s", matching_filename, dest)
                        link_or_copy(matching_filename, dest)

        # Determine where to write conversion results.
        dest = notebook_name + output_extension if output_extension is not None else notebook_name
        dest_path = Path(build_directory) / dest

        # Write conversion results.
        self.log.info("Writing %i bytes to %s", len(output), dest_path)
        if isinstance(output, str):
            with open(dest_path, "w", encoding="utf-8") as f:
                f.write(output)
        else:
            with open(dest_path, "wb") as f:
                f.write(output)

        return dest_path