initial commit

This commit is contained in:
teraflops 2025-05-29 22:58:53 +02:00
commit c8bf562b04
Signed by: teraflops
GPG Key ID: 2B77D97AF6F8968C
59 changed files with 3996 additions and 0 deletions

86
Dockerfile Normal file
View File

@ -0,0 +1,86 @@
FROM ubuntu:latest
# Configurar el entorno para instalación sin interacción
ENV DEBIAN_FRONTEND=noninteractive
# Crear usuario y grupo icecast2 (sin privilegios)
RUN useradd -m -d /home/icecast2 -s /bin/bash icecast2
# Instalar dependencias del sistema
RUN apt update && apt install -y --no-install-recommends \
python3 python3-pip ffmpeg liquidsoap icecast2 socat curl gettext vim\
&& rm -rf /var/lib/apt/lists/*
# Descargar e instalar la última versión de yt-dlp manualmente
RUN curl -L https://github.com/yt-dlp/yt-dlp/releases/download/2025.03.27/yt-dlp_linux -o /usr/local/bin/yt-dlp \
&& chmod a+rx /usr/local/bin/yt-dlp
# Crear carpetas necesarias y dar permisos adecuados
RUN mkdir -p \
/config \
/var/lib/radio-data \
/usr/local/icecast/logs \
/app \
/var/log/liquidsoap \
/var/lib/bot \
&& chown -R icecast2:icecast2 \
/config \
/var/lib/radio-data \
/usr/local/icecast/logs \
/app \
/var/log/liquidsoap \
/var/lib/bot \
&& chmod -R 755 \
/config \
/var/lib/radio-data \
/usr/local/icecast/logs \
/app \
/var/log/liquidsoap \
/var/lib/bot
RUN for f in /etc/icecast2/web/*.xsl; do \
realpath=$(realpath "$f"); \
dest="/usr/share/icecast2/web/$(basename "$f")"; \
if [ "$(realpath "$dest")" = "$realpath" ]; then \
echo "Skipping $f, already points to real file"; \
else \
cp "$realpath" "$dest"; \
fi; \
done && \
chmod -R a+rX /usr/share/icecast2/web
RUN mkdir -p /var/lib/radio-data/music /var/lib/radio && \
ln -sf /var/lib/radio-data/music /var/lib/radio/music
# Copiar y instalar dependencias de Python
COPY requirements.txt /app/requirements.txt
RUN pip install --no-cache-dir --break-system-packages -r /app/requirements.txt
# Copiar configuraciones al contenedor
COPY config/icecast.xml /config/icecast.xml
COPY config/liquidsoap.liq /config/liquidsoap.liq
COPY config/entrypoint.sh /entrypoint.sh
RUN chmod +x /entrypoint.sh
# Copiar el bot y sus plugins
COPY bot.py /app/bot.py
COPY plugins /app/plugins
# Definir directorio de trabajo
WORKDIR /app
# Crear directorio para el socket y asignar permisos correctos
RUN mkdir -p /run/liquidsoap && \
chown -R icecast2:icecast2 /run/liquidsoap && \
chmod 777 /run/liquidsoap
# Cambiar a usuario sin privilegios
USER icecast2
# Exponer puertos necesarios para Icecast
EXPOSE 8000
# Ejecutar el entrypoint
CMD ["sh", "/entrypoint.sh"]

93
README.md Normal file
View File

@ -0,0 +1,93 @@
# myircbot
## Getting started
To make it easy for you to get started with GitLab, here's a list of recommended next steps.
Already a pro? Just edit this README.md and make it your own. Want to make it easy? [Use the template at the bottom](#editing-this-readme)!
## Add your files
- [ ] [Create](https://docs.gitlab.com/ee/user/project/repository/web_editor.html#create-a-file) or [upload](https://docs.gitlab.com/ee/user/project/repository/web_editor.html#upload-a-file) files
- [ ] [Add files using the command line](https://docs.gitlab.com/ee/gitlab-basics/add-file.html#add-a-file-using-the-command-line) or push an existing Git repository with the following command:
```
cd existing_repo
git remote add origin https://gitlab.com/teraflops/myircbot.git
git branch -M main
git push -uf origin main
```
## Integrate with your tools
- [ ] [Set up project integrations](https://gitlab.com/teraflops/myircbot/-/settings/integrations)
## Collaborate with your team
- [ ] [Invite team members and collaborators](https://docs.gitlab.com/ee/user/project/members/)
- [ ] [Create a new merge request](https://docs.gitlab.com/ee/user/project/merge_requests/creating_merge_requests.html)
- [ ] [Automatically close issues from merge requests](https://docs.gitlab.com/ee/user/project/issues/managing_issues.html#closing-issues-automatically)
- [ ] [Enable merge request approvals](https://docs.gitlab.com/ee/user/project/merge_requests/approvals/)
- [ ] [Set auto-merge](https://docs.gitlab.com/ee/user/project/merge_requests/merge_when_pipeline_succeeds.html)
## Test and Deploy
Use the built-in continuous integration in GitLab.
- [ ] [Get started with GitLab CI/CD](https://docs.gitlab.com/ee/ci/quick_start/)
- [ ] [Analyze your code for known vulnerabilities with Static Application Security Testing (SAST)](https://docs.gitlab.com/ee/user/application_security/sast/)
- [ ] [Deploy to Kubernetes, Amazon EC2, or Amazon ECS using Auto Deploy](https://docs.gitlab.com/ee/topics/autodevops/requirements.html)
- [ ] [Use pull-based deployments for improved Kubernetes management](https://docs.gitlab.com/ee/user/clusters/agent/)
- [ ] [Set up protected environments](https://docs.gitlab.com/ee/ci/environments/protected_environments.html)
***
# Editing this README
When you're ready to make this README your own, just edit this file and use the handy template below (or feel free to structure it however you want - this is just a starting point!). Thanks to [makeareadme.com](https://www.makeareadme.com/) for this template.
## Suggestions for a good README
Every project is different, so consider which of these sections apply to yours. The sections used in the template are suggestions for most open source projects. Also keep in mind that while a README can be too long and detailed, too long is better than too short. If you think your README is too long, consider utilizing another form of documentation rather than cutting out information.
## Name
Choose a self-explaining name for your project.
## Description
Let people know what your project can do specifically. Provide context and add a link to any reference visitors might be unfamiliar with. A list of Features or a Background subsection can also be added here. If there are alternatives to your project, this is a good place to list differentiating factors.
## Badges
On some READMEs, you may see small images that convey metadata, such as whether or not all the tests are passing for the project. You can use Shields to add some to your README. Many services also have instructions for adding a badge.
## Visuals
Depending on what you are making, it can be a good idea to include screenshots or even a video (you'll frequently see GIFs rather than actual videos). Tools like ttygif can help, but check out Asciinema for a more sophisticated method.
## Installation
Within a particular ecosystem, there may be a common way of installing things, such as using Yarn, NuGet, or Homebrew. However, consider the possibility that whoever is reading your README is a novice and would like more guidance. Listing specific steps helps remove ambiguity and gets people to using your project as quickly as possible. If it only runs in a specific context like a particular programming language version or operating system or has dependencies that have to be installed manually, also add a Requirements subsection.
## Usage
Use examples liberally, and show the expected output if you can. It's helpful to have inline the smallest example of usage that you can demonstrate, while providing links to more sophisticated examples if they are too long to reasonably include in the README.
## Support
Tell people where they can go to for help. It can be any combination of an issue tracker, a chat room, an email address, etc.
## Roadmap
If you have ideas for releases in the future, it is a good idea to list them in the README.
## Contributing
State if you are open to contributions and what your requirements are for accepting them.
For people who want to make changes to your project, it's helpful to have some documentation on how to get started. Perhaps there is a script that they should run or some environment variables that they need to set. Make these steps explicit. These instructions could also be useful to your future self.
You can also document commands to lint the code or run tests. These steps help to ensure high code quality and reduce the likelihood that the changes inadvertently break something. Having instructions for running tests is especially helpful if it requires external setup, such as starting a Selenium server for testing in a browser.
## Authors and acknowledgment
Show your appreciation to those who have contributed to the project.
## License
For open source projects, say how it is licensed.
## Project status
If you have run out of energy or time for your project, put a note at the top of the README saying that development has slowed down or stopped completely. Someone may choose to fork your project or volunteer to step in as a maintainer or owner, allowing your project to keep going. You can also make an explicit request for maintainers.

657
bot.py Normal file
View File

@ -0,0 +1,657 @@
import socket
import ssl
import base64
import os
import threading
import importlib
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
import queue
import time
from plugins import news
from plugins import radio
from plugins.database import store_message
import traceback
from plugins.database import init_db as init_db_messages
from plugins.auth import init_db as init_db_auth
# Inicializar ambas bases de datos
init_db_messages() # Base de datos de mensajes
init_db_auth()
ADMIN_PASSWORD = os.getenv("IRC_ADMIN_PASSWORD")
password = os.getenv("IRC_BOT_PASSWORD")
if not password:
raise ValueError("Error: La variable de entorno IRC_BOT_PASSWORD no está definida.")
class IRCBot:
def __init__(self, server, port, nickname, username, password, channels,
command_prefix="!", control_socket="/tmp/ircbot.sock"):
self.server = server
self.port = port
self.nickname = nickname
self.username = username
self.password = password
self.channels = channels # Lista de canales
self.command_prefix = command_prefix
self.control_socket = control_socket
self.authenticated = False
self.joined = False
self.plugins = {}
self.admins = set() # Lista de administradores autenticados
self.msg_queue = queue.Queue()
self.running = True
# Inicia el hilo que envía los mensajes en cola (para throttling).
threading.Thread(target=self._message_worker, daemon=True).start()
# Carga plugins
self.load_plugins()
# Monitoreo en un hilo separado (ejemplo: radio, news)
threading.Thread(target=radio.monitor_liquidsoap, daemon=True).start()
news_plugin = news.NewsPlugin() # Instancia de la clase
threading.Thread(target=news_plugin.announce_news, args=(self.send_raw, self), daemon=True).start()
# Conexión TLS
context = ssl.create_default_context()
self.sock = socket.create_connection((self.server, self.port))
self.sock = context.wrap_socket(self.sock, server_hostname=self.server)
print(f" Conectado a {self.server}:{self.port}")
# SASL
self.send_raw("CAP REQ :sasl")
self.send_raw(f"NICK {self.nickname}")
self.send_raw(f"USER {self.username} 0 * :{self.nickname}")
# Iniciar socket UNIX (control externo)
threading.Thread(target=self.control_socket_server, daemon=True).start()
# Iniciar escucha del servidor IRC
self.listen()
def load_plugins(self, plugin_dir="plugins"):
"""Carga o recarga dinámicamente los plugins desde el directorio especificado."""
if not os.path.exists(plugin_dir):
os.makedirs(plugin_dir)
self.plugins = {} # Reinicia la lista de plugins
for file in os.listdir(plugin_dir):
if file.endswith(".py") and file != "__init__.py":
name = file[:-3] # Nombre sin `.py`
try:
module = importlib.import_module(f"{plugin_dir}.{name}")
# Plugins especiales con lógica personalizada
if name == "grab" and hasattr(module, "GrabPlugin"):
self.plugins["grab"] = module.GrabPlugin()
print(f" Plugin especial '{name}' instanciado.")
continue
if name == "alias" and hasattr(module, "AliasPlugin"):
alias_plugin = module.AliasPlugin(self.send_raw, self)
self.plugins["alias"] = alias_plugin
# Registrar alias como comandos
dynamic_cmds = alias_plugin.get_dynamic_commands()
for alias_name, alias_function in dynamic_cmds.items():
self.plugins[alias_name] = alias_function
print(f" Alias dinámico registrado: {alias_name}")
continue
if name == "uptime" and hasattr(module, "UptimePlugin"):
self.plugins["uptime"] = module.UptimePlugin(self)
print(f" Plugin especial '{name}' instanciado con `bot`.")
continue
if name == "give" and hasattr(module, "GivePlugin"):
self.plugins["give"] = module.GivePlugin(self.plugins)
print(f" Plugin especial '{name}' instanciado con referencia a otros plugins.")
continue
# Instanciar clases convencionales como XyzPlugin
class_name = f"{name.capitalize()}Plugin"
if hasattr(module, class_name):
self.plugins[name] = getattr(module, class_name)()
print(f" Plugin '{name}' instanciado como clase ({class_name}).")
continue
# Si tiene run() directamente, cargar como módulo plano
if hasattr(module, "run"):
self.plugins[name] = module
print(f" Plugin '{name}' cargado como módulo.")
continue
print(f" ⚠️ El plugin '{name}' no tiene 'run()' ni clase '{class_name}', se omitirá.")
except Exception as e:
error_trace = traceback.format_exc()
print(f" ❌ Error al cargar el plugin '{name}': {e}\n{error_trace}")
print(" ✅ Todos los plugins han sido cargados correctamente.")
def send_raw(self, message):
"""Envía un mensaje al servidor IRC inmediatamente (para conexiones, etc.)."""
try:
if self.sock and not self.sock._closed:
print(f">> {message}")
self.sock.send((message + "\r\n").encode("utf-8"))
else:
print(" Socket no disponible, intentando reconectar...")
self.connect()
except Exception as e:
print(f" Error al enviar mensaje: {e}")
self.handle_reconnect()
def send_queued_message(self, target, message):
"""Coloca un mensaje en cola para enviarlo con retraso (throttling)."""
max_length = 400 # Máx. caracteres para evitar flood
if len(message) > max_length:
parts = [message[i:i + max_length] for i in range(0, len(message), max_length)]
for part in parts:
self.msg_queue.put(f"PRIVMSG {target} :{part}")
else:
self.msg_queue.put(f"PRIVMSG {target} :{message}")
def _message_worker(self):
"""Hilo que gestiona la salida de mensajes con throttling."""
while self.running:
try:
message = self.msg_queue.get()
self.send_raw(message)
time.sleep(1) # Ajusta el delay si el servidor lo requiere
except Exception as e:
print(f" Error al enviar mensaje de la cola: {e}")
def join_channels(self):
"""Unirse a todos los canales configurados."""
for channel in self.channels:
self.send_raw(f"JOIN {channel}")
print(f" Unido a {channel}")
self.joined = True
def listen(self):
"""Escucha mensajes del servidor y maneja comandos."""
print(" Escuchando mensajes del servidor...")
while True:
try:
data = self.sock.recv(4096).decode("utf-8", errors="ignore").strip()
if not data:
print(" El servidor cerró la conexión.")
break
for line in data.split("\r\n"):
if not line:
continue
print(f"<< {line}") # Debug
if line.startswith("PING"):
self.send_raw(f"PONG {line.split()[1]}")
if "CAP" in line and "ACK :sasl" in line:
self.send_raw("AUTHENTICATE PLAIN")
if "AUTHENTICATE +" in line:
auth_string = base64.b64encode(
f"{self.username}\0{self.username}\0{self.password}".encode()
).decode()
self.send_raw(f"AUTHENTICATE {auth_string}")
if " 900 " in line or " 903 " in line:
print(" SASL autenticado correctamente.")
self.authenticated = True
self.send_raw("CAP END")
if ((" 376 " in line) or (" 422 " in line)) and self.authenticated and not self.joined:
self.join_channels()
# Mensajes PRIVMSG
if "PRIVMSG" in line:
self.handle_message(line)
except Exception as e:
print(f" Error en la conexión: {e}")
break
def handle_message(self, raw_line):
"""
Procesa los PRIVMSG del IRC y maneja autenticación para plugins restringidos.
"""
parts = raw_line.split()
if len(parts) < 4:
return
sender_info = parts[0]
command_type = parts[1]
target = parts[2]
msg_raw = " ".join(parts[3:])[1:] # Quitar el ':' inicial
sender_nick = sender_info.split("!")[0][1:]
# **Verificar si el mensaje es un privado (PM)**
if target.lower() == self.nickname.lower():
print(f"DEBUG: Recibido PM de {sender_nick}: {msg_raw}") # Log de depuración
self.handle_private_message(sender_nick, msg_raw)
return
# Evitar almacenar comandos sensibles como `.auth login usuario contraseña`
SENSITIVE_COMMANDS = ["auth", "login", "register"]
if msg_raw.startswith(self.command_prefix):
command_line = msg_raw[len(self.command_prefix):].strip()
tokens = command_line.split()
if tokens and tokens[0] in SENSITIVE_COMMANDS:
print(f" Mensaje de {sender_nick} omitido (comando sensible): {msg_raw}")
else:
store_message(sender_nick, target, msg_raw) # Guardar en SQLite si NO es un comando sensible
else:
store_message(sender_nick, target, msg_raw) # Guardar en SQLite
# **Guardar última frase del usuario para `.grab`**
if "grab" in self.plugins:
try:
self.plugins["grab"].store_last_message(sender_nick, msg_raw)
except Exception as e:
print(f" Error al almacenar mensaje en grab: {e}")
# **Procesar comandos si empiezan con el prefijo**
if command_type == "PRIVMSG" and msg_raw.startswith(self.command_prefix):
command_line = msg_raw[len(self.command_prefix):].strip()
tokens = command_line.split()
if not tokens:
return
cmd = tokens[0] # Comando (ej: "give", "alias", "grab", "ping")
args = tokens[1:]
# **Verificar si el usuario es admin y saltar autenticación**
if sender_nick in self.admins:
print(f" {sender_nick} es admin, ejecutando '.{cmd}' sin restricciones.")
else:
# **Verificar si el comando requiere autenticación**
restricted_plugins = ["grab"]
if cmd in restricted_plugins:
if "auth" in self.plugins and not self.plugins["auth"].is_authenticated(sender_nick):
self.send_queued_message(target, f" {sender_nick}, debes autenticarte para usar '.{cmd}'")
return
# **Si el comando existe en `self.plugins`, ejecutarlo**
if cmd in self.plugins:
plugin_or_func = self.plugins[cmd]
try:
if callable(plugin_or_func):
response = plugin_or_func(sender_nick, *args)
else:
response = plugin_or_func.run(sender_nick, *args)
if isinstance(response, str) and response:
for line in response.split("\n"):
self.send_queued_message(target, line)
except Exception as e:
error_trace = traceback.format_exc()
print(f" Error en '{cmd}': {e}\n{error_trace}")
self.send_queued_message(target, f" Error en '{cmd}': {str(e)}")
elif cmd == "help":
if args:
cmd_help = args[0]
if cmd_help in self.plugins and hasattr(self.plugins[cmd_help], "help"):
desc = self.plugins[cmd_help].help()
self.send_queued_message(target, desc)
else:
self.send_queued_message(target, f" No hay ayuda disponible para '.{cmd_help}'")
else:
# Listar comandos disponibles
disponibles = " ".join([f"`.{p}`" for p in self.plugins.keys()])
lines = [
" **Comandos disponibles:**",
disponibles,
" Usa .help <comando> para más detalles."
]
for l in lines:
self.send_queued_message(target, l)
else:
# Comando no encontrado
self.send_queued_message(target, f" No existe el comando '.{cmd}'")
def handle_private_message(self, sender, msg):
"""Maneja mensajes privados, autenticación de administradores y comandos de plugins."""
print(f" Mensaje privado recibido de {sender}: {msg}") # DEBUG
# **Manejar autenticación de administradores con `/msg bot op password`**
if msg.startswith("op "):
password_provided = msg.split(" ", 1)[1]
print(f" Intento de autenticación de {sender} con contraseña: {password_provided}") # DEBUG
if password_provided == ADMIN_PASSWORD:
self.admins.add(sender)
self.send_raw(f"PRIVMSG {sender} : Has sido autenticado como administrador.")
print(f" {sender} ahora es administrador.") # DEBUG
else:
self.send_raw(f"PRIVMSG {sender} : Contraseña incorrecta.")
print(f" {sender} intentó autenticarse con una contraseña incorrecta.") # DEBUG
return
# **Si el mensaje comienza con el prefijo de comandos, procesarlo**
if msg.startswith(self.command_prefix):
command_line = msg[len(self.command_prefix):].strip()
tokens = command_line.split()
if not tokens:
return
cmd = tokens[0] # Extraer el comando (`auth`, `grab`, etc.)
args = tokens[1:] # Extraer los argumentos
if cmd in self.plugins:
try:
print(f" Ejecutando {cmd} en PM con args={args}") # Debug
response = self.plugins[cmd].run(sender, *args) # FIX: Solo un `sender`
if response:
self.send_raw(f"PRIVMSG {sender} :{response}")
return
except Exception as e:
self.send_raw(f"PRIVMSG {sender} : Error en '{cmd}': {str(e)}")
print(f" Error en '{cmd}': {e}")
return
# **Si el usuario es administrador, manejar comandos especiales**
if sender in self.admins:
parts = msg.split()
if len(parts) == 0:
return # Si el mensaje está vacío, salir
command = parts[0] # Extraer el comando principal
print(f" {sender} ejecutando comando admin: {command}")
if command == "status":
self.send_raw(f"PRIVMSG {sender} :Bot activo | Plugins cargados: {len(self.plugins)}")
elif command == "join" and len(parts) > 1:
self.send_raw(f"JOIN {parts[1]}")
self.send_raw(f"PRIVMSG {sender} :Unido a {parts[1]}")
elif command == "part" and len(parts) > 1:
self.send_raw(f"PART {parts[1]}")
self.send_raw(f"PRIVMSG {sender} :Salió de {parts[1]}")
elif command == "list_plugins":
plist = ", ".join(self.plugins.keys()) if self.plugins else "No hay plugins cargados."
self.send_raw(f"PRIVMSG {sender} :Plugins cargados: {plist}")
elif command == "load" and len(parts) > 1:
plugin_name = parts[1]
try:
module = importlib.import_module(f"plugins.{plugin_name}")
self.plugins[plugin_name] = module
self.send_raw(f"PRIVMSG {sender} :Plugin {plugin_name} cargado.")
print(f" {sender} cargó el plugin {plugin_name}")
except Exception as e:
self.send_raw(f"PRIVMSG {sender} :Error al cargar plugin {plugin_name}: {e}")
elif command == "unload" and len(parts) > 1:
plugin_name = parts[1]
if plugin_name in self.plugins:
del self.plugins[plugin_name]
self.send_raw(f"PRIVMSG {sender} :Plugin {plugin_name} descargado.")
print(f" {sender} descargó el plugin {plugin_name}")
else:
self.send_raw(f"PRIVMSG {sender} :Plugin {plugin_name} no encontrado.")
elif command == "reload_plugins":
self.load_plugins()
self.send_raw(f"PRIVMSG {sender} :Plugins recargados correctamente.")
elif command == "quit":
self.send_raw("QUIT :Apagando bot...")
self.send_raw(f"PRIVMSG {sender} :Bot desconectado.")
os._exit(0)
elif command == "restart":
self.send_raw("QUIT :Reiniciando bot...")
self.send_raw(f"PRIVMSG {sender} :Bot reiniciándose.")
os.execv(__file__, [])
elif command == "reconnect":
self.send_raw("QUIT :Reconectando...")
self.send_raw(f"PRIVMSG {sender} :Reconectando al servidor IRC.")
self.sock.close()
self.__init__(self.server, self.port, self.nickname, self.username, self.password, self.channels)
elif command == "nick" and len(parts) > 1:
self.send_raw(f"NICK {parts[1]}")
self.send_raw(f"PRIVMSG {sender} :🆔 Nick cambiado a {parts[1]}")
elif command == "msg" and len(parts) > 2:
channel = parts[1]
message = " ".join(parts[2:])
self.send_raw(f"PRIVMSG {channel} :{message}")
self.send_raw(f"PRIVMSG {sender} :Mensaje enviado a {channel}")
elif command == "raw" and len(parts) > 1:
raw_cmd = " ".join(parts[1:])
self.send_raw(raw_cmd)
self.send_raw(f"PRIVMSG {sender} :Comando IRC enviado: {raw_cmd}")
elif command == "kick" and len(parts) > 2:
channel = parts[1]
user = parts[2]
self.send_raw(f"KICK {channel} {user} :Expulsado por {sender}")
self.send_raw(f"PRIVMSG {sender} :{user} ha sido expulsado de {channel}")
elif command == "ban" and len(parts) > 2:
channel = parts[1]
user = parts[2]
self.send_raw(f"MODE {channel} +b {user}")
self.send_raw(f"PRIVMSG {sender} :{user} ha sido baneado en {channel}")
elif command == "unban" and len(parts) > 2:
channel = parts[1]
user = parts[2]
self.send_raw(f"MODE {channel} -b {user}")
self.send_raw(f"PRIVMSG {sender} :{user} ha sido desbaneado en {channel}")
elif command == "op" and len(parts) > 2:
channel = parts[1]
user = parts[2]
self.send_raw(f"MODE {channel} +o {user}")
self.send_raw(f"PRIVMSG {sender} :{user} ahora tiene OP en {channel}")
elif command == "deop" and len(parts) > 2:
channel = parts[1]
user = parts[2]
self.send_raw(f"MODE {channel} -o {user}")
self.send_raw(f"PRIVMSG {sender} :{user} ya no tiene OP en {channel}")
elif command == "help":
comandos = [
" **Comandos disponibles:**",
"- `status` → Estado del bot y plugins",
"- `join <#canal>` → Unirse a un canal",
"- `part <#canal>` → Salir de un canal",
"- `list_plugins` → Listar plugins activos",
"- `load <plugin>` → Cargar un plugin",
"- `unload <plugin>` → Descargar un plugin",
"- `reload_plugins` → Recargar todos los plugins",
"- `quit` → Apagar el bot",
"- `restart` → Reiniciar el bot",
"- `reconnect` → Reconectar al servidor IRC",
"- `nick <nuevo_nick>` → Cambiar el apodo del bot",
"- `msg <#canal> <mensaje>` → Enviar mensaje a un canal",
"- `raw <comando>` → Enviar un comando IRC crudo",
"- `kick <#canal> <usuario>` → Expulsar usuario",
"- `ban <#canal> <usuario>` → Banear usuario",
"- `unban <#canal> <usuario>` → Desbanear usuario",
"- `op <#canal> <usuario>` → Dar OP a un usuario",
"- `deop <#canal> <usuario>` → Quitar OP a un usuario",
"- `logout` → Cerrar sesión de admin",
"🆘 Usa `.help <comando>` para obtener ayuda detallada.",
]
for line in comandos:
self.send_queued_message(sender, line)
elif command == "logout":
self.admins.discard(sender)
self.send_raw(f"PRIVMSG {sender} :Has cerrado sesión como administrador.")
print(f" {sender} ha cerrado sesión.")
else:
self.send_raw(f"PRIVMSG {sender} :Comando no reconocido. Usa 'help' para ver opciones.")
else:
self.send_raw(f"PRIVMSG {sender} :No tienes permisos. Usa 'op <contraseña>' para autenticarte.")
def control_socket_server(self):
"""Crea un socket UNIX para recibir comandos desde consola o socat."""
if os.path.exists(self.control_socket):
os.remove(self.control_socket)
server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
server.bind(self.control_socket)
server.listen(1)
print(f" Control socket activo en {self.control_socket}")
while True:
conn, _ = server.accept()
with conn:
data = conn.recv(1024).decode().strip()
if not data:
continue
print(f" Comando recibido desde consola: {data}")
parts = data.split()
if not parts:
continue
cmd = parts[0]
if cmd == "join" and len(parts) > 1:
self.send_raw(f"JOIN {parts[1]}")
print(f" Unido a {parts[1]}")
elif cmd == "part" and len(parts) > 1:
self.send_raw(f"PART {parts[1]}")
print(f" Salió de {parts[1]}")
elif cmd == "list_channels":
ch_list = ", ".join(self.channels)
conn.sendall(f"Canales actuales: {ch_list}\n".encode())
elif cmd == "load" and len(parts) > 1:
plug = parts[1]
try:
module = importlib.import_module(f"plugins.{plug}")
self.plugins[plug] = module
print(f" Plugin {plug} cargado")
conn.sendall(f"Plugin {plug} cargado\n".encode())
except Exception as e:
print(f" Error al cargar plugin {plug}: {e}")
conn.sendall(f"Error al cargar plugin {plug}: {e}\n".encode())
elif cmd == "unload" and len(parts) > 1:
plug = parts[1]
if plug in self.plugins:
del self.plugins[plug]
print(f" Plugin {plug} descargado")
conn.sendall(f"Plugin {plug} descargado\n".encode())
else:
conn.sendall(f"Plugin {plug} no encontrado\n".encode())
elif cmd == "list_plugins":
plist = ", ".join(self.plugins.keys()) if self.plugins else "No hay plugins cargados."
conn.sendall(f"Plugins cargados: {plist}\n".encode())
elif cmd == "quit":
self.send_raw("QUIT :Apagando bot...")
conn.sendall(" Bot desconectado.\n".encode())
os._exit(0)
elif cmd == "msg" and len(parts) > 2:
channel = parts[1]
message = " ".join(parts[2:])
self.send_raw(f"PRIVMSG {channel} :{message}")
conn.sendall(f" Mensaje enviado a {channel}\n".encode())
elif cmd == "raw" and len(parts) > 1:
raw_command = " ".join(parts[1:])
self.send_raw(raw_command)
conn.sendall(f" Comando IRC enviado: {raw_command}\n".encode())
elif cmd == "kick" and len(parts) > 2:
channel = parts[1]
user = parts[2]
self.send_raw(f"KICK {channel} {user} :Expulsado")
conn.sendall(f" {user} ha sido expulsado de {channel}\n".encode())
elif cmd == "ban" and len(parts) > 2:
channel = parts[1]
user = parts[2]
self.send_raw(f"MODE {channel} +b {user}")
conn.sendall(f" {user} ha sido baneado de {channel}\n".encode())
elif cmd == "unban" and len(parts) > 2:
channel = parts[1]
user = parts[2]
self.send_raw(f"MODE {channel} -b {user}")
conn.sendall(f" {user} ha sido desbaneado de {channel}\n".encode())
elif cmd == "op" and len(parts) > 2:
channel = parts[1]
user = parts[2]
self.send_raw(f"MODE {channel} +o {user}")
conn.sendall(f"{user} ahora tiene OP en {channel}\n".encode())
elif cmd == "deop" and len(parts) > 2:
channel = parts[1]
user = parts[2]
self.send_raw(f"MODE {channel} -o {user}")
conn.sendall(f" {user} ya no tiene OP en {channel}\n".encode())
else:
# Envía el texto tal cual
self.send_raw(data)
def handle_reconnect(self):
"""Reintenta la conexión al servidor IRC tras una desconexión."""
print(" 🔁 Intentando reconectar...")
try:
context = ssl.create_default_context()
self.sock = socket.create_connection((self.server, self.port))
self.sock = context.wrap_socket(self.sock, server_hostname=self.server)
# Reautenticación y rejoin
self.authenticated = False
self.joined = False
self.send_raw("CAP REQ :sasl")
self.send_raw(f"NICK {self.nickname}")
self.send_raw(f"USER {self.username} 0 * :{self.nickname}")
print(" ✅ Reconexión iniciada correctamente.")
except Exception as e:
print(f" ❌ Error durante la reconexión: {e}")
time.sleep(5)
self.handle_reconnect() # Reintento recursivo (con cuidado)
if __name__ == "__main__":
bot = IRCBot(
server="irc.libera.chat",
port=6697,
nickname="terryflaps",
username="terryflaps",
password=password,
channels=["#testtest", "#terryflaps"],
command_prefix="."
)

BIN
c.ogg Normal file

Binary file not shown.

43
config/entrypoint.sh Normal file
View File

@ -0,0 +1,43 @@
#!/bin/sh
set -e
# Si el Dockerfile ya crea /var/lib/radio-data/music con permisos icecast2:icecast2,
# solo hacemos el symlink.
#echo "[ENTRYPOINT] Creando enlace simbólico hacia /var/lib/radio/music"
#ln -sf /var/lib/radio-data/music /var/lib/radio/music
echo "[ENTRYPOINT] Sustituyendo credenciales..."
envsubst < /config/icecast.xml > /config/icecast_final.xml
envsubst < /config/liquidsoap.liq > /config/liquidsoap_final.liq
echo "[ENTRYPOINT] Copiando cookies..."
cp -f /cookies-secret/cookies.txt /app/cookies.txt
chmod 666 /app/cookies.txt
echo "[ENTRYPOINT] Iniciando Icecast..."
icecast2 -c /config/icecast_final.xml &
sleep 3
echo "[ENTRYPOINT] Asegurando que el directorio de sockets de Liquidsoap existe..."
mkdir -p /var/run/liquidsoap
chown -R icecast2:icecast2 /var/run/liquidsoap
chmod -R 755 /var/run/liquidsoap
echo "[ENTRYPOINT] Iniciando Liquidsoap..."
liquidsoap /config/liquidsoap_final.liq &
sleep 3
echo "[ENTRYPOINT] Verificando si Liquidsoap ya está corriendo..."
if pgrep -x "liquidsoap" > /dev/null; then
echo "[ENTRYPOINT] Liquidsoap ya está corriendo. Matándolo..."
pkill -9 liquidsoap
sleep 2 # Esperar un momento para que libere el puerto
fi
echo "[ENTRYPOINT] Iniciando Liquidsoap..."
liquidsoap /config/liquidsoap_final.liq &
sleep 3
echo "[ENTRYPOINT] Iniciando el bot..."
exec python3 bot.py

View File

@ -0,0 +1,43 @@
#!/bin/sh
set -e
# Si el Dockerfile ya crea /var/lib/radio-data/music con permisos icecast2:icecast2,
# solo hacemos el symlink.
#echo "[ENTRYPOINT] Creando enlace simbólico hacia /var/lib/radio/music"
#ln -sf /var/lib/radio-data/music /var/lib/radio/music
echo "[ENTRYPOINT] Sustituyendo credenciales..."
envsubst < /config/icecast.xml > /config/icecast_final.xml
envsubst < /config/liquidsoap.liq > /config/liquidsoap_final.liq
echo "[ENTRYPOINT] Copiando cookies..."
cp -f /cookies-secret/cookies.txt /app/cookies.txt
chmod 666 /app/cookies.txt
echo "[ENTRYPOINT] Iniciando Icecast..."
icecast2 -c /config/icecast_final.xml &
sleep 3
echo "[ENTRYPOINT] Asegurando que el directorio de sockets de Liquidsoap existe..."
mkdir -p /var/run/liquidsoap
chown -R icecast2:icecast2 /var/run/liquidsoap
chmod -R 755 /var/run/liquidsoap
echo "[ENTRYPOINT] Iniciando Liquidsoap..."
liquidsoap /config/liquidsoap_final.liq &
sleep 3
echo "[ENTRYPOINT] Verificando si Liquidsoap ya está corriendo..."
if pgrep -x "liquidsoap" > /dev/null; then
echo "[ENTRYPOINT] Liquidsoap ya está corriendo. Matándolo..."
pkill -9 liquidsoap
sleep 2 # Esperar un momento para que libere el puerto
fi
echo "[ENTRYPOINT] Iniciando Liquidsoap..."
liquidsoap /config/liquidsoap_final.liq &
sleep 3
echo "[ENTRYPOINT] Iniciando el bot..."
exec python3 bot.py

35
config/icecast.xml Normal file
View File

@ -0,0 +1,35 @@
<icecast>
<limits>
<clients>100</clients>
<sources>2</sources>
<queue-size>524288</queue-size>
</limits>
<authentication>
<admin-user>admin</admin-user>
<admin-password>${ICECAST_ADMIN_PASSWORD}</admin-password>
<source-password>${ICECAST_SOURCE_PASSWORD}</source-password>
</authentication>
<changeowner>
<user>icecast</user>
<group>icecast</group>
</changeowner>
<hostname>0.0.0.0</hostname>
<listen-socket>
<port>8000</port>
</listen-socket>
<paths>
<webroot>/usr/share/icecast2/web</webroot>
<adminroot>/usr/share/icecast2/admin</adminroot>
</paths>
<mount>
<mount-name>/stream.ogg</mount-name>
<password>${ICECAST_MOUNT_PASSWORD}</password>
<max-listeners>100</max-listeners>
<public>1</public>
</mount>
</icecast>

58
config/liquidsoap.liq Executable file
View File

@ -0,0 +1,58 @@
set("server.socket.path", "/var/run/liquidsoap/socket")
set("log.file.path", "/var/log/liquidsoap/liquidsoap.log")
set("log.stdout", true)
set("server.socket", true)
set("server.telnet", true)
set("server.telnet.bind_addr", "127.0.0.1")
set("server.telnet.port", 1234)
set("server.telnet.allow", "127.0.0.1")
def apply_metadata(m) =
filename = m["filename"]
raw_title = m["title"]
raw_artist = m["artist"]
title = if raw_title == "" then filename else raw_title end
artist = if raw_artist == "" then "Desconocido" else raw_artist end
[("stream_title", artist ^ " - " ^ title)]
end
def print_metadata(m) =
raw_title = m["title"]
raw_artist = m["artist"]
title = if raw_title == "" then "Sin título" else raw_title end
artist = if raw_artist == "" then "Desconocido" else raw_artist end
print("Reproduciendo ahora: #{artist} - #{title}")
end
radio_playlist = playlist.safe(mode="randomize", reload=10, "/var/lib/radio/")
radio_with_metadata = map_metadata(apply_metadata, radio_playlist)
default_audio = blank()
radio_stream = fallback(track_sensitive=false, [radio_with_metadata, default_audio])
radio_stream.on_metadata(print_metadata)
output.icecast(
%opus(
bitrate=320,
vbr="none",
application="audio",
complexity=10,
signal="music"
),
host = "localhost",
port = 8000,
password = "${ICECAST_SOURCE_PASSWORD}",
mount = "/stream.opus",
name = "My Radio Stream",
description = "Streaming en Opus desde Liquidsoap",
radio_stream
)

22
docker-compose.yml Normal file
View File

@ -0,0 +1,22 @@
version: "3.8"
services:
irc-bot:
build: .
container_name: irc-bot
restart: unless-stopped
volumes:
- /tmp:/tmp # Para compartir el socket UNIX (opcional)
environment:
- IRC_SERVER=irc.libera.chat
- IRC_NICK=terryflaps
- IRC_USER=terryflaps
- IRC_PASSWORD=wasamasa123
- IRC_CHANNEL=#testtest
networks:
- bot-network
networks:
bot-network:
driver: bridge

View File

@ -0,0 +1,9 @@
apiVersion: v1
kind: Secret
metadata:
name: calendarific-secret
namespace: default
type: Opaque
data:
CALENDARIFIC_API_KEY: BASE64 API KEY HERE

129
k8s/deployment.yaml Normal file
View File

@ -0,0 +1,129 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: irc-bot
labels:
app: irc-bot
spec:
replicas: 1
selector:
matchLabels:
app: irc-bot
template:
metadata:
labels:
app: irc-bot
spec:
# securityContext con fsGroup si lo deseas, pero no es obligatorio
securityContext:
fsGroup: 100 # Ajusta al GID correcto si usas un usuario con gid=100
initContainers:
- name: fix-permissions
image: busybox:latest
securityContext:
runAsUser: 0
command: ["sh", "-c"]
args:
- |
echo "Arreglando permisos en /var/lib/radio/radio-data..."
mkdir -p /var/lib/radio/
chown -R 1000:1000 /var/lib/radio/
chmod -R a+rwx /var/lib/radio
volumeMounts:
- name: radio-storage
mountPath: /var/lib/radio
containers:
- name: irc-bot
image: prietus/myircbot:1.0.38
resources:
requests:
cpu: "1000m"
memory: "512Mi"
limits:
cpu: "2000m"
memory: "1024Mi"
env:
- name: OLLAMA_URL
value: "http://ollama:11434/api/generate"
- name: IRC_BOT_PASSWORD
valueFrom:
secretKeyRef:
name: irc-bot-secret
key: irc_password
- name: OLLAMA_USER
valueFrom:
secretKeyRef:
name: irc-bot-secret
key: ollama_user
- name: OLLAMA_PASSWORD
valueFrom:
secretKeyRef:
name: irc-bot-secret
key: ollama_password
- name: WEATHER_API_KEY
valueFrom:
secretKeyRef:
name: irc-bot-secret
key: WEATHER_API_KEY
- name: ICECAST_PASSWORD
valueFrom:
secretKeyRef:
name: irc-bot-secret
key: icecast_password
- name: ICECAST_ADMIN_PASSWORD
valueFrom:
secretKeyRef:
name: irc-bot-secret
key: icecast_admin_password
- name: ICECAST_SOURCE_PASSWORD
valueFrom:
secretKeyRef:
name: irc-bot-secret
key: icecast_source_password
- name: ICECAST_MOUNT_PASSWORD
valueFrom:
secretKeyRef:
name: irc-bot-secret
key: icecast_mount_password
- name: LASTFM_API_KEY
valueFrom:
secretKeyRef:
name: lastfm-secret
key: LASTFM_API_KEY
- name: PASTE_USERNAME
valueFrom:
secretKeyRef:
name: pastebin-bot-secret
key: PASTE_USERNAME
- name: PASTE_PASSWORD
valueFrom:
secretKeyRef:
name: pastebin-bot-secret
key: PASTE_PASSWORD
- name: HUGGINGFACE_API_KEY
valueFrom:
secretKeyRef:
name: huggingface-secret
key: HUGGINGFACE_API_KEY
- name: CALENDARIFIC_API_KEY
valueFrom:
secretKeyRef:
name: calendarific-secret
key: CALENDARIFIC_API_KEY
volumeMounts:
- name: radio-storage
mountPath: /var/lib/radio
- name: control-socket
mountPath: /tmp
- name: youtube-cookies
mountPath: /cookies-secret
volumes:
- name: radio-storage
persistentVolumeClaim:
claimName: radio-storage
- name: control-socket
emptyDir: {}
- name: youtube-cookies
secret:
secretName: youtube-cookies

32
k8s/ingress.yaml Normal file
View File

@ -0,0 +1,32 @@
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: icecast-ingress
namespace: default
annotations:
kubernetes.io/ingress.class: traefik
cert-manager.io/cluster-issuer: letsencrypt-prod-dns
spec:
rules:
- host: radio.priet.us # Cambia esto por tu dominio real
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: icecast-service
port:
number: 8000
- path: /stream.mp3 # Ruta del stream
pathType: Prefix
backend:
service:
name: icecast-service
port:
number: 8000
tls:
- hosts:
- radio.priet.us
secretName: icecast-tls

12
k8s/pvc.yaml Normal file
View File

@ -0,0 +1,12 @@
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
name: radio-storage
spec:
accessModes:
- ReadWriteMany
resources:
requests:
storage: 10Gi # Ajusta según el tamaño necesario
storageClassName: longhorn # Asegúrate de que estás usando la StorageClass correcta

14
k8s/service.yaml Normal file
View File

@ -0,0 +1,14 @@
apiVersion: v1
kind: Service
metadata:
name: icecast-service
namespace: default
spec:
selector:
app: irc-bot # Ajusta esto si Icecast está en otro Deployment
ports:
- name: http
protocol: TCP
port: 8000 # PuertoClusterIPtargetPort: 8000 # Puerto en el contenedor
type: ClusterIP # Mantén ClusterIP si lo expones con Ingress

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

221
plugins/ai.py Normal file
View File

@ -0,0 +1,221 @@
import requests
import logging
import re
import os
from datetime import datetime
# Configuración de logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("AI-Plugin")
# Variables de entorno necesarias
HUGGINGFACE_API_KEY = os.getenv("HUGGINGFACE_API_KEY")
BRAVE_API_KEY = os.getenv("BRAVE_API_KEY")
MODEL_ID = os.getenv("HUGGINGFACE_MODEL", "deepseek/deepseek-v3-0324")
# Endpoints
HF_API_URL = "https://router.huggingface.co/novita/v3/openai/chat/completions"
BRAVE_URL = "https://api.search.brave.com/res/v1/web/search"
# Headers
HEADERS_HF = {
"Authorization": f"Bearer {HUGGINGFACE_API_KEY}",
"Content-Type": "application/json"
}
HEADERS_BRAVE = {
"Accept": "application/json",
"X-Subscription-Token": BRAVE_API_KEY
}
def respuesta_desactualizada(texto):
"""Detecta si la respuesta contiene fechas antiguas en relación al mes/año actual."""
texto = texto.lower()
ahora = datetime.now()
# Lista de años pasados (2020 hasta el año anterior)
anyos_pasados = [str(a) for a in range(2020, ahora.year)]
if any(a in texto for a in anyos_pasados):
return True
# Lista de meses pasados del mismo año (enero - mes anterior)
meses_ordenados = [
"enero", "febrero", "marzo", "abril", "mayo", "junio",
"julio", "agosto", "septiembre", "octubre", "noviembre", "diciembre"
]
mes_actual_idx = ahora.month - 1
meses_pasados = meses_ordenados[:mes_actual_idx]
for mes in meses_pasados:
if f"{mes} de {ahora.year}" in texto:
return True
return False
def sanitize_response(text, remove_first_line=True):
"""Limpia etiquetas HTML y fragmentos redundantes."""
clean_text = re.sub(r"<.*?>", "", text)
clean_text = re.sub(r"&quot;", '"', clean_text)
clean_text = re.sub(r"&amp;", "&", clean_text)
clean_text = clean_text.strip()
instrucciones_a_eliminar = [
"Resume y humaniza esta información en lenguaje natural:",
"Responde en español:",
"A continuación tienes un conjunto de textos o fragmentos sacados de sitios web. Tu tarea es analizarlos, eliminar información redundante o poco útil, y resumir todo en un único párrafo claro y natural para un humano. No repitas los títulos ni hagas listas. Explica como si le contaras a un amigo:"
]
for instruccion in instrucciones_a_eliminar:
if clean_text.lower().startswith(instruccion.lower()):
clean_text = clean_text[len(instruccion):].strip()
return clean_text
def query_huggingface(prompt, is_humanization=False):
"""Consulta el modelo DeepSeek en Hugging Face Router."""
if not HUGGINGFACE_API_KEY:
logger.error("No se ha definido la variable HUGGINGFACE_API_KEY.")
return None
# Incluir fecha real en el prompt del usuario
fecha_actual = datetime.now().strftime("%d de %B de %Y") # ej: "11 de abril de 2025"
prompt_con_fecha = f"Hoy es {fecha_actual}. {prompt.strip()}"
# Prompt de sistema personalizado
if is_humanization:
system_prompt = (
"Tu tarea es analizar fragmentos de sitios web y resumir la información relevante "
"en un único párrafo claro, sin repetir títulos ni hacer listas. Sé directo, preciso y "
"habla como si lo explicaras a un amigo en español."
)
else:
system_prompt = (
"Eres un asistente útil que siempre responde en español. "
"Ignora nombres de usuario como 'teraflops'. "
"Si no conoces información actualizada a la fecha de hoy, indica que no estás seguro. "
"No inventes datos si no los sabes."
)
payload = {
"model": MODEL_ID,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt_con_fecha}
],
"temperature": 0.7,
"max_tokens": 300
}
try:
response = requests.post(HF_API_URL, headers=HEADERS_HF, json=payload, timeout=120)
response.raise_for_status()
content = response.json()["choices"][0]["message"]["content"]
logger.info(f"[DEBUG] Respuesta cruda de Hugging Face:\n{content}")
return sanitize_response(content)
except requests.exceptions.RequestException as e:
logger.error(f"[ERROR] Fallo al conectar con Hugging Face: {e}")
return None
def buscar_en_brave(consulta):
"""Consulta Brave Search y decide si humanizar o mostrar mensaje de fallback."""
if not BRAVE_API_KEY:
logger.error("No se ha definido la variable BRAVE_API_KEY.")
return None
params = {"q": consulta, "count": 3, "safesearch": "moderate"}
try:
logger.info(f"[DEBUG] Consultando Brave Search con: {consulta}")
response = requests.get(BRAVE_URL, headers=HEADERS_BRAVE, params=params, timeout=30)
response.raise_for_status()
data = response.json()
resultados = data.get("web", {}).get("results", [])
if not resultados:
logger.warning("Brave no devolvió resultados.")
return "No encontré información relevante en Brave Search."
informacion_bruta = "\n".join(
f"{res.get('title', 'Sin título')}: {res.get('description', 'Sin descripción')}"
for res in resultados
)
logger.info(f"[DEBUG] Resultados de Brave crudos:\n{informacion_bruta}")
# Palabras clave genéricas que suelen indicar contenido útil
palabras_utiles = [
"2025", "abril", "marzo", "versión", "version", "released", "linux",
"resultado", "marcador", "gol", "victoria", "derrota", "ganó", "1-0", "2-1",
"empate", "directo", "esta noche", "hoy", "final", "resumen", "publicado",
"disponible", "estreno", "actualización", "presentación", "valencia", "sevilla"
]
texto_brave = informacion_bruta.lower()
pistas_utiles = any(p in texto_brave for p in palabras_utiles)
if not pistas_utiles:
logger.info("[DEBUG] Brave devolvió resultados ambiguos.")
return (
"📡 Brave Search encontró resultados poco claros. "
"No se halló información confirmada. Puedes revisar sitios oficiales para más detalles."
)
# Si hay contenido con pistas útiles, lo pasamos a Hugging Face para humanizar
resumen = query_huggingface(sanitize_response(informacion_bruta), is_humanization=True)
return resumen
except requests.exceptions.RequestException as e:
logger.error(f"[ERROR] Fallo en Brave Search: {e}")
return None
def procesar_consulta(prompt):
"""Intenta responder usando HF, y si falla, pregunta a Brave."""
logger.info(f"[DEBUG] ⏳ Consultando Hugging Face con: {prompt}")
respuesta_hf = query_huggingface(prompt)
if not respuesta_hf:
return "No se obtuvo respuesta de Hugging Face."
patron_generica = re.compile(
r"(no hay información.*?|"
r"no tengo.*?información|"
r"consulta.*?sitios? web oficiales|"
r"aún no se ha anunciado|"
r"lo siento, pero no puedo responder a eso|"
r"no tengo suficiente información|"
r"no puedo proporcionar esa información|"
r"te recomiendo verificar fuentes oficiales)",
re.IGNORECASE
)
# 🔧 Este bloque estaba fuera de la función — lo metemos correctamente dentro
if patron_generica.search(respuesta_hf) or respuesta_desactualizada(respuesta_hf):
logger.info("[DEBUG] Hugging Face dio respuesta genérica o desactualizada, consultando Brave...")
respuesta_brave = buscar_en_brave(prompt)
return f"📡 Fuente: Brave Search\n{respuesta_brave or 'No se encontró información útil.'}"
return f"🧠 Fuente: Hugging Face\n{respuesta_hf}"
def run(sender, *args):
"""Manejador del comando .ai"""
if not args:
return "Por favor, proporciona una consulta."
consulta = " ".join(args).strip()
logger.info(f"[.ai] Procesando consulta: {consulta}")
respuesta = procesar_consulta(consulta)
return respuesta or "No se encontró información disponible."
def help():
return "Uso: .ai <pregunta> - Responde con IA usando Hugging Face y Brave Search si es necesario."

195
plugins/alias.py Normal file
View File

@ -0,0 +1,195 @@
import os
import json
# Ajusta la ruta donde quieras guardar los aliases
ALIAS_FILE = os.path.expanduser("/var/lib/bot/aliases.json")
os.makedirs(os.path.dirname(ALIAS_FILE), exist_ok=True)
class AliasPlugin:
"""
Gestor de alias con soporte para variables posicionales ($1, $2, $*).
Cada alias se convierte en un comando independiente que el usuario
puede invocar con .<alias>.
"""
def __init__(self, send_raw, bot=None):
# Esta función se llama al instanciar el plugin en load_plugins().
self.send_raw = send_raw
self.bot = bot
self.aliases = self.load_aliases()
def load_aliases(self):
"""Carga los alias desde un archivo JSON."""
if os.path.exists(ALIAS_FILE):
try:
with open(ALIAS_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except json.JSONDecodeError:
return {}
return {}
def save_aliases(self):
"""Guarda los alias en un archivo JSON."""
with open(ALIAS_FILE, "w", encoding="utf-8") as f:
json.dump(self.aliases, f, indent=4, ensure_ascii=False)
def add_alias(self, name, command):
"""Añade un alias, lo guarda y recarga los plugins."""
if name in self.aliases:
return f" El alias '{name}' ya existe. Usa `.alias del {name}` para eliminarlo primero."
# Eliminar comillas dobles extra
if command.startswith('"') and command.endswith('"'):
command = command[1:-1] # Elimina la primera y última comilla
self.aliases[name] = command
self.save_aliases()
# Recargar plugins tras añadir un alias
if self.bot:
self.bot.load_plugins()
return f" Alias '{name}' creado para ejecutar: `{command}`"
def remove_alias(self, name):
"""Elimina un alias por su nombre."""
if name not in self.aliases:
return f" El alias '{name}' no existe."
del self.aliases[name]
self.save_aliases()
return f" Alias '{name}' eliminado."
def list_aliases(self):
"""Retorna una lista de todos los alias registrados."""
if not self.aliases:
return " No hay alias registrados."
lines = [f"`{name}` → `{cmd}`" for name, cmd in self.aliases.items()]
return " **Alias disponibles:**\n" + "\n".join(lines)
def execute_alias(self, sender, alias, *args):
if alias not in self.aliases:
return f" El alias '{alias}' no está registrado."
command = self.aliases[alias] # Ej: "echo Hola $1"
# Reemplazo de variables especiales
command = command.replace("$nick", sender) # Reemplaza $nick por el nombre del usuario
# Reemplazo de $1, $2, ..., $*
for i, arg in enumerate(args, start=1):
command = command.replace(f"${i}", arg)
command = command.replace("$*", " ".join(args[1:]))
# Ejecutar el comando procesado como si fuera un plugin del bot
tokens = command.split()
if not tokens:
return " Error: Alias vacío tras el procesamiento."
cmd = tokens[0] # Primer token es el comando (Ej: "echo")
cmd_args = tokens[1:] # El resto son argumentos
# Verificar si el comando existe en los plugins del bot
if cmd in self.bot.plugins:
plugin_or_func = self.bot.plugins[cmd]
try:
# Si es una función (alias dinámico), la llamamos
if callable(plugin_or_func):
response = plugin_or_func(sender, *cmd_args)
else:
# Si es un objeto plugin con método `.run()`, ejecutamos `.run(...)`
response = plugin_or_func.run(*cmd_args)
return response
except Exception as e:
return f" Error al ejecutar alias '{alias}': {str(e)}"
return f" El comando '{cmd}' no existe."
def get_help(self, alias_name):
"""Devuelve un texto con la ayuda de un alias concreto."""
if alias_name in self.aliases:
command = self.aliases[alias_name]
min_args = max(command.count("$1"), 1) # Asume al menos 1
return f"({alias_name} <al menos {min_args} argumento/s>) -- Alias for `{command}`."
return f" No hay ayuda disponible para '{alias_name}'."
def get_dynamic_commands(self):
"""
Retorna un dict con { alias: function } para registrarlos como comandos.
Cada alias quedará accesible con .<alias>.
"""
commands = {}
for alias_name in self.aliases:
# Creamos una función que llame a self.execute_alias
def alias_func(sender, *args, _alias=alias_name):
return self.execute_alias(sender, _alias, *args)
commands[alias_name] = alias_func
return commands
def run(self, sender, *args):
"""
Maneja el uso de `.alias <acción>`:
- .alias add <nombre> <comando>
- .alias del <nombre>
- .alias list
- .alias help <alias>
"""
if not args:
return " Uso: `.alias list` | `.alias add <nombre> <cmd>` | `.alias del <nombre>`"
action = args[0].lower()
if action == "add" and len(args) > 2:
name = args[1]
command = " ".join(args[2:])
return self.add_alias(name, command)
elif action == "del" and len(args) > 1:
return self.remove_alias(args[1])
elif action == "list":
return self.list_aliases()
elif action == "help" and len(args) > 1:
return self.get_help(args[1])
else:
return " Comando no reconocido. Usa `.alias list`."
#
# Función 'run' global para cuando el bot llama a 'alias' como plugin.
#
def run(sender, *args):
"""
Función que se invoca cuando el usuario hace .alias ...
Recuerda que en el bot, normalmente le pasamos: plugin.run(sender, *args)
"""
# Como no necesitamos send_raw aquí, podemos instanciar con None
alias_plugin = AliasPlugin(None)
# Simplemente delegamos la lógica de los comandos a alias_plugin.run
# Como no sabemos si lo primero que llega es 'sender' o no, ajusta según tu bot:
if len(args) == 0:
return alias_plugin.run("", *args)
# Asumiendo que 'sender' es el primer argumento:
sender = args[0]
more_args = args[1:]
return alias_plugin.run(sender, *more_args)
def help():
"""Texto de ayuda para .help alias"""
return (
"Usa `.alias add <nombre> <comando>` para crear un alias.\n"
"Usa `.alias del <nombre>` para eliminarlo.\n"
"Usa `.alias list` para ver todos los alias.\n"
"Usa `.alias help <nombre>` para ver la definición de un alias.\n"
"Los alias pueden usar `$1`, `$2`, `$*` para insertar argumentos."
)

13
plugins/ascii.py Normal file
View File

@ -0,0 +1,13 @@
import pyfiglet
def run(*text):
"""Convierte texto en arte ASCII."""
try:
ascii_art = pyfiglet.figlet_format(" ".join(text))
return f" **ASCII Art:**\n```{ascii_art}```"
except Exception as e:
return f" Error generando ASCII: {str(e)}"
def help():
return "Uso: .ascii <texto> - Convierte el texto en arte ASCII."

24
plugins/aspect.py Normal file
View File

@ -0,0 +1,24 @@
import math
def calcular_aspecto(ancho: int, alto: int):
"""Calcula la relación de aspecto más cercana."""
try:
gcd = math.gcd(ancho, alto)
aspecto_ancho = ancho // gcd
aspecto_alto = alto // gcd
return f" La relación de aspecto de {ancho}x{alto} es **{aspecto_ancho}:{aspecto_alto}**."
except Exception as e:
return f" Error: {e}"
def run(sender, *args):
"""Función principal que será ejecutada por el bot."""
if len(args) != 2:
return " Uso: `.aspect <ancho> <alto>` (Ej: `.aspect 1920 1080`)"
try:
return calcular_aspecto(int(args[0]), int(args[1]))
except ValueError:
return " Los valores deben ser números enteros. Uso: `.aspect <ancho> <alto>`"
def help():
return " Usa `.aspect <ancho> <alto>` para calcular la relación de aspecto de una resolución. Ejemplo: `.aspect 1920 1080`"

117
plugins/auth.py Normal file
View File

@ -0,0 +1,117 @@
import os
import sqlite3
DB_FILE = "/var/lib/bot/users.db"
def get_db_connection():
"""Obtiene una conexión a la base de datos SQLite."""
conn = sqlite3.connect(DB_FILE)
conn.row_factory = sqlite3.Row # Permitir acceder a columnas por nombre
return conn
def init_db():
"""Inicializa la base de datos de autenticación si no existe."""
os.makedirs(os.path.dirname(DB_FILE), exist_ok=True)
conn = get_db_connection()
cursor = conn.cursor()
# Tabla para usuarios registrados
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
nick TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL
)
""")
# Tabla para sesiones activas
cursor.execute("""
CREATE TABLE IF NOT EXISTS authenticated_users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
nick TEXT UNIQUE NOT NULL
)
""")
conn.commit()
conn.close()
def is_authenticated(nick):
"""Verifica si un usuario está autenticado."""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT nick FROM authenticated_users WHERE nick = ?", (nick,))
row = cursor.fetchone()
conn.close()
return row is not None # Si el usuario está en la tabla, está autenticado
def authenticate_user(nick):
"""Autentica a un usuario agregándolo a la base de datos."""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("INSERT OR IGNORE INTO authenticated_users (nick) VALUES (?)", (nick,))
conn.commit()
conn.close()
return f" {nick} ha iniciado sesión correctamente."
def logout_user(nick):
"""Cierra la sesión de un usuario."""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("DELETE FROM authenticated_users WHERE nick = ?", (nick,))
conn.commit()
conn.close()
return f" {nick} ha cerrado sesión."
def register_user(nick, password):
"""Registra un nuevo usuario en la base de datos con una contraseña."""
conn = get_db_connection()
cursor = conn.cursor()
try:
cursor.execute("INSERT INTO users (nick, password_hash) VALUES (?, ?)", (nick, password))
conn.commit()
response = f" Usuario {nick} registrado correctamente."
except sqlite3.IntegrityError:
response = f" El usuario {nick} ya está registrado."
conn.close()
return response
def login_user(nick, password):
"""Autentica al usuario verificando su contraseña."""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT password_hash FROM users WHERE nick = ?", (nick,))
row = cursor.fetchone()
if row and row["password_hash"] == password:
return authenticate_user(nick)
else:
return f" Contraseña incorrecta para {nick}."
def run(sender, *args):
"""Ejecuta el comando `.auth`."""
if not args:
return " Uso: `.auth register <nick> <password>`, `.auth login <nick> <password>`, `.auth logout <nick>`"
action = args[0].lower()
if action == "register" and len(args) == 3:
return register_user(args[1], args[2])
elif action == "login" and len(args) == 3:
return login_user(args[1], args[2])
elif action == "logout" and len(args) == 2:
return logout_user(args[1])
return " Uso incorrecto. Prueba `.auth register <nick> <password>` o `.auth login <nick> <password>`."
def help():
"""Muestra la ayuda para el módulo de autenticación."""
return (" Comandos de autenticación:\n"
" - `.auth register <nick> <password>` → Registra un usuario.\n"
" - `.auth login <nick> <password>` → Inicia sesión.\n"
" - `.auth logout <nick>` → Cierra sesión.")

19
plugins/base64.py Normal file
View File

@ -0,0 +1,19 @@
import base64
def run(mode, text):
"""Codifica o decodifica en Base64."""
try:
if mode == "encode":
encoded = base64.b64encode(text.encode()).decode()
return f"🔐 **Base64 Enc:** {encoded}"
elif mode == "decode":
decoded = base64.b64decode(text).decode()
return f"🔓 **Base64 Dec:** {decoded}"
else:
return "❌ Uso: .base64 encode|decode <texto>"
except Exception as e:
return f"❌ Error: {str(e)}"
def help():
return "Uso: .base64 encode|decode <texto> - Codifica o decodifica en Base64."

37
plugins/base64_utils.py Normal file
View File

@ -0,0 +1,37 @@
import base64
class Base64_utilsPlugin:
def encode(self, text):
"""Codifica un texto en Base64."""
try:
encoded = base64.b64encode(text.encode()).decode()
return f" **Base64 Enc:** {encoded}"
except Exception as e:
return f" Error al codificar: {str(e)}"
def decode(self, text):
"""Decodifica un texto en Base64."""
try:
decoded = base64.b64decode(text).decode()
return f" **Base64 Dec:** {decoded}"
except Exception as e:
return f" Error al decodificar: {str(e)}"
def run(self, sender, *args):
"""Función principal que será ejecutada por el bot."""
if len(args) < 2:
return " Uso: `.base64_utils encode|decode <texto>`"
mode, text = args[0], " ".join(args[1:]) # Unir en caso de que haya espacios en el texto
if mode == "encode":
return self.encode(text)
elif mode == "decode":
return self.decode(text)
else:
return " Modo inválido. Usa `.base64_utils encode <texto>` o `.base64_utils decode <texto>`."
def help(self):
"""Descripción del plugin para el comando .help"""
return " Uso: `.base64_utils encode|decode <texto>` - Codifica o decodifica en Base64."

113
plugins/blackjack.py Normal file
View File

@ -0,0 +1,113 @@
import random
# Definir los valores de las cartas
CARD_VALUES = {
"A": 11, "2": 2, "3": 3, "4": 4, "5": 5, "6": 6,
"7": 7, "8": 8, "9": 9, "10": 10, "J": 10, "Q": 10, "K": 10
}
# Definir los palos de las cartas
SUITS = ["", "", "", ""]
class BlackjackPlugin:
def __init__(self):
self.active_games = {} # {usuario: {"cartas": [], "total": int}}
def deal_card(self):
"""Reparte una carta aleatoria."""
card = random.choice(list(CARD_VALUES.keys()))
suit = random.choice(SUITS)
return f"{card}{suit}", CARD_VALUES[card]
def calculate_total(self, cards):
"""Calcula el valor total de la mano."""
total = sum(value for _, value in cards)
num_ases = sum(1 for card, value in cards if card.startswith("A"))
# Ajustar Ases de 11 a 1 si es necesario
while total > 21 and num_ases:
total -= 10 # Cambiar un As de 11 a 1
num_ases -= 1
return total
def bot_turn(self):
"""El bot juega automáticamente su turno después de que el jugador se planta."""
bot_cards = []
while self.calculate_total(bot_cards) < 17: # El bot se planta con 17 o más
card, value = self.deal_card()
bot_cards.append((card, value))
return bot_cards
def run(self, sender, *args):
"""Controla el juego de Blackjack."""
if not args:
return " Uso: `.blackjack start` para jugar, `.hit` para pedir carta, `.stand` para plantarte."
command = args[0].lower()
if command == "start":
# Comenzar una nueva partida
card1, value1 = self.deal_card()
card2, value2 = self.deal_card()
self.active_games[sender] = {"cartas": [(card1, value1), (card2, value2)]}
total = self.calculate_total(self.active_games[sender]["cartas"])
if total == 21:
self.active_games.pop(sender) # Eliminar partida
return f" {sender} tiene **Blackjack** con {card1} {card2}! ¡Ganaste!"
return f"🃏 {sender} ha comenzado una partida.\nCartas: {card1} {card2} (Total: {total})\nUsa `.hit` para pedir carta o `.stand` para plantarte."
elif command == "hit":
# Pedir una carta extra
if sender not in self.active_games:
return " No tienes una partida activa. Usa `.blackjack start` para comenzar."
card, value = self.deal_card()
self.active_games[sender]["cartas"].append((card, value))
total = self.calculate_total(self.active_games[sender]["cartas"])
if total > 21:
self.active_games.pop(sender) # Eliminar partida
return f" {sender} pidió {card} y ahora tiene {total}. ¡Te pasaste!"
return f"🃏 {sender} pidió {card}. **Total:** {total}\nUsa `.hit` para otra carta o `.stand` para plantarte."
elif command == "stand":
# Plantarse y terminar la partida
if sender not in self.active_games:
return " No tienes una partida activa. Usa `.blackjack start` para comenzar."
player_total = self.calculate_total(self.active_games[sender]["cartas"])
# 🃏 TURNO DEL BOT
bot_cards = self.bot_turn()
bot_total = self.calculate_total(bot_cards)
# Determinar el resultado final
if bot_total > 21:
result = f" El bot se pasó con {bot_total}. ¡{sender} gana!"
elif player_total > bot_total:
result = f" {sender} gana con {player_total} contra {bot_total} del bot. ¡Felicidades!"
elif player_total < bot_total:
result = f" El bot gana con {bot_total} contra {player_total} de {sender}."
else:
result = f" Empate con {player_total}."
# Borrar la partida una vez terminado
self.active_games.pop(sender)
return (f"🃏 {sender} se planta con {player_total}.\n\n"
f" **Turno del bot...**\nCartas del bot: {', '.join(card for card, _ in bot_cards)} (Total: {bot_total})\n\n"
f"{result}")
else:
return " Uso: `.blackjack start` para jugar, `.hit` para pedir carta, `.stand` para plantarte."
def help(self):
"""Descripción del plugin para el comando .help"""
return ("🃏 **Blackjack** - Juego de cartas\n"
" Usa `.blackjack start` para comenzar.\n"
" Luego `.hit` para pedir carta o `.stand` para plantarte.\n"
" ¡Gana si consigues 21 o el bot se pasa!")

37
plugins/bytes.py Normal file
View File

@ -0,0 +1,37 @@
class BytesPlugin:
"""Conversor de unidades de almacenamiento."""
unidades = ["B", "KB", "MB", "GB", "TB", "PB"]
def convertir_bytes(self, valor: float, unidad: str):
try:
unidad = unidad.upper()
if unidad not in self.unidades:
return " Unidad no válida. Usa: B, KB, MB, GB, TB, PB."
index = self.unidades.index(unidad)
resultados = []
for i, u in enumerate(self.unidades):
conversion = valor * (1024 ** (index - i))
resultados.append(f" {conversion:.2f} {u}")
return "\n".join(resultados)
except Exception as e:
return f" Error en conversión: {e}"
def run(self, sender, *args):
if len(args) != 2:
return " Uso: `.bytes <valor> <unidad>` (Ej: `.bytes 5 GB`)"
try:
valor = float(args[0])
unidad = args[1]
return self.convertir_bytes(valor, unidad)
except ValueError:
return " Error: El valor debe ser un número. Ejemplo correcto: `.bytes 1024 KB`"
def help(self):
return (" **Conversor de Bytes**\n"
" Usa `.bytes <valor> <unidad>` para convertir unidades de almacenamiento.\n"
" Ejemplo: `.bytes 1024 KB` para convertir 1024 KB a otras unidades.\n"
" Unidades disponibles: `B, KB, MB, GB, TB, PB`.")

44
plugins/chiste.py Normal file
View File

@ -0,0 +1,44 @@
import requests
from bs4 import BeautifulSoup
import random
import re
class ChistePlugin:
"""Plugin para obtener un chiste aleatorio desde la web."""
URL = "https://chistescortos.yavendras.com/"
def obtener_chiste(self):
"""Obtiene un chiste aleatorio desde la web y lo formatea correctamente."""
try:
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36"
}
response = requests.get(self.URL, headers=headers, timeout=10)
response.raise_for_status()
soup = BeautifulSoup(response.text, "html.parser")
chistes = soup.find_all("p", class_="description")
if not chistes:
return " No encontré chistes en este momento."
chiste_html = random.choice(chistes)
chiste = chiste_html.decode_contents()
# Reemplazar etiquetas <br> por saltos de línea
chiste = re.sub(r'<br\s*/?>', '\n', chiste).strip()
return f" {chiste}"
except requests.exceptions.RequestException as e:
return f" Error al obtener chiste: {str(e)}"
def run(self, sender, *args):
"""Función principal ejecutada por el bot."""
return self.obtener_chiste()
def help(self):
"""Descripción del plugin para el comando .help"""
return " Usa `.chiste` para obtener un chiste aleatorio."

