JSONCompressContext._options() - Code Metrics - Inspection of "ongoing extension refactor, favicons" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — 5.2-unstable ( 1c6b27 )

by Felipe A.

created 2016-12-09 13:40 UTC

JSONCompressContext._options() A

↳ Parent: JSONCompressContext

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	4
c	1
b	0
f	0
dl	0
loc	9
rs	9.2

import re

import jinja2
import jinja2.ext
import jinja2.lexer


class StateMachineCompressContext(object):
    '''
    Base non-compressing implementation for state-machine-based compression
    contexts compatible with :class:`CompressExtension`.
    '''
    re_whitespace = re.compile('[ \\t\\r\\n]+')
    token_class = jinja2.lexer.Token
    block_tokens = {
        'variable_begin': 'variable_end',
        'block_begin': 'block_end'
        }
    jumps = {}  # character-based state machine jumps
    current = None  # current state
    start = ''  # character which started current state
    pending = ''  # buffer of current state data
    lineno = 0  # current token lineno
    skip_until_token = None  # inside token until this is met

    def _minify(self, data, current, start, partial=False):
        return data

    def _options(self, value, current, start):
        offset = len(start)
        for mark, next in self.jumps[current].items():
            index = value.find(mark, offset)
            if -1 != index:
                yield index, mark, next
        yield len(value), '', None  # avoid value errors on empty min()

    def feed(self, token):
        if self.skip_until_token:
            yield token
            if token.type == self.skip_until_token:
                self.skip_until_token = None
            return

        if token.type in self.block_tokens:
            for data in self.finalize():
                yield data
            yield token
            self.skip_until_token = self.block_tokens[token.type]
            return

        size = len(token.value)
        lineno = token.lineno
        self.pending += token.value
        while True:
            index, mark, next = min(
                self._options(self.pending, self.current, self.start),
                key=lambda x: (x[0], -len(x[1]))
                )
            if next is None:
                break
            data = self._minify(self.pending[:index], self.current, self.start)
            self.lineno = lineno if size > len(self.pending) else self.lineno
            self.start = mark
            self.current = next
            self.pending = self.pending[index:]
            yield self.token_class(self.lineno, 'data', data)

    def finalize(self):
        if self.pending:
            data = self._minify(self.pending, self.current, self.start, True)
            yield self.token_class(self.lineno, 'data', data)
        self.start = ''
        self.pending = ''


class SGMLCompressContext(StateMachineCompressContext):
    block_tags = {}  # block content will be treated as literal text
    jumps = {  # state machine jumps
        'text': {
            '<': 'tag',
            '<!--': 'comment',
            '<![CDATA[': 'cdata',
            },
        'lit1': {'"': 'tag'},
        'lit2': ("'", 'tag'),
        'tag': {
            '>': 'text',
            '"': 'lit1',
            "'": 'lit2'
            },
        'comment': {'-->': 'text'},
        'cdata': {']]>': 'text'}
        }
    current = 'text'
    skip_until = None  # inside literal tag until this is met

    def _minify(self, data, current, start, partial=False):
        if current == 'tag':
            tagstart = start == '<'
            data = self.re_whitespace.sub(' ', data[1:] if tagstart else data)
            if tagstart:
                data = data.lstrip() if partial else data.strip()
                tagname = data.split(' ', 1)[0]
                self.skip_until = self.block_tags.get(tagname)
                return '<' + data
            elif partial:
                return data.rstrip()
            return start if data.strip() == start else data
        elif current == 'text':
            if not self.skip_until:
                return start if data.strip() == start else data
            elif not partial:
                self.skip_until = None
            return data
        return data

    def _options(self, value, current, start):
        offset = len(start)
        if self.skip_until and current == 'text':
            mark = self.skip_until
            index = value.find(mark, offset)
            if -1 != index:
                yield index, mark, current
        else:
            supa = super(SGMLCompressContext, self)
            for option in supa._options(value, current, start):
                yield option
        yield len(value), '', None  # avoid value errors on empty min()


class HTMLCompressContext(SGMLCompressContext):
    block_tags = {
        'textarea': '</textarea>',
        'pre': '</pre>',
        'script': '</script>',
        'style': '</style>',
        }


class JSONCompressContext(StateMachineCompressContext):
    jumps = {
        'object': {
            '"': 'string'
            },
        'string': {
            '"': 'object',
            '\\': 'escape'
            }
        }
    current = 'object'

    def _minify(self, data, current, start, partial=False):
        return data.replace(' ', '') if current == 'object' else data

    def _options(self, value, current, start):
        if current == 'escape':
            if value:
                yield 0, value[0], current
        else:
            supa = super(SGMLCompressContext, self)
            for option in supa._options(value, current, start):
                yield option
        yield len(value), '', None  # avoid value errors on empty min()


class CompressExtension(jinja2.ext.Extension):
    context_class = StateMachineCompressContext
    extensions = {}

    def supported(self, filename):
        print(filename or 'SIN FILENAME')
        return filename and filename.rsplit('.', 1)[-1] in self.extensions

    def supported_filter_stream(self, stream):
        feed = self.context_class()
        for token in stream:
            for data in feed.feed(token):
                yield data
        for data in feed.finalize():
            yield data

    def filter_stream(self, stream):
        if self.supported(stream.filename):
            return self.supported_filter_stream(stream)
        return stream


class HTMLCompress(CompressExtension):
    context_class = HTMLCompressContext
    extensions = {'html', 'xml'}


class JSONCompress(CompressExtension):
    context_class = JSONCompressContext
    extensions = {'json'}


