"""Telegram Notifications for Brief Approval.
Sends DMs to Matt for content brief approvals.
"""
import os
import logging
from typing import Optional
from datetime import datetime
import requests
logger = logging.getLogger(name)
Matt's Telegram user ID
MATT_TELEGRAM_ID = "8386527252"
Bot token from environment
BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") # Same as HoffmannButlerBot
def _send_message(chat_id: str, text: str, reply_markup: Optional[dict] = None) -> bool:
"""Send Telegram message via Bot API."""
if not BOT_TOKEN:
logger.error("TELEGRAM_BOT_TOKEN not set")
return False
url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
payload = {
"chat_id": chat_id,
"text": text,
"parse_mode": "HTML",
}
if reply_markup:
payload["reply_markup"] = reply_markup
try:
response = requests.post(url, json=payload, timeout=30)
response.raise_for_status()
return True
except Exception as e:
logger.error(f"Failed to send Telegram message: {e}")
return False
def notify_brief_pending(
brief_id: str,
title: str,
struggle_angle: str,
created_by: str
) -> bool:
"""
Send approval request to Matt via Telegram DM.
Args:
brief_id: The brief UUID
title: Brief title
struggle_angle: Brief struggle angle (truncated)
created_by: Who created the brief
Returns:
True if sent successfully
"""
# Truncate struggle angle for Telegram preview
preview = struggle_angle[:200] + "..." if len(struggle_angle) > 200 else struggle_angle
message = f"""📝 <b>New Content Brief Awaiting Approval</b>
Title: {title}
Created by: {created_by}
Struggle Angle:
{preview}
What would you like to do?"""
# Inline keyboard with approve/reject/edit buttons
reply_markup = {
"inline_keyboard": [
[
{"text": "✅ Approve", "callback_data": f"brief:approve:{brief_id}"},
{"text": "❌ Reject", "callback_data": f"brief:reject:{brief_id}"}
],
[
{"text": "✏️ Edit", "callback_data": f"brief:edit:{brief_id}"},
{"text": "📄 View Full", "callback_data": f"brief:view:{brief_id}"}
]
]
}
return _send_message(MATT_TELEGRAM_ID, message, reply_markup)
def notify_brief_approved(
brief_id: str,
title: str,
approved_by: str
) -> bool:
"""Notify that brief was approved and generation started."""
message = f"""✅ Brief Approved
Title: {title}
Approved by: {approved_by}
Content generation started. You'll be notified when it's ready."""
return _send_message(MATT_TELEGRAM_ID, message)
def notify_brief_rejected(
brief_id: str,
title: str,
reason: str
) -> bool:
"""Notify that brief was rejected with feedback."""
message = f"""❌ Brief Rejected
Title: {title}
Feedback:
{reason}
The brief author has been notified."""
return _send_message(MATT_TELEGRAM_ID, message)
def notify_generation_complete(
brief_id: str,
title: str,
struggle_score: float,
preview_url: str
) -> bool:
"""Notify that content generation is complete."""
score_emoji = "🟢" if struggle_score >= 75 else "🟡" if struggle_score >= 50 else "🔴"
message = f"""✨ <b>Content Generation Complete</b>
Title: {title}
{score_emoji} Struggle Score: {struggle_score:.1f}/100
return _send_message(MATT_TELEGRAM_ID, message)
def notify_author_brief_submitted(
chat_id: str,
title: str
) -> bool:
"""Notify brief author that submission was received."""
message = f"""📝 Brief Submitted
Title: {title}
Your brief has been sent for approval. You'll be notified when it's reviewed."""
return _send_message(chat_id, message)
def notify_author_brief_rejected(
chat_id: str,
title: str,
feedback: str
) -> bool:
"""Notify author that brief was rejected with feedback."""
message = f"""❌ Brief Needs Revision
Title: {title}
Feedback:
{feedback}
Please revise and resubmit."""
return _send_message(chat_id, message)