52
plugins/convert.py Normal file
View File

@ -0,0 +1,52 @@
class ConvertPlugin:
"""Plugin para conversión de unidades."""
conversiones = {
"m": {"km": 0.001, "cm": 100, "mm": 1000, "ft": 3.28084, "in": 39.3701},
"km": {"m": 1000, "cm": 100000, "mm": 1000000, "mi": 0.621371},
"cm": {"m": 0.01, "mm": 10, "in": 0.393701},
"mm": {"m": 0.001, "cm": 0.1},
"ft": {"m": 0.3048, "in": 12},
"in": {"m": 0.0254, "cm": 2.54, "ft": 0.0833333},
"kg": {"g": 1000, "lb": 2.20462, "oz": 35.274},
"g": {"kg": 0.001, "lb": 0.00220462, "oz": 0.035274},
"lb": {"kg": 0.453592, "g": 453.592, "oz": 16},
"oz": {"kg": 0.0283495, "g": 28.3495, "lb": 0.0625},
"c": {"f": lambda x: (x * 9/5) + 32, "k": lambda x: x + 273.15},
"f": {"c": lambda x: (x - 32) * 5/9, "k": lambda x: ((x - 32) * 5/9) + 273.15},
"k": {"c": lambda x: x - 273.15, "f": lambda x: ((x - 273.15) * 9/5) + 32},
"b": {"kb": 0.001, "mb": 0.000001, "gb": 0.000000001},
"kb": {"b": 1000, "mb": 0.001, "gb": 0.000001},
"mb": {"b": 1000000, "kb": 1000, "gb": 0.001},
"gb": {"b": 1000000000, "kb": 1000000, "mb": 1000}
}
def convertir_unidad(self, valor: float, unidad_origen: str, unidad_destino: str):
"""Convierte unidades de medida entre sí."""
unidad_origen = unidad_origen.lower()
unidad_destino = unidad_destino.lower()
if unidad_origen in self.conversiones and unidad_destino in self.conversiones[unidad_origen]:
conversion = self.conversiones[unidad_origen][unidad_destino]
resultado = conversion(valor) if callable(conversion) else valor * conversion
return f" {valor} {unidad_origen} equivale a **{resultado:.4f} {unidad_destino}**."
else:
return " Conversión no soportada. Usa `.convert help` para ver opciones."
def run(self, sender, *args):
"""Función principal ejecutada por el bot."""
if len(args) != 3:
return " Uso: `.convert <valor> <unidad_origen> <unidad_destino>` (Ej: `.convert 10 km m`)"
try:
valor = float(args[0])
return self.convertir_unidad(valor, args[1], args[2])
except ValueError:
return " El valor debe ser un número."
def help(self):
"""Descripción del plugin para el comando .help"""
return (" Usa `.convert <valor> <unidad_origen> <unidad_destino>` para convertir unidades.\n"
"Ejemplo: `.convert 100 cm m` → 100 cm equivale a 1.0 m.\n"
"**Unidades soportadas:** m, km, cm, mm, ft, in, kg, g, lb, oz, c, f, k, b, kb, mb, gb.")

