doorstop.common.copy_dir_contents() - Code Metrics - Inspection of "YAML frontmatter format" - doorstop-dev/doorstop - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — develop (#566)

unknown

created 2022-04-11 07:53 UTC

doorstop.common.copy_dir_contents() A

↳ Parent: doorstop.common

Complexity

Conditions

Size

Total Lines	19
Code Lines	13

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	13
dl	0
loc	19
rs	9.2833
c	0
b	0
f	0
cc	5
nop	2

# SPDX-License-Identifier: LGPL-3.0-only

"""Common exceptions, classes, and functions for Doorstop."""

import argparse
import codecs
import csv
import glob
import io
import logging
import os
import re
import shutil

import frontmatter
import yaml

verbosity = 0  # global verbosity setting for controlling string formatting
PRINT_VERBOSITY = 0  # minimum verbosity to using `print`
STR_VERBOSITY = 3  # minimum verbosity to use verbose `__str__`
MAX_VERBOSITY = 4  # maximum verbosity level implemented


def _trace(self, message, *args, **kws):
    if self.isEnabledFor(logging.DEBUG - 1):
        self._log(logging.DEBUG - 1, message, args, **kws)  # pylint: disable=W0212


logging.addLevelName(logging.DEBUG - 1, "TRACE")
logging.Logger.trace = _trace  # type: ignore

logger = logging.getLogger
log = logger(__name__)

# exception classes ##########################################################


class DoorstopError(Exception):
    """Generic Doorstop error."""


class DoorstopFileError(DoorstopError, IOError):
    """Raised on IO errors."""


class DoorstopWarning(DoorstopError, Warning):
    """Generic Doorstop warning."""


class DoorstopInfo(DoorstopWarning, Warning):
    """Generic Doorstop info."""


# logging classes ############################################################


class HelpFormatter(argparse.ArgumentDefaultsHelpFormatter):
    """Command-line help text formatter with wider help text."""

    def __init__(self, *args, **kwargs):
        kwargs["max_help_position"] = 40
        super().__init__(*args, **kwargs)


class WarningFormatter(logging.Formatter):
    """Logging formatter that displays verbose formatting for WARNING+."""

    def __init__(self, default_format, verbose_format, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.default_format = default_format
        self.verbose_format = verbose_format

    def format(self, record):
        """Python 3 hack to change the formatting style dynamically."""
        if record.levelno > logging.INFO:
            self._style._fmt = self.verbose_format  # pylint: disable=W0212
        else:
            self._style._fmt = self.default_format  # pylint: disable=W0212
        return super().format(record)


# disk helper functions ######################################################


def create_dirname(path):
    """Ensure a parent directory exists for a path."""
    dirpath = os.path.dirname(path)
    if dirpath and not os.path.isdir(dirpath):
        log.info("creating directory {}...".format(dirpath))
        os.makedirs(dirpath)


def read_lines(path, encoding="utf-8"):
    """Read lines of text from a file.

    :param path: file to write lines
    :param encoding: output file encoding

    :return: path of new file

    """
    log.trace("reading lines from '{}'...".format(path))  # type: ignore
    with open(path, "r", encoding=encoding) as stream:
        for line in stream:
            yield line


def read_text(path):
    """Read text from a file.

    :param path: file path to read from
    :param encoding: input file encoding

    :return: string

    """
    log.trace("reading text from '{}'...".format(path))  # type: ignore
    try:
        with codecs.open(path, "r", encoding="utf-8") as f:
            return f.read()
    except Exception as e:
        msg = "reading '{}' failed: {}".format(path, e)
        raise DoorstopError(msg)


def load_yaml(text, path, loader=yaml.SafeLoader):
    """Parse a dictionary from YAML text.

    :param text: string containing dumped YAML data
    :param path: file path for error messages

    :return: dictionary

    """
    # Load the YAML data
    try:
        data = yaml.load(text, Loader=loader) or {}
    except yaml.error.YAMLError as exc:
        msg = "invalid contents: {}:\n{}".format(path, exc)
        raise DoorstopError(msg) from None
    # Ensure data is a dictionary
    if not isinstance(data, dict):
        msg = "invalid contents: {}".format(path)
        raise DoorstopError(msg)
    # Return the parsed data
    return data


def load_markdown(text, path, textattributekeys):
    """Parse a dictionary from Markdown file with YAML frontmatter.

    :param text: string containing markdown data with yaml frontmatter
    :param path: file path for error messages

    :return: dictionary

    """
    # Load YAML-frontmatter data from text
    try:
        data, content = frontmatter.parse(text, handler=frontmatter.YAMLHandler())
    except yaml.error.YAMLError as exc:
        msg = "invalid yaml contents: {}:\n{}".format(path, exc)
        raise DoorstopError(msg) from None
    # Ensure data is a dictionary
    if not isinstance(data, dict):
        msg = "invalid contents: {}".format(path)
        raise DoorstopError(msg)

    # parse content and update data dictionariy accordingly
    update_data_from_markdown_content(data, content, textattributekeys)

    # Return the parsed data
    return data


def write_lines(lines, path, end="\n", encoding="utf-8", *, executable=False):
    """Write lines of text to a file.

    :param lines: iterator of strings
    :param path: file to write lines
    :param end: string to end lines
    :param encoding: output file encoding

    :return: path of new file

    """
    log.trace("writing lines to '{}'...".format(path))  # type: ignore
    with open(path, "wb") as stream:
        for line in lines:
            data = (line + end).encode(encoding)
            stream.write(data)
    if executable and os.path.isfile(path):
        os.chmod(path, 0o775)
    return path


def write_text(text, path, end="\n"):
    """Write text to a file.

    :param text: string
    :param path: file to write text
    :param end: string to end lines
    :param encoding: output file encoding

    :return: path of new file

    """
    if text:
        log.trace("writing text to '{}'...".format(path))  # type: ignore
    with open(path, "w", encoding="utf-8", newline=end) as f:
        f.write(text)
    return path


def write_csv(table, path, delimiter=",", newline="", encoding="utf-8"):
    """Write table to a file.

    :param table: iterator of rows
    :param path: file to write lines
    :param delimiter: string to end cells
    :param newline: string to end lines
    :param encoding: output file encoding

    :return: path of new file

    """
    log.trace("writing table to '{}'...".format(path))  # type: ignore
    with open(path, "w", newline=newline, encoding=encoding) as stream:
        writer = csv.writer(stream, delimiter=delimiter)
        for row in table:
            writer.writerow(row)
    return path


def touch(path):
    """Ensure a file exists."""
    if not os.path.exists(path):
        log.trace("creating empty '{}'...".format(path))  # type: ignore
        write_text("", path)


def copy_dir_contents(src, dst):
    """Copy the contents of a directory."""
    for fpath in glob.glob("{}/*".format(src)):
        dest_path = os.path.join(dst, os.path.split(fpath)[-1])
        if os.path.exists(dest_path):
            if os.path.basename(fpath) == "doorstop":
                msg = "Skipping '{}' as this directory name is required by doorstop".format(
                    fpath
                )
            else:
                msg = "Skipping '{}' as a file or directory with this name already exists".format(
                    fpath
                )
            log.warning(msg)
        else:
            if os.path.isdir(fpath):
                shutil.copytree(fpath, dest_path)
            else:
                shutil.copyfile(fpath, dest_path)


def delete(path):
    """Delete a file or directory with error handling."""
    if os.path.isdir(path):
        try:
            log.trace("deleting '{}'...".format(path))  # type: ignore
            shutil.rmtree(path)
        except IOError:
            # bug: http://code.activestate.com/lists/python-list/159050
            msg = "unable to delete: {}".format(path)
            log.warning(msg)
    elif os.path.isfile(path):
        log.trace("deleting '{}'...".format(path))  # type: ignore
        os.remove(path)


def delete_contents(dirname):
    """Delete the contents of a directory."""
    for file in glob.glob("{}/*".format(dirname)):
        if os.path.isdir(file):
            shutil.rmtree(os.path.join(dirname, file))
        else:
            try:
                os.remove(os.path.join(dirname, file))
            except FileExistsError:
                log.warning(
                    "Two assets folders have files or directories " "with the same name"
                )
                raise


def update_data_from_markdown_content(data, content, textattributekeys):
    """Update data dictionary based on text content and attribute keys to look for within the content."""
    h1 = re.compile(r"^#{1}\s+(.*)")
    # for line based iteration
    s = io.StringIO(content)
    # final text
    header = None
    text = ""

    if "header" in textattributekeys:
        # search for first content line and check
        # if it is a h1 header
        for l in s:
            # skip empty lines
            if len(l.strip()) == 0:
                continue
            # check if first found line is a header
            m = h1.match(l.strip())
            if m:
                # header found
                header = m.group(1)
            else:
                # no header found, add to normal text
                text += l
            break

        # if header was found, skip empty lines before main text
        if header:
            for l in s:
                if len(l.strip()) != 0:
                    text += l
                    break

    # remaining content is normal text
    for l in s:
        text += l

    if "header" in textattributekeys and header:
        data["header"] = header

    if "text" in textattributekeys:
        data["text"] = text


def dump_markdown(data, textattr):
    content = ""
    if "header" in textattr and textattr["header"].strip() != "":
        content += "# {}\n".format(textattr["header"].strip())
        content += "\n"

    content += textattr["text"]

    text = frontmatter.dumps(frontmatter.Post(content, **data))
    return text


1			# SPDX-License-Identifier: LGPL-3.0-only
2
3			"""Common exceptions, classes, and functions for Doorstop."""
4
5			import argparse
6			import codecs
7			import csv
8			import glob
9			import io
10			import logging
11			import os
12			import re
13			import shutil
14
15			import frontmatter
16			import yaml
17
18			verbosity = 0 # global verbosity setting for controlling string formatting
19			PRINT_VERBOSITY = 0 # minimum verbosity to using `print`
20			STR_VERBOSITY = 3 # minimum verbosity to use verbose `__str__`
21			MAX_VERBOSITY = 4 # maximum verbosity level implemented
22
23
24			def _trace(self, message, args, *kws):
25			if self.isEnabledFor(logging.DEBUG - 1):
26			self._log(logging.DEBUG - 1, message, args, **kws) # pylint: disable=W0212
27
28
29			logging.addLevelName(logging.DEBUG - 1, "TRACE")
30			logging.Logger.trace = _trace # type: ignore
31
32			logger = logging.getLogger
33			log = logger(__name__)
34
35			# exception classes ##########################################################
36
37
38			class DoorstopError(Exception):
39			"""Generic Doorstop error."""
40
41
42			class DoorstopFileError(DoorstopError, IOError):
43			"""Raised on IO errors."""
44
45
46			class DoorstopWarning(DoorstopError, Warning):
47			"""Generic Doorstop warning."""
48
49
50			class DoorstopInfo(DoorstopWarning, Warning):
51			"""Generic Doorstop info."""
52
53
54			# logging classes ############################################################
55
56
57			class HelpFormatter(argparse.ArgumentDefaultsHelpFormatter):
58			"""Command-line help text formatter with wider help text."""
59
60			def __init__(self, args, *kwargs):
61			kwargs["max_help_position"] = 40
62			super().__init__(args, *kwargs)
63
64
65			class WarningFormatter(logging.Formatter):
66			"""Logging formatter that displays verbose formatting for WARNING+."""
67
68			def __init__(self, default_format, verbose_format, args, *kwargs):
69			super().__init__(args, *kwargs)
70			self.default_format = default_format
71			self.verbose_format = verbose_format
72
73			def format(self, record):
74			"""Python 3 hack to change the formatting style dynamically."""
75			if record.levelno > logging.INFO:
76			self._style._fmt = self.verbose_format # pylint: disable=W0212
77			else:
78			self._style._fmt = self.default_format # pylint: disable=W0212
79			return super().format(record)
80
81
82			# disk helper functions ######################################################
83
84
85			def create_dirname(path):
86			"""Ensure a parent directory exists for a path."""
87			dirpath = os.path.dirname(path)
88			if dirpath and not os.path.isdir(dirpath):
89			log.info("creating directory {}...".format(dirpath))
90			os.makedirs(dirpath)
91
92
93			def read_lines(path, encoding="utf-8"):
94			"""Read lines of text from a file.
95
96			:param path: file to write lines
97			:param encoding: output file encoding
98
99			:return: path of new file
100
101			"""
102			log.trace("reading lines from '{}'...".format(path)) # type: ignore
103			with open(path, "r", encoding=encoding) as stream:
104			for line in stream:
105			yield line
106
107
108			def read_text(path):
109			"""Read text from a file.
110
111			:param path: file path to read from
112			:param encoding: input file encoding
113
114			:return: string
115
116			"""
117			log.trace("reading text from '{}'...".format(path)) # type: ignore
118			try:
119			with codecs.open(path, "r", encoding="utf-8") as f:
120			return f.read()
121			except Exception as e:
122			msg = "reading '{}' failed: {}".format(path, e)
123			raise DoorstopError(msg)
124
125
126			def load_yaml(text, path, loader=yaml.SafeLoader):
127			"""Parse a dictionary from YAML text.
128
129			:param text: string containing dumped YAML data
130			:param path: file path for error messages
131
132			:return: dictionary
133
134			"""
135			# Load the YAML data
136			try:
137			data = yaml.load(text, Loader=loader) or {}
138			except yaml.error.YAMLError as exc:
139			msg = "invalid contents: {}:\n{}".format(path, exc)
140			raise DoorstopError(msg) from None
141			# Ensure data is a dictionary
142			if not isinstance(data, dict):
143			msg = "invalid contents: {}".format(path)
144			raise DoorstopError(msg)
145			# Return the parsed data
146			return data
147
148
149			def load_markdown(text, path, textattributekeys):
150			"""Parse a dictionary from Markdown file with YAML frontmatter.
151
152			:param text: string containing markdown data with yaml frontmatter
153			:param path: file path for error messages
154
155			:return: dictionary
156
157			"""
158			# Load YAML-frontmatter data from text
159			try:
160			data, content = frontmatter.parse(text, handler=frontmatter.YAMLHandler())
161			except yaml.error.YAMLError as exc:
162			msg = "invalid yaml contents: {}:\n{}".format(path, exc)
163			raise DoorstopError(msg) from None
164			# Ensure data is a dictionary
165			if not isinstance(data, dict):
166			msg = "invalid contents: {}".format(path)
167			raise DoorstopError(msg)
168
169			# parse content and update data dictionariy accordingly
170			update_data_from_markdown_content(data, content, textattributekeys)
171
172			# Return the parsed data
173			return data
174
175
176			def write_lines(lines, path, end="\n", encoding="utf-8", *, executable=False):
177			"""Write lines of text to a file.
178
179			:param lines: iterator of strings
180			:param path: file to write lines
181			:param end: string to end lines
182			:param encoding: output file encoding
183
184			:return: path of new file
185
186			"""
187			log.trace("writing lines to '{}'...".format(path)) # type: ignore
188			with open(path, "wb") as stream:
189			for line in lines:
190			data = (line + end).encode(encoding)
191			stream.write(data)
192			if executable and os.path.isfile(path):
193			os.chmod(path, 0o775)
194			return path
195
196
197			def write_text(text, path, end="\n"):
198			"""Write text to a file.
199
200			:param text: string
201			:param path: file to write text
202			:param end: string to end lines
203			:param encoding: output file encoding
204
205			:return: path of new file
206
207			"""
208			if text:
209			log.trace("writing text to '{}'...".format(path)) # type: ignore
210			with open(path, "w", encoding="utf-8", newline=end) as f:
211			f.write(text)
212			return path
213
214
215			def write_csv(table, path, delimiter=",", newline="", encoding="utf-8"):
216			"""Write table to a file.
217
218			:param table: iterator of rows
219			:param path: file to write lines
220			:param delimiter: string to end cells
221			:param newline: string to end lines
222			:param encoding: output file encoding
223
224			:return: path of new file
225
226			"""
227			log.trace("writing table to '{}'...".format(path)) # type: ignore
228			with open(path, "w", newline=newline, encoding=encoding) as stream:
229			writer = csv.writer(stream, delimiter=delimiter)
230			for row in table:
231			writer.writerow(row)
232			return path
233
234
235			def touch(path):
236			"""Ensure a file exists."""
237			if not os.path.exists(path):
238			log.trace("creating empty '{}'...".format(path)) # type: ignore
239			write_text("", path)
240
241
242			def copy_dir_contents(src, dst):
243			"""Copy the contents of a directory."""
244			for fpath in glob.glob("{}/*".format(src)):
245			dest_path = os.path.join(dst, os.path.split(fpath)[-1])
246			if os.path.exists(dest_path):
247			if os.path.basename(fpath) == "doorstop":
248			msg = "Skipping '{}' as this directory name is required by doorstop".format(
249			fpath
250			)
251			else:
252			msg = "Skipping '{}' as a file or directory with this name already exists".format(
253			fpath
254			)
255			log.warning(msg)
256			else:
257			if os.path.isdir(fpath):
258			shutil.copytree(fpath, dest_path)
259			else:
260			shutil.copyfile(fpath, dest_path)
261
262
263			def delete(path):
264			"""Delete a file or directory with error handling."""
265			if os.path.isdir(path):
266			try:
267			log.trace("deleting '{}'...".format(path)) # type: ignore
268			shutil.rmtree(path)
269			except IOError:
270			# bug: http://code.activestate.com/lists/python-list/159050
271			msg = "unable to delete: {}".format(path)
272			log.warning(msg)
273			elif os.path.isfile(path):
274			log.trace("deleting '{}'...".format(path)) # type: ignore
275			os.remove(path)
276
277
278			def delete_contents(dirname):
279			"""Delete the contents of a directory."""
280			for file in glob.glob("{}/*".format(dirname)):
281			if os.path.isdir(file):
282			shutil.rmtree(os.path.join(dirname, file))
283			else:
284			try:
285			os.remove(os.path.join(dirname, file))
286			except FileExistsError:
287			log.warning(
288			"Two assets folders have files or directories " "with the same name"
289			)
290			raise
291
292
293			def update_data_from_markdown_content(data, content, textattributekeys):
294			"""Update data dictionary based on text content and attribute keys to look for within the content."""
295			h1 = re.compile(r"^#{1}\s+(.*)")
296			# for line based iteration
297			s = io.StringIO(content)
298			# final text
299			header = None
300			text = ""
301
302			if "header" in textattributekeys:
303			# search for first content line and check
304			# if it is a h1 header
305			for l in s:
306			# skip empty lines
307			if len(l.strip()) == 0:
308			continue
309			# check if first found line is a header
310			m = h1.match(l.strip())
311			if m:
312			# header found
313			header = m.group(1)
314			else:
315			# no header found, add to normal text
316			text += l
317			break
318
319			# if header was found, skip empty lines before main text
320			if header:
321			for l in s:
322			if len(l.strip()) != 0:
323			text += l
324			break
325
326			# remaining content is normal text
327			for l in s:
328			text += l
329
330			if "header" in textattributekeys and header:
331			data["header"] = header
332
333			if "text" in textattributekeys:
334			data["text"] = text
335
336
337			def dump_markdown(data, textattr):
338			content = ""
339			if "header" in textattr and textattr["header"].strip() != "":
340			content += "# {}\n".format(textattr["header"].strip())
341			content += "\n"
342
343			content += textattr["text"]
344
345			text = frontmatter.dumps(frontmatter.Post(content, **data))
346			return text
347

doorstop-dev / doorstop

Pull Request — develop (#566)

doorstop.common.copy_dir_contents() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like