






import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


if TYPE_CHECKING:
							from transformers.pipelines.conversational import Conversation

__UpperCamelCase   :						Any							=  logging.get_logger(__name__)

__UpperCamelCase   :						str							=  {
    """vocab_file""": """vocab.json""",
    """merges_file""": """merges.txt""",
}

__UpperCamelCase   :						Optional[Any]							=  {
    """vocab_file""": {
        """gpt2""": """https://huggingface.co/gpt2/resolve/main/vocab.json""",
        """gpt2-medium""": """https://huggingface.co/gpt2-medium/resolve/main/vocab.json""",
        """gpt2-large""": """https://huggingface.co/gpt2-large/resolve/main/vocab.json""",
        """gpt2-xl""": """https://huggingface.co/gpt2-xl/resolve/main/vocab.json""",
        """distilgpt2""": """https://huggingface.co/distilgpt2/resolve/main/vocab.json""",
    },
    """merges_file""": {
        """gpt2""": """https://huggingface.co/gpt2/resolve/main/merges.txt""",
        """gpt2-medium""": """https://huggingface.co/gpt2-medium/resolve/main/merges.txt""",
        """gpt2-large""": """https://huggingface.co/gpt2-large/resolve/main/merges.txt""",
        """gpt2-xl""": """https://huggingface.co/gpt2-xl/resolve/main/merges.txt""",
        """distilgpt2""": """https://huggingface.co/distilgpt2/resolve/main/merges.txt""",
    },
}

__UpperCamelCase   :						Union[str, Any]							=  {
    """gpt2""": 1024,
    """gpt2-medium""": 1024,
    """gpt2-large""": 1024,
    """gpt2-xl""": 1024,
    """distilgpt2""": 1024,
}
@lru_cache()
def        a_	( )     ->     int:







				"""simple docstring"""
				snake_case__													=       (
				    list(range(ord('!'					) ,				ord('~'					) + 1					)					) + list(range(ord('¡'					) ,				ord('¬'					) + 1					)					) + list(range(ord('®'					) ,				ord('ÿ'					) + 1					)					)
				)
				snake_case__													=       bs[:]
				snake_case__													=       0
				for b in range(2**8					):
								if b not in bs:
												bs.append(_A					)
												cs.append(2**8 + n					)
												n += 1
				snake_case__													=       [chr(_A					) for n in cs]
				return dict(zip(_A ,				_A					)					)
def        a_	( _A					)     ->     str:







				"""simple docstring"""
				snake_case__													=       set()
				snake_case__													=       word[0]
				for char in word[1:]:
								pairs.add((prev_char, char)					)
								snake_case__													=       char
				return pairs