191
plugins/database.py Normal file
View File

@ -0,0 +1,191 @@
import sqlite3
import os
import time
DB_FILE = "/var/lib/bot/messages.db"
def get_db_connection():
"""Obtiene una conexión a la base de datos SQLite con soporte de diccionario."""
conn = sqlite3.connect(DB_FILE)
conn.row_factory = sqlite3.Row # Permite acceder a columnas por nombre
return conn
def init_db():
"""Inicializa la base de datos y crea las tablas necesarias."""
os.makedirs(os.path.dirname(DB_FILE), exist_ok=True)
conn = get_db_connection()
cursor = conn.cursor()
# Tabla para almacenar mensajes generales
cursor.execute("""
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
nick TEXT NOT NULL,
target TEXT NOT NULL,
message TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
# Tabla para almacenar grabs (frases guardadas)
cursor.execute("""
CREATE TABLE IF NOT EXISTS grabs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
nick TEXT NOT NULL,
message TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
# Tabla para usuarios autenticados (ejecución de plugins)
cursor.execute("""
CREATE TABLE IF NOT EXISTS authorized_users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
nick TEXT UNIQUE NOT NULL,
auth_time INTEGER NOT NULL
)
""")
conn.commit()
conn.close()
# ====================== GESTIÓN DE MENSAJES ======================
def store_message(nick, target, message, max_messages=10000):
"""Guarda un mensaje en la base de datos SQLite y elimina los más antiguos si es necesario."""
if message.startswith("."): # Ignorar comandos
return
try:
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("INSERT INTO messages (nick, target, message) VALUES (?, ?, ?)", (nick, target, message))
cursor.execute("SELECT COUNT(*) FROM messages")
total_messages = cursor.fetchone()[0]
if total_messages > max_messages:
cursor.execute("""
DELETE FROM messages WHERE id IN (
SELECT id FROM messages ORDER BY id ASC LIMIT ?
)
""", (total_messages - max_messages,))
conn.commit()
except Exception as e:
print(f" Error al almacenar mensaje en la base de datos: {e}")
finally:
conn.close()
def get_last_message(nick):
"""Obtiene el último mensaje enviado por un usuario."""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT message FROM messages WHERE nick = ? ORDER BY id DESC LIMIT 1", (nick,))
row = cursor.fetchone()
conn.close()
return row["message"] if row else None
def get_messages_by_nick(nick):
"""Obtiene todos los mensajes enviados por un usuario."""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT message FROM messages WHERE nick = ? ORDER BY id", (nick,))
rows = cursor.fetchall()
conn.close()
return [row["message"] for row in rows]
def get_message_by_index(nick, index):
"""Obtiene un mensaje específico de un usuario por índice."""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT message FROM messages WHERE nick = ? ORDER BY id LIMIT 1 OFFSET ?",
(nick, index - 1))
row = cursor.fetchone()
conn.close()
return row["message"] if row else None
# ====================== GESTIÓN DE GRABS ======================
def store_grab(nick, message):
"""Guarda un grab (frase destacada) en la base de datos."""
try:
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("INSERT INTO grabs (nick, message) VALUES (?, ?)", (nick, message))
conn.commit()
return f" Grab guardado para {nick}."
except Exception as e:
return f" Error al guardar grab: {e}"
finally:
conn.close()
def get_grabs_by_nick(nick):
"""Obtiene todos los grabs guardados de un usuario."""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT id, message FROM grabs WHERE nick = ? ORDER BY id", (nick,))
rows = cursor.fetchall()
conn.close()
return {row["id"]: row["message"] for row in rows}
def get_grab_by_index(nick, index):
"""Obtiene un grab específico de un usuario según el índice."""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT message FROM grabs WHERE nick = ? ORDER BY id LIMIT 1 OFFSET ?",
(nick, index - 1))
row = cursor.fetchone()
conn.close()
return row["message"] if row else None
def get_random_grab(nick):
"""Obtiene un grab aleatorio de un usuario."""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT message FROM grabs WHERE nick = ? ORDER BY RANDOM() LIMIT 1", (nick,))
row = cursor.fetchone()
conn.close()
return row["message"] if row else None
# ====================== AUTENTICACIÓN DE USUARIOS ======================
def authenticate_user(nick):
"""Guarda al usuario como autenticado con timestamp."""
try:
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("INSERT OR REPLACE INTO authorized_users (nick, auth_time) VALUES (?, ?)",
(nick, int(time.time())))
conn.commit()
return f" {nick}, ahora puedes ejecutar comandos de plugins restringidos."
except Exception as e:
return f" Error al autenticar usuario: {e}"
finally:
conn.close()
def is_user_authenticated(nick):
"""Verifica si un usuario está autenticado."""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT auth_time FROM authorized_users WHERE nick = ?", (nick,))
row = cursor.fetchone()
conn.close()
return row is not None # Retorna True si está autenticado
def remove_authentication(nick):
"""Elimina la autenticación del usuario."""
try:
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("DELETE FROM authorized_users WHERE nick = ?", (nick,))
conn.commit()
return f" {nick}, has cerrado sesión de los comandos restringidos."
except Exception as e:
return f" Error al cerrar sesión: {e}"
finally:
conn.close()

