Refactor literalinclude directive

sphinx/directives/code.py

@@ -11,8 +11,6 @@ import sys
 import codecs
 from difflib import unified_diff
 
-from six import string_types
-
 from docutils import nodes
 from docutils.parsers.rst import Directive, directives
 from docutils.statemachine import ViewList
@@ -24,8 +22,9 @@ from sphinx.util.nodes import set_source_info
 
 if False:
     # For type annotation
-    from typing import Any  # NOQA
+    from typing import Any, Tuple  # NOQA
     from sphinx.application import Sphinx  # NOQA
+    from sphinx.config import Config  # NOQA
 
 
 class Highlight(Directive):
@@ -78,7 +77,8 @@ def container_wrapper(directive, literal_node, caption):
     directive.state.nested_parse(ViewList([caption], source=''),
                                  directive.content_offset, parsed)
     if isinstance(parsed[0], nodes.system_message):
-        raise ValueError(parsed[0])
+        msg = _('Invalid caption: %s' % parsed[0].astext())
+        raise ValueError(msg)
     caption_node = nodes.caption(parsed[0].rawsource, '',
                                  *parsed[0].children)
     caption_node.source = literal_node.source
@@ -110,6 +110,7 @@ class CodeBlock(Directive):
 
     def run(self):
         # type: () -> List[nodes.Node]
+        document = self.state.document
         code = u'\n'.join(self.content)
 
         linespec = self.options.get('emphasize-lines')
@@ -118,7 +119,6 @@ class CodeBlock(Directive):
                 nlines = len(self.content)
                 hl_lines = [x + 1 for x in parselinenos(linespec, nlines)]
             except ValueError as err:
-                document = self.state.document
                 return [document.reporter.warning(str(err), line=self.lineno)]
         else:
             hl_lines = None
@@ -145,9 +145,7 @@ class CodeBlock(Directive):
             try:
                 literal = container_wrapper(self, literal, caption)
             except ValueError as exc:
-                document = self.state.document
+                return [document.reporter.warning(str(exc), line=self.lineno)]
-                errmsg = _('Invalid caption: %s' % exc[0][0].astext())  # type: ignore
-                return [document.reporter.warning(errmsg, line=self.lineno)]
 
         # literal will be note_implicit_target that is linked from caption and numref.
         # when options['name'] is provided, it should be primary ID.
@@ -156,6 +154,194 @@ class CodeBlock(Directive):
         return [literal]
 
 
