# Copyright 2004-2005 Elemental Security, Inc. All Rights Reserved.
# Licensed to PSF under a Contributor Agreement.

# Modifications:
# Copyright 2006 Google, Inc. All Rights Reserved.
# Licensed to PSF under a Contributor Agreement.
"""Parser driver.

This provides a high-level interface to parse a file into a syntax tree.

"""

__author__ = 'Guido van Rossum <guido@python.org>'

__all__ = ['Driver', 'load_grammar']

import io
import logging
import os
import pkgutil
import sys
# Python imports
from contextlib import contextmanager
from dataclasses import dataclass
from dataclasses import field
from pathlib import Path
from typing import Any
from typing import Iterator
from typing import List
from typing import Optional

from platformdirs import user_cache_dir

from yapf._version import __version__ as yapf_version

# Pgen imports
from . import grammar
from . import parse
from . import pgen
from . import token
from . import tokenize


@dataclass
class ReleaseRange:
  start: int
  end: Optional[int] = None
  tokens: List[Any] = field(default_factory=list)

  def lock(self) -> None:
    total_eaten = len(self.tokens)
    self.end = self.start + total_eaten


class TokenProxy:

  def __init__(self, generator: Any) -> None:
    self._tokens = generator
    self._counter = 0
    self._release_ranges: List[ReleaseRange] = []

  @contextmanager
  def release(self) -> Iterator['TokenProxy']:
    release_range = ReleaseRange(self._counter)
    self._release_ranges.append(release_range)
    try:
      yield self
    finally:
      # Lock the last release range to the final position that
      # has been eaten.
      release_range.lock()

  def eat(self, point: int) -> Any:
    eaten_tokens = self._release_ranges[-1].tokens
    if point < len(eaten_tokens):
      return eaten_tokens[point]
    else:
      while point >= len(eaten_tokens):
        token = next(self._tokens)
        eaten_tokens.append(token)
      return token

  def __iter__(self) -> 'TokenProxy':
    return self

  def __next__(self) -> Any:
    # If the current position is already compromised (looked up)
    # return the eaten token, if not just go further on the given
    # token producer.
    for release_range in self._release_ranges:
      assert release_range.end is not None

      start, end = release_range.start, release_range.end
      if start <= self._counter < end:
        token = release_range.tokens[self._counter - start]
        break
    else:
      token = next(self._tokens)
    self._counter += 1
    return token

  def can_advance(self, to: int) -> bool:
    # Try to eat, fail if it can't. The eat operation is cached
    # so there wont be any additional cost of eating here
    try:
      self.eat(to)
    except StopIteration:
      return False
    else:
      return True


class Driver(object):

  def __init__(self, grammar, convert=None, logger=None):
    self.grammar = grammar
    if logger is None:
      logger = logging.getLogger()
    self.logger = logger
    self.convert = convert

  def parse_tokens(self, tokens, debug=False):
    """Parse a series of tokens and return the syntax tree."""
    # XXX Move the prefix computation into a wrapper around tokenize.
    p = parse.Parser(self.grammar, self.convert)
    proxy = TokenProxy(tokens)
    p.setup(proxy=proxy)
    lineno = 1
    column = 0
    type = value = start = end = line_text = None
    prefix = ''
    for quintuple in proxy:
      type, value, start, end, line_text = quintuple
      if start != (lineno, column):
        assert (lineno, column) <= start, ((lineno, column), start)
        s_lineno, s_column = start
        if lineno < s_lineno:
          prefix += '\n' * (s_lineno - lineno)
          lineno = s_lineno
          column = 0
        if column < s_column:
          prefix += line_text[column:s_column]
          column = s_column
      if type in (tokenize.COMMENT, tokenize.NL):
        prefix += value
        lineno, column = end
        if value.endswith('\n'):
          lineno += 1
          column = 0
        continue
      if type == token.OP:
        type = grammar.opmap[value]
      if debug:
        self.logger.debug('%s %r (prefix=%r)', token.tok_name[type], value,
                          prefix)
      if p.addtoken(type, value, (prefix, start)):
        if debug:
          self.logger.debug('Stop.')
        break
      prefix = ''
      lineno, column = end
      if value.endswith('\n'):
        lineno += 1
        column = 0
    else:
      # We never broke out -- EOF is too soon (how can this happen???)
      raise parse.ParseError('incomplete input', type, value, (prefix, start))
    return p.rootnode

  def parse_stream_raw(self, stream, debug=False):
    """Parse a stream and return the syntax tree."""
    tokens = tokenize.generate_tokens(stream.readline)
    return self.parse_tokens(tokens, debug)

  def parse_stream(self, stream, debug=False):
    """Parse a stream and return the syntax tree."""
    return self.parse_stream_raw(stream, debug)

  def parse_file(self, filename, encoding=None, debug=False):
    """Parse a file and return the syntax tree."""
    with io.open(filename, 'r', encoding=encoding) as stream:
      return self.parse_stream(stream, debug)

  def parse_string(self, text, debug=False):
    """Parse a string and return the syntax tree."""
    tokens = tokenize.generate_tokens(io.StringIO(text).readline)
    return self.parse_tokens(tokens, debug)


