


import collections
import json
import math
import os
import re
import time
from fnmatch import fnmatch
from typing import Dict

import requests
from slack_sdk import WebClient


snake_case__				:	List[Any]					       =					WebClient(token=os.environ['CI_SLACK_BOT_TOKEN'])


def    __lowerCamelCase   (      A__						: int					)     ->    Union[str, Any]:
   lowerCamelCase_						:   Dict		= test_results.split(""" """					)

   lowerCamelCase_						:   int		= 0
   lowerCamelCase_						:   Tuple		= 0

   # When the output is short enough, the output is surrounded by = signs: "== OUTPUT =="
   # When it is too long, those signs are not present.
   lowerCamelCase_						:   List[str]		= expressions[-2] if """=""" in expressions[-1] else expressions[-1]

   for i, expression in enumerate(A__					):
      if "failed" in expression:
         failed += int(expressions[i - 1]					)
      if "passed" in expression:
         success += int(expressions[i - 1]					)

   return failed, success, time_spent


def    __lowerCamelCase   (      A__						: Tuple					)     ->    int:
   lowerCamelCase_						:   List[str]		= {}
   lowerCamelCase_						:   Optional[Any]		= None
   lowerCamelCase_						:   Tuple		= False
   for line in failures_short_lines.split("""\n"""					):
      if re.search(R"""_ \[doctest\]"""						,						A__					):
         lowerCamelCase_						:   Union[str, Any]		= True
         lowerCamelCase_						:   Any		= line.split(""" """					)[2]
      elif in_error and not line.split(""" """					)[0].isdigit():
         lowerCamelCase_						:   List[Any]		= line
         lowerCamelCase_						:   Union[str, Any]		= False

   return failures