36
plugins/dns.py Normal file
View File

@ -0,0 +1,36 @@
import dns.resolver
class DnsPlugin:
"""Plugin para consultar registros DNS de un dominio."""
def run(self, sender, *args):
"""Consulta registros DNS de un dominio."""
if not args:
return " Uso: `.dns <dominio>` - Consulta registros DNS A, AAAA, MX, CNAME y TXT."
domain = args[0]
records = ["A", "AAAA", "MX", "CNAME", "TXT"]
response = [f"** Registros DNS de {domain}**"]
try:
for record in records:
try:
answers = dns.resolver.resolve(domain, record)
registros = ", ".join([r.to_text() for r in answers])
response.append(f" **{record}:** {registros}")
except dns.resolver.NoAnswer:
continue # Si no hay respuesta, omitimos el registro
except dns.resolver.NXDOMAIN:
return f" El dominio `{domain}` no existe."
except dns.resolver.Timeout:
return f"⏳ Tiempo de espera agotado consultando `{domain}`."
return "\n".join(response) if len(response) > 1 else f" No se encontraron registros para `{domain}`."
except Exception as e:
return f" Error en consulta DNS: {str(e)}"
def help(self):
"""Muestra la ayuda para el comando .dns"""
return " Uso: `.dns <dominio>` - Muestra los registros DNS A, AAAA, MX, CNAME y TXT de un dominio."

