import os import re import json import pandas as pd from transcricao.transcribe import Transcriber from conversor.converter import Converter from analise_ia.processar import IaAnalyzer from conversor.converter import GoogleDriveDownload class VideoProcessor: def __init__(self, drive_handler): self.downloader = drive_handler def processar_videos(self, pasta_selecionada, videos_selecionado, video_upload, modo): resultados = []#Texto que será exibido como log no front do Gradio pasta_videos = "data/videos" pasta_audio = "data/audio" pasta_transcricoes = "data/transcricoes" pasta_resumos = "data/resumos" arquivo_download = None def verificar_existencia(arquivo): """ Verifica se um arquivo já existe. """ return os.path.exists(arquivo) # Processar vídeos carregados localmente if video_upload: video_upload = [video_upload] # Garante que seja uma lista print("Upload de vídeo selecionado") for video_path in video_upload: try: print(f"🔍 Verificando vídeo: {video_path}") video_nome = os.path.basename(video_path) audio_path = os.path.join(pasta_audio, f"{os.path.splitext(video_nome)[0]}.wav") transcricao_path = os.path.join(pasta_transcricoes, f"{video_nome}.txt") # Se a transcrição já existe, pula a conversão, mas continua a análise if not verificar_existencia(transcricao_path): Converter(video_path, audio_path).convert() transcriber = Transcriber(model_size="base") transcription = transcriber.transcribe_audio(audio_path) transcriber.save_transcription(transcription, transcricao_path) else: with open(transcricao_path, "r", encoding="utf-8") as f: transcription = f.read() # Escolher modo de processamento analyzer = IaAnalyzer() arquivo_download = None if modo == "Consultor de Negócios (Recomendar Benefícios)": resultado = analyzer.analisar_transcricao(transcription, modo) arquivo_download = self.salvar_analise_pdf(video_nome, resultado) resultados.append(f"✅ {video_nome} processado com sucesso (Consultor de Negócios)") elif modo == "Resumir Reunião (Gerar Resumo)": resumo = analyzer.analisar_transcricao(transcription, modo) resumo_path = os.path.join(pasta_resumos, f"resumo_{video_nome}.txt") with open(resumo_path, "w", encoding="utf-8") as resumo_file: resumo_file.write(resumo) arquivos_txt = [ os.path.join(pasta_resumos, f) for f in os.listdir(pasta_resumos) if f.startswith("resumo_") and f.endswith(".txt") ] if arquivos_txt: arquivo_download = arquivos_txt[-1] resultados.append(f"✅ {video_nome} processado com sucesso (Resumo de Reunião)") elif modo == "Gerar Resumo em Excel": resultado = analyzer.analisar_transcricao(transcription,modo) self.salvar_resumo_excel(video_nome, resultado, "data/analises/dados_resumos.xlsx") arquivo_download = "data/analises/dados_resumos.xlsx" resultados.append(f"✅ {video_nome} processado com sucesso (Resumo salvo em Excel)") if not os.path.exists(arquivo_download or ""): arquivo_download = None except Exception as e: resultados.append(f"❌ Erro ao processar {video_nome}: {e}") # Processar vídeos do Google Drive for video_selecionado in videos_selecionado: try: print(f"🔍 Verificando vídeo: {video_selecionado}") video_path = os.path.join(pasta_videos, video_selecionado) audio_path = os.path.join(pasta_audio, f"{os.path.splitext(video_selecionado)[0]}.wav") transcricao_path = os.path.join(pasta_transcricoes, f"{video_selecionado}.txt") # Verifica se o vídeo já foi baixado if not verificar_existencia(video_path): print(f"⬇️ Baixando vídeo {video_selecionado}...") pastas = self.downloader.downloader.list_folders() pasta_id = pastas.get(pasta_selecionada) if not pasta_id: resultados.append(f"❌ Erro: Pasta '{pasta_selecionada}' não encontrada.") continue videos = self.downloader.downloader.list_videos(pasta_id) video_id = next((v["id"] for v in videos if v["name"] == video_selecionado), None) if not video_id: resultados.append(f"❌ Erro: Vídeo '{video_selecionado}' não encontrado.") continue self.downloader.downloader.download_video(video_id, video_path) # Verifica se a transcrição já foi feita if not verificar_existencia(transcricao_path): Converter(video_path, audio_path).convert() transcriber = Transcriber(model_size="base") transcription = transcriber.transcribe_audio(audio_path) transcriber.save_transcription(transcription, transcricao_path) else: with open(transcricao_path, "r", encoding="utf-8") as f: transcription = f.read() # Escolher modo de processamento analyzer = IaAnalyzer() arquivo_download = None upload_drive = GoogleDriveDownload() if modo == "Consultor de Negócios (Recomendar Benefícios)": resultado = analyzer.analisar_transcricao(transcription, modo) arquivo_download = self.salvar_analise_pdf(video_selecionado, resultado) nome_destino = os.path.basename(arquivo_download) upload_drive.upload_para_drive(arquivo_download, nome_destino=nome_destino, pasta_id="1uDYsmHqltJ6euXVH_Fh7OiWgrhvZlpxB") resultados.append(f"✅ {video_selecionado} processado com sucesso (Consultor de Negócios)") elif modo == "Resumir Reunião (Gerar Resumo)": resumo = analyzer.analisar_transcricao(transcription, modo) resumo_path = os.path.join(pasta_resumos, f"resumo_{video_selecionado}.txt") with open(resumo_path, "w", encoding="utf-8") as resumo_file: resumo_file.write(resumo) # Define explicitamente qual arquivo será usado como o último gerado arquivo_download = resumo_path resultados.append(f"✅ {video_selecionado} processado com sucesso (Resumo de Reunião)") elif modo == "Gerar Resumo em Excel": resultado = analyzer.analisar_transcricao(transcription, modo) nome_limpo = re.sub(r'[<>:"/\\|?*]', '_', video_selecionado) self.salvar_resumo_excel(nome_limpo, resultado, "data/analises/dados_resumos.xlsx") arquivo_download = "data/analises/dados_resumos.xlsx" resultados.append(f"✅ {video_selecionado} processado com sucesso (Resumo salvo em Excel)") if not os.path.exists(arquivo_download or ""): arquivo_download = None except Exception as e: resultados.append(f"❌ Erro ao processar {video_selecionado}: {e}") return "\n".join(resultados), arquivo_download """ Funções personalizadas para cada caso de uso. Auxiliam no processamento do video """ @staticmethod def salvar_analise_pdf(nome_arquivo, resultado, arquivo_excel=None): from fpdf import FPDF import os import re from datetime import datetime from openpyxl import Workbook, load_workbook class PDF(FPDF): def header(self): self.set_font("Arial", "B", 14) self.cell(0, 10, "Relatório Consultor de Negócios", ln=True, align="C") self.ln(10) def add_section(self, titulo, texto): self.set_font("Arial", "B", 12) self.set_text_color(0) self.cell(0, 8, titulo, ln=True) self.set_font("Arial", "", 11) self.multi_cell(0, 7, texto) self.ln(5) # Remove extensão do nome nome_limpo = re.sub(r'[()]', '', os.path.splitext(nome_arquivo)[0]) print("nome limpo:", nome_limpo) # Extrai data no formato YYYY_MM_DD data_match = re.search(r"(\d{4})_(\d{2})_(\d{2})", nome_limpo) if data_match: data_formatada = f"{data_match.group(3)}/{data_match.group(2)}/{data_match.group(1)}" data_para_arquivo = f"{data_match.group(1)}_{data_match.group(2)}_{data_match.group(3)}" else: data_formatada = "Data não encontrada" data_para_arquivo = "sem_data" # Extrai nome do cliente entre "Reuniao Contabilizei" e a data cliente_match = re.search(r"Reuni[aã]o Contabilizei\s+(.+?)\s+-\s+\d{4}_\d{2}_\d{2}", nome_limpo) nome_cliente = cliente_match.group(1).strip() if cliente_match else "Cliente_nao_identificado" # Gera nome seguro para o PDF nome_pdf_seguro = re.sub(r"[^\w\-]", "_", f"{nome_cliente}_{data_para_arquivo}") # Criar pasta de saída pasta_saida = "data/pdf/" pasta_saida_excel = "data/analises/" os.makedirs(pasta_saida, exist_ok=True) caminho_pdf = os.path.join(pasta_saida, f"{nome_pdf_seguro}.pdf") caminho_excel = os.path.join(pasta_saida_excel, "analises_completas.xlsx") # Criar PDF pdf = PDF() pdf.add_page() # Título do cliente e data pdf.set_font("Arial", "B", 12) pdf.cell(0, 8, f"Cliente: {nome_cliente}", ln=True) pdf.cell(0, 8, f"Data: {data_formatada}", ln=True) pdf.ln(10) # Conteúdo organizado mapeamento = { "tipo_atuacao": "Tipo de Atuação", "desafios": "Desafios", "dores_necessidades": "Dores / Necessidades Identificadas", "acoes_recomendadas" : "Ações Recomendadas", "observacoes": "Observações" } for chave, titulo in mapeamento.items(): if chave in resultado: pdf.set_font("Arial", "B", 12) pdf.set_text_color(0) pdf.cell(0, 8, f"{titulo}:", ln=True) pdf.set_font("Arial", "", 11) if isinstance(resultado[chave], list): for item in resultado[chave]: pdf.cell(5) # identação pdf.multi_cell(0, 7, f"{item}") else: pdf.multi_cell(0, 7, resultado[chave]) pdf.ln(5) pdf.output(caminho_pdf) print(f"PDF salvo em: {caminho_pdf}") # === Atualização do Excel Compartilhado === # Verifica se o arquivo existe if os.path.exists(caminho_excel): wb = load_workbook(caminho_excel) ws = wb.active # Verifica se o cabeçalho está correto cabecalho_atual = [cell.value for cell in ws[1]] cabecalho_esperado = ["Nome", "Data"] + list(mapeamento.values()) if cabecalho_atual != cabecalho_esperado: ws.delete_rows(1, 1) ws.append(cabecalho_esperado) else: wb = Workbook() ws = wb.active ws.title = "Análises" ws.append(["Nome", "Data"] + list(mapeamento.values())) # Monta a nova linha com todos os campos linha = [nome_cliente, data_formatada] for chave in mapeamento: valor = resultado.get(chave, "") if isinstance(valor, list): # Adiciona bullets e quebra de linha para melhor visualização valor = "\n".join(f"• {item.strip()}" for item in valor if item.strip()) linha.append(valor) ws.append(linha) wb.save(caminho_excel) print(f"Excel atualizado com todos os campos em: {caminho_excel}") return caminho_pdf @staticmethod def salvar_resumo_excel(nome_arquivo, resumo_dict, arquivo_excel): """ Salva o resumo gerado em um arquivo Excel. """ dados = {"Arquivo": nome_arquivo} if isinstance(resumo_dict, dict): dados.update({ "Principais Dúvidas": ", ".join(resumo_dict.get("duvidas", [])), "Tópicos Discutidos": ", ".join(resumo_dict.get("topicos", [])), "Soluções Sugeridas": ", ".join(resumo_dict.get("solucoes", [])), "Ações Necessárias": ", ".join(resumo_dict.get("acoes", [])), "Satisfação do Cliente": resumo_dict.get("satisfacao", "Não disponível"), "Objetivo da Reunião": resumo_dict.get("objetivo", "Não disponível"), }) # Cria um DataFrame com os dados do resumo df_novo = pd.DataFrame([dados]) # Verifica se o arquivo já existe e concatena os dados if os.path.exists(arquivo_excel): df_existente = pd.read_excel(arquivo_excel, engine="openpyxl") df_final = pd.concat([df_existente, df_novo], ignore_index=True) else: df_final = df_novo # Salva o DataFrame atualizado no arquivo Excel df_final.to_excel(arquivo_excel, index=False, engine="openpyxl") print(f"Resumo salvo em {arquivo_excel}")