class   SCREAMING_SNAKE_CASE_			:




       '''simple docstring'''



       def __init__(						self				:				Dict		,				__a				:				str		,				__a				:				Dict							)					->Tuple:
          lowerCamelCase_						:   Optional[int]		= title

          lowerCamelCase_						:   List[str]		= doc_test_results["""time_spent"""].split(""","""							)[0]
          lowerCamelCase_						:   List[str]		= doc_test_results["""success"""]
          lowerCamelCase_						:   Tuple		= doc_test_results["""failures"""]
          lowerCamelCase_						:   Optional[Any]		= self.n_success + self.n_failures

          # Failures and success of the modeling tests
          lowerCamelCase_						:   Dict		= doc_test_results



       @property
       def 				_lowerCAmelCase							(						self				:				str							)					->str:
          lowerCamelCase_						:   List[Any]		= [self._time_spent]
          lowerCamelCase_						:   Optional[Any]		= 0

          for time in time_spent:
             lowerCamelCase_						:   Tuple		= time.split(""":"""							)

             # Time can be formatted as xx:xx:xx, as .xx, or as x.xx if the time spent was less than a minute.
             if len(__a							) == 1:
                lowerCamelCase_						:   Optional[int]		= [0, 0, time_parts[0]]

             lowerCamelCase_,	lowerCamelCase_,	lowerCamelCase_						:   Any		= int(time_parts[0]							), int(time_parts[1]							), float(time_parts[2]							)
             total_secs += hours * 3_600 + minutes * 60 + seconds

          lowerCamelCase_,	lowerCamelCase_,	lowerCamelCase_						:   Optional[int]		= total_secs // 3_600, (total_secs % 3_600) // 60, total_secs % 60
          return F'''{int(__a							)}h{int(__a							)}m{int(__a							)}s'''



       @property
       def 				_lowerCAmelCase							(						self				:				str							)					->Dict:
          return {"type": "header", "text": {"type": "plain_text", "text": self.title}}



       @property
       def 				_lowerCAmelCase							(						self				:				Tuple							)					->Dict:
          return {
              "type": "section",
              "text": {
                  "type": "plain_text",
                  "text": F'''🌞 There were no failures: all {self.n_tests} tests passed. The suite ran in {self.time}.''',
                  "emoji": True,
              },
              "accessory": {
                  "type": "button",
                  "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
                  "url": F'''https://github.com/huggingface/transformers/actions/runs/{os.environ['GITHUB_RUN_ID']}''',
              },
          }



       @property
       def 				_lowerCAmelCase							(						self				:				Dict							)					->Dict:
          return {
              "type": "section",
              "text": {
                  "type": "plain_text",
                  "text": (
                      F'''There were {self.n_failures} failures, out of {self.n_tests} tests.\nThe suite ran in'''
                      F''' {self.time}.'''
                  ),
                  "emoji": True,
              },
              "accessory": {
                  "type": "button",
                  "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
                  "url": F'''https://github.com/huggingface/transformers/actions/runs/{os.environ['GITHUB_RUN_ID']}''',
              },
          }



       @property
       def 				_lowerCAmelCase							(						self				:				int							)					->Dict:
          lowerCamelCase_						:   Any		= 40
          lowerCamelCase_						:   Tuple		= {k: v["""failed"""] for k, v in doc_test_results.items() if isinstance(__a		,				__a							)}

          lowerCamelCase_						:   Optional[int]		= """"""
          for category, failures in category_failures.items():
             if len(__a							) == 0:
                continue

             if report != "":
                report += "\n\n"

             report += F'''*{category} failures*:'''.ljust(line_length // 2							).rjust(line_length // 2							) + "\n"
             report += "`"
             report += "`\n`".join(__a							)
             report += "`"

          return {
              "type": "section",
              "text": {
                  "type": "mrkdwn",
                  "text": F'''The following examples had failures:\n\n\n{report}\n''',
              },
          }



       @property
       def 				_lowerCAmelCase							(						self				:				Any							)					->str:
          lowerCamelCase_						:   int		= [self.header]

          if self.n_failures > 0:
             blocks.append(self.failures							)

          if self.n_failures > 0:
             blocks.extend([self.category_failures]							)

          if self.n_failures == 0:
             blocks.append(self.no_failures							)

          return json.dumps(__a							)



       @staticmethod
       def 				_lowerCAmelCase							(						)					->List[Any]:
          lowerCamelCase_						:   Dict		= [
              {
                  """type""": """section""",
                  """text""": {
                      """type""": """plain_text""",
                      """text""": """There was an issue running the tests.""",
                  },
                  """accessory""": {
                      """type""": """button""",
                      """text""": {"""type""": """plain_text""", """text""": """Check Action results""", """emoji""": True},
                      """url""": F'''https://github.com/huggingface/transformers/actions/runs/{os.environ['GITHUB_RUN_ID']}''',
                  },
              }
          ]

          print("""Sending the following payload"""							)
          print(json.dumps({"""blocks""": json.loads(__a							)}							)							)

          client.chat_postMessage(
              channel=os.environ["""CI_SLACK_CHANNEL_ID_DAILY"""]		,				text="""There was an issue running the tests."""		,				blocks=__a		,				)



       def 				_lowerCAmelCase							(						self				:				Any							)					->Optional[int]:
          print("""Sending the following payload"""							)
          print(json.dumps({"""blocks""": json.loads(self.payload							)}							)							)

          lowerCamelCase_						:   str		= F'''{self.n_failures} failures out of {self.n_tests} tests,''' if self.n_failures else """All tests passed."""

          lowerCamelCase_						:   Union[str, Any]		= client.chat_postMessage(
              channel=os.environ["""CI_SLACK_CHANNEL_ID_DAILY"""]		,				blocks=self.payload		,				text=__a		,				)



       def 				_lowerCAmelCase							(						self				:				Tuple		,				__a				:				List[Any]		,				__a				:				int		,				__a				:				int		,				__a				:				Optional[Any]							)					->List[str]:
          lowerCamelCase_						:   Any		= """"""
          for key, value in failures.items():
             lowerCamelCase_						:   Tuple		= value[:200] + """ [Truncated]""" if len(__a							) > 250 else value
             failures_text += F'''*{key}*\n_{value}_\n\n'''

          lowerCamelCase_						:   Optional[Any]		= job_name
          lowerCamelCase_						:   Dict		= {"""type""": """section""", """text""": {"""type""": """mrkdwn""", """text""": text}}

          if job_link is not None:
             lowerCamelCase_						:   Optional[Any]		= {
                 """type""": """button""",
                 """text""": {"""type""": """plain_text""", """text""": """GitHub Action job""", """emoji""": True},
                 """url""": job_link,
             }

          return [
              {"type": "header", "text": {"type": "plain_text", "text": title.upper(), "emoji": True}},
              content,
              {"type": "section", "text": {"type": "mrkdwn", "text": failures_text}},
          ]




       def 				_lowerCAmelCase							(						self				:				Any							)					->int:
          if self.thread_ts is None:
             raise ValueError("""Can only post reply if a post has been made."""							)

          lowerCamelCase_						:   List[str]		= self.doc_test_results.pop("""job_link"""							)
          self.doc_test_results.pop("""failures"""							)
          self.doc_test_results.pop("""success"""							)
          self.doc_test_results.pop("""time_spent"""							)

          lowerCamelCase_						:   str		= sorted(self.doc_test_results.items()		,				key=lambda __a							: t[0]							)
          for job, job_result in sorted_dict:
             if len(job_result["""failures"""]							):
                lowerCamelCase_						:   Union[str, Any]		= F'''*Num failures* :{len(job_result['failed']							)} \n'''
                lowerCamelCase_						:   Tuple		= job_result["""failures"""]
                lowerCamelCase_						:   Optional[int]		= self.get_reply_blocks(__a		,				__a		,				__a		,				text=__a							)

                print("""Sending the following reply"""							)
                print(json.dumps({"""blocks""": blocks}							)							)

                client.chat_postMessage(
                    channel=os.environ["""CI_SLACK_CHANNEL_ID_DAILY"""]		,				text=F'''Results for {job}'''		,				blocks=__a		,				thread_ts=self.thread_ts["""ts"""]		,				)

                time.sleep(1							)







def    __lowerCamelCase   (      )     ->    str:
   lowerCamelCase_						:   List[str]		= os.environ["""GITHUB_RUN_ID"""]
   lowerCamelCase_						:   Optional[int]		= f'''https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100'''
   lowerCamelCase_						:   Optional[int]		= requests.get(A__					).json()
   lowerCamelCase_						:   Any		= {}

   try:
      jobs.update({job["""name"""]: job["""html_url"""] for job in result["""jobs"""]}					)
      lowerCamelCase_						:   Tuple		= math.ceil((result["""total_count"""] - 100) / 100					)

      for i in range(A__					):
         lowerCamelCase_						:   List[Any]		= requests.get(url + f'''&page={i + 2}'''					).json()
         jobs.update({job["""name"""]: job["""html_url"""] for job in result["""jobs"""]}					)

      return jobs
   except Exception as e:
      print("""Unknown error, could not fetch links."""						,						A__					)

   return {}


def    __lowerCamelCase   (      A__						: str					)     ->    int:
   lowerCamelCase_						:   Any		= {}

   if os.path.exists(A__					):
      lowerCamelCase_						:   Any		= os.listdir(A__					)
      for file in files:
         try:
            with open(os.path.join(A__						,						A__					)						,						encoding="""utf-8"""					) as f:
               lowerCamelCase_						:   List[str]		= f.read()
         except UnicodeDecodeError as e:
            raise ValueError(f'''Could not open {os.path.join(A__						,						A__					)}.'''					) from e

   return _artifact


def    __lowerCamelCase   (      )     ->    Optional[int]:





   class   SCREAMING_SNAKE_CASE_			:




          '''simple docstring'''



          def __init__(						self				:				Dict		,				__a				:				str							)					->Optional[Any]:
             lowerCamelCase_						:   Dict		= name
             lowerCamelCase_						:   int		= []



          def __str__(						self				:				str							)					->List[Any]:
             return self.name




          def 				_lowerCAmelCase							(						self				:				Any		,				__a				:				str							)					->Any:
             self.paths.append({"""name""": self.name, """path""": path}							)







   lowerCamelCase_						:   Dict[str, Artifact]		= {}

   lowerCamelCase_						:   int		= filter(os.path.isdir						,						os.listdir()					)
   for directory in directories:
      lowerCamelCase_						:   Optional[Any]		= directory
      if artifact_name not in _available_artifacts:
         lowerCamelCase_						:   Any		= Artifact(A__					)

         _available_artifacts[artifact_name].add_path(A__					)

   return _available_artifacts


if __name__ == "__main__":
      snake_case__				:	List[str]					       =					get_job_links()
      snake_case__				:	List[Any]					       =					retrieve_available_artifacts()

      snake_case__				:	Tuple					       =					collections.OrderedDict(
          [
              ('*.py', 'API Examples'),
              ('*.md', 'MD Examples'),
          ]
      )

      # This dict will contain all the information relative to each doc test category:
      # - failed: list of failed tests
      # - failures: dict in the format 'test': 'error_message'
      snake_case__				:	str					       =					{
          v: {
              'failed': [],
              'failures': {},
          }
          for v in docs.values()
      }

      # Link to the GitHub Action job
      snake_case__				:	Any					       =					github_actions_job_links.get('run_doctests')

      snake_case__				:	List[str]					       =					available_artifacts['doc_tests_gpu_test_reports'].paths[0]
      snake_case__				:	Any					       =					retrieve_artifact(artifact_path['name'])
      if "stats" in artifact:
            snake_case__       ,   snake_case__       ,   snake_case__				:	int					       =					handle_test_results(artifact['stats'])
            snake_case__				:	Tuple					       =					failed
            snake_case__				:	Optional[Any]					       =					success
            snake_case__				:	Union[str, Any]					       =					time_spent[1:-1] + ', '

            snake_case__				:	Tuple					       =					extract_first_line_failure(artifact['failures_short'])
            for line in artifact["summary_short"].split('\n'):
                  if re.search('FAILED', line):
                        snake_case__				:	Union[str, Any]					       =					line.replace('FAILED ', '')
                        snake_case__				:	str					       =					line.split()[0].replace('\n', '')

                        if "::" in line:
                              snake_case__       ,   snake_case__				:	Optional[int]					       =					line.split('::')
                        else:
                              snake_case__       ,   snake_case__				:	Any					       =					line, line

                        for file_regex in docs.keys():
                              if fnmatch(file_path, file_regex):
                                    snake_case__				:	str					       =					docs[file_regex]
                                    doc_test_results[category]["failed"].append(test)

                                    snake_case__				:	Dict					       =					all_failures[test] if test in all_failures else 'N/A'
                                    snake_case__				:	Optional[int]					       =					failure
                                    break

      snake_case__				:	Union[str, Any]					       =					Message('🤗 Results of the doc tests.', doc_test_results)
      message.post()
      message.post_reply()









from ...configuration_utils import PretrainedConfig
from ...utils import logging


snake_case__				:	List[Any]					       =					logging.get_logger(__name__)

snake_case__				:	str					       =					{
    'facebook/dpr-ctx_encoder-single-nq-base': (
        'https://huggingface.co/facebook/dpr-ctx_encoder-single-nq-base/resolve/main/config.json'
    ),
    'facebook/dpr-question_encoder-single-nq-base': (
        'https://huggingface.co/facebook/dpr-question_encoder-single-nq-base/resolve/main/config.json'
    ),
    'facebook/dpr-reader-single-nq-base': (
        'https://huggingface.co/facebook/dpr-reader-single-nq-base/resolve/main/config.json'
    ),
    'facebook/dpr-ctx_encoder-multiset-base': (
        'https://huggingface.co/facebook/dpr-ctx_encoder-multiset-base/resolve/main/config.json'
    ),
    'facebook/dpr-question_encoder-multiset-base': (
        'https://huggingface.co/facebook/dpr-question_encoder-multiset-base/resolve/main/config.json'
    ),
    'facebook/dpr-reader-multiset-base': (
        'https://huggingface.co/facebook/dpr-reader-multiset-base/resolve/main/config.json'
    ),
}





class   SCREAMING_SNAKE_CASE_			(a__						):




       '''simple docstring'''



       _a   							=							"dpr"



       def __init__(						self				:				Union[str, Any]		,				__a				:				Optional[Any]=30_522		,				__a				:				List[Any]=768		,				__a				:				List[Any]=12		,				__a				:				Dict=12		,				__a				:				Union[str, Any]=3_072		,				__a				:				Any="gelu"		,				__a				:				Any=0.1		,				__a				:				Any=0.1		,				__a				:				Tuple=512		,				__a				:				int=2		,				__a				:				Optional[Any]=0.02		,				__a				:				List[Any]=1e-12		,				__a				:				int=0		,				__a				:				int="absolute"		,				__a				:				int = 0		,				**__a				:				Optional[int]		,				)					->Tuple:
          super().__init__(pad_token_id=__a		,				**__a							)

          lowerCamelCase_						:   str		= vocab_size
          lowerCamelCase_						:   List[Any]		= hidden_size
          lowerCamelCase_						:   Optional[int]		= num_hidden_layers
          lowerCamelCase_						:   Optional[Any]		= num_attention_heads
          lowerCamelCase_						:   List[Any]		= hidden_act
          lowerCamelCase_						:   Optional[Any]		= intermediate_size
          lowerCamelCase_						:   List[Any]		= hidden_dropout_prob
          lowerCamelCase_						:   Any		= attention_probs_dropout_prob
          lowerCamelCase_						:   Any		= max_position_embeddings
          lowerCamelCase_						:   Tuple		= type_vocab_size
          lowerCamelCase_						:   Dict		= initializer_range
          lowerCamelCase_						:   List[Any]		= layer_norm_eps
          lowerCamelCase_						:   List[str]		= projection_dim
          lowerCamelCase_						:   Optional[int]		= position_embedding_type





