Ekaropolus bc7a8c5625
All checks were successful
continuous-integration/drone/push Build is passing
Telegram Bot integration Messagees
2025-05-20 02:21:19 -06:00

85 lines
3.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()}")
from django.db import models
class TelegramConversation(models.Model):
bot = models.ForeignKey(
'TelegramBot',
on_delete=models.CASCADE,
related_name='conversations'
)
user_id = models.CharField(max_length=64)
started_at = models.DateTimeField(auto_now_add=True)
def __str__(self):
return f"{self.user_id} @ {self.started_at:%Y-%m-%d %H:%M}"
class TelegramMessage(models.Model):
IN = 'in'
OUT = 'out'
DIRECTION_CHOICES = [
(IN, 'In'),
(OUT, 'Out'),
]
conversation = models.ForeignKey(
TelegramConversation,
on_delete=models.CASCADE,
related_name='messages'
)
direction = models.CharField(max_length=4, choices=DIRECTION_CHOICES)
content = models.TextField()
timestamp = models.DateTimeField(auto_now_add=True)
response_time_ms = models.IntegerField(null=True, blank=True)
def __str__(self):
return f"[{self.direction}] {self.content[:30]}"