mirror of https://github.com/morpheus65535/bazarr
81 lines
3.8 KiB
Python
81 lines
3.8 KiB
Python
_E='original'
|
|
_D='Binding'
|
|
_C='string'
|
|
_B='Original'
|
|
_A=None
|
|
import codecs,re
|
|
from.compat import IS_TYPE_CHECKING,to_text
|
|
if IS_TYPE_CHECKING:from typing import IO,Iterator,Match,NamedTuple,Optional,Pattern,Sequence,Text,Tuple
|
|
def make_regex(string,extra_flags=0):return re.compile(to_text(string),re.UNICODE|extra_flags)
|
|
_newline=make_regex('(\\r\\n|\\n|\\r)')
|
|
_multiline_whitespace=make_regex('\\s*',extra_flags=re.MULTILINE)
|
|
_whitespace=make_regex('[^\\S\\r\\n]*')
|
|
_export=make_regex('(?:export[^\\S\\r\\n]+)?')
|
|
_single_quoted_key=make_regex("'([^']+)'")
|
|
_unquoted_key=make_regex('([^=\\#\\s]+)')
|
|
_equal_sign=make_regex('(=[^\\S\\r\\n]*)')
|
|
_single_quoted_value=make_regex("'((?:\\\\'|[^'])*)'")
|
|
_double_quoted_value=make_regex('"((?:\\\\"|[^"])*)"')
|
|
_unquoted_value_part=make_regex('([^ \\r\\n]*)')
|
|
_comment=make_regex('(?:[^\\S\\r\\n]*#[^\\r\\n]*)?')
|
|
_end_of_line=make_regex('[^\\S\\r\\n]*(?:\\r\\n|\\n|\\r|$)')
|
|
_rest_of_line=make_regex('[^\\r\\n]*(?:\\r|\\n|\\r\\n)?')
|
|
_double_quote_escapes=make_regex('\\\\[\\\\\'\\"abfnrtv]')
|
|
_single_quote_escapes=make_regex("\\\\[\\\\']")
|
|
try:import typing;Original=typing.NamedTuple(_B,[(_C,typing.Text),('line',int)]);Binding=typing.NamedTuple(_D,[('key',typing.Optional[typing.Text]),('value',typing.Optional[typing.Text]),(_E,Original),('error',bool)])
|
|
except ImportError:from collections import namedtuple;Original=namedtuple(_B,[_C,'line']);Binding=namedtuple(_D,['key','value',_E,'error'])
|
|
class Position:
|
|
def __init__(A,chars,line):A.chars=chars;A.line=line
|
|
@classmethod
|
|
def start(A):return A(chars=0,line=1)
|
|
def set(A,other):B=other;A.chars=B.chars;A.line=B.line
|
|
def advance(A,string):B=string;A.chars+=len(B);A.line+=len(re.findall(_newline,B))
|
|
class Error(Exception):0
|
|
class Reader:
|
|
def __init__(A,stream):A.string=stream.read();A.position=Position.start();A.mark=Position.start()
|
|
def has_next(A):return A.position.chars<len(A.string)
|
|
def set_mark(A):A.mark.set(A.position)
|
|
def get_marked(A):return Original(string=A.string[A.mark.chars:A.position.chars],line=A.mark.line)
|
|
def peek(A,count):return A.string[A.position.chars:A.position.chars+count]
|
|
def read(A,count):
|
|
C=count;B=A.string[A.position.chars:A.position.chars+C]
|
|
if len(B)<C:raise Error('read: End of string')
|
|
A.position.advance(B);return B
|
|
def read_regex(A,regex):
|
|
B=regex.match(A.string,A.position.chars)
|
|
if B is _A:raise Error('read_regex: Pattern not found')
|
|
A.position.advance(A.string[B.start():B.end()]);return B.groups()
|
|
def decode_escapes(regex,string):
|
|
def A(match):return codecs.decode(match.group(0),'unicode-escape')
|
|
return regex.sub(A,string)
|
|
def parse_key(reader):
|
|
A=reader;B=A.peek(1)
|
|
if B=='#':return
|
|
elif B=="'":C,=A.read_regex(_single_quoted_key)
|
|
else:C,=A.read_regex(_unquoted_key)
|
|
return C
|
|
def parse_unquoted_value(reader):
|
|
A=reader;B=''
|
|
while True:
|
|
D,=A.read_regex(_unquoted_value_part);B+=D;C=A.peek(2)
|
|
if len(C)<2 or C[0]in'\r\n'or C[1]in' #\r\n':return B
|
|
B+=A.read(2)
|
|
def parse_value(reader):
|
|
A=reader;B=A.peek(1)
|
|
if B=="'":C,=A.read_regex(_single_quoted_value);return decode_escapes(_single_quote_escapes,C)
|
|
elif B=='"':C,=A.read_regex(_double_quoted_value);return decode_escapes(_double_quote_escapes,C)
|
|
elif B in('','\n','\r'):return''
|
|
else:return parse_unquoted_value(A)
|
|
def parse_binding(reader):
|
|
C=False;A=reader;A.set_mark()
|
|
try:
|
|
A.read_regex(_multiline_whitespace)
|
|
if not A.has_next():return Binding(key=_A,value=_A,original=A.get_marked(),error=C)
|
|
A.read_regex(_export);D=parse_key(A);A.read_regex(_whitespace)
|
|
if A.peek(1)=='=':A.read_regex(_equal_sign);B=parse_value(A)
|
|
else:B=_A
|
|
A.read_regex(_comment);A.read_regex(_end_of_line);return Binding(key=D,value=B,original=A.get_marked(),error=C)
|
|
except Error:A.read_regex(_rest_of_line);return Binding(key=_A,value=_A,original=A.get_marked(),error=True)
|
|
def parse_stream(stream):
|
|
A=Reader(stream)
|
|
while A.has_next():yield parse_binding(A) |