48 lines
2.0 KiB
Python
48 lines
2.0 KiB
Python
from django.db import models
|
|
from pxy_langchain.models import AIAssistant # Now referencing LangChain AI assistants
|
|
|
|
class TelegramBot(models.Model):
|
|
"""
|
|
Represents a Telegram bot that interacts with users using a LangChain AI assistant.
|
|
"""
|
|
name = models.CharField(max_length=50, unique=True, help_text="Bot name (e.g., 'SupportBot').")
|
|
username = models.CharField(max_length=50, unique=True, help_text="Bot username (e.g., 'SupportBot').")
|
|
token = models.CharField(max_length=200, unique=True, help_text="Telegram bot token.")
|
|
is_active = models.BooleanField(default=True, help_text="Indicates if this bot is active.")
|
|
assistant = models.ForeignKey(
|
|
AIAssistant,
|
|
on_delete=models.CASCADE,
|
|
related_name="telegram_bots",
|
|
help_text="The LangChain AI assistant associated with this Telegram bot.",
|
|
)
|
|
|
|
def __str__(self):
|
|
return f"{self.name} (@{self.username})"
|
|
|
|
@staticmethod
|
|
def get_bot_token(bot_name):
|
|
"""Retrieve the token for the given bot name."""
|
|
try:
|
|
bot = TelegramBot.objects.get(name=bot_name, is_active=True)
|
|
return bot.token
|
|
except TelegramBot.DoesNotExist:
|
|
raise ValueError(f"Bot with name '{bot_name}' not found or inactive.")
|
|
|
|
def set_webhook(self, base_url):
|
|
"""
|
|
Set the webhook for this bot dynamically based on the server's base URL.
|
|
"""
|
|
if not self.is_active:
|
|
raise ValueError(f"Bot '{self.name}' is inactive. Activate it before setting the webhook.")
|
|
|
|
webhook_url = f"{base_url}/bots/webhook/{self.name}/"
|
|
response = requests.post(
|
|
f"https://api.telegram.org/bot{self.token}/setWebhook",
|
|
data={"url": webhook_url}
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return response.json()
|
|
else:
|
|
raise ValueError(f"Failed to set webhook for {self.name}: {response.json()}")
|