import os import re import shutil from typing import List import requests from constantes import AGENTS_FILES_PATH, HUGGINGFACE_DATASET_URL_TEMPLATE class File_Util: """ Manipulação de diretórios e arquivos """ @staticmethod def create_or_clear_output_directory(output_dir: str): """ Cria o diretório de saída se não existir. """ if not os.path.exists(output_dir): os.makedirs(output_dir) print(f"Diretório criado: {output_dir}") else: # Limpa todos os arquivos e subdiretórios for filename in os.listdir(output_dir): file_path = os.path.join(output_dir, filename) try: if os.path.isfile(file_path) or os.path.islink(file_path): os.unlink(file_path) elif os.path.isdir(file_path): shutil.rmtree(file_path) except Exception as e: print(f"Erro ao excluir {file_path}: {e}") print(f"Diretório limpo: {output_dir}") @staticmethod def retirar_sufixo_codec_arquivo(directory: str) -> List[str]: """ Os arquivos de audio e video quando baixados ficam com o codec embutido no nome, dificultando identificar o nome do arquivo a ser processado. O objetivo é remover o sufixo do nome do arquivo. """ return_list = [] for filename in os.listdir(directory): # Procura padrão como ".f123" antes da extensão new_filename = re.sub(r'\.f\d{3}(?=\.\w+$)', '', filename) if new_filename != filename: old_path = os.path.join(directory, filename) new_path = os.path.join(directory, new_filename) os.rename(old_path, new_path) print(f"Renomeado: {filename} → {new_filename}") return_list.append(new_filename) return return_list @staticmethod def baixa_arquivo_task(task_file_name: str) -> str: """ Baixa o arquivo da task para o diretório temporario local Parâmetros: - task_file_name (str): Caminho completo ou nome do arquivo. Retorna: - str: Caminho válido para o arquivo, ou None se não encontrado. """ # Verifica se o arquivo já existe no caminho informado task_file_path= os.path.join(AGENTS_FILES_PATH, task_file_name) if os.path.isfile(task_file_path): return task_file_path #Tenta baixar do dataset url = HUGGINGFACE_DATASET_URL_TEMPLATE.format(filename=task_file_name) try: response = requests.get(url) response.raise_for_status() with open(task_file_path, "wb") as f: f.write(response.content) return task_file_path except Exception as e: print(f"[ERRO] Problemas ao baixar arquivo {task_file_name}") return None