+class LiteralIncludeReader(object):
+    INVALID_OPTIONS_PAIR = [
+        ('pyobject', 'lines'),
+        ('lineno-match', 'lineno-start'),
+        ('lineno-match', 'append'),
+        ('lineno-match', 'prepend'),
+        ('start-after', 'start-at'),
+        ('end-before', 'end-at'),
+        ('diff', 'pyobject'),
+        ('diff', 'lineno-start'),
+        ('diff', 'lineno-match'),
+        ('diff', 'lines'),
+        ('diff', 'start-after'),
+        ('diff', 'end-before'),
+        ('diff', 'start-at'),
+        ('diff', 'end-at'),
+    ]
+
+    def __init__(self, filename, options, config):
+        # type: (unicode, Dict, Config) -> None
+        self.filename = filename
+        self.options = options
+        self.encoding = options.get('encoding', config.source_encoding)
+        self.lineno_start = self.options.get('lineno-start', 1)
+
+        self.parse_options()
+
+    def parse_options(self):
+        # type: () -> None
+        for option1, option2 in self.INVALID_OPTIONS_PAIR:
+            if option1 in self.options and option2 in self.options:
+                raise ValueError(_('Cannot use both "%s" and "%s" options') %
+                                 (option1, option2))
+
+    def read_file(self, filename):
+        # type: (unicode) -> List[unicode]
+        try:
+            with codecs.open(filename, 'r', self.encoding, errors='strict') as f:  # type: ignore  # NOQA
+                text = f.read()  # type: unicode
+                if 'tab-width' in self.options:
+                    text = text.expandtabs(self.options['tab-width'])
+
+                lines = text.splitlines(True)
+                if 'dedent' in self.options:
+                    return dedent_lines(lines, self.options.get('dedent'))
+                else:
+                    return lines
+        except (IOError, OSError) as exc:
+            raise IOError(_('Include file %r not found or reading it failed') % filename)
+        except UnicodeError as exc:
+            raise UnicodeError(_('Encoding %r used for reading included file %r seems to '
+                                 'be wrong, try giving an :encoding: option') %
+                               (self.encoding, filename))
+
+    def read(self):
+        # type: () -> Tuple[unicode, int]
+        if 'diff' in self.options:
+            lines = self.show_diff()
+        else:
+            filters = [self.pyobject_filter,
+                       self.lines_filter,
+                       self.start_filter,
+                       self.end_filter,
+                       self.prepend_filter,
+                       self.append_filter]
+            lines = self.read_file(self.filename)
+            for func in filters:
+                lines = func(lines)
+
+        return ''.join(lines), len(lines)
+
+    def show_diff(self):
+        # type: () -> List[unicode]
+        new_lines = self.read_file(self.filename)
+        old_filename = self.options.get('diff')
+        old_lines = self.read_file(old_filename)
+        diff = unified_diff(old_lines, new_lines, old_filename, self.filename)  # type: ignore
+        return list(diff)
+
+    def pyobject_filter(self, lines):
+        # type: (List[unicode]) -> List[unicode]
+        pyobject = self.options.get('pyobject')
+        if pyobject:
+            from sphinx.pycode import ModuleAnalyzer
+            analyzer = ModuleAnalyzer.for_file(self.filename, '')
+            tags = analyzer.find_tags()
+            if pyobject not in tags:
+                raise ValueError(_('Object named %r not found in include file %r') %
+                                 (pyobject, self.filename))
+            else:
+                start = tags[pyobject][1]
+                end = tags[pyobject][2]
+                lines = lines[start - 1:end - 1]
+                if 'lineno-match' in self.options:
+                    self.lineno_start = start
+
+        return lines
+
+    def lines_filter(self, lines):
+        # type: (List[unicode]) -> List[unicode]
+        linespec = self.options.get('lines')
+        if linespec:
+            linelist = parselinenos(linespec, len(lines))
+            if 'lineno-match' in self.options:
+                # make sure the line list is not "disjoint".
+                first = linelist[0]
+                if all(first + i == n for i, n in enumerate(linelist)):
+                    self.lineno_start = linelist[0] + 1
+                else:
+                    raise ValueError(_('Cannot use "lineno-match" with a disjoint '
+                                       'set of "lines"'))
+
+            # just ignore non-existing lines
+            lines = [lines[n] for n in linelist if n < len(lines)]
+            if not lines:
+                raise ValueError(_('Line spec %r: no lines pulled from include file %r') %
+                                 (linespec, self.filename))
+
+        return lines
+
+    def start_filter(self, lines):
+        # type: (List[unicode]) -> List[unicode]
+        if 'start-at' in self.options:
+            start = self.options.get('start-at')
+            inclusive = False
+        elif 'start-after' in self.options:
+            start = self.options.get('start-after')
+            inclusive = True
+        else:
+            start = None
+
+        if start:
+            for lineno, line in enumerate(lines):
+                if start in line:
+                    if 'lineno-match' in self.options:
+                        self.lineno_start += lineno + 1
+
+                    if inclusive:
+                        return lines[lineno + 1:]
+                    else:
+                        return lines[lineno:]
+
+        return lines
+
+    def end_filter(self, lines):
+        # type: (List[unicode]) -> List[unicode]
+        if 'end-at' in self.options:
+            end = self.options.get('end-at')
+            inclusive = True
+        elif 'end-before' in self.options:
+            end = self.options.get('end-before')
+            inclusive = False
+        else:
+            end = None
+
+        if end:
+            for lineno, line in enumerate(lines):
+                if end in line:
+                    if 'lineno-match' in self.options:
+                        self.lineno_start += lineno + 1
+
+                    if inclusive:
+                        return lines[:lineno + 1]
+                    else:
+                        if lineno == 0:
+                            return []
+                        else:
+                            return lines[:lineno]
+
+        return lines
+
+    def prepend_filter(self, lines):
+        # type: (List[unicode]) -> List[unicode]
+        prepend = self.options.get('prepend')
+        if prepend:
+            lines.insert(0, prepend + '\n')
+
+        return lines
+
+    def append_filter(self, lines):
+        # type: (List[unicode]) -> List[unicode]
+        append = self.options.get('append')
+        if append:
+            lines.append(append + '\n')
+
+        return lines
+
+
 class LiteralInclude(Directive):
     """
     Like ``.. include:: :literal:``, but only warns if the include file is
@@ -190,24 +376,6 @@ class LiteralInclude(Directive):
         'diff': directives.unchanged_required,
     }
 
-    def read_with_encoding(self, filename, document, codec_info, encoding):
-        # type: (unicode, nodes.Node, Any, unicode) -> List
-        try:
-            with codecs.StreamReaderWriter(open(filename, 'rb'), codec_info[2],
-                                           codec_info[3], 'strict') as f:
-                lines = f.readlines()
-                lines = dedent_lines(lines, self.options.get('dedent'))  # type: ignore
-                return lines
-        except (IOError, OSError):
-            return [document.reporter.warning(
-                'Include file %r not found or reading it failed' % filename,
-                line=self.lineno)]
-        except UnicodeError:
-            return [document.reporter.warning(
-                'Encoding %r used for reading included file %r seems to '
-                'be wrong, try giving an :encoding: option' %
-                (encoding, filename))]
-
     def run(self):
         # type: () -> List[nodes.Node]
         document = self.state.document
@@ -215,177 +383,46 @@ class LiteralInclude(Directive):
             return [document.reporter.warning('File insertion disabled',
                                               line=self.lineno)]
         env = document.settings.env
+
+        # convert options['diff'] to absolute path
+        if 'diff' in self.options:
+            _, path = env.relfn2path(self.options['diff'])
+            self.options['diff'] = path
+
+        try:
             rel_filename, filename = env.relfn2path(self.arguments[0])
-
-        if 'pyobject' in self.options and 'lines' in self.options:
-            return [document.reporter.warning(
-                'Cannot use both "pyobject" and "lines" options',
-                line=self.lineno)]
-
-        if 'lineno-match' in self.options and 'lineno-start' in self.options:
-            return [document.reporter.warning(
-                'Cannot use both "lineno-match" and "lineno-start"',
-                line=self.lineno)]
-
-        if 'lineno-match' in self.options and \
-           (set(['append', 'prepend']) & set(self.options.keys())):
-            return [document.reporter.warning(
-                'Cannot use "lineno-match" and "append" or "prepend"',
-                line=self.lineno)]
-
-        if 'start-after' in self.options and 'start-at' in self.options:
-            return [document.reporter.warning(
-                'Cannot use both "start-after" and "start-at" options',
-                line=self.lineno)]
-
-        if 'end-before' in self.options and 'end-at' in self.options:
-            return [document.reporter.warning(
-                'Cannot use both "end-before" and "end-at" options',
-                line=self.lineno)]
-
-        encoding = self.options.get('encoding', env.config.source_encoding)
-        codec_info = codecs.lookup(encoding)
-
-        lines = self.read_with_encoding(filename, document,
-                                        codec_info, encoding)
-        if lines and not isinstance(lines[0], string_types):
-            return lines
-
-        diffsource = self.options.get('diff')
-        if diffsource is not None:
-            tmp, fulldiffsource = env.relfn2path(diffsource)
-
-            difflines = self.read_with_encoding(fulldiffsource, document,
-                                                codec_info, encoding)
-            if not isinstance(difflines[0], string_types):
-                return difflines
-            diff = unified_diff(
-                difflines,
-                lines,
-                diffsource,
-                self.arguments[0])
-            lines = list(diff)
-
-        linenostart = self.options.get('lineno-start', 1)
-        objectname = self.options.get('pyobject')
-        if objectname is not None:
-            from sphinx.pycode import ModuleAnalyzer
-            analyzer = ModuleAnalyzer.for_file(filename, '')
-            tags = analyzer.find_tags()
-            if objectname not in tags:
-                return [document.reporter.warning(
-                    'Object named %r not found in include file %r' %
-                    (objectname, filename), line=self.lineno)]
-            else:
-                lines = lines[tags[objectname][1] - 1: tags[objectname][2] - 1]
-                if 'lineno-match' in self.options:
-                    linenostart = tags[objectname][1]
-
-        linespec = self.options.get('lines')
-        if linespec:
-            try:
-                linelist = parselinenos(linespec, len(lines))
-            except ValueError as err:
-                return [document.reporter.warning(str(err), line=self.lineno)]
-
-            if 'lineno-match' in self.options:
-                # make sure the line list is not "disjoint".
-                previous = linelist[0]
-                for line_number in linelist[1:]:
-                    if line_number == previous + 1:
-                        previous = line_number
-                        continue
-                    return [document.reporter.warning(
-                        'Cannot use "lineno-match" with a disjoint set of '
-                        '"lines"', line=self.lineno)]
-                linenostart = linelist[0] + 1
-            # just ignore non-existing lines
-            lines = [lines[i] for i in linelist if i < len(lines)]
-            if not lines:
-                return [document.reporter.warning(
-                    'Line spec %r: no lines pulled from include file %r' %
-                    (linespec, filename), line=self.lineno)]
-
-        linespec = self.options.get('emphasize-lines')
-        if linespec:
-            try:
-                hl_lines = [x + 1 for x in parselinenos(linespec, len(lines))]
-            except ValueError as err:
-                return [document.reporter.warning(str(err), line=self.lineno)]
-        else:
-            hl_lines = None
-
-        start_str = self.options.get('start-after')
-        start_inclusive = False
-        if self.options.get('start-at') is not None:
-            start_str = self.options.get('start-at')
-            start_inclusive = True
-        end_str = self.options.get('end-before')
-        end_inclusive = False
-        if self.options.get('end-at') is not None:
-            end_str = self.options.get('end-at')
-            end_inclusive = True
-        if start_str is not None or end_str is not None:
-            use = not start_str
-            res = []
-            for line_number, line in enumerate(lines):
-                if not use and start_str and start_str in line:
-                    if 'lineno-match' in self.options:
-                        linenostart += line_number + 1
-                    use = True
-                    if start_inclusive:
-                        res.append(line)
-                elif use and end_str and end_str in line:
-                    if end_inclusive:
-                        res.append(line)
-                    break
-                elif use:
-                    res.append(line)
-            lines = res
-
-        prepend = self.options.get('prepend')
-        if prepend:
-            lines.insert(0, prepend + '\n')
-
-        append = self.options.get('append')
-        if append:
-            lines.append(append + '\n')
-
-        text = ''.join(lines)
-        if self.options.get('tab-width'):
-            text = text.expandtabs(self.options['tab-width'])
-        retnode = nodes.literal_block(text, text, source=filename)
-        set_source_info(self, retnode)
-        if diffsource:  # if diff is set, set udiff
-            retnode['language'] = 'udiff'
-        if 'language' in self.options:
-            retnode['language'] = self.options['language']
-        retnode['linenos'] = 'linenos' in self.options or \
-                             'lineno-start' in self.options or \
-                             'lineno-match' in self.options
-        retnode['classes'] += self.options.get('class', [])
-        extra_args = retnode['highlight_args'] = {}
-        if hl_lines is not None:
-            extra_args['hl_lines'] = hl_lines
-        extra_args['linenostart'] = linenostart
             env.note_dependency(rel_filename)
 
-        caption = self.options.get('caption')
+            reader = LiteralIncludeReader(filename, self.options, env.config)
-        if caption is not None:
+            text, lines = reader.read()
-            if not caption:
+
-                caption = self.arguments[0]
+            retnode = nodes.literal_block(text, text, source=filename)
-            try:
+            set_source_info(self, retnode)
+            if self.options.get('diff'):  # if diff is set, set udiff
+                retnode['language'] = 'udiff'
+            elif 'language' in self.options:
+                retnode['language'] = self.options['language']
+            retnode['linenos'] = ('linenos' in self.options or
+                                  'lineno-start' in self.options or
+                                  'lineno-match' in self.options)
+            retnode['classes'] += self.options.get('class', [])
+            extra_args = retnode['highlight_args'] = {}
+            if 'empahsize-lines' in self.options:
+                hl_lines = parselinenos(self.options['emphasize-lines'], lines)
+                extra_args['hl_lines'] = [x + 1 for x in hl_lines]
+            extra_args['linenostart'] = reader.lineno_start
+
+            if 'caption' in self.options:
+                caption = self.options['caption'] or self.arguments[0]
                 retnode = container_wrapper(self, retnode, caption)
-            except ValueError as exc:
-                document = self.state.document
-                errmsg = _('Invalid caption: %s' % exc[0][0].astext())  # type: ignore
-                return [document.reporter.warning(errmsg, line=self.lineno)]
 
             # retnode will be note_implicit_target that is linked from caption and numref.
             # when options['name'] is provided, it should be primary ID.
             self.add_name(retnode)
 
             return [retnode]
+        except Exception as exc:
+            return [document.reporter.warning(str(exc), line=self.lineno)]
 
 
 def setup(app):