class 					__SCREAMING_SNAKE_CASE(		a_		):
							_UpperCAmelCase	      =    VOCAB_FILES_NAMES
							_UpperCAmelCase	      =    PRETRAINED_VOCAB_FILES_MAP
							_UpperCAmelCase	      =    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							_UpperCAmelCase	      =    ["input_ids", "attention_mask"]
							def __init__(       self: Optional[int]    ,		UpperCamelCase: Optional[int]    ,		UpperCamelCase: Union[str, Any]    ,		UpperCamelCase: str="replace"    ,		UpperCamelCase: int="<|endoftext|>"    ,		UpperCamelCase: List[str]="<|endoftext|>"    ,		UpperCamelCase: Any="<|endoftext|>"    ,		UpperCamelCase: Optional[int]=None    ,		UpperCamelCase: Optional[Any]=False    ,		UpperCamelCase: Dict=False    ,		**UpperCamelCase: Dict    ,		)						->						List[str]:
											snake_case__													=       AddedToken(UpperCamelCase    ,		lstrip=UpperCamelCase    ,		rstrip=UpperCamelCase  ) if isinstance(UpperCamelCase    ,		UpperCamelCase  ) else bos_token
											snake_case__													=       AddedToken(UpperCamelCase    ,		lstrip=UpperCamelCase    ,		rstrip=UpperCamelCase  ) if isinstance(UpperCamelCase    ,		UpperCamelCase  ) else eos_token
											snake_case__													=       AddedToken(UpperCamelCase    ,		lstrip=UpperCamelCase    ,		rstrip=UpperCamelCase  ) if isinstance(UpperCamelCase    ,		UpperCamelCase  ) else unk_token
											snake_case__													=       AddedToken(UpperCamelCase    ,		lstrip=UpperCamelCase    ,		rstrip=UpperCamelCase  ) if isinstance(UpperCamelCase    ,		UpperCamelCase  ) else pad_token
											super().__init__(
											    errors=UpperCamelCase    ,		unk_token=UpperCamelCase    ,		bos_token=UpperCamelCase    ,		eos_token=UpperCamelCase    ,		pad_token=UpperCamelCase    ,		add_prefix_space=UpperCamelCase    ,		add_bos_token=UpperCamelCase    ,		**UpperCamelCase    ,		)
											snake_case__													=       add_bos_token

											with open(UpperCamelCase    ,		encoding='utf-8'  ) as vocab_handle:
															snake_case__													=       json.load(UpperCamelCase  )
											snake_case__													=       {v: k for k, v in self.encoder.items()}
											snake_case__													=       errors  # how to handle errors in decoding
											snake_case__													=       bytes_to_unicode()
											snake_case__													=       {v: k for k, v in self.byte_encoder.items()}
											with open(UpperCamelCase    ,		encoding='utf-8'  ) as merges_handle:
															snake_case__													=       merges_handle.read().split('\n'  )[1:-1]
											snake_case__													=       [tuple(merge.split()  ) for merge in bpe_merges]
											snake_case__													=       dict(zip(UpperCamelCase    ,		range(len(UpperCamelCase  )  )  )  )
											snake_case__													=       {}
											snake_case__													=       add_prefix_space

											# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
											snake_case__													=       re.compile(R'\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'  )
							@property
							def lowerCAmelCase_	(       self: List[Any]  )						->						List[str]:
											return len(self.encoder  )
							def lowerCAmelCase_	(       self: List[str]  )						->						Optional[int]:
											return dict(self.encoder    ,		**self.added_tokens_encoder  )
							def lowerCAmelCase_	(       self: Dict    ,		UpperCamelCase: int  )						->						Optional[int]:
											if token in self.cache:
															return self.cache[token]
											snake_case__													=       tuple(UpperCamelCase  )
											snake_case__													=       get_pairs(UpperCamelCase  )

											if not pairs:
															return token

											while True:
															snake_case__													=       min(UpperCamelCase    ,		key=lambda UpperCamelCase  : self.bpe_ranks.get(UpperCamelCase    ,		float('inf'  )  )  )
															if bigram not in self.bpe_ranks:
																			break
															snake_case__							,  snake_case__													=       bigram
															snake_case__													=       []
															snake_case__													=       0
															while i < len(UpperCamelCase  ):
																			try:
																							snake_case__													=       word.index(UpperCamelCase    ,		UpperCamelCase  )
																			except ValueError:
																							new_word.extend(word[i:]  )
																							break
																			else:
																							new_word.extend(word[i:j]  )
																							snake_case__													=       j

																			if word[i] == first and i < len(UpperCamelCase  ) - 1 and word[i + 1] == second:
																							new_word.append(first + second  )
																							i += 2
																			else:
																							new_word.append(word[i]  )
																							i += 1
															snake_case__													=       tuple(UpperCamelCase  )
															snake_case__													=       new_word
															if len(UpperCamelCase  ) == 1:
																			break
															else:
																			snake_case__													=       get_pairs(UpperCamelCase  )
											snake_case__													=       ' '.join(UpperCamelCase  )
											snake_case__													=       word
											return word
							def lowerCAmelCase_	(       self: int    ,		UpperCamelCase: Union[str, Any]    ,		UpperCamelCase: Tuple=None  )						->						int:
											if self.add_bos_token:
															snake_case__													=       [self.bos_token_id]
											else:
															snake_case__													=       []

											snake_case__													=       bos_token_ids + token_ids_a

											if token_ids_a is None:
															return output

											return output + bos_token_ids + token_ids_a
							def lowerCAmelCase_	(       self: Optional[int]    ,		UpperCamelCase: List[int]    ,		UpperCamelCase: Optional[List[int]] = None    ,		UpperCamelCase: bool = False  )						->						List[int]:
											if already_has_special_tokens:
															return super().get_special_tokens_mask(
															    token_ids_a=UpperCamelCase    ,		token_ids_a=UpperCamelCase    ,		already_has_special_tokens=UpperCamelCase  )

											if not self.add_bos_token:
															return super().get_special_tokens_mask(
															    token_ids_a=UpperCamelCase    ,		token_ids_a=UpperCamelCase    ,		already_has_special_tokens=UpperCamelCase  )

											if token_ids_a is None:
															return [1] + ([0] * len(UpperCamelCase  ))
											return [1] + ([0] * len(UpperCamelCase  )) + [1] + ([0] * len(UpperCamelCase  ))
							def lowerCAmelCase_	(       self: Union[str, Any]    ,		UpperCamelCase: Tuple  )						->						Tuple:
											snake_case__													=       []
											for token in re.findall(self.pat    ,		UpperCamelCase  ):
															snake_case__													=       ''.join(
															    self.byte_encoder[b] for b in token.encode('utf-8'  )  )  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
															bpe_tokens.extend(bpe_token for bpe_token in self.bpe(UpperCamelCase  ).split(' '  )  )
											return bpe_tokens
							def lowerCAmelCase_	(       self: List[Any]    ,		UpperCamelCase: Dict  )						->						int:
											return self.encoder.get(UpperCamelCase    ,		self.encoder.get(self.unk_token  )  )
							def lowerCAmelCase_	(       self: Union[str, Any]    ,		UpperCamelCase: Optional[int]  )						->						Optional[Any]:
											return self.decoder.get(UpperCamelCase  )
							def lowerCAmelCase_	(       self: List[str]    ,		UpperCamelCase: int  )						->						Tuple:
											snake_case__													=       ''.join(UpperCamelCase  )
											snake_case__													=       bytearray([self.byte_decoder[c] for c in text]  ).decode('utf-8'    ,		errors=self.errors  )
											return text
							def lowerCAmelCase_	(       self: List[Any]    ,		UpperCamelCase: str    ,		UpperCamelCase: Optional[str] = None  )						->						Tuple[str]:
											if not os.path.isdir(UpperCamelCase  ):
															logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''  )
															return
											snake_case__													=       os.path.join(
											    UpperCamelCase    ,		(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']  )
											snake_case__													=       os.path.join(
											    UpperCamelCase    ,		(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']  )

											with open(UpperCamelCase    ,		'w'    ,		encoding='utf-8'  ) as f:
															f.write(json.dumps(self.encoder    ,		indent=2    ,		sort_keys=UpperCamelCase    ,		ensure_ascii=UpperCamelCase  ) + '\n'  )

											snake_case__													=       0
											with open(UpperCamelCase    ,		'w'    ,		encoding='utf-8'  ) as writer:
															writer.write('#version: 0.2\n'  )
															for bpe_tokens, token_index in sorted(self.bpe_ranks.items()    ,		key=lambda UpperCamelCase  : kv[1]  ):
																			if index != token_index:
																							logger.warning(
																							    F'''Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'''
																							    ' Please check that the tokenizer is not corrupted!'  )
																							snake_case__													=       token_index
																			writer.write(' '.join(UpperCamelCase  ) + '\n'  )
																			index += 1

											return vocab_file, merge_file
							def lowerCAmelCase_	(       self: int    ,		UpperCamelCase: List[Any]    ,		UpperCamelCase: Any=False    ,		**UpperCamelCase: str  )						->						List[str]:
											snake_case__													=       kwargs.pop('add_prefix_space'    ,		self.add_prefix_space  )
											if is_split_into_words or add_prefix_space:
															snake_case__													=       ' ' + text
											return (text, kwargs)


							def lowerCAmelCase_	(       self: List[Any]    ,		UpperCamelCase: "Conversation"  )						->						List[int]:
											snake_case__													=       []
											for is_user, text in conversation.iter_texts():
															input_ids.extend(self.encode(UpperCamelCase    ,		add_special_tokens=UpperCamelCase  ) + [self.eos_token_id]  )
											if len(UpperCamelCase  ) > self.model_max_length:
															snake_case__													=       input_ids[-self.model_max_length :]
											return input_ids












import argparse
import logging
import os
import sys
from collections import namedtuple

import torch
from modeling_bertabs import BertAbs, build_predictor
from torch.utils.data import DataLoader, SequentialSampler
from tqdm import tqdm

from transformers import BertTokenizer

from .utils_summarization import (
    CNNDMDataset,
    build_mask,
    compute_token_type_ids,
    encode_for_summarization,
    truncate_or_pad,
)


__UpperCamelCase   :						List[str]							=  logging.getLogger(__name__)
logging.basicConfig(stream=sys.stdout, level=logging.INFO)


__UpperCamelCase   :						Tuple							=  namedtuple("""Batch""", ["""document_names""", """batch_size""", """src""", """segs""", """mask_src""", """tgt_str"""])
def        a_	( _A					)     ->     Union[str, Any]:







				"""simple docstring"""
				snake_case__													=       BertTokenizer.from_pretrained('bert-base-uncased' ,				do_lower_case=_A					)
				snake_case__													=       BertAbs.from_pretrained('remi/bertabs-finetuned-extractive-abstractive-summarization'					)
				model.to(args.device					)
				model.eval()

				snake_case__													=       {
				    'BOS': tokenizer.vocab['[unused0]'],
				    'EOS': tokenizer.vocab['[unused1]'],
				    'PAD': tokenizer.vocab['[PAD]'],
				}

				if args.compute_rouge:
								snake_case__													=       []
								snake_case__													=       []

								import nltk
								import rouge

								nltk.download('punkt'					)
								snake_case__													=       rouge.Rouge(
								    metrics=['rouge-n', 'rouge-l'] ,				max_n=2 ,				limit_length=_A ,				length_limit=args.beam_size ,				length_limit_type='words' ,				apply_avg=_A ,				apply_best=_A ,				alpha=0.5 ,				weight_factor=1.2 ,				stemming=_A ,				)

				# these (unused) arguments are defined to keep the compatibility
				# with the legacy code and will be deleted in a next iteration.
				snake_case__													=       ''
				snake_case__													=       ''

				snake_case__													=       build_data_iterator(_A ,				_A					)
				snake_case__													=       build_predictor(_A ,				_A ,				_A ,				_A					)

				logger.info('***** Running evaluation *****'					)
				logger.info('  Number examples = %d' ,				len(data_iterator.dataset					)					)
				logger.info('  Batch size = %d' ,				args.batch_size					)
				logger.info(''					)
				logger.info('***** Beam Search parameters *****'					)
				logger.info('  Beam size = %d' ,				args.beam_size					)
				logger.info('  Minimum length = %d' ,				args.min_length					)
				logger.info('  Maximum length = %d' ,				args.max_length					)
				logger.info('  Alpha (length penalty) = %.2f' ,				args.alpha					)
				logger.info('  Trigrams %s be blocked' ,				('will' if args.block_trigram else 'will NOT')					)

				for batch in tqdm(_A					):
								snake_case__													=       predictor.translate_batch(_A					)
								snake_case__													=       predictor.from_batch(_A					)
								snake_case__													=       [format_summary(_A					) for t in translations]
								save_summaries(_A ,				args.summaries_output_dir ,				batch.document_names					)

								if args.compute_rouge:
												reference_summaries += batch.tgt_str
												generated_summaries += summaries

				if args.compute_rouge:
								snake_case__													=       rouge_evaluator.get_scores(_A ,				_A					)
								snake_case__													=       format_rouge_scores(_A					)
								save_rouge_scores(_A					)
								print(_A					)
def        a_	( _A ,				_A ,				_A					)     ->     Optional[Any]:







				"""simple docstring"""
				for summary, document_name in zip(_A ,				_A					):
								# Prepare the summary file's name
								if "." in document_name:
												snake_case__													=       '.'.join(document_name.split('.'					)[:-1]					)
												snake_case__													=       document_name.split('.'					)[-1]
												snake_case__													=       bare_document_name + '_summary.' + extension
								else:
												snake_case__													=       document_name + '_summary'

								snake_case__													=       os.path.join(_A ,				_A					)
								with open(_A ,				'w'					) as output:
												output.write(_A					)
def        a_	( _A					)     ->     Union[str, Any]:







				"""simple docstring"""
				snake_case__							,  snake_case__							,  snake_case__													=       translation
				snake_case__													=       (
				    raw_summary.replace('[unused0]' ,				''					)
				    .replace('[unused3]' ,				''					)
				    .replace('[PAD]' ,				''					)
				    .replace('[unused1]' ,				''					)
				    .replace(R' +' ,				' '					)
				    .replace(' [unused2] ' ,				'. '					)
				    .replace('[unused2]' ,				''					)
				    .strip()
				)

				return summary
def        a_	( _A					)     ->     Any:







				"""simple docstring"""
				return """\n
****** ROUGE SCORES ******

** ROUGE 1
F1        >> {:.3f}
Precision >> {:.3f}
Recall    >> {:.3f}

** ROUGE 2
F1        >> {:.3f}
Precision >> {:.3f}
Recall    >> {:.3f}

** ROUGE L
F1        >> {:.3f}
Precision >> {:.3f}
Recall    >> {:.3f}""".format(
				    scores['rouge-1']['f'] ,				scores['rouge-1']['p'] ,				scores['rouge-1']['r'] ,				scores['rouge-2']['f'] ,				scores['rouge-2']['p'] ,				scores['rouge-2']['r'] ,				scores['rouge-l']['f'] ,				scores['rouge-l']['p'] ,				scores['rouge-l']['r'] ,				)
def        a_	( _A					)     ->     int:







				"""simple docstring"""
				with open('rouge_scores.txt' ,				'w'					) as output:
								output.write(_A					)
def        a_	( _A ,				_A					)     ->     Optional[Any]:







				"""simple docstring"""
				snake_case__													=       load_and_cache_examples(_A ,				_A					)
				snake_case__													=       SequentialSampler(_A					)

				def collate_fn(_A					):
								return collate(_A ,				_A ,				block_size=512 ,				device=args.device					)

				snake_case__													=       DataLoader(
				    _A ,				sampler=_A ,				batch_size=args.batch_size ,				collate_fn=_A ,				)

				return iterator
def        a_	( _A ,				_A					)     ->     Any:







				"""simple docstring"""
				snake_case__													=       CNNDMDataset(args.documents_dir					)
				return dataset
def        a_	( _A ,				_A ,				_A ,				_A					)     ->     str:







				"""simple docstring"""
				snake_case__													=       [x for x in data if not len(x[1]					) == 0]  # remove empty_files
				snake_case__													=       [name for name, _, _ in data]
				snake_case__													=       [' '.join(_A					) for _, _, summary_list in data]

				snake_case__													=       [encode_for_summarization(_A ,				_A ,				_A					) for _, story, summary in data]
				snake_case__													=       torch.tensor(
				    [truncate_or_pad(_A ,				_A ,				tokenizer.pad_token_id					) for story, _ in encoded_text]					)
				snake_case__													=       compute_token_type_ids(_A ,				tokenizer.cls_token_id					)
				snake_case__													=       build_mask(_A ,				tokenizer.pad_token_id					)

				snake_case__													=       Batch(
				    document_names=_A ,				batch_size=len(_A					) ,				src=encoded_stories.to(_A					) ,				segs=encoder_token_type_ids.to(_A					) ,				mask_src=encoder_mask.to(_A					) ,				tgt_str=_A ,				)

				return batch
def        a_	( _A ,				_A					)     ->     Optional[Any]:







				"""simple docstring"""
				snake_case__													=       summary_tokens.to('cpu'					).numpy()
				snake_case__													=       tokenizer.decode(_A					)
				snake_case__													=       summary.split('.'					)
				snake_case__													=       [s + '.' for s in sentences]
				return sentences
def        a_	( )     ->     int:







				"""simple docstring"""
				snake_case__													=       argparse.ArgumentParser()
				parser.add_argument(
				    '--documents_dir' ,				default=_A ,				type=_A ,				required=_A ,				help='The folder where the documents to summarize are located.' ,				)
				parser.add_argument(
				    '--summaries_output_dir' ,				default=_A ,				type=_A ,				required=_A ,				help='The folder in wich the summaries should be written. Defaults to the folder where the documents are' ,				)
				parser.add_argument(
				    '--compute_rouge' ,				default=_A ,				type=_A ,				required=_A ,				help='Compute the ROUGE metrics during evaluation. Only available for the CNN/DailyMail dataset.' ,				)
				# EVALUATION options
				parser.add_argument(
				    '--no_cuda' ,				default=_A ,				type=_A ,				help='Whether to force the execution on CPU.' ,				)
				parser.add_argument(
				    '--batch_size' ,				default=4 ,				type=_A ,				help='Batch size per GPU/CPU for training.' ,				)
				# BEAM SEARCH arguments
				parser.add_argument(
				    '--min_length' ,				default=50 ,				type=_A ,				help='Minimum number of tokens for the summaries.' ,				)
				parser.add_argument(
				    '--max_length' ,				default=200 ,				type=_A ,				help='Maixmum number of tokens for the summaries.' ,				)
				parser.add_argument(
				    '--beam_size' ,				default=5 ,				type=_A ,				help='The number of beams to start with for each example.' ,				)
				parser.add_argument(
				    '--alpha' ,				default=0.95 ,				type=_A ,				help='The value of alpha for the length penalty in the beam search.' ,				)
				parser.add_argument(
				    '--block_trigram' ,				default=_A ,				type=_A ,				help='Whether to block the existence of repeating trigrams in the text generated by beam search.' ,				)
				snake_case__													=       parser.parse_args()

				# Select device (distibuted not available)
				snake_case__													=       torch.device('cuda' if torch.cuda.is_available() and not args.no_cuda else 'cpu'					)

				# Check the existence of directories
				if not args.summaries_output_dir:
								snake_case__													=       args.documents_dir

				if not documents_dir_is_valid(args.documents_dir					):
								raise FileNotFoundError(
								    'We could not find the directory you specified for the documents to summarize, or it was empty. Please'
								    ' specify a valid path.'					)
				os.makedirs(args.summaries_output_dir ,				exist_ok=_A					)

				evaluate(_A					)
def        a_	( _A					)     ->     str:







				"""simple docstring"""
				if not os.path.exists(_A					):
								return False

				snake_case__													=       os.listdir(_A					)
				if len(_A					) == 0:
								return False

				return True


if __name__ == "__main__":
							main()