1			import re
2
3			import jinja2
4			import jinja2.ext
5			import jinja2.lexer
6
7
8			class StateMachineCompressContext(object):
9			'''
10			Base non-compressing implementation for state-machine-based compression
11			contexts compatible with :class:`CompressExtension`.
12			'''
13			re_whitespace = re.compile('[ \\t\\r\\n]+')
14			token_class = jinja2.lexer.Token
15			block_tokens = {
16			'variable_begin': 'variable_end',
17			'block_begin': 'block_end'
18			}
19			jumps = {} # character-based state machine jumps
20			current = None # current state
21			start = '' # character which started current state
22			pending = '' # buffer of current state data
23			lineno = 0 # current token lineno
24			skip_until_token = None # inside token until this is met
25
26			def _minify(self, data, current, start, partial=False):
27			return data
28
29			def _options(self, value, current, start):
30			offset = len(start)
31			for mark, next in self.jumps[current].items():
32			index = value.find(mark, offset)
33			if -1 != index:
34			yield index, mark, next
35			yield len(value), '', None # avoid value errors on empty min()
36
37			def feed(self, token):
38			if self.skip_until_token:
39			yield token
40			if token.type == self.skip_until_token:
41			self.skip_until_token = None
42			return
43
44			if token.type in self.block_tokens:
45			for data in self.finalize():
46			yield data
47			yield token
48			self.skip_until_token = self.block_tokens[token.type]
49			return
50
51			size = len(token.value)
52			lineno = token.lineno
53			self.pending += token.value
54			while True:
55			index, mark, next = min(
56			self._options(self.pending, self.current, self.start),
57			key=lambda x: (x[0], -len(x[1]))
58			)
59			if next is None:
60			break
61			data = self._minify(self.pending[:index], self.current, self.start)
62			self.lineno = lineno if size > len(self.pending) else self.lineno
63			self.start = mark
64			self.current = next
65			self.pending = self.pending[index:]
66			yield self.token_class(self.lineno, 'data', data)
67
68			def finalize(self):
69			if self.pending:
70			data = self._minify(self.pending, self.current, self.start, True)
71			yield self.token_class(self.lineno, 'data', data)
72			self.start = ''
73			self.pending = ''
74
75
76			class SGMLCompressContext(StateMachineCompressContext):
77			block_tags = {} # block content will be treated as literal text
78			jumps = { # state machine jumps
79			'text': {
80			'<': 'tag',
81			'<!--': 'comment',
82			'<![CDATA[': 'cdata',
83			},
84			'lit1': {'"': 'tag'},
85			'lit2': ("'", 'tag'),
86			'tag': {
87			'>': 'text',
88			'"': 'lit1',
89			"'": 'lit2'
90			},
91			'comment': {'-->': 'text'},
92			'cdata': {']]>': 'text'}
93			}
94			current = 'text'
95			skip_until = None # inside literal tag until this is met
96
97			def _minify(self, data, current, start, partial=False):
98			if current == 'tag':
99			tagstart = start == '<'
100			data = self.re_whitespace.sub(' ', data[1:] if tagstart else data)
101			if tagstart:
102			data = data.lstrip() if partial else data.strip()
103			tagname = data.split(' ', 1)[0]
104			self.skip_until = self.block_tags.get(tagname)
105			return '<' + data
106			elif partial:
107			return data.rstrip()
108			return start if data.strip() == start else data
109			elif current == 'text':
110			if not self.skip_until:
111			return start if data.strip() == start else data
112			elif not partial:
113			self.skip_until = None
114			return data
115			return data
116
117			def _options(self, value, current, start):
118			offset = len(start)
119			if self.skip_until and current == 'text':
120			mark = self.skip_until
121			index = value.find(mark, offset)
122			if -1 != index:
123			yield index, mark, current
124			else:
125			supa = super(SGMLCompressContext, self)
126			for option in supa._options(value, current, start):
127			yield option
128			yield len(value), '', None # avoid value errors on empty min()
129
130
131			class HTMLCompressContext(SGMLCompressContext):
132			block_tags = {
133			'textarea': '</textarea>',
134			'pre': '</pre>',
135			'script': '</script>',
136			'style': '</style>',
137			}
138
139
140			class JSONCompressContext(StateMachineCompressContext):
141			jumps = {
142			'object': {
143			'"': 'string'
144			},
145			'string': {
146			'"': 'object',
147			'\\': 'escape'
148			}
149			}
150			current = 'object'
151
152			def _minify(self, data, current, start, partial=False):
153			return data.replace(' ', '') if current == 'object' else data
154
155			def _options(self, value, current, start):
156			if current == 'escape':
157			if value:
158			yield 0, value[0], current
159			else:
160			supa = super(SGMLCompressContext, self)
161			for option in supa._options(value, current, start):
162			yield option
163			yield len(value), '', None # avoid value errors on empty min()
164
165
166			class CompressExtension(jinja2.ext.Extension):
167			context_class = StateMachineCompressContext
168			extensions = {}
169
170			def supported(self, filename):
171			print(filename or 'SIN FILENAME')
172			return filename and filename.rsplit('.', 1)[-1] in self.extensions
173
174			def supported_filter_stream(self, stream):
175			feed = self.context_class()
176			for token in stream:
177			for data in feed.feed(token):
178			yield data
179			for data in feed.finalize():
180			yield data
181
182			def filter_stream(self, stream):
183			if self.supported(stream.filename):
184			return self.supported_filter_stream(stream)
185			return stream
186
187
188			class HTMLCompress(CompressExtension):
189			context_class = HTMLCompressContext
190			extensions = {'html', 'xml'}
191
192
193			class JSONCompress(CompressExtension):
194			context_class = JSONCompressContext
195			extensions = {'json'}
196

ergoithz / browsepy

Push — 5.2-unstable ( 1c6b27 )

JSONCompressContext._options() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like