69 lines
1.9 KiB
Python
69 lines
1.9 KiB
Python
"""
|
||
Async Telegram notification service.
|
||
Replaces the synchronous urllib-based notify() from foxy-autopilot.py.
|
||
"""
|
||
|
||
import logging
|
||
import httpx
|
||
|
||
from app.config import settings
|
||
|
||
log = logging.getLogger("foxy.notifications")
|
||
|
||
|
||
async def send_telegram(message: str) -> bool:
|
||
"""Send a message to the configured Telegram chat."""
|
||
if not settings.TELEGRAM_BOT_TOKEN or not settings.TELEGRAM_CHAT_ID:
|
||
log.warning("Telegram not configured — skipping notification")
|
||
return False
|
||
|
||
url = f"https://api.telegram.org/bot{settings.TELEGRAM_BOT_TOKEN}/sendMessage"
|
||
payload = {
|
||
"chat_id": settings.TELEGRAM_CHAT_ID,
|
||
"text": message,
|
||
"parse_mode": "HTML",
|
||
}
|
||
|
||
try:
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
resp = await client.post(url, data=payload)
|
||
if resp.status_code == 200:
|
||
return True
|
||
log.warning(f"Telegram API returned {resp.status_code}: {resp.text[:200]}")
|
||
return False
|
||
except Exception as e:
|
||
log.warning(f"Telegram notification error (ignored): {e}")
|
||
return False
|
||
|
||
|
||
async def notify_project_event(
|
||
project_name: str,
|
||
event: str,
|
||
details: str = "",
|
||
agent: str = "",
|
||
) -> bool:
|
||
"""Send a formatted project event notification."""
|
||
msg_parts = [f"🦊 <b>Foxy Dev Team</b>", f"📋 <b>{project_name}</b>"]
|
||
|
||
if agent:
|
||
msg_parts.append(f"🤖 {agent}")
|
||
msg_parts.append(f"📊 {event}")
|
||
if details:
|
||
msg_parts.append(f"ℹ️ {details}")
|
||
|
||
return await send_telegram("\n".join(msg_parts))
|
||
|
||
|
||
async def notify_error(
|
||
project_name: str,
|
||
agent_name: str,
|
||
error: str,
|
||
) -> bool:
|
||
"""Send an error notification."""
|
||
return await send_telegram(
|
||
f"🦊 ⚠️ <b>Erreur</b>\n"
|
||
f"📋 {project_name}\n"
|
||
f"❌ {agent_name}\n"
|
||
f"<code>{error[:300]}</code>"
|
||
)
|