Diseña una interfaz para transcribir videos de TikTok e Instagram sin Slack

Diseña una interfaz para este código de Python para ya no necesitar usar Slack: # app.py import os import logging import threading from slack_bolt import App from slack_bolt.adapter.socket_mode import SocketModeHandler from dotenv import load_dotenv import yt_dlp import whisper import boto3 from botocore.exceptions import NoCredentialsError, ClientError import re import requests import urllib.parse # Cargar variables de entorno load_dotenv() # Configuración del logging logging.basicConfig(level=logging.INFO) # Inicializar la aplicación de Slack Bolt con el token del bot slack_app = App(token=os.getenv("SLACK_BOT_TOKEN")) # Estado para almacenar el nivel de precisión user_precision_level = {} # Almacenar event_ids procesados processed_event_ids = set() # Descargar el modelo de Whisper una vez y reutilizarlo downloaded_models = {} def clean_url(url): """Limpia y normaliza la URL del video.""" # Elimina caracteres de marcado como '<' y '>' url = url.strip('<>') # Elimina el fragmento al final de la URL url = url.split('#')[0] # Decodifica caracteres de URL como %3E url = urllib.parse.unquote(url) # Elimina parámetros de consulta url = url.split('?')[0] # Elimina barras inclinadas al final url = url.rstrip('/') return url def expand_url(url): """Expande una URL acortada si es necesario.""" try: response = requests.get(url, allow_redirects=True, timeout=5) return response.url except requests.RequestException as e: logging.error(f"No se pudo expandir la URL. Error: {str(e)}") return url def download_video(url, video_id, platform): """Descargar video usando yt_dlp con manejo de cookies para Instagram.""" ydl_opts = { 'outtmpl': f'videos/{video_id}.mp4', 'format': 'best', 'quiet': False, # Cambia a False para obtener más detalles en los logs 'logger': logging.getLogger(), # Utiliza el logger existente } if platform == "Instagram": # Utilizar el archivo de cookies ydl_opts['cookiefile'] = 'instagram_cookies.txt' try: os.makedirs('videos', exist_ok=True) logging.info(f"ydl_opts utilizados: {ydl_opts}") with yt_dlp.YoutubeDL(ydl_opts) as ydl: ydl.download([url]) return f'videos/{video_id}.mp4' except Exception as e: logging.error(f"No se pudo descargar el video de {platform}. Error: {str(e)}") return None def transcribe_video(video_path, precision): """Transcribir el audio del video usando Whisper.""" model_size = "tiny" if precision == "medio": model_size = "small" elif precision == "alto": model_size = "base" if model_size not in downloaded_models: try: downloaded_models[model_size] = whisper.load_model(model_size) except Exception as e: logging.error(f"No se pudo cargar el modelo de Whisper. Error: {str(e)}") return None try: model = downloaded_models[model_size] result = model.transcribe(video_path) return result["text"] except Exception as e: logging.error(f"No se pudo transcribir el video. Error: {str(e)}") return None def upload_to_s3(file_path, bucket_name, object_name): """Subir un archivo a un bucket de S3.""" s3_client = boto3.client('s3') try: s3_client.upload_file(file_path, bucket_name, object_name) logging.info(f"Archivo {file_path} subido a S3 en {bucket_name}/{object_name}") return True except ClientError as e: logging.error(f"No se pudo subir el archivo a S3. Error: {str(e)}") return False except NoCredentialsError: logging.error("Credenciales de AWS no encontradas.") return False def process_transcription(client, channel_id, user_id, video_path, precision, message_ts): """Función para procesar la transcripción en un hilo separado.""" try: logging.info(f"Iniciando transcripción para el usuario {user_id} con precisión {precision}.") # Actualizar el mensaje con el progreso client.chat_update( channel=channel_id, ts=message_ts, text="", blocks=[ { "type": "section", "text": {"type": "mrkdwn", "text": f"*Nivel de precisión seleccionado:* {precision.capitalize()}"} }, { "type": "section", "text": {"type": "mrkdwn", "text": ":hourglass_flowing_sand: *Transcripción en progreso, por favor espera...*"} } ] ) # Transcribir el video transcript = transcribe_video(video_path, precision) if transcript: logging.info(f"Transcripción completada para el usuario {user_id}.") # Actualizar el mensaje con la transcripción client.chat_update( channel=channel_id, ts=message_ts, text="", blocks=[ { "type": "section", "text": {"type": "mrkdwn", "text": f"*Nivel de precisión seleccionado:* {precision.capitalize()}"} }, { "type": "section", "text": {"type": "mrkdwn", "text": ":white_check_mark: *Transcripción completada:*"} }, { "type": "section", "text": {"type": "mrkdwn", "text": transcript} } ] ) else: logging.error(f"Transcripción fallida para el usuario {user_id}.") client.chat_update( channel=channel_id, ts=message_ts, text="", blocks=[ { "type": "section", "text": {"type": "mrkdwn", "text": f"*Nivel de precisión seleccionado:* {precision.capitalize()}"} }, { "type": "section", "text": {"type": "mrkdwn", "text": ":x: *No se pudo transcribir el video. Por favor, inténtalo nuevamente más tarde.*"} } ] ) except Exception as e: logging.error(f"Error durante la transcripción para el usuario {user_id}: {e}") client.chat_update( channel=channel_id, ts=message_ts, text="", blocks=[ { "type": "section", "text": {"type": "mrkdwn", "text": f"*Nivel de precisión seleccionado:* {precision.capitalize()}"} }, { "type": "section", "text": {"type": "mrkdwn", "text": ":x: *Ocurrió un error durante la transcripción. Por favor, inténtalo más tarde.*"} } ] ) finally: # Eliminar el archivo local if os.path.exists(video_path): os.remove(video_path) logging.info(f"Archivo local {video_path} eliminado.") # Eliminar el estado del usuario if user_id in user_precision_level: del user_precision_level[user_id] @slack_app.event("app_mention") def handle_app_mentions(body, say, logger): event_id = body.get("event_id") if event_id in processed_event_ids: logging.info(f"Evento {event_id} ya procesado. Ignorando.") return else: processed_event_ids.add(event_id) user_id = body["event"]["user"] text = body["event"]["text"] logging.info(f"Evento de mención recibido del usuario {user_id}: {text}") url_match = re.search(r'(https?://[^\s]+)', text) if not url_match: say(text=f"<@{user_id}> No se encontró una URL válida en el mensaje. Por favor, proporciona una URL de TikTok o Instagram.") return video_url = url_match.group(0) video_url = clean_url(video_url) video_url = expand_url(video_url) logging.info(f"URL de video limpiada: {video_url}") platform = None if "tiktok.com" in video_url: platform = "TikTok" elif "instagram.com" in video_url: platform = "Instagram" if not platform: say(text=f"<@{user_id}> No se pudo determinar la plataforma del video. Por favor, proporciona una URL válida de TikTok o Instagram.") return say( text="", blocks=[ { "type": "section", "text": {"type": "mrkdwn", "text": ":arrow_forward: *Por favor, selecciona el nivel de precisión para la transcripción:*"} }, { "type": "actions", "elements": [ {"type": "button", "text": {"type": "plain_text", "text": "Bajo"}, "value": "bajo", "action_id": "select_precision_bajo"}, {"type": "button", "text": {"type": "plain_text", "text": "Medio"}, "value": "medio", "action_id": "select_precision_medio"}, {"type": "button", "text": {"type": "plain_text", "text": "Alto"}, "value": "alto", "action_id": "select_precision_alto"} ] } ] ) user_precision_level[user_id] = {"url": video_url, "platform": platform, "precision": None, "waiting_for_response": True} @slack_app.action("select_precision_bajo") @slack_app.action("select_precision_medio") @slack_app.action("select_precision_alto") def handle_precision_selection(ack, body, client, say, logger): ack() user_id = body["user"]["id"] precision = body["actions"][0]["value"] if user_id in user_precision_level and user_precision_level[user_id]["waiting_for_response"]: user_precision_level[user_id]["precision"] = precision user_precision_level[user_id]["waiting_for_response"] = False video_url = user_precision_level[user_id]["url"] platform = user_precision_level[user_id]["platform"] try: # Obtener el channel_id y ts correctamente channel_id = body["container"]["channel_id"] original_message_ts = body["container"]["message_ts"] # Actualizar el mensaje original para mostrar el nivel de precisión response = client.chat_update( channel=channel_id, ts=original_message_ts, text="", blocks=[ { "type": "section", "text": {"type": "mrkdwn", "text": f":sparkles: *Nivel de precisión seleccionado:* {precision.capitalize()}"} }, { "type": "section", "text": {"type": "mrkdwn", "text": ":hourglass: *Obteniendo la transcripción... Esto puede tomar un momento.*"} } ] ) message_ts = response["ts"] except Exception as e: logger.error(f"Error al actualizar el mensaje: {e}") return video_id_match = None if platform == "TikTok": video_id_match = re.search(r'video/([\w ]+)', video_url) elif platform == "Instagram": video_id_match = re.search(r'/p/([\w ]+)|/reel/([\w ]+)|/tv/([\w ]+)', video_url) if video_id_match: video_id = next(filter(None, video_id_match.groups())) else: client.chat_update( channel=channel_id, ts=message_ts, text="", blocks=[ { "type": "section", "text": {"type": "mrkdwn", "text": f":x: *No se pudo extraer el ID del video de la URL proporcionada. Por favor, verifica la URL e inténtalo nuevamente.*"} } ] ) del user_precision_level[user_id] return video_path = download_video(video_url, video_id, platform) if video_path: # Actualizar el mensaje para indicar que el video se obtuvo correctamente client.chat_update( channel=channel_id, ts=message_ts, text="", blocks=[ { "type": "section", "text": {"type": "mrkdwn", "text": f":sparkles: *Nivel de precisión seleccionado:* {precision.capitalize()}"} }, { "type": "section", "text": {"type": "mrkdwn", "text": ":white_check_mark: *El video se ha obtenido correctamente. Generando la transcripción...*"} } ] ) # Subir el video a S3 bucket_name = os.getenv("AWS_S3_BUCKET") upload_success = upload_to_s3(video_path, bucket_name, f"{video_id}.mp4") if not upload_success: client.chat_update( channel=channel_id, ts=message_ts, text="", blocks=[ { "type": "section", "text": {"type": "mrkdwn", "text": f":x: *No se pudo subir el video a S3. Por favor, inténtalo nuevamente.*"} } ] ) del user_precision_level[user_id] return # Iniciar la transcripción en un hilo separado transcription_thread = threading.Thread(target=process_transcription, args=(client, channel_id, user_id, video_path, precision, message_ts)) transcription_thread.start() else: client.chat_update( channel=channel_id, ts=message_ts, text="", blocks=[ { "type": "section", "text": {"type": "mrkdwn", "text": ":x: *No se pudo descargar el video. Por favor, verifica la URL e inténtalo nuevamente.*"} } ] ) del user_precision_level[user_id] else: # Si el usuario ya ha hecho una selección, ignoramos clics adicionales pass # Iniciar el Socket Mode Handler para ejecutar la aplicación if __name__ == "__main__": handler = SocketModeHandler(slack_app, os.getenv("SLACK_APP_TOKEN")) handler.start()