27
plugins/dpi.py Normal file
View File

@ -0,0 +1,27 @@
import math
class DpiPlugin:
"""Plugin para calcular DPI de una pantalla dados el ancho, alto en píxeles y el tamaño en pulgadas."""
def run(self, sender, *args):
"""Ejecuta el cálculo de DPI con los argumentos dados."""
if len(args) != 3:
return " Uso: `.dpi <ancho> <alto> <pulgadas>` (Ej: `.dpi 1920 1080 24`)"
try:
ancho = int(args[0])
alto = int(args[1])
pulgadas = float(args[2])
# Cálculo de DPI
diagonal_pixeles = math.sqrt(ancho**2 + alto**2)
dpi = diagonal_pixeles / pulgadas
return f" **DPI Calculado:** {dpi:.2f} dpi"
except ValueError:
return " Error: Asegúrate de ingresar números válidos para el ancho, alto y pulgadas."
def help(self):
"""Muestra la ayuda para el comando .dpi"""
return " Uso: `.dpi <ancho> <alto> <pulgadas>` - Calcula los DPI de una pantalla. Ejemplo: `.dpi 1920 1080 24`"

11
plugins/echo.py Normal file
View File

@ -0,0 +1,11 @@
def run(*args):
"""Devuelve el mensaje tal cual."""
if not args:
return " Uso: `.echo <mensaje>`"
return " ".join(args)
def help():
"""Descripción del plugin para el comando .help"""
return " Usa `.echo <mensaje>` para repetir el mensaje tal cual."

65
plugins/fecha.py Normal file
View File

@ -0,0 +1,65 @@
import os
import requests
# API Keys
CALENDARIFIC_API_KEY = os.getenv("CALENDARIFIC_API_KEY")
OLLAMA_URL = os.getenv("OLLAMA_URL", "http://ollama.default.svc.cluster.local:11434/api/generate")
def translate_with_ollama(text):
"""Traduce el texto en inglés a español usando Ollama, eliminando introducciones innecesarias."""
payload = {
"model": "gemma2:2b",
"prompt": f"Traduce al español este texto de manera concisa, sin introducciones: {text}",
"stream": False
}
try:
response = requests.post(OLLAMA_URL, json=payload, timeout=30)
response.raise_for_status()
translated_text = response.json().get("response", "").strip()
# Eliminar frases como "El texto en español sería:"
if translated_text.lower().startswith("el texto en español sería:"):
translated_text = translated_text.split(":", 1)[1].strip()
return translated_text
except requests.exceptions.RequestException as e:
return f"Error al conectar con Ollama: {e}"
def get_holidays(day, month, country):
"""Consulta los festivos en Calendarific y los traduce con Ollama."""
url = f"https://calendarific.com/api/v2/holidays?api_key={CALENDARIFIC_API_KEY}&country={country}&year=2025&month={month}&day={day}"
try:
response = requests.get(url)
data = response.json()
if "error" in data:
return f"Error en Calendarific: {data['error']}"
holidays = data.get("response", {}).get("holidays", [])
if not holidays:
return f"No hay festivos en {country} el {day}/{month}."
festivos_en = "\n".join([f"- {h['name']} ({h['description']})" for h in holidays])
festivos_es = translate_with_ollama(festivos_en)
return f"Festivos en {country} el {day}/{month}:\n{festivos_es}"
except Exception as e:
return f"Error al consultar Calendarific: {e}"
def run(sender, *args):
"""Ejecuta la consulta de festivos"""
if len(args) != 3:
return "Uso: .fecha <día> <mes> <país (código ISO)>"
day, month, country = args
return get_holidays(day, month, country)
def help():
return "Uso: .fecha <día> <mes> <país (código ISO)> - Consulta si es festivo en ese país."

14
plugins/flip.py Normal file
View File

@ -0,0 +1,14 @@
import random
class FlipPlugin:
"""Plugin para lanzar una moneda y obtener Cara o Cruz"""
def run(self, sender, *args):
"""Lanza una moneda y devuelve Cara o Cruz."""
resultado = random.choice(["🪙 Cara", "🪙 Cruz"])
return f"{sender}, el resultado es: {resultado}"
def help(self):
"""Muestra la ayuda del comando .flip"""
return "🪙 Uso: `.flip` - Lanza una moneda al aire y devuelve Cara o Cruz."

41
plugins/geoip.py Normal file
View File

