


'''simple docstring'''



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


__A		=WebClient(token=os.environ['CI_SLACK_BOT_TOKEN'])







def       _UpperCamelCase (						UpperCamelCase__   ):
			UpperCAmelCase__					:   Any       				=       test_results.split(""" """   )

			UpperCAmelCase__					:   Dict       				=       0
			UpperCAmelCase__					:   int       				=       0

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

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

			return failed, success, time_spent







def       _UpperCamelCase (						UpperCamelCase__   ):
			UpperCAmelCase__					:   Union[str, Any]       				=       {}
			UpperCAmelCase__					:   Dict       				=       None
			UpperCAmelCase__					:   List[str]       				=       False
			for line in failures_short_lines.split("""\n"""   ):
						if re.search(R"""_ \[doctest\]"""	, UpperCamelCase__   ):
									UpperCAmelCase__					:   str       				=       True
									UpperCAmelCase__					:   Tuple       				=       line.split(""" """   )[2]
						elif in_error and not line.split(""" """   )[0].isdigit():
									UpperCAmelCase__					:   Any       				=       line
									UpperCAmelCase__					:   Dict       				=       False

			return failures


class 						_snake_case   :




					def __init__( self							,							_lowerCamelCase							,							_lowerCamelCase):
								UpperCAmelCase__					:   str       				=       title

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

								# Failures and success of the modeling tests
								UpperCAmelCase__					:   List[Any]       				=       doc_test_results




					@property
					def 			snake_case__  ( self):
								UpperCAmelCase__					:   str       				=       [self._time_spent]
								UpperCAmelCase__					:   Any       				=       0

								for time in time_spent:
											UpperCAmelCase__					:   List[str]       				=       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(_lowerCamelCase) == 1:
														UpperCAmelCase__					:   Dict       				=       [0, 0, time_parts[0]]

											UpperCAmelCase__				,    UpperCAmelCase__				,    UpperCAmelCase__					:   str       				=       int(time_parts[0]), int(time_parts[1]), float(time_parts[2])
											total_secs += hours * 3600 + minutes * 60 + seconds

								UpperCAmelCase__				,    UpperCAmelCase__				,    UpperCAmelCase__					:   Any       				=       total_secs // 3600, (total_secs % 3600) // 60, total_secs % 60
								return f'''{int(_lowerCamelCase)}h{int(_lowerCamelCase)}m{int(_lowerCamelCase)}s'''




					@property
					def 			snake_case__  ( self):
								return {"type": "header", "text": {"type": "plain_text", "text": self.title}}




					@property
					def 			snake_case__  ( self):
								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 			snake_case__  ( self):
								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 			snake_case__  ( self):
								UpperCAmelCase__					:   Any       				=       40
								UpperCAmelCase__					:   List[str]       				=       {k: v["""failed"""] for k, v in doc_test_results.items() if isinstance(_lowerCamelCase							,							_lowerCamelCase)}

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

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

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

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




					@property
					def 			snake_case__  ( self):
								UpperCAmelCase__					:   List[Any]       				=       [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(_lowerCamelCase)




					@staticmethod
					def 			snake_case__  ( ):
								UpperCAmelCase__					:   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(_lowerCamelCase)}))

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




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

								UpperCAmelCase__					:   Union[str, Any]       				=       f'''{self.n_failures} failures out of {self.n_tests} tests,''' if self.n_failures else """All tests passed."""

								UpperCAmelCase__					:   Optional[Any]       				=       client.chat_postMessage(
								    channel=os.environ["""CI_SLACK_CHANNEL_ID_DAILY"""]							,							blocks=self.payload							,							text=_lowerCamelCase							,							)




					def 			snake_case__  ( self							,							_lowerCamelCase							,							_lowerCamelCase							,							_lowerCamelCase							,							_lowerCamelCase):
								UpperCAmelCase__					:   str       				=       """"""
								for key, value in failures.items():
											UpperCAmelCase__					:   List[str]       				=       value[:200] + """ [Truncated]""" if len(_lowerCamelCase) > 250 else value
											failures_text += f'''*{key}*\n_{value}_\n\n'''

								UpperCAmelCase__					:   List[Any]       				=       job_name
								UpperCAmelCase__					:   List[str]       				=       {"""type""": """section""", """text""": {"""type""": """mrkdwn""", """text""": text}}

								if job_link is not None:
											UpperCAmelCase__					:   Tuple       				=       {
											    """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 			snake_case__  ( self):
								if self.thread_ts is None:
											raise ValueError("""Can only post reply if a post has been made.""")

								UpperCAmelCase__					:   List[Any]       				=       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""")

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

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

														time.sleep(1)







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

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

						for i in range(UpperCamelCase__   ):
									UpperCAmelCase__					:   Optional[int]       				=       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."""	, UpperCamelCase__   )

			return {}







def       _UpperCamelCase (						UpperCamelCase__   ):
			UpperCAmelCase__					:   Dict       				=       {}

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

			return _artifact







def       _UpperCamelCase (						):


			class 						_snake_case   :




								def __init__( self							,							_lowerCamelCase):
											UpperCAmelCase__					:   Dict       				=       name
											UpperCAmelCase__					:   int       				=       []




								def __str__( self):
											return self.name




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

			UpperCAmelCase__					:   Dict[str, Artifact]       				=       {}

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

									_available_artifacts[artifact_name].add_path(UpperCamelCase__   )

			return _available_artifacts


if __name__ == "__main__":
		__A		=get_job_links()
		__A		=retrieve_available_artifacts()

		__A		=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'
		__A		={
		    v: {
		        'failed': [],
		        'failures': {},
		    }
		    for v in docs.values()
		}

		# Link to the GitHub Action job
		__A		=github_actions_job_links.get('run_doctests')

		__A		=available_artifacts['doc_tests_gpu_test_reports'].paths[0]
		__A		=retrieve_artifact(artifact_path['name'])
		if "stats" in artifact:
				__A		,	__A		,	__A		=handle_test_results(artifact['stats'])
				__A		=failed
				__A		=success
				__A		=time_spent[1:-1] + ', '

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

								if "::" in line:
										__A		,	__A		=line.split('::')
								else:
										__A		,	__A		=line, line

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

												__A		=all_failures[test] if test in all_failures else 'N/A'
												__A		=failure
												break

		__A		=Message('🤗 Results of the doc tests.', doc_test_results)
		message.post()
		message.post_reply()



'''simple docstring'''



import os
from typing import Optional

import fsspec
from fsspec.archive import AbstractArchiveFileSystem
from fsspec.utils import DEFAULT_BLOCK_SIZE


class 						_snake_case   (       a__		):
					lowerCAmelCase :Optional[int]						=   ''''''
					lowerCAmelCase :str						=   (
					    None  # protocol passed in prefix to the url. ex: "gzip", for gzip://file.txt::http://foo.bar/file.txt.gz
					)
					lowerCAmelCase :str						=   None  # compression type in fsspec. ex: "gzip"
					lowerCAmelCase :str						=   None  # extension of the filename to strip. ex: "".gz" to get file.txt from file.txt.gz




					def __init__( self							,							_lowerCamelCase = ""							,							_lowerCamelCase = None							,							_lowerCamelCase = None							,							**_lowerCamelCase):
								super().__init__(self							,							**_lowerCamelCase)
								# always open as "rb" since fsspec can then use the TextIOWrapper to make it work for "r" mode
								UpperCAmelCase__					:   Optional[Any]       				=       fsspec.open(
								    _lowerCamelCase							,							mode="""rb"""							,							protocol=_lowerCamelCase							,							compression=self.compression							,							client_kwargs={
								        """requote_redirect_url""": False,  # see https://github.com/huggingface/datasets/pull/5459
								        """trust_env""": True,  # Enable reading proxy env variables.
								        **(target_options or {}).pop("""client_kwargs"""							,							{}),  # To avoid issues if it was already passed.
								    }							,							**(target_options or {})							,							)
								UpperCAmelCase__					:   List[Any]       				=       os.path.basename(self.file.path.split("""::""")[0])
								UpperCAmelCase__					:   Dict       				=       (
								    self.compressed_name[: self.compressed_name.rindex(""".""")]
								    if """.""" in self.compressed_name
								    else self.compressed_name
								)
								UpperCAmelCase__					:   Tuple       				=       None




					@classmethod
					def 			snake_case__  ( cls							,							_lowerCamelCase):
								# compressed file paths are always relative to the archive root
								return super()._strip_protocol(_lowerCamelCase).lstrip("""/""")




					def 			snake_case__  ( self):
								if self.dir_cache is None:
											UpperCAmelCase__					:   Optional[Any]       				=       {**self.file.fs.info(self.file.path), """name""": self.uncompressed_name}
											UpperCAmelCase__					:   Union[str, Any]       				=       {f["""name"""]: f}




					def 			snake_case__  ( self							,							_lowerCamelCase):
								return self.file.open().read()




					def 			snake_case__  ( self							,							_lowerCamelCase							,							_lowerCamelCase = "rb"							,							_lowerCamelCase=None							,							_lowerCamelCase=True							,							_lowerCamelCase=None							,							**_lowerCamelCase							,							):
								UpperCAmelCase__					:   List[str]       				=       self._strip_protocol(_lowerCamelCase)
								if mode != "rb":
											raise ValueError(f'''Tried to read with mode {mode} on file {self.file.path} opened with mode \'rb\'''')
								return self.file.open()


class 						_snake_case   (       a__		):
					lowerCAmelCase :Dict						=   '''bz2'''
					lowerCAmelCase :List[str]						=   '''bz2'''
					lowerCAmelCase :Dict						=   '''.bz2'''


class 						_snake_case   (       a__		):
					lowerCAmelCase :int						=   '''gzip'''
					lowerCAmelCase :Tuple						=   '''gzip'''
					lowerCAmelCase :str						=   '''.gz'''


class 						_snake_case   (       a__		):
					lowerCAmelCase :List[str]						=   '''lz4'''
					lowerCAmelCase :Any						=   '''lz4'''
					lowerCAmelCase :int						=   '''.lz4'''


class 						_snake_case   (       a__		):
					lowerCAmelCase :Union[str, Any]						=   '''xz'''
					lowerCAmelCase :int						=   '''xz'''
					lowerCAmelCase :List[Any]						=   '''.xz'''


class 						_snake_case   (       a__		):
					lowerCAmelCase :Tuple						=   '''zstd'''
					lowerCAmelCase :List[str]						=   '''zstd'''
					lowerCAmelCase :Union[str, Any]						=   '''.zst'''




					def __init__( self							,							_lowerCamelCase							,							_lowerCamelCase = "rb"							,							_lowerCamelCase = None							,							_lowerCamelCase = None							,							_lowerCamelCase = DEFAULT_BLOCK_SIZE							,							**_lowerCamelCase							,							):
								super().__init__(
								    fo=_lowerCamelCase							,							mode=_lowerCamelCase							,							target_protocol=_lowerCamelCase							,							target_options=_lowerCamelCase							,							block_size=_lowerCamelCase							,							**_lowerCamelCase							,							)
								# We need to wrap the zstd decompressor to avoid this error in fsspec==2021.7.0 and zstandard==0.15.2:
								#
								# File "/Users/user/.virtualenvs/hf-datasets/lib/python3.7/site-packages/fsspec/core.py", line 145, in open
								#     out.close = close
								# AttributeError: 'zstd.ZstdDecompressionReader' object attribute 'close' is read-only
								#
								# see https://github.com/intake/filesystem_spec/issues/725
								UpperCAmelCase__					:   Dict       				=       self.file.__enter__


								class 						_snake_case   :




													def __init__( self							,							_lowerCamelCase):
																UpperCAmelCase__					:   Optional[int]       				=       file_




													def __enter__( self):
																self._file.__enter__()
																return self




													def __exit__( self							,							*_lowerCamelCase							,							**_lowerCamelCase):
																self._file.__exit__(*_lowerCamelCase							,							**_lowerCamelCase)




													def __iter__( self):
																return iter(self._file)




													def 			snake_case__  ( self):
																return next(self._file)




													def __getattr__( self							,							_lowerCamelCase):
																return getattr(self._file							,							_lowerCamelCase)

								def fixed_enter(*_lowerCamelCase							,							**_lowerCamelCase):
											return WrappedFile(_enter(*_lowerCamelCase							,							**_lowerCamelCase))

								UpperCAmelCase__					:   List[Any]       				=       fixed_enter