1  import codecs
   2  import re
   3  from typing import (IO, Iterator, Match, NamedTuple, Optional,  # noqa:F401
   4                      Pattern, Sequence, Tuple)
   5  
   6  
   7  def make_regex(string: str, extra_flags: int = 0) -> Pattern[str]:
   8      return re.compile(string, re.UNICODE | extra_flags)
   9  
  10  
  11  _newline = make_regex(r"(\r\n|\n|\r)")
  12  _multiline_whitespace = make_regex(r"\s*", extra_flags=re.MULTILINE)
  13  _whitespace = make_regex(r"[^\S\r\n]*")
  14  _export = make_regex(r"(?:export[^\S\r\n]+)?")
  15  _single_quoted_key = make_regex(r"'([^']+)'")
  16  _unquoted_key = make_regex(r"([^=\#\s]+)")
  17  _equal_sign = make_regex(r"(=[^\S\r\n]*)")
  18  _single_quoted_value = make_regex(r"'((?:\\'|[^'])*)'")
  19  _double_quoted_value = make_regex(r'"((?:\\"|[^"])*)"')
  20  _unquoted_value = make_regex(r"([^\r\n]*)")
  21  _comment = make_regex(r"(?:[^\S\r\n]*#[^\r\n]*)?")
  22  _end_of_line = make_regex(r"[^\S\r\n]*(?:\r\n|\n|\r|$)")
  23  _rest_of_line = make_regex(r"[^\r\n]*(?:\r|\n|\r\n)?")
  24  _double_quote_escapes = make_regex(r"\\[\\'\"abfnrtv]")
  25  _single_quote_escapes = make_regex(r"\\[\\']")
  26  
  27  
  28  class Original(NamedTuple):
  29      string: str
  30      line: int
  31  
  32  
  33  class Binding(NamedTuple):
  34      key: Optional[str]
  35      value: Optional[str]
  36      original: Original
  37      error: bool
  38  
  39  
  40  class Position:
  41      def __init__(self, chars: int, line: int) -> None:
  42          self.chars = chars
  43          self.line = line
  44  
  45      @classmethod
  46      def start(cls) -> "Position":
  47          return cls(chars=0, line=1)
  48  
  49      def set(self, other: "Position") -> None:
  50          self.chars = other.chars
  51          self.line = other.line
  52  
  53      def advance(self, string: str) -> None:
  54          self.chars += len(string)
  55          self.line += len(re.findall(_newline, string))
  56  
  57  
  58  class Error(Exception):
  59      pass
  60  
  61  
  62  class Reader:
  63      def __init__(self, stream: IO[str]) -> None:
  64          self.string = stream.read()
  65          self.position = Position.start()
  66          self.mark = Position.start()
  67  
  68      def has_next(self) -> bool:
  69          return self.position.chars < len(self.string)
  70  
  71      def set_mark(self) -> None:
  72          self.mark.set(self.position)
  73  
  74      def get_marked(self) -> Original:
  75          return Original(
  76              string=self.string[self.mark.chars:self.position.chars],
  77              line=self.mark.line,
  78          )
  79  
  80      def peek(self, count: int) -> str:
  81          return self.string[self.position.chars:self.position.chars + count]
  82  
  83      def read(self, count: int) -> str:
  84          result = self.string[self.position.chars:self.position.chars + count]
  85          if len(result) < count:
  86              raise Error("read: End of string")
  87          self.position.advance(result)
  88          return result
  89  
  90      def read_regex(self, regex: Pattern[str]) -> Sequence[str]:
  91          match = regex.match(self.string, self.position.chars)
  92          if match is None:
  93              raise Error("read_regex: Pattern not found")
  94          self.position.advance(self.string[match.start():match.end()])
  95          return match.groups()
  96  
  97  
  98  def decode_escapes(regex: Pattern[str], string: str) -> str:
  99      def decode_match(match: Match[str]) -> str:
 100          return codecs.decode(match.group(0), 'unicode-escape')  # type: ignore
 101  
 102      return regex.sub(decode_match, string)
 103  
 104  
 105  def parse_key(reader: Reader) -> Optional[str]:
 106      char = reader.peek(1)
 107      if char == "#":
 108          return None
 109      elif char == "'":
 110          (key,) = reader.read_regex(_single_quoted_key)
 111      else:
 112          (key,) = reader.read_regex(_unquoted_key)
 113      return key
 114  
 115  
 116  def parse_unquoted_value(reader: Reader) -> str:
 117      (part,) = reader.read_regex(_unquoted_value)
 118      return re.sub(r"\s+#.*", "", part).rstrip()
 119  
 120  
 121  def parse_value(reader: Reader) -> str:
 122      char = reader.peek(1)
 123      if char == u"'":
 124          (value,) = reader.read_regex(_single_quoted_value)
 125          return decode_escapes(_single_quote_escapes, value)
 126      elif char == u'"':
 127          (value,) = reader.read_regex(_double_quoted_value)
 128          return decode_escapes(_double_quote_escapes, value)
 129      elif char in (u"", u"\n", u"\r"):
 130          return u""
 131      else:
 132          return parse_unquoted_value(reader)
 133  
 134  
 135  def parse_binding(reader: Reader) -> Binding:
 136      reader.set_mark()
 137      try:
 138          reader.read_regex(_multiline_whitespace)
 139          if not reader.has_next():
 140              return Binding(
 141                  key=None,
 142                  value=None,
 143                  original=reader.get_marked(),
 144                  error=False,
 145              )
 146          reader.read_regex(_export)
 147          key = parse_key(reader)
 148          reader.read_regex(_whitespace)
 149          if reader.peek(1) == "=":
 150              reader.read_regex(_equal_sign)
 151              value: Optional[str] = parse_value(reader)
 152          else:
 153              value = None
 154          reader.read_regex(_comment)
 155          reader.read_regex(_end_of_line)
 156          return Binding(
 157              key=key,
 158              value=value,
 159              original=reader.get_marked(),
 160              error=False,
 161          )
 162      except Error:
 163          reader.read_regex(_rest_of_line)
 164          return Binding(
 165              key=None,
 166              value=None,
 167              original=reader.get_marked(),
 168              error=True,
 169          )
 170  
 171  
 172  def parse_stream(stream: IO[str]) -> Iterator[Binding]:
 173      reader = Reader(stream)
 174      while reader.has_next():
 175          yield parse_binding(reader)
 176