def _generate_pickle_name(gt):
  # type:(str) -> str
  """Get the filepath to write a pickle file to
  given the path of a grammar textfile.

  The returned filepath should be in a user-specific cache directory.

  Args:
      gt (str): path to grammar text file

  Returns:
      str: path to pickle file
  """

  grammar_textfile_name = os.path.basename(gt)
  head, tail = os.path.splitext(grammar_textfile_name)
  if tail == '.txt':
    tail = ''
  cache_dir = user_cache_dir(
      appname='YAPF', appauthor='Google', version=yapf_version)
  return cache_dir + os.sep + head + tail + '-py' + '.'.join(
      map(str, sys.version_info)) + '.pickle'


def load_grammar(gt='Grammar.txt',
                 gp=None,
                 save=True,
                 force=False,
                 logger=None):
  # type:(str, str | None, bool, bool, logging.Logger | None) -> grammar.Grammar
  """Load the grammar (maybe from a pickle)."""
  if logger is None:
    logger = logging.getLogger()
  gp = _generate_pickle_name(gt) if gp is None else gp
  grammar_text = gt
  try:
    newer = _newer(gp, gt)
  except OSError as err:
    logger.debug('OSError, could not check if newer: %s', err.args)
    newer = True
  if not os.path.exists(gt):
    # Assume package data
    gt_basename = os.path.basename(gt)
    pd = pkgutil.get_data('yapf_third_party._ylib2to3', gt_basename)
    if pd is None:
      raise RuntimeError('Failed to load grammer %s from package' % gt_basename)
    grammar_text = io.StringIO(pd.decode(encoding='utf-8'))
  if force or not newer:
    g = pgen.generate_grammar(grammar_text)
    if save:
      try:
        Path(gp).parent.mkdir(parents=True, exist_ok=True)
        g.dump(gp)
      except OSError:
        # Ignore error, caching is not vital.
        pass
  else:
    g = grammar.Grammar()
    g.load(gp)
  return g


def _newer(a, b):
  """Inquire whether file a was written since file b."""
  if not os.path.exists(a):
    return False
  if not os.path.exists(b):
    return True
  return os.path.getmtime(a) >= os.path.getmtime(b)


def load_packaged_grammar(package, grammar_source):
  """Normally, loads a pickled grammar by doing
        pkgutil.get_data(package, pickled_grammar)
    where *pickled_grammar* is computed from *grammar_source* by adding the
    Python version and using a ``.pickle`` extension.

    However, if *grammar_source* is an extant file, load_grammar(grammar_source)
    is called instead. This facilitates using a packaged grammar file when needed
    but preserves load_grammar's automatic regeneration behavior when possible.

    """  # noqa: E501
  if os.path.isfile(grammar_source):
    return load_grammar(grammar_source)
  pickled_name = _generate_pickle_name(os.path.basename(grammar_source))
  data = pkgutil.get_data(package, pickled_name)
  g = grammar.Grammar()
  g.loads(data)
  return g


def main(*args):
  """Main program, when run as a script: produce grammar pickle files.

    Calls load_grammar for each argument, a path to a grammar text file.
    """
  if not args:
    args = sys.argv[1:]
  logging.basicConfig(
      level=logging.INFO, stream=sys.stdout, format='%(message)s')
  for gt in args:
    load_grammar(gt, save=True, force=True)
  return True


if __name__ == '__main__':
  sys.exit(int(not main()))