Webhook for telegram bot fix to validate asigned of langchain and handling exceptions
Some checks failed
continuous-integration/drone Build is failing

This commit is contained in:
Ekaropolus 2025-07-05 21:11:36 -06:00
parent 69af6a9458
commit 3abe84ca2b
2 changed files with 25 additions and 36 deletions

View File

@ -9,32 +9,37 @@ class TelegramBotAdmin(admin.ModelAdmin):
list_filter = ("is_active",) list_filter = ("is_active",)
actions = ["set_webhooks"] actions = ["set_webhooks"]
@admin.action(description="Set webhooks for selected bots") @admin.action(description="Set webhooks for selected bots")
def set_webhooks(self, request, queryset): def set_webhooks(self, request, queryset):
base_url = request.build_absolute_uri("/")[:-1] # Get base server URL base_url = f"{request.scheme}://{request.get_host()}"
results = [] for bot in queryset:
for bot in queryset: if bot.is_active:
if bot.is_active: try:
try: if not bot.assistant:
result = bot.set_webhook(base_url)
self.message_user( self.message_user(
request, request,
f"Webhook set for {bot.name}: {result}", f"Bot {bot.name} has no assistant configured.",
level="success", level="warning",
) )
except Exception as e: continue
self.message_user( result = bot.set_webhook(base_url)
request,
f"Failed to set webhook for {bot.name}: {str(e)}",
level="error",
)
results.append(result)
else:
self.message_user( self.message_user(
request, request,
f"Skipped inactive bot: {bot.name}", f"Webhook set for {bot.name}: {result}",
level="warning", level="success",
) )
except Exception as e:
self.message_user(
request,
f"Failed to set webhook for {bot.name}: {str(e)}",
level="error",
)
else:
self.message_user(
request,
f"Skipped inactive bot: {bot.name}",
level="warning",
)
def get_assistant_name(self, obj): def get_assistant_name(self, obj):
"""Show the name of the assistant linked to the bot.""" """Show the name of the assistant linked to the bot."""

View File

@ -1,16 +0,0 @@
import requests
from pxy_bots.models import TelegramBot
BASE_URL = "https://your-domain.com/bots/webhook/"
def set_telegram_webhooks():
"""Sets webhooks for all active bots."""
bots = TelegramBot.objects.filter(is_active=True)
for bot in bots:
webhook_url = f"{BASE_URL}{bot.name}/"
response = requests.post(
f"https://api.telegram.org/bot{bot.token}/setWebhook",
data={"url": webhook_url}
)
print(f"Webhook for {bot.name} ({bot.username}) set to: {webhook_url}")
print("Response:", response.json())