text: import os
  import sys
  from copy import copy
  from functools import partial
  from tempfile import mkstemp
  
  import attr
  import logging
  import json
  from pathlib2 import Path
  from threading import Thread, Event
  
  from .util import get_command_output, remove_user_pass_from_url
  from ....backend_api import Session
  from ....debugging import get_logger
  from .detectors import GitEnvDetector, GitDetector, HgEnvDetector, HgDetector, Result as DetectionResult
  
  _logger = get_logger("Repository Detection")
  
  
  class ScriptInfoError(Exception):
      pass
  
  
  class ScriptRequirements(object):
      _max_requirements_size = 512 * 1024
  
      def __init__(self, root_folder):
          self._root_folder = root_folder
  
      def get_requirements(self, entry_point_filename=None):
          # noinspection PyBroadException
          try:
              from ....utilities.pigar.reqs import get_installed_pkgs_detail
              from ....utilities.pigar.__main__ import GenerateReqs
              installed_pkgs = get_installed_pkgs_detail()
              gr = GenerateReqs(save_path='', project_path=self._root_folder, installed_pkgs=installed_pkgs,
                                ignores=['.git', '.hg', '.idea', '__pycache__', '.ipynb_checkpoints',
                                         'site-packages', 'dist-packages'])
              reqs, try_imports, guess, local_pks = gr.extract_reqs(
                  module_callback=ScriptRequirements.add_trains_used_packages, entry_point_filename=entry_point_filename)
              return self.create_requirements_txt(reqs, local_pks)
          except Exception:
              return '', ''
  
      @staticmethod
      def add_trains_used_packages(modules):
          # hack: forcefully insert storage modules if we have them
          # noinspection PyBroadException
          try:
              # noinspection PyPackageRequirements,PyUnresolvedReferences
              import boto3  # noqa: F401
              modules.add('boto3', 'trains.storage', 0)
          except Exception:
              pass
          # noinspection PyBroadException
          try:
              # noinspection PyPackageRequirements,PyUnresolvedReferences
              from google.cloud import storage  # noqa: F401
              modules.add('google_cloud_storage', 'trains.storage', 0)
          except Exception:
              pass
          # noinspection PyBroadException
          try:
              # noinspection PyPackageRequirements,PyUnresolvedReferences
              from azure.storage.blob import ContentSettings  # noqa: F401
              modules.add('azure_storage_blob', 'trains.storage', 0)
          except Exception:
              pass
  
          # bugfix, replace sklearn with scikit-learn name
          if 'sklearn' in modules:
              sklearn = modules.pop('sklearn', {})
              for fname, lines in sklearn.items():
                  modules.add('scikit_learn', fname, lines)
  
          # if we have torch and it supports tensorboard, we should add that as well
          # (because it will not be detected automatically)
          if 'torch' in modules and 'tensorboard' not in modules:
              # noinspection PyBroadException
              try:
                  # see if this version of torch support tensorboard
                  # noinspection PyPackageRequirements,PyUnresolvedReferences
                  import torch.utils.tensorboard  # noqa: F401
                  # noinspection PyPackageRequirements,PyUnresolvedReferences
                  import tensorboard  # noqa: F401
                  modules.add('tensorboard', 'torch', 0)
              except Exception:
                  pass
  
          # remove setuptools, we should not specify this module version. It is installed by default
          if 'setuptools' in modules:
              modules.pop('setuptools', {})
  
          # add forced requirements:
          # noinspection PyBroadException
          try:
              from ..task import Task
              # noinspection PyProtectedMember
              for package, version in Task._force_requirements.items():
                  modules.add(package, 'trains', 0)
          except Exception:
              pass
  
          return modules
  
      @staticmethod
      def create_requirements_txt(reqs, local_pks=None):
          # write requirements.txt
          # noinspection PyBroadException
          try:
              conda_requirements = ''
              conda_prefix = os.environ.get('CONDA_PREFIX')
              if conda_prefix and not conda_prefix.endswith(os.path.sep):
                  conda_prefix += os.path.sep
              if conda_prefix and sys.executable.startswith(conda_prefix):
                  conda_packages_json = get_command_output(['conda', 'list', '--json'])
                  conda_packages_json = json.loads(conda_packages_json)
                  reqs_lower = {k.lower(): (k, v) for k, v in reqs.items()}
                  for r in conda_packages_json:
                      # check if this is a pypi package, if it is, leave it outside
                      if not r.get('channel') or r.get('channel') == 'pypi':
                          continue
                      # check if we have it in our required packages
                      name = r['name'].lower().replace('-', '_')
                      # hack support pytorch/torch different naming convention
                      if name == 'pytorch':
                          name = 'torch'
                      k, v = reqs_lower.get(name, (None, None))
                      if k and v is not None:
                          conda_requirements += '{0} {1} {2}\n'.format(k, '==', v.version)
          except Exception:
              conda_requirements = ''
  
          # add forced requirements:
          # noinspection PyBroadException
          try:
              from ..task import Task
              # noinspection PyProtectedMember
              forced_packages = copy(Task._force_requirements)
          except Exception:
              forced_packages = {}
  
          # python version header
          requirements_txt = '# Python ' + sys.version.replace('\n', ' ').replace('\r', ' ') + '\n'
  
          if local_pks:
              requirements_txt += '\n# Local modules found - skipping:\n'
              for k, v in local_pks.sorted_items():
                  requirements_txt += '# {0} == {1}\n'.format(k, v.version)
  
          # requirement summary
          requirements_txt += '\n'
          for k, v in reqs.sorted_items():
              version = v.version
              if k in forced_packages:
                  forced_version = forced_packages.pop(k, None)
                  if forced_version:
                      version = forced_version
              # requirements_txt += ''.join(['# {0}\n'.format(c) for c in v.comments.sorted_items()])
              if k == '-e':
                  requirements_txt += '{0} {1}\n'.format(k, version)
              elif v:
                  requirements_txt += '{0} {1} {2}\n'.format(k, '==', version)
              else:
                  requirements_txt += '{0}\n'.format(k)
  
          # add forced requirements that we could not find installed on the system
          for k in sorted(forced_packages.keys()):
              if forced_packages[k]:
                  requirements_txt += '{0} {1} {2}\n'.format(k, '==', forced_packages[k])
              else:
                  requirements_txt += '{0}\n'.format(k)
  
          requirements_txt_packages_only = \
              requirements_txt + '\n# Skipping detailed import analysis, it is too large\n'
  
          # requirements details (in comments)
          requirements_txt += '\n' + \
                              '# Detailed import analysis\n' \
                              '# **************************\n'
  
          if local_pks:
              for k, v in local_pks.sorted_items():
                  requirements_txt += '\n'
                  requirements_txt += '# IMPORT LOCAL PACKAGE {0}\n'.format(k)
                  requirements_txt += ''.join(['# {0}\n'.format(c) for c in v.comments.sorted_items()])
  
          for k, v in reqs.sorted_items():
              requirements_txt += '\n'
              if k == '-e':
                  requirements_txt += '# IMPORT PACKAGE {0} {1}\n'.format(k, v.version)
              else:
                  requirements_txt += '# IMPORT PACKAGE {0}\n'.format(k)
              requirements_txt += ''.join(['# {0}\n'.format(c) for c in v.comments.sorted_items()])
  
          # make sure we do not exceed the size a size limit
          return (requirements_txt if len(requirements_txt) < ScriptRequirements._max_requirements_size
                  else requirements_txt_packages_only,
                  conda_requirements)
  
  
  class _JupyterObserver(object):
      _thread = None
      _exit_event = Event()
      _sync_event = Event()
      _sample_frequency = 30.
      _first_sample_frequency = 3.
      _jupyter_history_logger = None
  
      @classmethod
      def observer(cls, jupyter_notebook_filename, log_history):
          if cls._thread is not None:
              # order of signaling is important!
              cls._exit_event.set()
              cls._sync_event.set()
              cls._thread.join()
  
          if log_history and cls._jupyter_history_logger is None:
              cls._jupyter_history_logger = _JupyterHistoryLogger()
              cls._jupyter_history_logger.hook()
  
          cls._sync_event.clear()
          cls._exit_event.clear()
          cls._thread = Thread(target=cls._daemon, args=(jupyter_notebook_filename, ))
          cls._thread.daemon = True
          cls._thread.start()
  
      @classmethod
      def signal_sync(cls, *_, **__):
          cls._sync_event.set()
  
      @classmethod
      def close(cls):
          if not cls._thread:
              return
          cls._exit_event.set()
          cls._sync_event.set()
          cls._thread.join()
          cls._thread = None
  
      @classmethod
      def _daemon(cls, jupyter_notebook_filename):
          from trains import Task
  
          # load jupyter notebook package
          # noinspection PyBroadException
          try:
              # noinspection PyPackageRequirements
              from nbconvert.exporters.script import ScriptExporter
              _script_exporter = ScriptExporter()
          except Exception:
              return
          # load pigar
          # noinspection PyBroadException
          try:
              from ....utilities.pigar.reqs import get_installed_pkgs_detail, file_import_modules
              from ....utilities.pigar.modules import ReqsModules
              from ....utilities.pigar.log import logger
              logger.setLevel(logging.WARNING)
          except Exception:
              file_import_modules = None
          # load IPython
          # noinspection PyBroadException
          try:
              # noinspection PyPackageRequirements
              from IPython import get_ipython
          except Exception:
              # should not happen
              get_ipython = None
  
          # setup local notebook files
          if jupyter_notebook_filename:
              notebook = Path(jupyter_notebook_filename)
              local_jupyter_filename = jupyter_notebook_filename
          else:
              notebook = None
              fd, local_jupyter_filename = mkstemp(suffix='.ipynb')
              os.close(fd)
          last_update_ts = None
          counter = 0
          prev_script_hash = None
  
          # noinspection PyBroadException
          try:
              from ....version import __version__
              our_module = cls.__module__.split('.')[0], __version__
          except Exception:
              our_module = None
  
          # noinspection PyBroadException
          try:
              import re
              replace_ipython_pattern = re.compile(r'\n([ \t]*)get_ipython\(\)')
          except Exception:
              replace_ipython_pattern = None
  
          # main observer loop, check if we need to exit
          while not cls._exit_event.wait(timeout=0.):
              # wait for timeout or sync event
              cls._sync_event.wait(cls._sample_frequency if counter else cls._first_sample_frequency)
  
              cls._sync_event.clear()
              counter += 1
              # noinspection PyBroadException
              try:
                  # if there is no task connected, do nothing
                  task = Task.current_task()
                  if not task:
                      continue
  
                  script_code = None
                  fmodules = None
                  current_cell = None
                  # if we have a local file:
                  if notebook:
                      if not notebook.exists():
                          continue
                      # check if notebook changed
                      if last_update_ts is not None and notebook.stat().st_mtime - last_update_ts <= 0:
                          continue
                      last_update_ts = notebook.stat().st_mtime
                  else:
                      # serialize notebook to a temp file
                      if cls._jupyter_history_logger:
                          script_code, current_cell = cls._jupyter_history_logger.history_to_str()
                      else:
                          # noinspection PyBroadException
                          try:
                              # noinspection PyBroadException
                              try:
                                  os.unlink(local_jupyter_filename)
                              except Exception:
                                  pass
                              get_ipython().run_line_magic('history', '-t -f {}'.format(local_jupyter_filename))
                              with open(local_jupyter_filename, 'r') as f:
                                  script_code = f.read()
                              # load the modules
                              from ....utilities.pigar.modules import ImportedModules
                              fmodules = ImportedModules()
                              for nm in set([str(m).split('.')[0] for m in sys.modules]):
                                  fmodules.add(nm, 'notebook', 0)
                          except Exception:
                              continue
  
                  # get notebook python script
                  if script_code is None:
                      script_code, _ = _script_exporter.from_filename(local_jupyter_filename)
  
                  current_script_hash = hash(script_code + (current_cell or ''))
                  if prev_script_hash and prev_script_hash == current_script_hash:
                      continue
  
                  # remove ipython direct access from the script code
                  # we will not be able to run them anyhow
                  if replace_ipython_pattern:
                      script_code = replace_ipython_pattern.sub(r'\n# \g<1>get_ipython()', script_code)
  
                  requirements_txt = ''
                  conda_requirements = ''
                  # parse jupyter python script and prepare pip requirements (pigar)
                  # if backend supports requirements
                  if file_import_modules and Session.check_min_api_version('2.2'):
                      if fmodules is None:
                          fmodules, _ = file_import_modules(
                              notebook.parts[-1] if notebook else 'notebook', script_code)
                          if current_cell:
                              cell_fmodules, _ = file_import_modules(
                                  notebook.parts[-1] if notebook else 'notebook', current_cell)
                              # noinspection PyBroadException
                              try:
                                  fmodules |= cell_fmodules
                              except Exception:
                                  pass
                      # add current cell to the script
                      if current_cell:
                          script_code += '\n' + current_cell
                      fmodules = ScriptRequirements.add_trains_used_packages(fmodules)
                      # noinspection PyUnboundLocalVariable
                      installed_pkgs = get_installed_pkgs_detail()
                      # make sure we are in installed packages
                      if our_module and (our_module[0] not in installed_pkgs):
                          installed_pkgs[our_module[0]] = our_module
  
                      # noinspection PyUnboundLocalVariable
                      reqs = ReqsModules()
                      for name in fmodules:
                          if name in installed_pkgs:
                              pkg_name, version = installed_pkgs[name]
                              reqs.add(pkg_name, version, fmodules[name])
                      requirements_txt, conda_requirements = ScriptRequirements.create_requirements_txt(reqs)
  
                  # update script
                  prev_script_hash = current_script_hash
                  data_script = task.data.script
                  data_script.diff = script_code
                  data_script.requirements = {'pip': requirements_txt, 'conda': conda_requirements}
                  # noinspection PyProtectedMember
                  task._update_script(script=data_script)
                  # update requirements
                  # noinspection PyProtectedMember
                  task._update_requirements(requirements=requirements_txt)
              except Exception:
                  pass
  
  
  class ScriptInfo(object):
      max_diff_size_bytes = 500000
  
      plugins = [GitEnvDetector(), HgEnvDetector(), HgDetector(), GitDetector()]
      """ Script info detection plugins, in order of priority """
  
      @classmethod
      def _jupyter_install_post_store_hook(cls, jupyter_notebook_filename, log_history=False):
          # noinspection PyBroadException
          try:
              if 'IPython' in sys.modules:
                  # noinspection PyPackageRequirements
                  from IPython import get_ipython
                  if get_ipython():
                      _JupyterObserver.observer(jupyter_notebook_filename, log_history)
                      get_ipython().events.register('pre_run_cell', _JupyterObserver.signal_sync)
                      if log_history:
                          get_ipython().events.register('post_run_cell', _JupyterObserver.signal_sync)
          except Exception:
              pass
  
      @classmethod
      def _get_jupyter_notebook_filename(cls):
          if not (sys.argv[0].endswith(os.path.sep + 'ipykernel_launcher.py') or
                  sys.argv[0].endswith(os.path.join(os.path.sep, 'ipykernel', '__main__.py'))) \
                  or len(sys.argv) < 3 or not sys.argv[2].endswith('.json'):
              return None
  
          # we can safely assume that we can import the notebook package here
          # noinspection PyBroadException
          try:
              # noinspection PyPackageRequirements
              from notebook.notebookapp import list_running_servers
              import requests
              current_kernel = sys.argv[2].split(os.path.sep)[-1].replace('kernel-', '').replace('.json', '')
              # noinspection PyBroadException
              try:
                  server_info = next(list_running_servers())
              except Exception:
                  # on some jupyter notebook versions this function can crash on parsing the json file,
                  # we will parse it manually here
                  # noinspection PyPackageRequirements
                  import ipykernel
                  from glob import glob
                  import json
                  for f in glob(os.path.join(os.path.dirname(ipykernel.get_connection_file()), 'nbserver-*.json')):
                      # noinspection PyBroadException
                      try:
                          with open(f, 'r') as json_data:
                              server_info = json.load(json_data)
                      except Exception:
                          server_info = None
                      if server_info:
                          break
              try:
                  r = requests.get(
                      url=server_info['url'] + 'api/sessions',
                      headers={'Authorization': 'token {}'.format(server_info.get('token', '')), })
              except requests.exceptions.SSLError:
                  # disable SSL check warning
                  from urllib3.exceptions import InsecureRequestWarning
                  # noinspection PyUnresolvedReferences
                  requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)
                  # fire request
                  r = requests.get(
                      url=server_info['url'] + 'api/sessions',
                      headers={'Authorization': 'token {}'.format(server_info.get('token', '')), }, verify=False)
                  # enable SSL check warning
                  import warnings
                  warnings.simplefilter('default', InsecureRequestWarning)
  
              r.raise_for_status()
              notebooks = r.json()
  
              cur_notebook = None
              for n in notebooks:
                  if n['kernel']['id'] == current_kernel:
                      cur_notebook = n
                      break
  
              notebook_path = cur_notebook['notebook'].get('path', '')
              notebook_name = cur_notebook['notebook'].get('name', '')
  
              is_google_colab = False
              # check if this is google.colab, then there is no local file
              # noinspection PyBroadException
              try:
                  # noinspection PyPackageRequirements
                  from IPython import get_ipython
                  if get_ipython() and 'google.colab' in get_ipython().extension_manager.loaded:
                      is_google_colab = True
              except Exception:
                  pass
  
              if is_google_colab:
                  script_entry_point = str(notebook_name or 'notebook').replace(
                      '>', '_').replace('<', '_').replace('.ipynb', '.py')
                  if not script_entry_point.lower().endswith('.py'):
                      script_entry_point += '.py'
                  local_ipynb_file = None
              else:
                  # always slash, because this is from uri (so never backslash not even oon windows)
                  entry_point_filename = notebook_path.split('/')[-1]
  
                  # now we should try to find the actual file
                  entry_point = (Path.cwd() / entry_point_filename).absolute()
                  if not entry_point.is_file():
                      entry_point = (Path.cwd() / notebook_path).absolute()
  
                  # get local ipynb for observer
                  local_ipynb_file = entry_point.as_posix()
  
                  # now replace the .ipynb with .py
                  # we assume we will have that file available with the Jupyter notebook plugin
                  entry_point = entry_point.with_suffix('.py')
  
                  script_entry_point = entry_point.as_posix()
  
              # install the post store hook,
              # notice that if we do not have a local file we serialize/write every time the entire notebook
              cls._jupyter_install_post_store_hook(local_ipynb_file, is_google_colab)
  
              return script_entry_point
          except Exception:
              return None
  
      @classmethod
      def _get_entry_point(cls, repo_root, script_path):
          repo_root = Path(repo_root).absolute()
  
          try:
              # Use os.path.relpath as it calculates up dir movements (../)
              entry_point = os.path.relpath(str(script_path), str(Path.cwd()))
          except ValueError:
              # Working directory not under repository root
              entry_point = script_path.relative_to(repo_root)
  
          return Path(entry_point).as_posix()
  
      @classmethod
      def _get_working_dir(cls, repo_root):
          repo_root = Path(repo_root).absolute()
  
          try:
              return Path.cwd().relative_to(repo_root).as_posix()
          except ValueError:
              # Working directory not under repository root
              return os.path.curdir
  
      @classmethod
      def _get_script_code(cls, script_path):
          # noinspection PyBroadException
          try:
              with open(script_path, 'r') as f:
                  script_code = f.read()
              return script_code
          except Exception:
              pass
          return ''
  
      @classmethod
      def _get_script_info(cls, filepaths, check_uncommitted=True, create_requirements=True, log=None):
          jupyter_filepath = cls._get_jupyter_notebook_filename()
          if jupyter_filepath:
              scripts_path = [Path(os.path.normpath(jupyter_filepath)).absolute()]
          else:
              scripts_path = [Path(os.path.normpath(f)).absolute() for f in filepaths if f]
              if all(not f.is_file() for f in scripts_path):
                  raise ScriptInfoError(
                      "Script file {} could not be found".format(scripts_path)
                  )
  
          scripts_dir = [f.parent for f in scripts_path]
  
          def _log(msg, *args, **kwargs):
              if not log:
                  return
              log.warning(
                  "Failed auto-detecting task repository: {}".format(
                      msg.format(*args, **kwargs)
                  )
              )
  
          plugin = next((p for p in cls.plugins if any(p.exists(d) for d in scripts_dir)), None)
          repo_info = DetectionResult()
          script_dir = scripts_dir[0]
          script_path = scripts_path[0]
          messages = []
          auxiliary_git_diff = None
  
          if not plugin:
              log.info("No repository found, storing script code instead")
          else:
              try:
                  for i, d in enumerate(scripts_dir):
                      repo_info = plugin.get_info(str(d), include_diff=check_uncommitted)
                      if not repo_info.is_empty():
                          script_dir = d
                          script_path = scripts_path[i]
                          break
              except Exception as ex:
                  _log("no info for {} ({})", scripts_dir, ex)
              else:
                  if repo_info.is_empty():
                      _log("no info for {}", scripts_dir)
  
          repo_root = repo_info.root or script_dir
          if not plugin:
              working_dir = '.'
              entry_point = str(script_path.name)
          else:
              working_dir = cls._get_working_dir(repo_root)
              entry_point = cls._get_entry_point(repo_root, script_path)
  
          if check_uncommitted:
              diff = cls._get_script_code(script_path.as_posix()) \
                  if not plugin or not repo_info.commit else repo_info.diff
              # make sure diff is not too big:
              if len(diff) > cls.max_diff_size_bytes:
                  messages.append(
                      "======> WARNING! Git diff to large to store "
                      "({}kb), skipping uncommitted changes <======".format(len(diff)//1024))
                  auxiliary_git_diff = diff
                  diff = '# WARNING! git diff too large to store, clear this section to execute without it.\n' \
                         '# full git diff available in Artifacts/auxiliary_git_diff\n' \
                         '# Clear the section before enqueueing Task!\n'
  
          else:
              diff = ''
          # if this is not jupyter, get the requirements.txt
          requirements = ''
          conda_requirements = ''
          # create requirements if backend supports requirements
          # if jupyter is present, requirements will be created in the background, when saving a snapshot
          if not jupyter_filepath and Session.check_min_api_version('2.2'):
              script_requirements = ScriptRequirements(
                  Path(repo_root).as_posix() if repo_info.url else script_path.as_posix())
              if create_requirements:
                  requirements, conda_requirements = script_requirements.get_requirements()
          else:
              script_requirements = None
  
          script_info = dict(
              repository=remove_user_pass_from_url(repo_info.url),
              branch=repo_info.branch,
              version_num=repo_info.commit,
              entry_point=entry_point,
              working_dir=working_dir,
              diff=diff,
              requirements={'pip': requirements, 'conda': conda_requirements} if requirements else None,
              binary='python{}.{}'.format(sys.version_info.major, sys.version_info.minor),
              repo_root=repo_root,
              jupyter_filepath=jupyter_filepath,
          )
  
          if repo_info.modified:
              messages.append(
                  "======> WARNING! UNCOMMITTED CHANGES IN REPOSITORY {} <======".format(
                      script_info.get("repository", "")
                  )
              )
  
          if not any(script_info.values()):
              script_info = None
  
          return (ScriptInfoResult(script=script_info, warning_messages=messages, auxiliary_git_diff=auxiliary_git_diff),
                  script_requirements)
  
      @classmethod
      def get(cls, filepaths=None, check_uncommitted=True, create_requirements=True, log=None):
          try:
              if not filepaths:
                  filepaths = [sys.argv[0], ]
              return cls._get_script_info(
                  filepaths=filepaths, check_uncommitted=check_uncommitted,
                  create_requirements=create_requirements, log=log)
          except Exception as ex:
              if log:
                  log.warning("Failed auto-detecting task repository: {}".format(ex))
          return ScriptInfoResult(), None
  
      @classmethod
      def is_running_from_module(cls):
          # noinspection PyBroadException
          try:
              return '__main__' in sys.modules and vars(sys.modules['__main__'])['__package__']
          except Exception:
              return False
  
      @classmethod
      def detect_running_module(cls, script_dict):
          # noinspection PyBroadException
          try:
              # If this is jupyter, do not try to detect the running module, we know what we have.
              if script_dict.get('jupyter_filepath'):
                  return script_dict
  
              if cls.is_running_from_module():
                  argvs = ''
                  git_root = os.path.abspath(script_dict['repo_root']) if script_dict['repo_root'] else None
                  for a in sys.argv[1:]:
                      if git_root and os.path.exists(a):
                          # check if common to project:
                          a_abs = os.path.abspath(a)
                          if os.path.commonpath([a_abs, git_root]) == git_root:
                              # adjust path relative to working dir inside git repo
                              a = ' ' + os.path.relpath(a_abs, os.path.join(git_root, script_dict['working_dir']))
                      argvs += ' {}'.format(a)
                  # update the script entry point to match the real argv and module call
                  script_dict['entry_point'] = '-m {}{}'.format(
                      vars(sys.modules['__main__'])['__package__'], (' ' + argvs) if argvs else '')
          except Exception:
              pass
          return script_dict
  
      @classmethod
      def close(cls):
          _JupyterObserver.close()
  
  
  @attr.s
  class ScriptInfoResult(object):
      script = attr.ib(default=None)
      warning_messages = attr.ib(factory=list)
      auxiliary_git_diff = attr.ib(default=None)
  
  
  class _JupyterHistoryLogger(object):
      _reg_replace_ipython = r'\n([ \t]*)get_ipython\(\)'
      _reg_replace_magic = r'\n([ \t]*)%'
      _reg_replace_bang = r'\n([ \t]*)!'
  
      def __init__(self):
          self._exception_raised = False
          self._cells_code = {}
          self._counter = 0
          self._ip = None
          self._current_cell = None
          # noinspection PyBroadException
          try:
              import re
              self._replace_ipython_pattern = re.compile(self._reg_replace_ipython)
              self._replace_magic_pattern = re.compile(self._reg_replace_magic)
              self._replace_bang_pattern = re.compile(self._reg_replace_bang)
          except Exception:
              self._replace_ipython_pattern = None
              self._replace_magic_pattern = None
              self._replace_bang_pattern = None
  
      def hook(self, ip=None):
          if not ip:
              # noinspection PyBroadException
              try:
                  # noinspection PyPackageRequirements
                  from IPython import get_ipython
              except Exception:
                  return
              self._ip = get_ipython()
          else:
              self._ip = ip
  
          # noinspection PyBroadException
          try:
              # if this is colab, the callbacks do not contain the raw_cell content, so we have to patch it
              if 'google.colab' in self._ip.extension_manager.loaded:
                  self._ip._org_run_cell = self._ip.run_cell
                  self._ip.run_cell = partial(self._patched_run_cell, self._ip)
          except Exception:
              pass
  
          # start with the current history
          self._initialize_history()
          self._ip.events.register('post_run_cell', self._post_cell_callback)
          self._ip.events.register('pre_run_cell', self._pre_cell_callback)
          self._ip.set_custom_exc((Exception,), self._exception_callback)
  
      def _patched_run_cell(self, shell, *args, **kwargs):
          # noinspection PyBroadException
          try:
              raw_cell = kwargs.get('raw_cell') or args[0]
              self._current_cell = raw_cell
          except Exception:
              pass
          # noinspection PyProtectedMember
          return shell._org_run_cell(*args, **kwargs)
  
      def history(self, filename):
          with open(filename, 'wt') as f:
              for k, v in sorted(self._cells_code.items(), key=lambda p: p[0]):
                  f.write(v)
  
      def history_to_str(self):
          # return a pair: (history as str, current cell if we are in still in cell execution otherwise None)
          return '\n'.join(v for k, v in sorted(self._cells_code.items(), key=lambda p: p[0])), self._current_cell
  
      # noinspection PyUnusedLocal
      def _exception_callback(self, shell, etype, value, tb, tb_offset=None):
          self._exception_raised = True
          return shell.showtraceback()
  
      def _pre_cell_callback(self, *args, **_):
          # noinspection PyBroadException
          try:
              if args:
                  self._current_cell = args[0].raw_cell
              # we might have this value from somewhere else
              if self._current_cell:
                  self._current_cell = self._conform_code(self._current_cell, replace_magic_bang=True)
          except Exception:
              pass
  
      def _post_cell_callback(self, *_, **__):
          # noinspection PyBroadException
          try:
              self._current_cell = None
              if self._exception_raised:
                  # do nothing
                  self._exception_raised = False
                  return
  
              self._exception_raised = False
              # add the cell history
              # noinspection PyBroadException
              try:
                  cell_code = '\n' + self._ip.history_manager.input_hist_parsed[-1]
              except Exception:
                  return
  
              # fix magic / bang in code
              cell_code = self._conform_code(cell_code)
  
              self._cells_code[self._counter] = cell_code
              self._counter += 1
          except Exception:
              pass
  
      def _initialize_history(self):
          # only once
          if -1 in self._cells_code:
              return
          # noinspection PyBroadException
          try:
              cell_code = '\n' + '\n'.join(self._ip.history_manager.input_hist_parsed[:-1])
          except Exception:
              return
  
          cell_code = self._conform_code(cell_code)
          self._cells_code[-1] = cell_code
  
      def _conform_code(self, cell_code, replace_magic_bang=False):
          # fix magic / bang in code
          if self._replace_ipython_pattern:
              cell_code = self._replace_ipython_pattern.sub(r'\n# \g<1>get_ipython()', cell_code)
          if replace_magic_bang and self._replace_magic_pattern and self._replace_bang_pattern:
              cell_code = self._replace_magic_pattern.sub(r'\n# \g<1>%', cell_code)
              cell_code = self._replace_bang_pattern.sub(r'\n# \g<1>!', cell_code)
          return cell_code
  