import os
import logging
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes, CallbackQueryHandler
from wordpress_xmlrpc import Client, WordPressPost
from wordpress_xmlrpc.methods import posts, media
from wordpress_xmlrpc.compat import xmlrpc_client
from openai import OpenAI
import bs4
import requests
import re
from datetime import datetime
import urllib.parse
from bs4 import BeautifulSoup
import webbrowser
# Configuración de logging
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO
)
logger = logging.getLogger(__name__)
# Configuración
TELEGRAM_TOKEN = '7744861281:AAFV8lUPKH7PCNXrcGufQ6e8W6NMc--M5Lo' # Reemplaza con tu token de Telegram
OPENAI_API_KEY = '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' # Tu API key de OpenAI
WORDPRESS_URL = 'https://iaranda.es/xmlrpc.php' # URL de tu WordPress
WORDPRESS_USER = 'admin' # Usuario de WordPress
WORDPRESS_PASS = '8pE0 NscB KDEB 8rfy jyF8 KQUu' # Contraseña de aplicación de WordPress
# Configurar OpenAI
client = OpenAI(api_key=OPENAI_API_KEY)
# Categorías disponibles
CATEGORIAS = {
'audio': 43,
'imagen': 46,
'otros': 47,
'texto': 45,
'video': 44
}
class NewsBot:
def __init__(self):
"""Inicializa el bot con las configuraciones necesarias"""
self.wp = Client(WORDPRESS_URL, WORDPRESS_USER, WORDPRESS_PASS)
self.pending_posts = {} # Almacena posts pendientes por usuario
self.last_message = None
async def start(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Maneja el comando /start"""
await update.message.reply_text(
'¡Hola! Soy tu bot de noticias IA. 🤖\n\n'
'Envíame un enlace de Twitter/X, YouTube o cualquier artículo web '
'y te guiaré paso a paso para procesarlo y publicarlo en tu blog.\n\n'
'Comandos disponibles:\n'
'/start - Muestra este mensaje\n'
'/help - Muestra ayuda detallada\n'
'/inspiraccion - Procesa el último enlace como InspirAcción\n'
'/cancelar - Cancela el proceso actual'
)
async def help(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Maneja el comando /help"""
await update.message.reply_text(
'📱 Cómo usar el bot:\n\n'
'1. Envía cualquier enlace de:\n'
' - Twitter/X\n'
' - YouTube\n'
' - Artículos web\n\n'
'2. El bot procesará automáticamente el contenido\n\n'
'3. Para marcar como InspirAcción:\n'
' Usa /inspiraccion después de enviar el enlace\n\n'
'4. El contenido se publicará automáticamente en tu blog'
)
def extract_url(self, text):
"""Extrae la URL del texto del mensaje"""
url_pattern = r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+'
urls = re.findall(url_pattern, text)
return urls[0] if urls else None
async def process_new_link(self, message):
"""Procesa un nuevo enlace enviado por el usuario"""
try:
self.last_message = message
# Extraer URL del mensaje
url = self.extract_url(message.text)
if not url:
await message.reply_text(
'Por favor, envía un enlace válido de Twitter/X, YouTube o un artículo web.'
)
return
await message.reply_text("🔍 Analizando el enlace...")
# Procesar el enlace según su tipo
content_data = None
content_type = None
if 'twitter.com' in url or 'x.com' in url:
await message.reply_text("🐦 Procesando tweet...")
content_type = 'tweet'
content_data = self.process_tweet(url)
elif 'youtube.com' in url or 'youtu.be' in url:
await message.reply_text("📺 Procesando video de YouTube...")
content_type = 'youtube'
content_data = self.process_youtube(url)
else:
await message.reply_text("📄 Procesando artículo web...")
content_type = 'article'
content_data = self.process_article(url)
if not content_data:
await message.reply_text('❌ No se pudo procesar el contenido del enlace.')
return
# Añadir chat_id a content_data para poder guardar las etiquetas después
content_data['chat_id'] = message.chat.id
# Generar contenido con GPT
article_content = await self.generate_article(content_data, content_type)
if content_type == 'tweet' and not article_content:
# Si es un tweet sin texto significativo, solo usar el embed
final_content = content_data['embed_code']
else:
# Mostrar el contenido generado al usuario y permitir edición
keyboard = [
[
InlineKeyboardButton("✏️ Editar texto", callback_data="edit_content"),
InlineKeyboardButton("✅ Mantener texto", callback_data="keep_content")
]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await message.reply_text(
"✨ He generado el siguiente texto para el artículo:\n\n"
f"{article_content}\n\n"
"¿Quieres editar el texto o lo dejamos así?",
reply_markup=reply_markup
)
# Guardar datos temporalmente
if message.chat.id not in self.pending_posts:
self.pending_posts[message.chat.id] = {}
self.pending_posts[message.chat.id].update({
'content': article_content,
'embed_code': content_data['embed_code'],
'waiting_for': 'content_decision',
'is_inspiraccion': False
})
except Exception as e:
logger.error(f"Error en process_new_link: {str(e)}")
await message.reply_text(
f'❌ Ocurrió un error al procesar el enlace: {str(e)}'
)
def process_youtube(self, url):
"""Procesa un video de YouTube y obtiene su información"""
try:
logger.info(f"Procesando video de YouTube: {url}")
# Extraer ID del video de YouTube
video_id = None
if 'youtube.com/watch?v=' in url:
video_id = url.split('watch?v=')[1].split('&')[0]
elif 'youtu.be/' in url:
video_id = url.split('youtu.be/')[1].split('?')[0]
if not video_id:
raise ValueError("No se pudo extraer el ID del video")
logger.info(f"ID del video extraído: {video_id}")
# Usar la API de noembed para obtener información del video
api_url = f"https://noembed.com/embed?url=https://www.youtube.com/watch?v={video_id}"
response = requests.get(api_url)
data = response.json()
# Crear el código embebido con dimensiones más proporcionadas
embed_code = f'''
'''
logger.info("Información de YouTube extraída correctamente")
return {
'title': data.get('title', ''),
'author': data.get('author_name', ''),
'description': data.get('description', ''),
'url': url,
'embed_code': embed_code,
'type': 'youtube',
'content': f'Video de YouTube: {data.get("title", "")}'
}
except Exception as e:
logger.error(f"Error detallado procesando YouTube: {str(e)}")
logger.error(f"URL del video: {url}")
return None
def process_tweet(self, url):
"""Procesa un tweet y retorna su contenido embebido y descripción"""
try:
logger.info(f"Procesando tweet: {url}")
# Obtener el código embebido de Twitter
api_url = f"https://publish.twitter.com/oembed?url={url}&omit_script=true"
response = requests.get(api_url)
if response.status_code != 200:
logger.error(f"Error en la API de Twitter. Status code: {response.status_code}")
logger.error(f"Respuesta: {response.text}")
raise Exception("Error al obtener el tweet")
data = response.json()
# Extraer el texto del tweet del HTML
soup = BeautifulSoup(data['html'], 'html.parser')
tweet_text = soup.get_text()
# Centrar el tweet y darle un ancho máximo
embed_code = f'''
{data['html']}
'''
logger.info("Tweet procesado correctamente")
logger.info(f"Texto del tweet: {tweet_text}")
return {
'text': tweet_text,
'embed_code': embed_code,
'type': 'tweet',
'url': url
}
except Exception as e:
logger.error(f"Error al procesar tweet: {str(e)}")
return None
def process_article(self, url):
"""Procesa un artículo web y retorna su contenido"""
try:
logger.info(f"Procesando artículo web: {url}")
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
}
response = requests.get(url, headers=headers)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'html.parser')
# Extraer metadatos OpenGraph
og_title = soup.find('meta', property='og:title')
og_description = soup.find('meta', property='og:description')
og_image = soup.find('meta', property='og:image')
# Extraer título (OpenGraph o alternativas)
title = None
if og_title:
title = og_title.get('content')
else:
title_tag = soup.find('title')
if title_tag:
title = title_tag.text.strip()
logger.info(f"Título encontrado: {title}")
# Extraer descripción (OpenGraph o alternativas)
description = None
if og_description:
description = og_description.get('content')
else:
meta_desc = soup.find('meta', {'name': 'description'})
if meta_desc:
description = meta_desc.get('content')
# Extraer imagen (OpenGraph o alternativas)
image_url = None
if og_image:
image_url = og_image.get('content')
else:
# Buscar la primera imagen relevante
for img in soup.find_all('img', src=True):
if img.get('width', '').isdigit() and int(img['width']) > 200:
image_url = img['src']
break
# Intentar obtener el contenido principal
content = ""
main_content = soup.find('main')
if main_content:
logger.info("Contenido extraído de ")
content = main_content.get_text(strip=True)
else:
article = soup.find('article')
if article:
logger.info("Contenido extraído de ")
content = article.get_text(strip=True)
else:
body = soup.find('body')
if body:
logger.info("Contenido extraído de ")
content = body.get_text(strip=True)
logger.info(f"Contenido extraído (primeros 100 caracteres): {content[:100]}...")
# Crear un embed code más rico con imagen y descripción
embed_code = f'''
'''
return {
'title': title,
'description': description,
'image_url': image_url,
'content': content,
'url': url,
'type': 'article',
'embed_code': embed_code
}
except Exception as e:
logger.error(f"Error al procesar artículo web: {str(e)}")
logger.error(f"URL del artículo: {url}")
return None
async def generate_article(self, content_data, content_type):
"""Genera un artículo usando GPT"""
try:
# Informar al usuario
await self.last_message.reply_text("🤖 Generando contenido con GPT...")
if content_type == 'tweet' and len(content_data.get('text', '')) < 50:
return None
prompt = f"""Genera un artículo informativo y profesional sobre este contenido:
Tipo: {content_type}
Título: {content_data.get('title', '')}
Descripción: {content_data.get('description', '')}
URL: {content_data.get('url', '')}
El artículo debe:
- Ser informativo y objetivo
- Tener 2-3 párrafos
- Incluir detalles relevantes del contenido
- No usar frases genéricas ni llamadas a la acción
- No incluir etiquetas ni secciones adicionales"""
response = client.chat.completions.create(
model="gpt-4o-mini-2024-07-18",
messages=[{"role": "user", "content": prompt}],
temperature=0.7
)
return response.choices[0].message.content.strip()
except Exception as e:
logger.error(f"Error en generate_article: {str(e)}")
await self.last_message.reply_text("❌ Error al generar el contenido. Intentándolo de nuevo...")
return None
async def handle_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Maneja los mensajes recibidos"""
message = update.message
user_id = update.effective_user.id
self.last_message = message
if user_id not in self.pending_posts:
# Si no hay un post pendiente, procesar como nuevo enlace
await self.process_new_link(message)
return
post_data = self.pending_posts[user_id]
if post_data['waiting_for'] == 'edit_content':
await message.reply_text("✍️ Actualizando el contenido...")
post_data['content'] = message.text
post_data['waiting_for'] = 'category'
# Mostrar botones de categoría
keyboard = [
[
InlineKeyboardButton("Audio", callback_data="cat_audio"),
InlineKeyboardButton("Imagen", callback_data="cat_imagen"),
InlineKeyboardButton("Texto", callback_data="cat_texto")
],
[
InlineKeyboardButton("Video", callback_data="cat_video"),
InlineKeyboardButton("Otros", callback_data="cat_otros")
]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await message.reply_text(
'Selecciona la categoría del post:',
reply_markup=reply_markup
)
elif post_data['waiting_for'] == 'title':
await message.reply_text("📝 Guardando el título...")
post_data['title'] = message.text
post_data['waiting_for'] = 'tags'
# Mostrar etiquetas sugeridas si existen
suggested_tags = post_data.get('suggested_tags', '')
if suggested_tags:
await message.reply_text(
'Te sugiero estas etiquetas:\n'
f'{suggested_tags}\n\n'
'Puedes usar estas etiquetas o escribir otras diferentes, separadas por comas:'
)
else:
await message.reply_text(
'Por favor, introduce las etiquetas separadas por comas:'
)
elif post_data['waiting_for'] == 'tags':
await message.reply_text("🏷️ Procesando las etiquetas...")
# Procesar las etiquetas
tags = [tag.strip() for tag in message.text.split(',')]
post_data['tags'] = tags
# Preguntar si es InspirAcción
keyboard = [
[
InlineKeyboardButton("Sí", callback_data="inspiraccion_si"),
InlineKeyboardButton("No", callback_data="inspiraccion_no")
]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await message.reply_text(
'¿Es este post un InspirAcción?',
reply_markup=reply_markup
)
async def handle_callback(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Maneja las respuestas de los botones"""
query = update.callback_query
await query.answer()
user_id = update.effective_user.id
if user_id not in self.pending_posts:
await query.message.reply_text("La sesión ha expirado. Por favor, empieza de nuevo.")
return
post_data = self.pending_posts[user_id]
if query.data == "edit_content":
post_data['waiting_for'] = 'edit_content'
# Enviar el texto en formato código para facilitar la copia
await query.message.reply_text(
f"`{post_data['content']}`",
parse_mode='MarkdownV2'
)
elif query.data == "keep_content":
# Combinar el contenido con el embed code antes de continuar
post_data['content'] = post_data['content'] + "\n\n" + post_data['embed_code']
post_data['waiting_for'] = 'category'
# Mostrar botones de categoría
keyboard = [
[
InlineKeyboardButton("Audio", callback_data="cat_audio"),
InlineKeyboardButton("Imagen", callback_data="cat_imagen"),
InlineKeyboardButton("Texto", callback_data="cat_texto")
],
[
InlineKeyboardButton("Video", callback_data="cat_video"),
InlineKeyboardButton("Otros", callback_data="cat_otros")
]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.message.reply_text(
'Selecciona la categoría del post:',
reply_markup=reply_markup
)
# Manejo de categorías
elif query.data.startswith("cat_"):
category = query.data.replace("cat_", "")
post_data['category'] = category
# Generar título sugerido
suggested_title = await self.generate_title_with_gpt(post_data['content'])
post_data['suggested_title'] = suggested_title
post_data['waiting_for'] = 'title'
# Mostrar botones para el título
keyboard = [
[
InlineKeyboardButton("Aceptar título", callback_data="accept_title"),
InlineKeyboardButton("Escribir otro", callback_data="modify_title")
]
]
reply_markup = InlineKeyboardMarkup(keyboard)
await query.message.reply_text(
f'Te sugiero este título:\n"{suggested_title}"\n\n'
'Puedes aceptar este título o escribir uno nuevo:',
reply_markup=reply_markup
)
elif query.data == "accept_title":
post_data['title'] = post_data['suggested_title']
post_data['waiting_for'] = 'tags'
await query.edit_message_text(
f'Título aceptado: "{post_data["title"]}"\n\n'
'Por favor, introduce las etiquetas separadas por comas:'
)
elif query.data == "modify_title":
post_data['waiting_for'] = 'title'
await query.edit_message_text(
'Por favor, escribe el nuevo título:'
)
elif query.data == "inspiraccion_si":
post_data['is_inspiraccion'] = True
await self.prepare_post_for_publishing(query.message, post_data)
elif query.data == "inspiraccion_no":
post_data['is_inspiraccion'] = False
await self.prepare_post_for_publishing(query.message, post_data)
async def prepare_post_for_publishing(self, message, post_data):
"""Prepara el post para publicación"""
try:
await message.reply_text("📤 Preparando el post para publicación...")
# Crear el post
post = WordPressPost()
# Establecer el título y tags
if post_data['is_inspiraccion']:
post.title = "InspirAcción"
if 'InspirAcción' not in post_data['tags']:
post_data['tags'].append('InspirAcción')
else:
post.title = post_data['title']
# Establecer el contenido
# Asegurarse de que el embed_code se añada solo una vez
post.content = post_data['content']
if 'embed_code' in post_data and post_data['embed_code'] not in post.content:
post.content = post.content.rstrip() + "\n\n" + post_data['embed_code']
# Establecer la categoría
category_id = CATEGORIAS.get(post_data['category'].lower(), CATEGORIAS['otros'])
post.terms_names = {
'category': [post_data['category']],
'post_tag': post_data['tags']
}
# Establecer el estado como publicado
post.post_status = 'publish'
await message.reply_text("🎨 Generando imagen con DALL-E...")
# Generar imagen con DALL-E
image_path = await self.generate_image_with_dalle(post.title)
# Crear cliente WordPress
wp = Client(WORDPRESS_URL, WORDPRESS_USER, WORDPRESS_PASS)
# Subir la imagen destacada
if image_path and os.path.exists(image_path):
try:
await message.reply_text("🖼️ Subiendo imagen destacada...")
# Leer el archivo de imagen
with open(image_path, 'rb') as img:
image_data = img.read()
# Preparar datos para WordPress
data = {
'name': 'imagen.jpeg',
'type': 'image/jpeg',
'bits': xmlrpc_client.Binary(image_data)
}
# Subir la imagen
response = wp.call(media.UploadFile(data))
image_id = response['id']
# Establecer como imagen destacada
post.thumbnail = image_id
await message.reply_text("✅ Imagen destacada añadida correctamente")
# Limpiar el archivo temporal
os.remove(image_path)
except Exception as img_error:
logger.error(f"Error al subir la imagen destacada: {str(img_error)}")
await message.reply_text("⚠️ No se pudo añadir la imagen destacada")
else:
await message.reply_text("⚠️ No se pudo generar la imagen con DALL-E")
await message.reply_text("🌐 Publicando en WordPress...")
# Publicar el post
post_id = wp.call(posts.NewPost(post))
# Obtener la URL del post publicado
post_info = wp.call(posts.GetPost(post_id))
post_url = post_info.link
# Abrir el post en el navegador
webbrowser.open(post_url)
# Confirmar al usuario
await message.reply_text(
'✅ ¡Post publicado correctamente!\n\n'
f'Título: {post.title}\n'
f'Categoría: {post_data["category"]}\n'
f'Tags: {", ".join(post_data["tags"])}\n'
f'URL: {post_url}'
)
# Limpiar los datos del post
if message.chat.id in self.pending_posts:
del self.pending_posts[message.chat.id]
except Exception as e:
logger.error(f"Error al publicar: {str(e)}")
await message.reply_text(
f'❌ Error al publicar el post: {str(e)}'
)
async def generate_image_with_dalle(self, title):
"""Genera una imagen con DALL-E basada en el título"""
try:
# Crear el prompt combinando el título con el estilo preferido
prompt = f"{title}, vector illustration, in the style of studyblr, interactive art, simple vector, minimal composition"
# Generar imagen con DALL-E
response = client.images.generate(
model="dall-e-3",
prompt=prompt,
size="1024x1024",
quality="standard",
n=1,
)
# Obtener la URL de la imagen generada
image_url = response.data[0].url
# Descargar la imagen
image_response = requests.get(image_url)
image_response.raise_for_status()
# Guardar la imagen localmente
image_path = os.path.join(os.path.dirname(__file__), 'imagen.jpeg')
with open(image_path, 'wb') as f:
f.write(image_response.content)
return image_path
except Exception as e:
logger.error(f"Error generando imagen con DALL-E: {str(e)}")
return None
async def generate_title_with_gpt(self, content):
"""Genera un título sugerido usando GPT"""
try:
response = client.chat.completions.create(
model="gpt-4o-mini-2024-07-18",
messages=[
{"role": "system", "content": "Eres un experto en generar títulos atractivos y concisos para posts de blog."},
{"role": "user", "content": f"Genera un título atractivo y conciso (máximo 60 caracteres) para un post que contiene este contenido: {content}"}
]
)
return response.choices[0].message.content.strip('"')
except Exception as e:
logger.error(f"Error generando título con GPT: {str(e)}")
return "Nuevo post"
async def cancel(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Cancela el proceso actual"""
user_id = update.effective_user.id
if user_id in self.pending_posts:
del self.pending_posts[user_id]
await update.message.reply_text('Proceso cancelado. Puedes empezar de nuevo cuando quieras.')
else:
await update.message.reply_text('No hay ningún proceso activo para cancelar.')
async def inspiraccion(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Marca el último enlace como InspirAcción"""
user_id = update.effective_user.id
if user_id in self.pending_posts:
self.pending_posts[user_id]['is_inspiraccion'] = True
self.pending_posts[user_id]['title'] = 'InspirAcción'
await update.message.reply_text(
'Post marcado como InspirAcción.\n'
'Por favor, introduce las etiquetas separadas por comas:'
)
self.pending_posts[user_id]['waiting_for'] = 'tags'
else:
await update.message.reply_text(
'No hay ningún enlace pendiente de procesar. '
'Primero envía un enlace y luego usa este comando.'
)
def main():
"""Inicia el bot"""
application = Application.builder().token(TELEGRAM_TOKEN).build()
# Añadir manejadores
bot = NewsBot()
application.add_handler(CommandHandler("start", bot.start))
application.add_handler(CommandHandler("help", bot.help))
application.add_handler(CommandHandler("inspiraccion", bot.inspiraccion))
application.add_handler(CommandHandler("cancelar", bot.cancel))
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, bot.handle_message))
application.add_handler(CallbackQueryHandler(bot.handle_callback))
# Iniciar el bot
application.run_polling()
if __name__ == '__main__':
main()