@ -0,0 +1,41 @@
import requests
class GeoipPlugin:
"""Plugin para obtener información de geolocalización de una IP o dominio."""
def run(self, sender, *args):
"""Ejecuta la consulta de GeoIP."""
if not args:
return "Uso: `.geoip <IP o dominio>` (Ejemplo: `.geoip 8.8.8.8`)"
ip_o_dominio = args[0]
return self.obtener_geoip(ip_o_dominio)
def obtener_geoip(self, ip_o_dominio: str) -> str:
"""Consulta la geolocalización de una IP o dominio."""
try:
response = requests.get(f"https://ipwhois.app/json/{ip_o_dominio}", timeout=5)
data = response.json()
if "error" in data:
return f"Error: {data.get('reason', 'No se pudo obtener información de geolocalización.')}"
ciudad = data.get("city", "Desconocida")
region = data.get("region", "Desconocida")
pais = data.get("country_name", "Desconocido")
latitud = data.get("latitude", "N/A")
longitud = data.get("longitude", "N/A")
isp = data.get("org", "Desconocida")
return (f" **GeoIP de {ip_o_dominio}:**\n"
f"- Ubicación: {ciudad}, {region}, {pais}\n"
f"- ISP: {isp}\n"
f"- Coordenadas: {latitud}, {longitud}")
except requests.RequestException as e:
return f" Error de conexión: {str(e)}"
def help(self):
"""Muestra la ayuda para el comando .geoip"""
return " Uso: `.geoip <IP o dominio>` - Obtiene información de geolocalización."

36
plugins/give.py Normal file
View File

@ -0,0 +1,36 @@
class GivePlugin:
"""Plugin para ejecutar un comando y enviar el resultado a otro usuario."""
def __init__(self, plugins):
self.plugins = plugins
def run(self, sender, *args):
"""Ejecuta un comando y envía el resultado como '<usuario>: <salida_del_comando>'."""
if len(args) < 2:
return " Uso: `.give <usuario> <comando> [argumentos]`"
usuario = args[0] # Primer argumento es el usuario
comando = args[1] # Segundo argumento es el comando a ejecutar
parametros = args[2:] # El resto son los argumentos del comando
if comando == "give":
return " No puedes usar 'give' dentro de 'give'."
if comando not in self.plugins:
return f" El comando '{comando}' no existe."
try:
plugin_or_func = self.plugins[comando]
if callable(plugin_or_func):
resultado = plugin_or_func(sender, *parametros)
else:
resultado = plugin_or_func.run(sender, *parametros)
return f" {usuario}: {resultado}"
except Exception as e:
return f" Error al ejecutar '{comando}': {str(e)}"
def help(self):
return " Usa `.give <usuario> <comando> [argumentos]` para ejecutar un comando y enviar el resultado como '<usuario>: <resultado>'."

137
plugins/grab.py Normal file
View File

