text: #!/usr/bin/env python3
  # Copyright lowRISC contributors.
  # Licensed under the Apache License, Version 2.0, see LICENSE for details.
  # SPDX-License-Identifier: Apache-2.0
  
  import argparse
  import fnmatch
  import logging as log
  import os
  import re
  import shutil
  import subprocess
  import sys
  import tempfile
  import textwrap
  from pathlib import Path
  
  import hjson
  
  DESC = """vendor, copy source code from upstream into this repository"""
  
  EXCLUDE_ALWAYS = ['.git']
  
  LOCK_FILE_HEADER = """// Copyright lowRISC contributors.
  // Licensed under the Apache License, Version 2.0, see LICENSE for details.
  // SPDX-License-Identifier: Apache-2.0
  
  // This file is generated by the util/vendor script. Please do not modify it
  // manually.
  
  """
  
  verbose = False
  
  
  def git_is_clean_workdir(git_workdir):
      """Check if the git working directory is clean (no unstaged or staged changes)"""
      cmd = ['git', 'status', '--untracked-files=no', '--porcelain']
      modified_files = subprocess.run(cmd,
                                      cwd=str(git_workdir),
                                      check=True,
                                      stdout=subprocess.PIPE,
                                      stderr=subprocess.PIPE).stdout.strip()
      return not modified_files
  
  
  def path_resolve(path, base_dir=Path.cwd()):
      """Create an absolute path. Relative paths are resolved using base_dir as base."""
  
      if isinstance(path, str):
          path = Path(path)
  
      if path.is_absolute():
          return path
  
      return (base_dir / path).resolve()
  
  
  def github_qualify_references(log, repo_userorg, repo_name):
      """ Replace "unqualified" GitHub references with "fully qualified" one
  
      GitHub automatically links issues and pull requests if they have a specific
      format. Links can be qualified with the user/org name and the repository
      name, or unqualified, if they only contain the issue or pull request number.
  
      This function converts all unqualified references to qualified ones.
  
      See https://help.github.com/en/articles/autolinked-references-and-urls#issues-and-pull-requests
      for a documentation of all supported formats.
      """
  
      r = re.compile(r"(^|[^\w])(?:#|[gG][hH]-)(\d+)\b")
      repl_str = r'\1%s/%s#\2' % (repo_userorg, repo_name)
      return [r.sub(repl_str, l) for l in log]
  
  
  def test_github_qualify_references():
      repo_userorg = 'lowRISC'
      repo_name = 'ibex'
  
      # Unqualified references, should be replaced
      items_unqualified = [
          '#28',
          'GH-27',
          'klaus #27',
          'Fixes #27',
          'Fixes #27 and #28',
          '(#27)',
          'something (#27) done',
          '#27 and (GH-38)',
      ]
      exp_items_unqualified = [
          'lowRISC/ibex#28',
          'lowRISC/ibex#27',
          'klaus lowRISC/ibex#27',
          'Fixes lowRISC/ibex#27',
          'Fixes lowRISC/ibex#27 and lowRISC/ibex#28',
          '(lowRISC/ibex#27)',
          'something (lowRISC/ibex#27) done',
          'lowRISC/ibex#27 and (lowRISC/ibex#38)',
      ]
      assert github_qualify_references(items_unqualified, repo_userorg,
                                       repo_name) == exp_items_unqualified
  
      # Qualified references, should stay as they are
      items_qualified = [
          'Fixes lowrisc/ibex#27',
          'lowrisc/ibex#2',
      ]
      assert github_qualify_references(items_qualified, repo_userorg,
                                       repo_name) == items_qualified
  
      # Invalid references, should stay as they are
      items_invalid = [
          'something#27',
          'lowrisc/ibex#',
      ]
      assert github_qualify_references(items_invalid, repo_userorg,
                                       repo_name) == items_invalid
  
  
  def test_github_parse_url():
      assert github_parse_url('https://example.com/something/asdf.git') is None
      assert github_parse_url('https://github.com/lowRISC/ibex.git') == (
          'lowRISC', 'ibex')
      assert github_parse_url('https://github.com/lowRISC/ibex') == ('lowRISC',
                                                                     'ibex')
      assert github_parse_url('git@github.com:lowRISC/ibex.git') == ('lowRISC',
                                                                     'ibex')
  
  
  def github_parse_url(github_repo_url):
      """Parse a GitHub repository URL into its parts.
  
      Return a tuple (userorg, name), or None if the parsing failed.
      """
  
      regex = r"(?:@github\.com\:|\/github\.com\/)([a-zA-Z\d-]+)\/([a-zA-Z\d-]+)(?:\.git)?$"
      m = re.search(regex, github_repo_url)
      if m is None:
          return None
      return (m.group(1), m.group(2))
  
  
  def produce_shortlog(clone_dir, old_rev, new_rev):
      """ Produce a list of changes between two revisions, one revision per line
  
      Merges are excluded"""
      cmd = [
          'git', '-C',
          str(clone_dir), 'log', '--pretty=format:%s (%aN)', '--no-merges',
          old_rev + '..' + new_rev, '.'
      ]
      try:
          proc = subprocess.run(cmd,
                                cwd=str(clone_dir),
                                check=True,
                                stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE,
                                universal_newlines=True)
          return proc.stdout.splitlines()
      except subprocess.CalledProcessError as e:
          log.error("Unable to capture shortlog: %s", e.stderr)
          return ""
  
  
  def format_list_to_str(list, width=70):
      """ Create Markdown-style formatted string from a list of strings """
      wrapper = textwrap.TextWrapper(initial_indent="* ",
                                     subsequent_indent="  ",
                                     width=width)
      return '\n'.join([wrapper.fill(s) for s in list])
  
  
  def refresh_patches(desc):
      if not 'patch_repo' in desc:
          log.fatal('Unable to refresh patches, patch_repo not set in config.')
          sys.exit(1)
  
      patch_dir_abs = path_resolve(desc['patch_dir'], desc['_base_dir'])
      log.info('Refreshing patches in %s' % (str(patch_dir_abs), ))
  
      # remove existing patches
      for patch in patch_dir_abs.glob('*.patch'):
          os.unlink(str(patch))
  
      # get current patches
      _export_patches(desc['patch_repo']['url'], patch_dir_abs,
                      desc['patch_repo']['rev_base'],
                      desc['patch_repo']['rev_patched'])
  
  
  def _export_patches(patchrepo_clone_url, target_patch_dir, upstream_rev,
                      patched_rev):
      clone_dir = Path(tempfile.mkdtemp())
      try:
          clone_git_repo(patchrepo_clone_url, clone_dir, patched_rev)
          rev_range = 'origin/' + upstream_rev + '..' + 'origin/' + patched_rev
          cmd = ['git', 'format-patch', '-o', str(target_patch_dir.resolve()), rev_range]
          if not verbose:
              cmd += ['-q']
          subprocess.run(cmd, cwd=str(clone_dir), check=True)
  
      finally:
          shutil.rmtree(str(clone_dir), ignore_errors=True)
  
  
  def import_from_upstream(upstream_path, target_path, exclude_files=[]):
      log.info('Copying upstream sources to %s', target_path)
      # remove existing directories before importing them again
      shutil.rmtree(str(target_path), ignore_errors=True)
  
      # import new contents for rtl directory
      _cp_from_upstream(upstream_path, target_path, exclude_files)
  
  
  def apply_patch(basedir, patchfile, strip_level=1):
      cmd = ['git', 'apply', '-p' + str(strip_level), str(patchfile)]
      if verbose:
          cmd += ['--verbose']
      subprocess.run(cmd, cwd=str(basedir), check=True)
  
  
  def clone_git_repo(repo_url, clone_dir, rev='master'):
      log.info('Cloning upstream repository %s @ %s', repo_url, rev)
  
      # Clone the whole repository
      cmd = ['git', 'clone', '--no-single-branch']
      if not verbose:
          cmd += ['-q']
      cmd += [repo_url, str(clone_dir)]
      subprocess.run(cmd, check=True)
  
      # Check out exactly the revision requested
      cmd = ['git', '-C', str(clone_dir), 'reset', '--hard', rev]
      if not verbose:
          cmd += ['-q']
      subprocess.run(cmd, check=True)
  
      # Get revision information
      cmd = ['git', '-C', str(clone_dir), 'rev-parse', 'HEAD']
      rev = subprocess.run(cmd,
                           stdout=subprocess.PIPE,
                           stderr=subprocess.PIPE,
                           check=True,
                           universal_newlines=True).stdout.strip()
      log.info('Cloned at revision %s', rev)
      return rev
  
  
  def git_get_short_rev(clone_dir, rev):
      """ Get the shortened SHA-1 hash for a revision """
      cmd = ['git', '-C', str(clone_dir), 'rev-parse', '--short', rev]
      short_rev = subprocess.run(cmd,
                                 stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE,
                                 check=True,
                                 universal_newlines=True).stdout.strip()
      return short_rev
  
  
  def git_add_commit(repo_base, paths, commit_msg):
      """ Stage and commit all changes in paths"""
  
      # Stage all changes
      for p in paths:
          cmd_add = ['git', '-C', str(repo_base), 'add', str(p)]
          subprocess.run(cmd_add, check=True)
  
      cmd_commit = ['git', '-C', str(repo_base), 'commit', '-s', '-F', '-']
      try:
          subprocess.run(cmd_commit,
                         check=True,
                         universal_newlines=True,
                         input=commit_msg)
      except subprocess.CalledProcessError as e:
          log.warning("Unable to create commit. Are there no changes?")
  
  
  def ignore_patterns(base_dir, *patterns):
      """Similar to shutil.ignore_patterns, but with support for directory excludes."""
      def _rel_to_base(path, name):
          return os.path.relpath(os.path.join(path, name), base_dir)
  
      def _ignore_patterns(path, names):
          ignored_names = []
          for pattern in patterns:
              pattern_matches = [
                  n for n in names
                  if fnmatch.fnmatch(_rel_to_base(path, n), pattern)
              ]
              ignored_names.extend(pattern_matches)
          return set(ignored_names)
  
      return _ignore_patterns
  
  
  def _cp_from_upstream(src, dest, exclude=[]):
      shutil.copytree(str(src),
                      str(dest),
                      ignore=ignore_patterns(str(src), *exclude))
  
  
  def main(argv):
      parser = argparse.ArgumentParser(prog="vendor", description=DESC)
      parser.add_argument(
          '--update',
          '-U',
          dest='update',
          action='store_true',
          help='Update locked version of repository with upstream changes')
      parser.add_argument('--refresh-patches',
                          action='store_true',
                          help='Refresh the patches from the patch repository')
      parser.add_argument('--commit',
                          '-c',
                          action='store_true',
                          help='Commit the changes')
      parser.add_argument('desc_file',
                          metavar='file',
                          type=argparse.FileType('r', encoding='UTF-8'),
                          help='vendoring description file (*.vendor.hjson)')
      parser.add_argument('--verbose', '-v', action='store_true', help='Verbose')
      args = parser.parse_args()
  
      global verbose
      verbose = args.verbose
      if (verbose):
          log.basicConfig(format="%(levelname)s: %(message)s", level=log.DEBUG)
      else:
          log.basicConfig(format="%(levelname)s: %(message)s")
  
      desc_file_path = Path(args.desc_file.name).resolve()
      vendor_file_base_dir = desc_file_path.parent
  
      # Precondition: Ensure description file matches our naming rules
      if not str(desc_file_path).endswith('.vendor.hjson'):
          log.fatal("Description file names must have a .vendor.hjson suffix.")
          raise SystemExit(1)
  
      # Precondition: Check for a clean working directory when commit is requested
      if args.commit:
          if not git_is_clean_workdir(vendor_file_base_dir):
              log.fatal("A clean git working directory is required for "
                        "--commit/-c. git stash your changes and try again.")
              raise SystemExit(1)
  
      # Load description file
      try:
          desc = hjson.loads(args.desc_file.read(), use_decimal=True)
      except ValueError:
          raise SystemExit(sys.exc_info()[1])
      desc['_base_dir'] = vendor_file_base_dir
  
  
      desc_file_stem = desc_file_path.name.rsplit('.', 2)[0]
      lock_file_path = desc_file_path.with_name(desc_file_stem + '.lock.hjson')
  
      # Importing may use lock file upstream, information, so make a copy now
      # which we can overwrite with the upstream information from the lock file.
      import_desc = desc.copy()
  
      # Load lock file contents (if possible)
      try:
          with open(str(lock_file_path), 'r') as f:
              lock = hjson.loads(f.read(), use_decimal=True)
  
          # Use lock file information for import
          if not args.update:
              import_desc['upstream'] = lock['upstream'].copy()
      except FileNotFoundError:
          lock = None
          if not args.update:
              log.warning("Updating upstream repo as lock file %s not found.",
                          str(lock_file_path))
              args.update = True
  
      if args.refresh_patches:
          refresh_patches(import_desc)
  
      clone_dir = Path(tempfile.mkdtemp())
      try:
          # clone upstream repository
          upstream_new_rev = clone_git_repo(import_desc['upstream']['url'],
                                            clone_dir,
                                            rev=import_desc['upstream']['rev'])
  
          if not args.update:
              if upstream_new_rev != lock['upstream']['rev']:
                  log.fatal(
                      "Revision mismatch. Unable to re-clone locked version of repository."
                  )
                  log.fatal("Attempted revision: %s", import_desc['upstream']['rev'])
                  log.fatal("Re-cloned revision: %s", upstream_new_rev)
                  raise SystemExit(1)
  
          upstream_only_subdir = ''
          clone_subdir = clone_dir
          if 'only_subdir' in import_desc['upstream']:
              upstream_only_subdir = import_desc['upstream']['only_subdir']
              clone_subdir = clone_dir / upstream_only_subdir
              if not clone_subdir.is_dir():
                  log.fatal("subdir '%s' does not exist in repo",
                            upstream_only_subdir)
                  raise SystemExit(1)
  
          # apply patches to upstream sources
          if 'patch_dir' in import_desc:
              patches = path_resolve(import_desc['patch_dir'],
                                     vendor_file_base_dir).glob('*.patch')
              for patch in sorted(patches):
                  log.info("Applying patch %s" % str(patch))
                  apply_patch(clone_subdir, patch)
  
          # import selected (patched) files from upstream repo
          exclude_files = []
          if 'exclude_from_upstream' in import_desc:
              exclude_files += import_desc['exclude_from_upstream']
          exclude_files += EXCLUDE_ALWAYS
  
          import_from_upstream(
              clone_subdir, path_resolve(import_desc['target_dir'],
                                         vendor_file_base_dir), exclude_files)
  
          # get shortlog
          get_shortlog = bool(args.update)
          if lock is None:
              get_shortlog = False
              log.warning("No lock file %s: unable to summarize changes.", str(lock_file_path))
          elif lock['upstream']['url'] != import_desc['upstream']['url']:
              get_shortlog = False
              log.warning(
                  "The repository URL changed since the last run. Unable to get log of changes."
              )
  
          shortlog = None
          if get_shortlog:
              shortlog = produce_shortlog(clone_subdir, lock['upstream']['rev'],
                                          upstream_new_rev)
  
              # Ensure fully-qualified issue/PR references for GitHub repos
              gh_repo_info = github_parse_url(import_desc['upstream']['url'])
              if gh_repo_info:
                  shortlog = github_qualify_references(shortlog, gh_repo_info[0],
                                                       gh_repo_info[1])
  
              log.info("Changes since the last import:\n" +
                       format_list_to_str(shortlog))
  
          # write lock file
          if args.update:
              lock = {}
              lock['upstream'] = import_desc['upstream'].copy()
              lock['upstream']['rev'] = upstream_new_rev
              with open(str(lock_file_path), 'w', encoding='UTF-8') as f:
                  f.write(LOCK_FILE_HEADER)
                  hjson.dump(lock, f)
                  f.write("\n")
                  log.info("Wrote lock file %s", str(lock_file_path))
  
          # Commit changes
          if args.commit:
              sha_short = git_get_short_rev(clone_subdir, upstream_new_rev)
  
              repo_info = github_parse_url(import_desc['upstream']['url'])
              if repo_info is not None:
                  sha_short = "%s/%s@%s" % (repo_info[0], repo_info[1],
                                            sha_short)
  
              commit_msg_subject = 'Update %s to %s' % (import_desc['name'], sha_short)
              subdir_msg = ' '
              if upstream_only_subdir:
                  subdir_msg = ' subdir %s in ' % upstream_only_subdir
              intro = 'Update code from%supstream repository %s to revision %s' % (
                  subdir_msg, import_desc['upstream']['url'], upstream_new_rev)
              commit_msg_body = textwrap.fill(intro, width=70)
  
              if shortlog:
                  commit_msg_body += "\n\n"
                  commit_msg_body += format_list_to_str(shortlog, width=70)
  
              commit_msg = commit_msg_subject + "\n\n" + commit_msg_body
  
              commit_paths = []
              commit_paths.append(
                  path_resolve(import_desc['target_dir'], vendor_file_base_dir))
              if args.refresh_patches:
                  commit_paths.append(
                      path_resolve(import_desc['patch_dir'], vendor_file_base_dir))
              commit_paths.append(lock_file_path)
  
              git_add_commit(vendor_file_base_dir, commit_paths, commit_msg)
  
      finally:
          shutil.rmtree(str(clone_dir), ignore_errors=True)
  
      log.info('Import finished')
  
  
  if __name__ == '__main__':
      try:
          main(sys.argv)
      except subprocess.CalledProcessError as e:
          log.fatal("Called program '%s' returned with %d.\n"
                    "STDOUT:\n%s\n"
                    "STDERR:\n%s\n" %
                    (" ".join(e.cmd), e.returncode, e.stdout, e.stderr))
          raise
  