



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    snake_case_				(    _SCREAMING_SNAKE_CASE   ):
	__lowercase		       =			test_results.split(" "   )

	__lowercase		       =			0
	__lowercase		       =			0

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

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

	return failed, success, time_spent



def    snake_case_				(    _SCREAMING_SNAKE_CASE   ):
	__lowercase		       =			{}
	__lowercase		       =			None
	__lowercase		       =			False
	for line in failures_short_lines.split("\n"   ):
		if re.search(R"_ \[doctest\]"  ,	lowerCamelCase_   ):
			__lowercase		       =			True
			__lowercase		       =			line.split(" "   )[2]
		elif in_error and not line.split(" "   )[0].isdigit():
			__lowercase		       =			line
			__lowercase		       =			False

	return failures

class   _A   :


						'''simple docstring'''





						def __init__(	self	:							List[Any]							,      lowerCamelCase	:							str							,      lowerCamelCase	:							Optional[Any]	):






							'''simple docstring'''



							__lowercase		       =			title

							__lowercase		       =			doc_test_results["time_spent"].split(","	)[0]
							__lowercase		       =			doc_test_results["success"]
							__lowercase		       =			doc_test_results["failures"]
							__lowercase		       =			self.n_success + self.n_failures

							# Failures and success of the modeling tests
							__lowercase		       =			doc_test_results



						@property
						def     _snake_case	(	self	:							Dict	):






							'''simple docstring'''



							__lowercase		       =			[self._time_spent]
							__lowercase		       =			0

							for time in time_spent:
								__lowercase		       =			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:
									__lowercase		       =			[0, 0, time_parts[0]]

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

							__lowercase       ,    __lowercase       ,    __lowercase		       =			total_secs // 3_600, (total_secs % 3_600) // 60, total_secs % 60
							return f"""{int(_lowerCamelCase	)}h{int(_lowerCamelCase	)}m{int(_lowerCamelCase	)}s"""



						@property
						def     _snake_case	(	self	:							Optional[int]	):






							'''simple docstring'''



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



						@property
						def     _snake_case	(	self	:							List[str]	):






							'''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     _snake_case	(	self	:							Union[str, Any]	):






							'''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     _snake_case	(	self	:							Optional[Any]	):






							'''simple docstring'''



							__lowercase		       =			40
							__lowercase		       =			{k: v["failed"] for k, v in doc_test_results.items() if isinstance(_lowerCamelCase							,      _lowerCamelCase	)}

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






							'''simple docstring'''



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






							'''simple docstring'''



							__lowercase		       =			[
							    {
							        "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	:							Any	):






							'''simple docstring'''



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

							__lowercase		       =			f"""{self.n_failures} failures out of {self.n_tests} tests,""" if self.n_failures else "All tests passed."

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



						def     _snake_case	(	self	:							Union[str, Any]							,      lowerCamelCase	:							Dict							,      lowerCamelCase	:							List[str]							,      lowerCamelCase	:							int							,      lowerCamelCase	:							List[Any]	):






							'''simple docstring'''



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

							__lowercase		       =			job_name
							__lowercase		       =			{"type": "section", "text": {"type": "mrkdwn", "text": text}}

							if job_link is not None:
								__lowercase		       =			{
								    "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	:							int	):






							'''simple docstring'''



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

							__lowercase		       =			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"	)

							__lowercase		       =			sorted(self.doc_test_results.items()							,      key=lambda lowerCamelCase	: t[0]	)
							for job, job_result in sorted_dict:
								if len(job_result["failures"]	):
									__lowercase		       =			f"""*Num failures* :{len(job_result['failed']	)} \n"""
									__lowercase		       =			job_result["failures"]
									__lowercase		       =			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    snake_case_				(    ):
	__lowercase		       =			os.environ["GITHUB_RUN_ID"]
	__lowercase		       =			F"""https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100"""
	__lowercase		       =			requests.get(lowerCamelCase_   ).json()
	__lowercase		       =			{}

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

		for i in range(lowerCamelCase_   ):
			__lowercase		       =			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."  ,	lowerCamelCase_   )

	return {}



def    snake_case_				(    _SCREAMING_SNAKE_CASE   ):
	__lowercase		       =			{}

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

	return _artifact



def    snake_case_				(    ):

	class   _A   :


							'''simple docstring'''





							def __init__(	self	:							str							,      lowerCamelCase	:							int	):






								'''simple docstring'''



								__lowercase		       =			name
								__lowercase		       =			[]



							def __str__(	self	:							Any	):






								'''simple docstring'''



								return self.name



							def     _snake_case	(	self	:							Union[str, Any]							,      lowerCamelCase	:							Dict	):






								'''simple docstring'''



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

	__lowercase		       =			{}

	__lowercase		       =			filter(os.path.isdir  ,	os.listdir()   )
	for directory in directories:
		__lowercase		       =			directory
		if artifact_name not in _available_artifacts:
			__lowercase		       =			Artifact(lowerCamelCase_   )

			_available_artifacts[artifact_name].add_path(lowerCamelCase_   )

	return _available_artifacts


if __name__ == "__main__":
		snake_case__					:     int    		=   get_job_links()
		snake_case__					:     Dict    		=   retrieve_available_artifacts()

		snake_case__					:     Union[str, Any]    		=   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__					:     Dict    		=   {
		    v: {
		        """failed""": [],
		        """failures""": {},
		    }
		    for v in docs.values()
		}

		# Link to the GitHub Action job
		snake_case__					:     Optional[int]    		=   github_actions_job_links.get("""run_doctests""")

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

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

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

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

												snake_case__					:     List[str]    		=   all_failures[test] if test in all_failures else """N/A"""
												snake_case__					:     Tuple    		=   failure
												break

		snake_case__					:     Optional[int]    		=   Message("""🤗 Results of the doc tests.""", doc_test_results)
		message.post()
		message.post_reply()








# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from ..models.whisper import WhisperForConditionalGeneration, WhisperProcessor
from .base import PipelineTool

class   _A   ( _lowercase	):


						'''simple docstring'''





						_snake_case			:     Dict      							=      """openai/whisper-base"""
						_snake_case			:     Union[str, Any]      							=      (
						    """This is a tool that transcribes an audio into text. It takes an input named `audio` and returns the """
						    """transcribed text."""
						)
						_snake_case			:     Any      							=      """transcriber"""
						_snake_case			:     Any      							=      WhisperProcessor
						_snake_case			:     Optional[int]      							=      WhisperForConditionalGeneration

						_snake_case			:     str      							=      ["""audio"""]
						_snake_case			:     Optional[int]      							=      ["""text"""]



						def     _snake_case	(	self	:							List[str]							,      lowerCamelCase	:							Optional[int]	):






							'''simple docstring'''



							return self.pre_processor(lowerCamelCase							,      return_tensors="pt"	).input_features



						def     _snake_case	(	self	:							str							,      lowerCamelCase	:							List[Any]	):






							'''simple docstring'''



							return self.model.generate(inputs=lowerCamelCase	)



						def     _snake_case	(	self	:							List[str]							,      lowerCamelCase	:							Optional[Any]	):






							'''simple docstring'''



							return self.pre_processor.batch_decode(lowerCamelCase							,      skip_special_tokens=lowerCamelCase	)[0]