@ -0,0 +1,137 @@
import sqlite3
import os
import random
from plugins.database import get_db_connection, get_last_message
DB_FILE = "/var/lib/bot/grabs.db"
class GrabPlugin:
"""Gestor de frases grabadas con soporte SQLite."""
def __init__(self):
self.last_messages = {} # Diccionario {nick: última frase}
self.init_db()
def get_db_connection(self):
"""Obtiene una conexión a la base de datos SQLite."""
return sqlite3.connect(DB_FILE)
def init_db(self):
"""Inicializa la base de datos si no existe."""
os.makedirs(os.path.dirname(DB_FILE), exist_ok=True)
conn = self.get_db_connection()
cursor = conn.cursor()
# Crear la tabla de grabs si no existe
cursor.execute("""
CREATE TABLE IF NOT EXISTS grabs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
nick TEXT NOT NULL,
message TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
conn.close()
def store_last_message(self, nick, message):
"""Guarda la última frase de un usuario antes de ser grabada."""
self.last_messages[nick] = message # Se asegura de que `self.last_messages` esté definido
def grab(self, nick):
"""Guarda la última frase de un usuario si aún no está grabada."""
if nick not in self.last_messages:
return f" No hay frases recientes de {nick} para grabar."
message = self.last_messages[nick]
# Verificar si el mensaje ya está en los grabs del usuario
conn = self.get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM grabs WHERE nick = ? AND message = ?", (nick, message))
exists = cursor.fetchone()[0]
conn.close()
if exists > 0:
return f" La última frase de {nick} ya está grabada."
# Guardar en la base de datos
conn = self.get_db_connection()
cursor = conn.cursor()
cursor.execute("INSERT INTO grabs (nick, message) VALUES (?, ?)", (nick, message))
conn.commit()
conn.close()
return f" Frase grabada de {nick}."
def list_grabs(self, nick):
"""Lista los índices de los grabs disponibles de un usuario."""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT id FROM grabs WHERE nick = ?", (nick,))
rows = cursor.fetchall()
conn.close()
if not rows:
return f" {nick} no tiene frases grabadas."
indices = ", ".join(str(row[0]) for row in rows)
return f" Grab indices de {nick}: {indices}"
def get_grab(self, nick, grab_id=None):
"""Devuelve un grab aleatorio o por ID."""
conn = get_db_connection()
cursor = conn.cursor()
if grab_id:
cursor.execute("SELECT message FROM grabs WHERE nick = ? AND id = ?", (nick, grab_id))
else:
cursor.execute("SELECT message FROM grabs WHERE nick = ? ORDER BY RANDOM() LIMIT 1", (nick,))
row = cursor.fetchone()
conn.close()
return f" Grab de {nick}: \"{row[0]}\"" if row else f" {nick} no tiene grabs."
def store_last_message(self, nick, message):
"""Guarda la última frase de un usuario, evitando duplicados."""
if not message.strip():
return # No almacenar mensajes vacíos
# Si el mensaje ya es el último almacenado, no hacer nada
if self.last_messages.get(nick) == message:
return
self.last_messages[nick] = message # Guardar solo si es nuevo
def run(self, sender, *args):
"""Maneja el comando `.grab`."""
if not args:
return " Uso: `.grab <nick>` o `.grab list <nick>` o `.grab rq <nick> [id]`"
action = args[0].lower()
if action == "list" and len(args) == 2:
return self.list_grabs(args[1])
elif action == "rq" and len(args) >= 2:
return self.get_grab(args[1], int(args[2]) if len(args) > 2 else None)
elif len(args) == 1:
return self.grab(args[0])
return " Uso incorrecto. Prueba `.grab <nick>`, `.grab list <nick>` o `.grab rq <nick> [id]`."
def help():
"""Muestra la ayuda para el plugin de grabs."""
return (" Comandos de `.grab`:\n"
" - `.grab <nick>` → Guarda la última frase del usuario.\n"
" - `.grab list <nick>` → Lista los índices de grabs guardados.\n"
" - `.grab rq <nick>` → Muestra un grab aleatorio.\n"
" - `.grab rq <nick> <id>` → Muestra el grab con ID específico.")

127
plugins/imagegen.py Normal file
View File

@ -0,0 +1,127 @@
import os
import requests
import logging
import json
# Configuración de logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
# Configuración de APIs
HF_API_KEY = os.getenv("HUGGINGFACE_API_KEY")
HF_MODEL = "stabilityai/stable-diffusion-3.5-large"
#HF_MODEL = "stabilityai/stable-diffusion-2-1" # Modelo de Stable Diffusion
OLLAMA_API_URL = "http://10.233.23.140:11434/api/generate" # URL de tu instancia de Ollama
# Configuración del servidor Bepasty
PASTE_SERVER = "http://10.233.37.160"
PASTE_USERNAME = os.getenv("PASTE_USERNAME")
PASTE_PASSWORD = os.getenv("PASTE_PASSWORD") # Debería almacenarse en un Secret de Kubernetes
# Ruta temporal para la imagen generada
TEMP_IMAGE_PATH = "/tmp/generated_image"
def translate_to_english(text):
"""Traduce el texto al inglés usando la instancia de Ollama y maneja respuestas en streaming."""
payload = {
"model": "gemma2:2b",
"prompt": f"Traduce al inglés UNA sola versión, lo mas sugerente posible sin agreagr titulos o texto superfluo solo la traduccion: {text}",
"stream": False # Aseguramos que Ollama no use streaming
}
try:
response = requests.post(OLLAMA_API_URL, json=payload, timeout=120)
response.raise_for_status()
response_json = response.json()
# Si Ollama responde correctamente, extraer el texto
return response_json.get("response", text).strip()
except requests.exceptions.Timeout:
logging.error("⏳ Ollama tardó demasiado en responder. Usando texto original.")
return text
except requests.exceptions.RequestException as e:
logging.error(f" Error en la traducción con Ollama: {e}")
return text
def get_paste_token():
"""Obtiene un token de autenticación en paste.priet.us"""
url = f"{PASTE_SERVER}/api/token"
payload = {"username": PASTE_USERNAME, "password": PASTE_PASSWORD}
try:
response = requests.post(url, json=payload, timeout=10)
response.raise_for_status()
return response.json().get("token", "Error: No se pudo autenticar en paste.priet.us.")
except requests.exceptions.RequestException as e:
logging.error(f"Error al obtener token: {e}")
return None
def upload_to_paste(image_path):
"""Sube la imagen generada y devuelve la URL pública"""
token = get_paste_token()
if not token:
return "Error: No se pudo obtener el token de autenticación."
url = f"{PASTE_SERVER}/paste"
headers = {"Authorization": f"Bearer {token}"}
with open(image_path, "rb") as image_file:
files = {"c": image_file}
try:
response = requests.post(url, headers=headers, files=files, data={"expire": "yes", "private": "no"})
response.raise_for_status()
paste_data = response.json()
paste_url = paste_data.get("url", "Error: No se pudo obtener la URL.")
return paste_url.replace("http://10.233.37.160", "https://paste.priet.us")
except requests.exceptions.RequestException as e:
return f"Error al subir la imagen: {e}"
def generate_image(prompt):
"""Genera una imagen con la API de Hugging Face y la sube a paste.priet.us"""
if not HF_API_KEY:
return "Error: No se encontró la API Key de Hugging Face."
translated_prompt = translate_to_english(prompt)
url = f"https://api-inference.huggingface.co/models/{HF_MODEL}"
headers = {"Authorization": f"Bearer {HF_API_KEY}"}
payload = {"inputs": translated_prompt}
try:
logging.info(f" Generando imagen con prompt traducido: {translated_prompt}")
response = requests.post(url, headers=headers, json=payload, timeout=120)
if response.status_code != 200:
logging.error(f" Error en la generación de la imagen: {response.text}")
return f"Error en la generación de la imagen: {response.text}"
content_type = response.headers.get("Content-Type", "")
image_ext = "png" if "png" in content_type else "jpg" if "jpeg" in content_type else None
if not image_ext:
return "Error: No se recibió una imagen válida."
image_path = f"{TEMP_IMAGE_PATH}.{image_ext}"
with open(image_path, "wb") as f:
f.write(response.content)
paste_url = upload_to_paste(image_path)
os.remove(image_path) # Eliminamos la imagen después de subirla
return paste_url
except requests.exceptions.RequestException as e:
logging.error(f" Error al generar la imagen: {e}")
return f"Error al generar la imagen: {e}"
def run(sender, *args):
"""Ejecuta el comando de generación de imágenes."""
if not args:
return "Uso: .image <descripción>"
prompt = " ".join(args).strip()
return generate_image(prompt)
def help():
return "Uso: .image <descripción> - Genera una imagen con Stable Diffusion y la sube a paste.priet.us."

14
plugins/info.py Normal file
View File

@ -0,0 +1,14 @@
def run(*args):
"""Devuelve información general sobre el bot."""
return (
"Soy un bot modular de IRC escrito en Python. \n"
" Desarrollado para gestionar comandos personalizados y automatizar tareas en IRC.\n"
" Soporte para múltiples plugins como clima, radio, búsqueda en Wikipedia, y consultas a IA.\n"
" Repositorio: https://gitlab.com/teraflops/myircbot\n"
" Usa .help para ver la lista de comandos disponibles."
)
def help():
"""Devuelve la ayuda del comando."""
return "Uso: .info - Muestra información sobre el bot y sus características."

14
plugins/ip.py Normal file
View File

@ -0,0 +1,14 @@
import requests
def run(sender, *args):
"""Muestra la dirección IP pública del bot."""
try:
response = requests.get("https://api64.ipify.org?format=json", timeout=5)
data = response.json()
return f" **IP Pública:** {data['ip']}"
except Exception as e:
return f" Error obteniendo IP: {str(e)}"
def help():
return "Uso: .ip - Muestra la dirección IP pública del bot."

90
plugins/music.py Normal file
View File

@ -0,0 +1,90 @@
import requests
BASE_URL = "https://musicbrainz.org/ws/2/"
HEADERS = {
"User-Agent": "IRC-Bot/1.0 (https://tu-bot.com)"
}
def buscar_musicbrainz(tipo, consulta):
"""Consulta MusicBrainz y devuelve información detallada."""
endpoint = {
"artista": "artist",
"album": "release",
"cancion": "recording"
}.get(tipo.lower())
if not endpoint:
return " Uso: `.music <artista|album|cancion> <nombre>`"
params = {
"query": consulta,
"fmt": "json",
"limit": 1
}
try:
response = requests.get(f"{BASE_URL}{endpoint}/", params=params, headers=HEADERS, timeout=5)
data = response.json()
if endpoint == "artist" and data.get("artists"):
artista = data["artists"][0]
nombre = artista.get("name", "Desconocido")
pais = artista.get("country", " Desconocido")
fecha_inicio = artista.get("life-span", {}).get("begin", " Desconocida")
fecha_fin = artista.get("life-span", {}).get("end", " Aún activo")
genero = ", ".join(tag["name"] for tag in artista.get("tags", [])) if artista.get("tags") else " Sin géneros registrados"
alias = ", ".join(ali["name"] for ali in artista.get("aliases", [])) if artista.get("aliases") else "Ninguno"
tipo = artista.get("type", "Desconocido")
area = artista.get("area", {}).get("name", "Desconocida")
return (f" **Artista:** {nombre} ({tipo})\n"
f" **País:** {pais} | **Región:** {area}\n"
f" **Activo desde:** {fecha_inicio} | **Fin:** {fecha_fin}\n"
f" **Géneros:** {genero}\n"
f"🆔 **Alias:** {alias}")
elif endpoint == "release" and data.get("releases"):
album = data["releases"][0]
titulo = album.get("title", "Desconocido")
artista = album.get("artist-credit", [{}])[0].get("name", "Desconocido")
fecha = album.get("date", " Fecha desconocida")
label = album.get("label-info", [{}])[0].get("label", {}).get("name", " Desconocida")
formatos = ", ".join(m["format"] for m in album.get("media", [])) if album.get("media") else " Desconocido"
tracks = album.get("track-count", "N/A")
barcode = album.get("barcode", "N/A")
return (f" **Álbum:** {titulo} | **Artista:** {artista}\n"
f" **Lanzamiento:** {fecha} | **Sello:** {label}\n"
f" **Formatos:** {formatos} | **Tracks:** {tracks}\n"
f" **Código de Barras:** {barcode}")
elif endpoint == "recording" and data.get("recordings"):
cancion = data["recordings"][0]
titulo = cancion.get("title", "Desconocida")
artista = cancion.get("artist-credit", [{}])[0].get("name", "Desconocido")
duracion_ms = cancion.get("length", 0)
duracion = f"{duracion_ms // 60000}:{(duracion_ms // 1000) % 60:02d}" if duracion_ms else "⏳ Desconocida"
albumes = ", ".join(r["title"] for r in cancion.get("releases", [])) if cancion.get("releases") else " No asociado a álbum"
isrc = ", ".join(cancion.get("isrcs", [])) if "isrcs" in cancion else "N/A"
return (f" **Canción:** {titulo} | **Artista:** {artista}\n"
f"⏳ **Duración:** {duracion} min | **Álbumes:** {albumes}\n"
f" **ISRC:** {isrc}")
return " No encontré resultados en MusicBrainz."
except requests.exceptions.RequestException as e:
return f" Error en la consulta a MusicBrainz: {e}"
def run(*args):
"""Ejecuta la búsqueda en MusicBrainz."""
if len(args) < 2:
return " Uso: `.music <artista|album|cancion> <nombre>`"
tipo, consulta = args[0], " ".join(args[1:])
return buscar_musicbrainz(tipo, consulta)
def help():
"""Descripción del plugin para el comando .help"""
return " Usa `.music <artista|album|cancion> <nombre>` para buscar información detallada en MusicBrainz."

67
plugins/news.py Normal file
View File

@ -0,0 +1,67 @@
import feedparser
import time
import logging
import threading
ARCH_NEWS_FEED = "https://archlinux.org/feeds/news/"
LAST_NEWS_TITLE = None
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
class NewsPlugin:
"""Plugin para obtener las últimas noticias de Arch Linux."""
def fetch_latest_news(self):
"""Obtiene la última noticia del feed de Arch Linux."""
try:
feed = feedparser.parse(ARCH_NEWS_FEED)
if not feed.entries:
return " No se encontraron noticias."
latest_news = feed.entries[0]
title = latest_news.title
link = latest_news.link
logging.info(f"[NEWS] Última noticia obtenida: {title} - {link}")
return f" {title} - {link}"
except Exception as e:
logging.error(f" Error al obtener noticias: {e}")
return " Error al obtener noticias."
def announce_news(self, send_message, bot):
"""Monitorea el feed de noticias y envía actualizaciones periódicas."""
global LAST_NEWS_TITLE
while True:
try:
# Verifica si el bot está conectado antes de enviar
if bot.authenticated and bot.joined:
latest_news = self.fetch_latest_news()
if latest_news.startswith(""):
title = latest_news.split(" - ")[0][2:].strip()
if title != LAST_NEWS_TITLE:
LAST_NEWS_TITLE = title
send_message(f"PRIVMSG #testtest :{latest_news}")
time.sleep(600) # Verifica cada 10 minutos
except Exception as e:
logging.error(f" Error en announce_news: {e}")
time.sleep(60) # Espera 1 minuto antes de intentar de nuevo
def run(self, sender=None, *args):
"""Ejecuta el comando .news para mostrar las últimas noticias."""
return self.fetch_latest_news()
def help(self):
"""Muestra ayuda para el comando .news"""
return " Usa `.news` para obtener las últimas noticias de Arch Linux."
# Iniciar el monitoreo de noticias en un hilo separado cuando se carga el plugin
def start_news_monitoring(send_message, bot):
plugin = NewsPlugin()
thread = threading.Thread(target=plugin.announce_news, args=(send_message, bot), daemon=True)
thread.start()
return plugin

96
plugins/ollama.py Normal file
View File

@ -0,0 +1,96 @@
import os
import time
import requests
# --- Configuración ---
HUGGINGFACE_API_KEY = os.getenv("HUGGINGFACE_API_KEY")
MODEL_ID = os.getenv("HUGGINGFACE_MODEL", "deepseek/deepseek-v3-0324")
HF_API_URL = "https://router.huggingface.co/novita/v3/openai/chat/completions"
if not HUGGINGFACE_API_KEY:
raise ValueError("Error: La variable de entorno HUGGINGFACE_API_KEY no está definida.")
# --- Throttling ---
LAST_CALL_TIME = 0.0
MIN_INTERVAL_SECONDS = 5.0
# --- Fragmentación de la respuesta ---
MAX_CHUNK_LENGTH = 200
DELAY_BETWEEN_CHUNKS = 1.0
def query_huggingface(prompt):
"""Envía un prompt estilo chat a HuggingFace Router y devuelve la respuesta."""
headers = {
"Authorization": f"Bearer {HUGGINGFACE_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": MODEL_ID,
"messages": [
{
"role": "system",
"content": "You are a helpful assistant that always responds in neutral Latin American Spanish. Ignore usernames or nicknames like 'teraflops'. Focus only on the user's input."
},
{
"role": "user",
"content": prompt
}
],
"temperature": 0.7,
"max_tokens": 300
}
try:
response = requests.post(HF_API_URL, headers=headers, json=payload, timeout=120)
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"].strip()
except requests.exceptions.Timeout:
return "⏱️ El servidor de HuggingFace tardó demasiado en responder."
except requests.exceptions.RequestException as e:
return f"❌ Error al conectar con HuggingFace Router: {e}"
def chunk_text(text, max_length):
"""Corta 'text' en fragmentos de longitud máxima 'max_length'."""
return [text[i:i + max_length] for i in range(0, len(text), max_length)]
def run(*args):
"""Consulta el modelo y devuelve la respuesta troceada, aplicando throttling."""
global LAST_CALL_TIME
if not args:
return help()
now = time.time()
elapsed = now - LAST_CALL_TIME
if elapsed < MIN_INTERVAL_SECONDS:
wait_time = round(MIN_INTERVAL_SECONDS - elapsed, 1)
return f"¡Espera {wait_time} seg antes de hacer otra consulta!"
LAST_CALL_TIME = now
prompt = " ".join(args)
try:
full_text = query_huggingface(prompt)
except Exception as e:
return f"Error al consultar HuggingFace: {e}"
result = []
for fragment in chunk_text(full_text, MAX_CHUNK_LENGTH):
result.append(fragment)
time.sleep(DELAY_BETWEEN_CHUNKS)
return "\n".join(result)
def help():
return f""".ollama <pregunta> - Consulta el modelo '{MODEL_ID}' en HuggingFace usando el router.
Ejemplos:
!ollama ¿Qué es la entropía en física?
!ollama Resume el argumento de Don Quijote en 3 líneas.
"""

6
plugins/ping.py Normal file
View File

@ -0,0 +1,6 @@
def run(*args):
return "Pong!"
def help():
return "Uso: !ping - Responde con 'Pong!'"

42
plugins/pwgen.py Normal file
View File

@ -0,0 +1,42 @@
import random
import string
def generar_password(longitud=12, mayusculas=True, numeros=True, simbolos=False):
"""Genera una contraseña aleatoria con las opciones indicadas."""
caracteres = string.ascii_lowercase
if mayusculas:
caracteres += string.ascii_uppercase
if numeros:
caracteres += string.digits
if simbolos:
caracteres += string.punctuation
if longitud < 4:
return " La contraseña debe tener al menos 4 caracteres."
password = ''.join(random.choice(caracteres) for _ in range(longitud))
return f" Contraseña generada: **{password}**"
def run(sender, *args):
"""Función principal que será ejecutada por el bot."""
try:
longitud = int(args[0]) if len(args) > 0 else 12
mayusculas = "nomayus" not in args
numeros = "nonum" not in args
simbolos = "simbolos" in args
return generar_password(longitud, mayusculas, numeros, simbolos)
except ValueError:
return " Uso: `.pwgen [longitud] [nomayus] [nonum] [simbolos]` (Ej: `.pwgen 16 simbolos`)"
def help():
"""Descripción del plugin para el comando .help"""
return (" Genera contraseñas aleatorias.\n"
"Uso: `.pwgen [longitud] [nomayus] [nonum] [simbolos]`\n"
"**Ejemplo:** `.pwgen 16 simbolos` → Contraseña de 16 caracteres con símbolos.\n"
"**Opciones:**\n"
"- `nomayus` → No incluir mayúsculas.\n"
"- `nonum` → No incluir números.\n"
"- `simbolos` → Incluir símbolos.")

507
plugins/radio.py Normal file
View File

@ -0,0 +1,507 @@
import os
import socket
import subprocess
import unicodedata
import time
import threading
import requests
import re
import html
import json
SAVE_PATH = "/var/lib/radio/"
COOKIES_PATH = "/app/cookies.txt"
LIQUIDSOAP_SOCKET = "/var/run/liquidsoap/socket"
LASTFM_API_KEY = os.getenv("LASTFM_API_KEY") # API Key de Last.fm
GENIUS_API_KEY = os.getenv("GENIUS_API_KEY")
# Extensiones de audio permitidas
AUDIO_EXTENSIONS = {".mp3", ".m4a", ".opus", ".flac", ".webm"}
HUGGINGFACE_API_KEY = os.getenv("HUGGINGFACE_API_KEY") # Tu API Key de Hugging Face
HF_API_URL = "https://api-inference.huggingface.co/models/google/gemma-2-2b-it"
OLLAMA_API_URL = "http://ollama:11434/api/generate"
HEADERS = {
"Authorization": f"Bearer {HUGGINGFACE_API_KEY}",
"Content-Type": "application/json"
}
def get_current_song_file():
"""Devuelve el nombre de archivo de la canción actual si está disponible"""
artist, title = get_current_song()
if not artist or not title:
return None
for file in os.listdir(SAVE_PATH):
if clean_song_title(title).lower() in clean_song_title(file).lower():
return f"🎵 Reproduciendo: {clean_filename(file)}"
return " No se pudo encontrar el archivo de la canción actual."
def remove_code_fence(text):
"""
Busca un bloque de código tipo:
```json
{ ... }
```
y devuelve solo el contenido interno. Si no lo encuentra,
retorna el texto tal cual.
"""
pattern = r"```json\s*(.*?)\s*```"
match = re.search(pattern, text, re.DOTALL)
if match:
return match.group(1).strip()
return text.strip()
def generate_playlist(prompt):
"""Genera una lista de canciones usando tu instancia local de Ollama."""
query = (
f"Genera una lista de 10 canciones recomendadas para: {prompt}. "
"Devuelve SOLO un objeto JSON con la estructura:\n"
"{\n"
' \"songs\": [\n'
' \"Artista - Canción\",\n'
' \"Artista - Canción\"\n'
" ]\n"
"}\n"
"Nada de texto adicional ni bloques de código."
)
payload = {
"model": "gemma2:2b", # Ajusta al nombre exacto de tu modelo si es distinto
"prompt": query,
"stream": False
}
try:
response = requests.post(OLLAMA_API_URL, json=payload, timeout=60)
response.raise_for_status()
response_json = response.json()
raw_text = response_json.get("response", "").strip()
print(f"[DEBUG] Respuesta cruda de Ollama:\n{raw_text}")
clean_text = remove_code_fence(raw_text)
# 1) Intentar parsear como JSON
try:
parsed_data = json.loads(clean_text)
return parsed_data
except json.JSONDecodeError:
print("[ERROR] JSON inválido. Intentando regex...")
json_match = re.search(r"\{.*?\}", clean_text, re.DOTALL)
if json_match:
try:
parsed_data = json.loads(json_match.group(0))
return parsed_data
except json.JSONDecodeError:
pass
songs = re.findall(r"([^-]+) - ([^\n]+)", clean_text)
if songs:
return {"songs": [f"{a.strip()} - {t.strip()}" for a, t in songs]}
return {"songs": []}
except requests.exceptions.RequestException as e:
print(f"[ERROR] No se pudo conectar a Ollama: {e}")
return {"songs": []}
def suggest_playlist(criterion):
"""Genera y descarga una playlist basada en el criterio dado por el usuario."""
songs_data = generate_playlist(criterion) # YA DEVUELVE UN DICCIONARIO
if not isinstance(songs_data, dict) or "songs" not in songs_data:
return " No se pudo generar una playlist válida."
songs = songs_data["songs"]
if not songs:
return " No se encontraron canciones en la lista generada."
responses = []
for song in songs[:8]: # Limitar a 5 canciones para evitar spam en el IRC
print(f" Añadiendo: {song}") # Debug para verificar qué canciones se están añadiendo
response = add_song(song) # Descarga cada canción individualmente
print(f" Respuesta de add_song: {response}") # Debug para ver la respuesta de yt-dlp
responses.append(response)
send_liquidsoap_command("radio.reload") # Recargar la playlist después de añadir
return " Playlist generada con éxito:\n" + "\n".join(responses)
def monitor_radio():
"""Monitorea la reproducción y activa la eliminación de canciones."""
last_song = None
while True:
artist, title = get_current_song()
if artist and title and title != last_song:
print(f" Nueva canción detectada: {title} - {artist}")
last_song = title
# Ejecutar la eliminación de la canción en un hilo separado
threading.Thread(target=delete_song_after_playing, daemon=True).start()
time.sleep(10) # Verificar cada 10 segundos
def monitor_liquidsoap():
"""Escucha eventos de Liquidsoap y elimina canciones terminadas."""
if not os.path.exists(LIQUIDSOAP_SOCKET):
print(" Error: El socket de Liquidsoap no existe.")
return
try:
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
sock.connect(LIQUIDSOAP_SOCKET)
print(" Conectado al socket de Liquidsoap para monitoreo.")
while True:
data = sock.recv(4096).decode("utf-8").strip()
if not data:
continue
print(f" Liquidsoap: {data}") # Debug: Ver lo que responde Liquidsoap
if "request.on_air" in data or "end of track" in data:
print(" Detectada canción terminada. Verificando para eliminar...")
delete_song_after_playing()
except Exception as e:
print(f" Error en monitor_liquidsoap: {e}")
def normalize_artist_name(artist):
"""Normaliza el nombre del artista eliminando caracteres extraños."""
replacements = {
"": "/", # Reemplaza el carácter Unicode extraño por "/"
"": "'", # Reemplaza apóstrofes extraños
"": "'",
"": '"',
"": '"'
}
for bad_char, good_char in replacements.items():
artist = artist.replace(bad_char, good_char)
return artist.strip()
def help():
"""Devuelve la ayuda del comando .radio"""
return "Uso: .radio <comando> [argumentos] - Control de radio por Liquidsoap."
def clear_playlist():
"""Elimina todas las canciones de la playlist, ignorando archivos .nfs* y directorios."""
if not os.path.exists(SAVE_PATH):
return f"Error: El directorio {SAVE_PATH} no existe."
files_removed = 0
for file in os.listdir(SAVE_PATH):
file_path = os.path.join(SAVE_PATH, file)
# Ignorar archivos .nfs* y directorios
if file.startswith(".nfs") or os.path.isdir(file_path):
continue
try:
os.remove(file_path)
files_removed += 1
except Exception as e:
print(f" No se pudo borrar {file}: {e}")
send_liquidsoap_command("radio.reload") # Recargar la playlist en Liquidsoap
return f" {files_removed} archivos eliminados de la playlist."
def clean_song_title(song_title):
"""Elimina información irrelevante entre paréntesis o corchetes"""
cleaned_title = re.sub(r"\s*[\(\[].*?[\)\]]", "", song_title) # Elimina (Official Video), [Live], etc.
return cleaned_title.strip()
def clean_filename(filename):
"""Elimina caracteres raros de los nombres de archivos"""
return unicodedata.normalize('NFKD', filename).encode('ASCII', 'ignore').decode('utf-8')
def extract_artist_and_title(song_title):
"""Intenta separar el artista y el título de la canción"""
match = re.match(r"(.+?)\s*[-]\s*(.+)", song_title) # Soporta "Artista - Canción"
if match:
artist, title = match.groups()
return artist.strip(), title.strip()
return None, song_title
def send_liquidsoap_command(command):
"""Envía un comando al socket de Liquidsoap y devuelve la respuesta"""
if not os.path.exists(LIQUIDSOAP_SOCKET):
return "Error: El socket de Liquidsoap no existe. ¿Está corriendo el servidor?"
try:
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
sock.connect(LIQUIDSOAP_SOCKET)
sock.sendall((command + "\n").encode("utf-8"))
response = sock.recv(4096).decode("utf-8").strip()
return response
except Exception as e:
return f"Error al enviar comando a Liquidsoap: {e}"
def get_current_song():
"""Obtiene el título y artista de la canción en reproducción con sanitización mejorada"""
index = send_liquidsoap_command("request.on_air")
if not index.isdigit():
return None, None
metadata_response = send_liquidsoap_command(f"request.metadata {index}")
if not metadata_response or "Error" in metadata_response:
return None, None
filename = None
for line in metadata_response.split("\n"):
if line.startswith("filename="):
filename = line.split("=", 1)[1].strip().replace('"', '')
if not filename:
return None, None
# Normalizar el nombre del archivo y quitar contenido irrelevante
song_title = os.path.basename(filename).rsplit(".", 1)[0]
cleaned_title = clean_song_title(song_title) # Elimina (Official Video) y demás
artist, title = extract_artist_and_title(cleaned_title)
# Si no se detecta artista, usar "Desconocido"
artist = artist if artist else "Desconocido"
artist = normalize_artist_name(artist) # Normaliza ACDC → AC/DC
return artist.strip(), title.strip()
def delete_song_after_playing():
"""Espera a que la canción termine y la elimina de la carpeta de radio."""
artist, title = get_current_song()
if not artist or not title:
return " No se está reproduciendo ninguna canción."
# Buscar el archivo correspondiente en el sistema
song_filename = None
for file in os.listdir(SAVE_PATH):
if clean_song_title(title).lower() in clean_song_title(file).lower():
song_filename = os.path.join(SAVE_PATH, file)
break
if not song_filename or not os.path.exists(song_filename):
return " No se encontró la canción en el sistema de archivos."
print(f" Monitoreando {song_filename} para eliminación...")
# Esperar hasta que la canción ya no esté en reproducción
while True:
current_artist, current_title = get_current_song()
if not current_title or current_title != title:
break
time.sleep(5) # Esperar 5 segundos antes de volver a verificar
# Borrar archivo de la carpeta de radio
try:
os.remove(song_filename)
print(f" Eliminado: {song_filename}")
send_liquidsoap_command("radio.reload") # Recargar playlist en Liquidsoap
return " Canción eliminada después de su reproducción."
except Exception as e:
return f" Error al eliminar la canción: {e}"
def get_playlist():
"""Lista directamente los archivos en el directorio de canciones"""
if not os.path.exists(SAVE_PATH):
return f"Error: El directorio {SAVE_PATH} no existe."
files = os.listdir(SAVE_PATH)
music_files = [f for f in files if os.path.splitext(f)[1].lower() in AUDIO_EXTENSIONS]
if not music_files:
return "La playlist está vacía."
return "Playlist actual:\n" + "\n".join(f"{i}. {clean_filename(os.path.splitext(f)[0])}" for i, f in enumerate(sorted(music_files)))
def add_song(query):
"""Añade una canción a la playlist usando yt-dlp"""
if not query:
return help()
yt_command = [
"yt-dlp",
"--cookies", COOKIES_PATH,
f"ytsearch1:{query}",
"-f", "bestaudio",
"-x",
"--audio-format", "opus",
"--ffmpeg-location", "/usr/bin/ffmpeg",
"-o", f"{SAVE_PATH}%(title)s.%(ext)s",
"--embed-metadata", # Forzar metadatos en el archivo final
"--ppa", "ffmpeg:-metadata comment='Descargado via yt-dlp'"
]
try:
subprocess.run(yt_command, capture_output=True, text=True, check=True)
send_liquidsoap_command("radio.reload")
return f"'{query}' añadida al stream y playlist recargada."
except subprocess.CalledProcessError as e:
return f"Error al añadir la canción: {e.stderr}"
def get_track_info():
"""Obtiene información del track desde Last.fm"""
artist, track_title = get_current_song()
if not artist or not track_title:
return "No se está reproduciendo ninguna canción."
url = "http://ws.audioscrobbler.com/2.0/"
params = {
"method": "track.getInfo",
"api_key": LASTFM_API_KEY,
"format": "json",
"track": track_title
}
if artist:
params["artist"] = artist
try:
response = requests.get(url, params=params)
data = response.json()
if "track" not in data:
return f"No se encontró información para '{track_title}'."
track = data["track"]
# **Obtener artista y normalizar caracteres**
artist_name = html.unescape(track.get("artist", {}).get("name", "Desconocido"))
# **Obtener álbum si está disponible**
album = track.get("album", {}).get("title", "No disponible")
# **Obtener número de oyentes con formato**
listeners = track.get("listeners", "N/A")
if listeners.isdigit():
listeners = f"{int(listeners):,}".replace(",", ".")
# **Obtener lista de géneros si hay más de uno**
genres = track.get("toptags", {}).get("tag", [])
genre_list = ", ".join([g["name"] for g in genres[:3]]) if genres else "Sin género"
return (
f"Ahora suena: {track_title}\n"
f"Artista: {artist_name}\n"
f"Álbum: {album}\n"
f"Oyentes en Last.fm: {listeners}\n"
f"Género(s): {genre_list}"
)
except Exception as e:
return f"Error al obtener información del track: {e}"
def get_lyrics():
"""Busca la letra de la canción en Genius."""
if not GENIUS_API_KEY:
return "Error: La API Key de Genius no está configurada."
artist, title = get_current_song()
if not artist or not title:
return "No se está reproduciendo ninguna canción."
# Normalizar el artista antes de buscar
artist = normalize_artist_name(artist)
query = f"{artist} {title}"
print(f"[DEBUG] Buscando en Genius: {query}") # Debug
url = "https://api.genius.com/search"
headers = {"Authorization": f"Bearer {GENIUS_API_KEY}"}
params = {"q": query}
try:
response = requests.get(url, headers=headers, params=params, timeout=10)
response.raise_for_status()
data = response.json()
hits = data.get("response", {}).get("hits", [])
if not hits:
return f"No se encontró la letra de '{title}' de {artist}."
for hit in hits:
result = hit["result"]
genius_title = result.get("title", "").lower()
genius_artist = normalize_artist_name(result.get("primary_artist", {}).get("name", "").lower())
if title.lower() in genius_title and artist.lower() in genius_artist:
return f"Letras de '{title}' - {artist}: {result['url']}"
return f"No se encontró la letra de '{title}' de {artist}."
except requests.exceptions.RequestException as e:
return f"Error al obtener la letra: {e}"
def run(sender, *args):
"""Manejador principal de comandos de radio"""
if not args: # Si no hay argumentos, mostrar ayuda
return help()
command = args[0].lower()
command_args = args[1:] if len(args) > 1 else []
# Mapeo de comandos
if command == "suggest":
if not command_args:
return " Uso: `.radio suggest <criterio>` (Ejemplo: `.radio suggest música relajante`)"
return suggest_playlist(" ".join(command_args))
elif command == "lyrics":
return get_lyrics()
elif command == "add":
if not command_args:
return " ¡Falta el nombre de la canción! Ejemplo: .radio add Nirvana - Smells Like Teen Spirit"
return add_song(" ".join(command_args))
elif command == "playing":
return get_current_song_file() or " No hay música reproduciéndose"
elif command == "clear":
return clear_playlist()
elif command == "trackinfo":
return get_track_info()
elif command == "list":
return get_playlist()
elif command in ["skip", "next"]:
return send_liquidsoap_command("radio.skip")
elif command == "reload":
return send_liquidsoap_command("radio.reload")
else:
return help() # Comando no reconocido
def help():
return (
" Comandos de Radio: .radio <comando>\n"
"──────────────────────────\n"
"add <canción> - Añade a la cola\n"
"playing - Canción actual\n"
"trackinfo - Detalles técnicos\n"
"list - Lista de reproducción\n"
"lyrics - Letra de la canción\n"
"skip/next - Saltar tema\n"
"reload - Recargar playlist\n"
"clear - Limpiar playlist\n"
"suggest <criterio> - Generar playlist\n"
"Stream: https://radio.priet.us/stream.opus\n"
"M3U: https://radio.priet.us/stream.m3u"
)

148
plugins/rutracker.py Normal file
View File

@ -0,0 +1,148 @@
import time
import cloudscraper
from bs4 import BeautifulSoup
import os
import pickle
# Configuración de acceso
USERNAME = os.getenv("RUTRACKER_USER")
PASSWORD = os.getenv("RUTRACKER_PASSWORD")
RUTRACKER_COOKIE = os.getenv("RUTRACKER_COOKIE") # Cookie desde el Secret de Kubernetes
LOGIN_URL = "https://rutracker.org/forum/login.php"
SEARCH_URL = "https://rutracker.org/forum/tracker.php"
BASE_URL = "https://rutracker.org/forum/"
COOKIES_FILE = "/app/rut_cookies.txt" # Archivo local donde guardamos las cookies
# Headers para evitar detección de bots
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36",
"Referer": "https://rutracker.org/forum/index.php",
"Origin": "https://rutracker.org",
}
# Estado global para rate limit
LAST_EXECUTION_TIME = 0 # Última ejecución en timestamp
RATE_LIMIT_SECONDS = 60 # Intervalo de tiempo permitido entre ejecuciones
def load_cookies(session):
"""Carga cookies desde el archivo o desde la variable de entorno."""
if RUTRACKER_COOKIE:
session.cookies.set("bb_session", RUTRACKER_COOKIE, domain=".rutracker.org")
print(" Cookie cargada desde Kubernetes Secret.")
return True
if os.path.exists(COOKIES_FILE):
try:
with open(COOKIES_FILE, "rb") as f:
session.cookies.update(pickle.load(f))
print(" Cookies cargadas con éxito desde archivo.")
return True
except Exception as e:
print(f" Error cargando cookies desde archivo: {e}")
return False
def save_cookies(session):
"""Guarda cookies en un archivo."""
with open(COOKIES_FILE, "wb") as f:
pickle.dump(session.cookies, f)
print(" Cookies guardadas localmente.")
def login(session):
"""Inicia sesión en Rutracker si no hay una cookie válida."""
if not USERNAME or not PASSWORD:
return False # Si no hay credenciales, no se puede hacer login
print(" Verificando sesión en Rutracker...")
# Si ya hay cookies, probamos si siguen siendo válidas
test_response = session.get("https://rutracker.org/forum/index.php", headers=HEADERS)
if "logout" in test_response.text:
print(" Sesión válida con cookies.")
return True
# No hay sesión válida, hacemos login
print(" Iniciando sesión en Rutracker...")
login_data = {
"login_username": USERNAME,
"login_password": PASSWORD,
"login": "Вход"
}
response = session.post(LOGIN_URL, data=login_data, headers=HEADERS)
if "logout" in response.text:
print(" Login exitoso, guardando cookies...")
save_cookies(session)
return True
else:
print(" Error en el inicio de sesión. Revisa tus credenciales.")
return False
def rutracker_search(query):
"""Realiza una búsqueda en Rutracker y devuelve los primeros 5 resultados con magnet links."""
global LAST_EXECUTION_TIME
# Verificar el rate limit
current_time = time.time()
if current_time - LAST_EXECUTION_TIME < RATE_LIMIT_SECONDS:
remaining_time = int(RATE_LIMIT_SECONDS - (current_time - LAST_EXECUTION_TIME))
return f"⏳ Espera {remaining_time} segundos antes de hacer otra búsqueda."
# Actualizar el tiempo de última ejecución
LAST_EXECUTION_TIME = current_time
# Crear sesión con CloudScraper
session = cloudscraper.create_scraper()
# Cargar cookies si existen
cookies_loaded = load_cookies(session)
# Si no había cookies o no eran válidas, iniciar sesión
if not cookies_loaded and not login(session):
return " No se pudo autenticar en Rutracker."
# Realizar la búsqueda
search_params = {"nm": query}
search_response = session.get(SEARCH_URL, params=search_params)
soup = BeautifulSoup(search_response.text, "html.parser")
# Buscar enlaces a temas de torrents
topics = soup.select("a.tLink")
if not topics:
return f" No se encontraron torrents para: `{query}`"
results = []
for topic in topics[:5]: # Limita a los primeros 5 resultados
topic_title = topic.text.strip()
topic_url = BASE_URL + topic["href"]
# Buscar el magnet link dentro de la página del torrent
topic_page = session.get(topic_url)
topic_soup = BeautifulSoup(topic_page.text, "html.parser")
magnet_link = topic_soup.select_one("a.magnet-link")
if magnet_link:
results.append(f"**{topic_title}**\n [Magnet]({magnet_link['href']})")
else:
results.append(f"**{topic_title}**\n No se encontró magnet link.")
return "\n\n".join(results)
def run(sender, *args):
"""Ejecuta la búsqueda de torrents en Rutracker."""
if not args:
return " Uso: `.rutracker <búsqueda>` (Ejemplo: `.torrent Metallica`)"
query = " ".join(args)
return rutracker_search(query)
def help():
"""Descripción del plugin para el comando .help"""
return " Usa `.rutracker <búsqueda>` para buscar torrents en Rutracker. Máximo 1 consulta cada 60s."

4
plugins/test.py Normal file
View File

@ -0,0 +1,4 @@
def run(*args):
"""Mensaje de prueba"""
return "Este es un mensaje de prueba. El bot funciona correctamente."

48
plugins/timezone.py Normal file
View File

@ -0,0 +1,48 @@
import requests
def obtener_hora_zona(zona):
"""Obtiene la hora actual de una zona horaria usando una API."""
try:
response = requests.get(f"http://worldtimeapi.org/api/timezone/{zona}", timeout=10)
if response.status_code == 200:
data = response.json()
datetime_str = data.get("datetime", "").split(".")[0]
return f" La hora en {zona} es {datetime_str.replace('T', ' ')}"
elif response.status_code == 404:
return " Zona horaria no encontrada. Usa `.timezone list` para ver opciones."
else:
return " Error al obtener la hora."
except requests.exceptions.RequestException as e:
return f" Error de conexión: {e}"
def listar_zonas():
"""Lista las zonas horarias disponibles."""
try:
response = requests.get("http://worldtimeapi.org/api/timezone", timeout=10)
if response.status_code == 200:
zonas = response.json()
return " Zonas horarias disponibles:\n" + ", ".join(zonas[:10]) + "...\nUsa `.timezone <zona>` para consultar una."
else:
return " No se pudieron obtener las zonas horarias."
except requests.exceptions.RequestException as e:
return f" Error de conexión: {e}"
def run(sender, *args):
"""Función principal que será ejecutada por el bot."""
if not args:
return " Debes especificar una zona horaria. Usa `.timezone list` para ver opciones."
zona = " ".join(args)
if zona.lower() == "list":
return listar_zonas()
return obtener_hora_zona(zona)
def help():
"""Descripción del plugin para el comando .help"""
return (" Consulta la hora en cualquier zona horaria.\n"
"Uso: `.timezone <zona>`\n"
"**Ejemplo:** `.timezone Europe/Madrid`\n"
"**Otras opciones:**\n"
"- `.timezone list` → Lista algunas zonas horarias disponibles.")

21
plugins/uptime.py Normal file
View File

@ -0,0 +1,21 @@
import time
class UptimePlugin:
def __init__(self, bot):
"""Guarda la referencia al bot y usa su tiempo de inicio."""
self.bot = bot
def run(self, sender=None, *args):
"""Muestra cuánto tiempo lleva corriendo el bot."""
if not hasattr(self.bot, "start_time"):
self.bot.start_time = time.time() # Asegurar que `start_time` exista
uptime_seconds = int(time.time() - self.bot.start_time)
hours, remainder = divmod(uptime_seconds, 3600)
minutes, seconds = divmod(remainder, 60)
return f"⏳ **Uptime:** {hours}h {minutes}m {seconds}s"
def help(self):
return "Uso: .uptime - Muestra cuánto tiempo lleva encendido el bot."

9
plugins/uuid.py Normal file
View File

@ -0,0 +1,9 @@
import uuid
def run():
"""Genera un UUID aleatorio."""
return f"🆔 **UUID:** {uuid.uuid4()}"
def help():
return "Uso: .uuid - Genera un UUID aleatorio."

10
plugins/uuid_utils.py Normal file
View File

@ -0,0 +1,10 @@
import uuid
class Uuid_utilsPlugin:
def run(self, *args):
"""Genera un UUID aleatorio."""
return f"🆔 **UUID:** {uuid.uuid4()}"
def help(self):
return "Uso: .uuid - Genera un UUID aleatorio."

65
plugins/weather.py Normal file
View File

@ -0,0 +1,65 @@
import os
import requests
import logging
# Configurar el logger principal
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s"
)
API_KEY = os.getenv("WEATHER_API_KEY")
BASE_URL = "http://api.openweathermap.org/data/2.5/weather"
def fetch_weather(city):
"""Obtiene el clima de una ciudad desde OpenWeather."""
if not API_KEY:
logging.error("[WEATHER] No se encontró WEATHER_API_KEY en el entorno.")
return "Error: La API Key no está configurada."
params = {
"q": city,
"appid": API_KEY,
"units": "metric", # °C
"lang": "es"
}
try:
response = requests.get(BASE_URL, params=params, timeout=10)
data = response.json()
if response.status_code == 200 and "main" in data:
temp = data["main"].get("temp", "N/A")
desc = data["weather"][0].get("description", "N/A").capitalize()
humidity = data["main"].get("humidity", "N/A")
wind_speed = data["wind"].get("speed", "N/A")
return (
f" Clima en {city}: {temp}°C, {desc}\n"
f" Humedad: {humidity}%\n"
f" Viento: {wind_speed} m/s"
)
elif response.status_code == 404:
logging.warning(f"[WEATHER] Ciudad no encontrada: {city}")
return f"No se encontró información del clima para '{city}'."
else:
logging.error(f"[WEATHER] Error en la API. Código: {response.status_code}, data: {data}")
return "Error al obtener los datos del clima."
except requests.exceptions.RequestException as e:
logging.exception("[WEATHER] Error conectando con OpenWeather.")
return f"Error al conectar con OpenWeather: {e}"
def run(sender, *args):
if not args:
return help()
ciudad = " ".join(args)
return fetch_weather(ciudad)
def help():
return """.weather <ciudad> - Muestra el clima actual.
Ejemplo: .weather Madrid"""

24
plugins/whois.py Normal file
View File

@ -0,0 +1,24 @@
# plugins/whois.py
import whois
class WhoisPlugin:
def run(self, sender, *args):
"""Consulta información WHOIS de un dominio."""
if not args:
return "Uso: .whois <dominio> - Ejemplo: .whois google.com"
domain = args[0]
try:
w = whois.whois(domain)
info = f" **WHOIS de {domain}**\n"
info += f" Registrante: {w.name or 'Desconocido'}\n"
info += f" Email: {w.emails or 'Desconocido'}\n"
info += f" Servidores DNS: {', '.join(w.nameservers) if w.nameservers else 'No disponibles'}\n"
info += f" Expira el: {w.expiration_date}\n"
return info
except Exception as e:
return f" Error obteniendo WHOIS: {str(e)}"
def help(self):
return "Uso: .whois <dominio> - Obtiene información WHOIS de un dominio."

29
plugins/wiki.py Normal file
View File

@ -0,0 +1,29 @@
import wikipediaapi
wiki_lang = "es"
wiki = wikipediaapi.Wikipedia(
language=wiki_lang,
user_agent="MyIRCBot/1.0 (https://github.com/tu-repo; contacto@example.com)"
)
def fetch_summary(query):
"""Busca un resumen en Wikipedia."""
try:
page = wiki.page(query)
if not page.exists():
return f" No encontré información sobre '{query}'."
return f" {query}:\n{page.summary[:300]}..." # Máximo 300 caracteres
except Exception as e:
return f" Error en la consulta: {e}"
def run(sender, *args):
"""Ejecuta la búsqueda en Wikipedia."""
if not args:
return " Uso: `.wiki <término>` (Ejemplo: `.wiki Arch Linux`)"
termino = " ".join(args)
return fetch_summary(termino)
def help():
return " Uso: `.wiki <término>` - Busca en Wikipedia.\nEjemplo: `.wiki Arch Linux`"

12
requirements.txt Normal file
View File

@ -0,0 +1,12 @@
requests
irc
aioconsole
wikipedia-api
yt-dlp
feedparser
ollama
beautifulsoup4
python-whois
dnspython
cloudscraper
pyfiglet