nikdoof/leggen
1
1
Fork 0
mirror of https://github.com/elisiariocouto/leggen.git synced 2025-12-13 11:22:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix notification channels returning null by connecting API to real implementations

Browse Source 
- Update notification service to handle both old (api-key/chat-id) and new (token/chat_id) config formats
- Connect API service to actual Discord/Telegram notification implementations from CLI codebase
- Fix API routes to properly detect configured services using correct config keys
- Telegram notifications now work correctly, Discord properly shows as not configured
This commit is contained in:
Elisiário Couto
2025-09-09 16:10:08 +01:00
committed by Elisiário Couto
parent 541cb262ee
commit 16afa1ed8a
2 changed files with 82 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
leggend/api/routes/notifications.py
View File

@@ -36,11 +36,14 @@ async def get_notification_settings() -> APIResponse:
if discord_config.get("webhook")
else None,
telegram=TelegramConfig(
token="***" if telegram_config.get("token") else "",
chat_id=telegram_config.get("chat_id", 0),
token="***"
if (telegram_config.get("token") or telegram_config.get("api-key"))
else "",
chat_id=telegram_config.get("chat_id")
or telegram_config.get("chat-id", 0),
enabled=telegram_config.get("enabled", True),
)
if telegram_config.get("token")
if (telegram_config.get("token") or telegram_config.get("api-key"))
else None,
filters=NotificationFilters(
case_insensitive=filters_config.get("case-insensitive", {}),
@@ -158,12 +161,24 @@ async def get_notification_services() -> APIResponse:
"telegram": {
"name": "Telegram",
"enabled": bool(
notifications_config.get("telegram", {}).get("token")
and notifications_config.get("telegram", {}).get("chat_id")
(
notifications_config.get("telegram", {}).get("token")
or notifications_config.get("telegram", {}).get("api-key")
)
and (
notifications_config.get("telegram", {}).get("chat_id")
or notifications_config.get("telegram", {}).get("chat-id")
)
),
"configured": bool(
notifications_config.get("telegram", {}).get("token")
and notifications_config.get("telegram", {}).get("chat_id")
(
notifications_config.get("telegram", {}).get("token")
or notifications_config.get("telegram", {}).get("api-key")
)
and (
notifications_config.get("telegram", {}).get("chat_id")
or notifications_config.get("telegram", {}).get("chat-id")
)
),
"active": notifications_config.get("telegram", {}).get("enabled", True),
},

63
leggend/services/notification_service.py
View File

@@ -95,7 +95,8 @@ class NotificationService:
telegram_config = self.notifications_config.get("telegram", {})
return bool(
telegram_config.get("token")
and telegram_config.get("chat_id")
or telegram_config.get("api-key")
and (telegram_config.get("chat_id") or telegram_config.get("chat-id"))
and telegram_config.get("enabled", True)
)
@@ -117,11 +118,67 @@ class NotificationService:
async def _send_discord_test(self, message: str) -> None:
"""Send Discord test notification"""
logger.info(f"Sending Discord test: {message}")
try:
from leggen.notifications.discord import send_expire_notification
import click
# Create a mock context with the webhook
ctx = click.Context(click.Command("test"))
ctx.obj = {
"notifications": {
"discord": {
"webhook": self.notifications_config.get("discord", {}).get(
"webhook"
)
}
}
}
# Send test notification using the actual implementation
test_notification = {
"bank": "Test",
"requisition_id": "test-123",
"status": "active",
"days_left": 30,
}
send_expire_notification(ctx, test_notification)
logger.info(f"Discord test notification sent: {message}")
except Exception as e:
logger.error(f"Failed to send Discord test notification: {e}")
raise
async def _send_telegram_test(self, message: str) -> None:
"""Send Telegram test notification"""
logger.info(f"Sending Telegram test: {message}")
try:
from leggen.notifications.telegram import send_expire_notification
import click
# Create a mock context with the telegram config
ctx = click.Context(click.Command("test"))
telegram_config = self.notifications_config.get("telegram", {})
ctx.obj = {
"notifications": {
"telegram": {
"api-key": telegram_config.get("token")
or telegram_config.get("api-key"),
"chat-id": telegram_config.get("chat_id")
or telegram_config.get("chat-id"),
}
}
}
# Send test notification using the actual implementation
test_notification = {
"bank": "Test",
"requisition_id": "test-123",
"status": "active",
"days_left": 30,
}
send_expire_notification(ctx, test_notification)
logger.info(f"Telegram test notification sent: {message}")
except Exception as e:
logger.error(f"Failed to send Telegram test notification: {e}")
raise
async def _send_discord_expiry(self, notification_data: Dict[str, Any]) -> None:
"""Send Discord expiry notification"""