



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


_lowercase		:				List[str]      		=WebClient(token=os.environ["CI_SLACK_BOT_TOKEN"])

def      lowerCAmelCase_					(  _lowercase    :       Optional[Any])							->	int:







			"""simple docstring"""
			a__       :						Dict							=  test_results.split(""" """)

			a__       :						Any							=  0
			a__       :						Optional[Any]							=  0

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

			for i, expression in enumerate(_lowercase):
						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_					(  _lowercase    :       List[str])							->	Optional[int]:







			"""simple docstring"""
			a__       :						Optional[Any]							=  {}
			a__       :						List[Any]							=  None
			a__       :						Optional[Any]							=  False
			for line in failures_short_lines.split("""\n"""):
						if re.search(R"""_ \[doctest\]"""						,       _lowercase):
									a__       :						List[Any]							=  True
									a__       :						List[str]							=  line.split(""" """)[2]
						elif in_error and not line.split(""" """)[0].isdigit():
									a__       :						Tuple							=  line
									a__       :						Optional[int]							=  False

			return failures

class        snake_case__    :







						"""simple docstring"""




						def __init__(       self ,       __lowercase ,       __lowercase  ) ->				Union[str, Any]:
									"""simple docstring"""




									a__       :						Union[str, Any]							=  title

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

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

						@property
						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				str:
									"""simple docstring"""




									a__       :						List[str]							=  [self._time_spent]
									a__       :						Any							=  0

									for time in time_spent:
												a__       :						Optional[Any]							=  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(__lowercase  ) == 1:
															a__       :						Dict							=  [0, 0, time_parts[0]]

												a__	,						a__	,						a__       :						Dict							=  int(time_parts[0]  ), int(time_parts[1]  ), float(time_parts[2]  )
												total_secs += hours * 3_6_0_0 + minutes * 6_0 + seconds

									a__	,						a__	,						a__       :						Tuple							=  total_secs // 3_6_0_0, (total_secs % 3_6_0_0) // 6_0, total_secs % 6_0
									return F'''{int(__lowercase  )}h{int(__lowercase  )}m{int(__lowercase  )}s'''

						@property
						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Dict:
									"""simple docstring"""




									return {"type": "header", "text": {"type": "plain_text", "text": self.title}}

						@property
						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Dict:
									"""simple docstring"""




									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 							SCREAMING_SNAKE_CASE__(       self  ) ->				Dict:
									"""simple docstring"""




									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 							SCREAMING_SNAKE_CASE__(       self  ) ->				Dict:
									"""simple docstring"""




									a__       :						Tuple							=  4_0
									a__       :						Optional[Any]							=  {k: v["""failed"""] for k, v in doc_test_results.items() if isinstance(__lowercase ,       __lowercase  )}

									a__       :						Optional[Any]							=  """"""
									for category, failures in category_failures.items():
												if len(__lowercase  ) == 0:
															continue

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

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

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

						@property
						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				str:
									"""simple docstring"""




									a__       :						Optional[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(__lowercase  )

						@staticmethod
						def 							SCREAMING_SNAKE_CASE__(       ) ->				Tuple:
									"""simple docstring"""




									a__       :						Tuple							=  [
									    {
									        """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(__lowercase  )}  )  )

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

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Optional[Any]:
									"""simple docstring"""




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

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

									a__       :						int							=  client.chat_postMessage(
									    channel=os.environ["""CI_SLACK_CHANNEL_ID_DAILY"""] ,       blocks=self.payload ,       text=__lowercase ,       )

						def 							SCREAMING_SNAKE_CASE__(       self ,       __lowercase ,       __lowercase ,       __lowercase ,       __lowercase  ) ->				Optional[Any]:
									"""simple docstring"""




									a__       :						Any							=  """"""
									for key, value in failures.items():
												a__       :						Dict							=  value[:2_0_0] + """ [Truncated]""" if len(__lowercase  ) > 2_5_0 else value
												failures_text += F'''*{key}*\n_{value}_\n\n'''

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

									if job_link is not None:
												a__       :						Union[str, 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 							SCREAMING_SNAKE_CASE__(       self  ) ->				int:
									"""simple docstring"""




									if self.thread_ts is None:
												raise ValueError("""Can only post reply if a post has been made."""  )

									a__       :						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"""  )

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

															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=__lowercase ,       thread_ts=self.thread_ts["""ts"""] ,       )

															time.sleep(1  )




def      lowerCAmelCase_					(  )							->	Optional[int]:







			"""simple docstring"""
			a__       :						Dict							=  os.environ["""GITHUB_RUN_ID"""]
			a__       :						Dict							=  F'''https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100'''
			a__       :						str							=  requests.get(_lowercase).json()
			a__       :						str							=  {}

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

						for i in range(_lowercase):
									a__       :						Tuple							=  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."""						,       _lowercase)

			return {}

def      lowerCAmelCase_					(  _lowercase    :       str)							->	Tuple:







			"""simple docstring"""
			a__       :						Tuple							=  {}

			if os.path.exists(_lowercase):
						a__       :						Tuple							=  os.listdir(_lowercase)
						for file in files:
									try:
												with open(os.path.join(_lowercase						,       _lowercase)						,       encoding="""utf-8""") as f:
															a__       :						Tuple							=  f.read()
									except UnicodeDecodeError as e:
												raise ValueError(F'''Could not open {os.path.join(_lowercase						,       _lowercase)}.''') from e

			return _artifact

def      lowerCAmelCase_					(  )							->	Any:







			"""simple docstring"""

			class        snake_case__    :







									"""simple docstring"""




									def __init__(       self ,       __lowercase  ) ->				List[str]:
												"""simple docstring"""




												a__       :						Dict							=  name
												a__       :						Optional[Any]							=  []

									def __str__(       self  ) ->				Tuple:
												"""simple docstring"""




												return self.name

									def 							SCREAMING_SNAKE_CASE__(       self ,       __lowercase  ) ->				Union[str, Any]:
												"""simple docstring"""




												self.paths.append({"""name""": self.name, """path""": path}  )




			a__       :						Dict[str, Artifact]							=  {}

			a__       :						List[str]							=  filter(os.path.isdir						,       os.listdir())
			for directory in directories:
						a__       :						str							=  directory
						if artifact_name not in _available_artifacts:
									a__       :						Optional[Any]							=  Artifact(_lowercase)

									_available_artifacts[artifact_name].add_path(_lowercase)

			return _available_artifacts


if __name__ == "__main__":
		_lowercase		:				Optional[Any]      		=get_job_links()
		_lowercase		:				Optional[Any]      		=retrieve_available_artifacts()

		_lowercase		:				List[str]      		=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'
		_lowercase		:				Optional[Any]      		={
		    v: {
		        "failed": [],
		        "failures": {},
		    }
		    for v in docs.values()
		}

		# Link to the GitHub Action job
		_lowercase		:				Union[str, Any]      		=github_actions_job_links.get("run_doctests")

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

				_lowercase		:				int      		=extract_first_line_failure(artifact["failures_short"])
				for line in artifact["summary_short"].split("\n"):
						if re.search("FAILED", line):
								_lowercase		:				Optional[Any]      		=line.replace("FAILED ", "")
								_lowercase		:				Any      		=line.split()[0].replace("\n", "")

								if "::" in line:
										_lowercase		,					_lowercase		:				List[str]      		=line.split("::")
								else:
										_lowercase		,					_lowercase		:				str      		=line, line

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

												_lowercase		:				int      		=all_failures[test] if test in all_failures else "N/A"
												_lowercase		:				str      		=failure
												break

		_lowercase		:				str      		=Message("🤗 Results of the doc tests.", doc_test_results)
		message.post()
		message.post_reply()









from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_sentencepiece_available


_lowercase		:				List[Any]      		={}

try:
		if not is_sentencepiece_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		_lowercase		:				int      		=["BartphoTokenizer"]

if TYPE_CHECKING:
		try:
				if not is_sentencepiece_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .tokenization_bartpho import BartphoTokenizer

else:
		import sys

		_lowercase		:				Tuple      		=_LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)




