Mikkel Georgsen
1cb16e6e8f
Single-process Python app that: - Runs a Telegram bot in a group chat, logging all messages/files to libsql - Exposes send_message, pull_updates, queue_status MCP tools over HTTP - Downloads and stores file attachments with Telegram file_id + local path - Accessible via NetBird mesh at mgmt.mg:8321 (no auth needed) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
297 lines
11 KiB
Python
297 lines
11 KiB
Python
"""Telegram bot: logs group messages to libsql, sends outbound messages."""
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from telegram import Bot, Update
|
|
from telegram.ext import (
|
|
Application,
|
|
MessageHandler,
|
|
filters,
|
|
ContextTypes,
|
|
)
|
|
|
|
from .config import get_bot_token, get_group_chat_id, get_homelab_bot_id, MEDIA_DIR
|
|
from .db import Database
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class BridgeBot:
|
|
def __init__(self, db: Database):
|
|
self.db = db
|
|
self.token = get_bot_token()
|
|
self.group_chat_id = get_group_chat_id()
|
|
self.homelab_bot_id = get_homelab_bot_id()
|
|
self.bot: Bot | None = None
|
|
self.app: Application | None = None
|
|
self._outbound_task: asyncio.Task | None = None
|
|
self._my_bot_id: int | None = None
|
|
|
|
def _classify_sender(self, user_id: int | None) -> str:
|
|
"""Classify who sent a message."""
|
|
if user_id is None:
|
|
return "unknown"
|
|
if user_id == self._my_bot_id:
|
|
return "mcp_bot"
|
|
if self.homelab_bot_id and user_id == self.homelab_bot_id:
|
|
return "homelab_bot"
|
|
# Check if user is a bot (we'll handle this in the handler with full user info)
|
|
return "user"
|
|
|
|
async def _log_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
|
|
"""Log any group message to the database."""
|
|
message = update.effective_message
|
|
if not message or message.chat_id != self.group_chat_id:
|
|
return
|
|
|
|
user = message.from_user
|
|
user_id = user.id if user else None
|
|
sender_name = None
|
|
if user:
|
|
if user.is_bot:
|
|
sender_name = user.first_name or user.username
|
|
else:
|
|
parts = [user.first_name or "", user.last_name or ""]
|
|
sender_name = " ".join(p for p in parts if p) or user.username
|
|
|
|
sender_type = self._classify_sender(user_id)
|
|
# Refine: if it's a bot we don't know, label it
|
|
if sender_type == "user" and user and user.is_bot:
|
|
sender_type = "bot"
|
|
|
|
reply_to_id = None
|
|
if message.reply_to_message:
|
|
reply_to_id = message.reply_to_message.message_id
|
|
|
|
# Determine if there are attachments
|
|
has_attachment = bool(
|
|
message.photo or message.document or message.video
|
|
or message.voice or message.audio or message.sticker
|
|
or message.video_note or message.animation
|
|
)
|
|
|
|
# Get text content
|
|
content = message.text or message.caption or None
|
|
|
|
# Parse message date
|
|
msg_date = message.date
|
|
if msg_date:
|
|
created_at = msg_date.astimezone(timezone.utc).isoformat()
|
|
else:
|
|
created_at = datetime.now(timezone.utc).isoformat()
|
|
|
|
# Insert message
|
|
msg_id = self.db.insert_message(
|
|
telegram_message_id=message.message_id,
|
|
chat_id=message.chat_id,
|
|
sender_type=sender_type,
|
|
sender_id=user_id,
|
|
sender_name=sender_name,
|
|
content=content,
|
|
reply_to_message_id=reply_to_id,
|
|
has_attachment=has_attachment,
|
|
created_at=created_at,
|
|
)
|
|
|
|
if msg_id is None:
|
|
logger.debug(f"Duplicate message {message.message_id}, skipping")
|
|
return
|
|
|
|
logger.info(
|
|
f"Logged message {message.message_id} from {sender_name} ({sender_type})"
|
|
)
|
|
|
|
# Process attachments
|
|
if has_attachment:
|
|
await self._process_attachments(message, msg_id)
|
|
|
|
async def _process_attachments(self, message, db_message_id: int):
|
|
"""Download and log attachments."""
|
|
attachments = []
|
|
|
|
if message.photo:
|
|
# Get highest resolution photo
|
|
photo = message.photo[-1]
|
|
attachments.append(("photo", photo.file_id, photo.file_unique_id,
|
|
None, "image/jpeg", photo.file_size, message.caption))
|
|
|
|
if message.document:
|
|
doc = message.document
|
|
attachments.append(("document", doc.file_id, doc.file_unique_id,
|
|
doc.file_name, doc.mime_type, doc.file_size, message.caption))
|
|
|
|
if message.video:
|
|
vid = message.video
|
|
attachments.append(("video", vid.file_id, vid.file_unique_id,
|
|
vid.file_name, vid.mime_type, vid.file_size, message.caption))
|
|
|
|
if message.voice:
|
|
voice = message.voice
|
|
attachments.append(("voice", voice.file_id, voice.file_unique_id,
|
|
None, voice.mime_type, voice.file_size, None))
|
|
|
|
if message.audio:
|
|
audio = message.audio
|
|
attachments.append(("audio", audio.file_id, audio.file_unique_id,
|
|
audio.file_name, audio.mime_type, audio.file_size, None))
|
|
|
|
if message.sticker:
|
|
sticker = message.sticker
|
|
attachments.append(("sticker", sticker.file_id, sticker.file_unique_id,
|
|
None, None, sticker.file_size, None))
|
|
|
|
if message.animation:
|
|
anim = message.animation
|
|
attachments.append(("animation", anim.file_id, anim.file_unique_id,
|
|
anim.file_name, anim.mime_type, anim.file_size, message.caption))
|
|
|
|
for file_type, file_id, file_unique_id, file_name, mime_type, file_size, caption in attachments:
|
|
# Download file
|
|
local_path = await self._download_file(file_id, file_unique_id, file_name, file_type)
|
|
|
|
self.db.insert_attachment(
|
|
message_id=db_message_id,
|
|
file_type=file_type,
|
|
file_id=file_id,
|
|
file_unique_id=file_unique_id,
|
|
file_name=file_name,
|
|
mime_type=mime_type,
|
|
file_size=file_size,
|
|
local_path=str(local_path) if local_path else None,
|
|
caption=caption,
|
|
)
|
|
logger.info(f"Saved attachment: {file_type} -> {local_path}")
|
|
|
|
async def _download_file(
|
|
self, file_id: str, file_unique_id: str, file_name: str | None, file_type: str
|
|
) -> Path | None:
|
|
"""Download a file from Telegram to local media directory."""
|
|
try:
|
|
tg_file = await self.bot.get_file(file_id)
|
|
|
|
# Organize by date
|
|
date_dir = MEDIA_DIR / datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
|
date_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Build filename
|
|
if file_name:
|
|
local_name = f"{file_unique_id}_{file_name}"
|
|
else:
|
|
ext_map = {
|
|
"photo": ".jpg", "voice": ".ogg", "sticker": ".webp",
|
|
"video": ".mp4", "animation": ".mp4", "audio": ".mp3",
|
|
}
|
|
ext = ext_map.get(file_type, "")
|
|
local_name = f"{file_unique_id}{ext}"
|
|
|
|
local_path = date_dir / local_name
|
|
await tg_file.download_to_drive(str(local_path))
|
|
return local_path
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to download file {file_id}: {e}")
|
|
return None
|
|
|
|
async def send_to_group(self, text: str, attribution: str = "claude.ai"):
|
|
"""Send an attributed message to the group chat."""
|
|
formatted = f"*\\[{attribution}\\]* {self._escape_markdown(text)}"
|
|
try:
|
|
await self.bot.send_message(
|
|
chat_id=self.group_chat_id,
|
|
text=formatted,
|
|
parse_mode="MarkdownV2",
|
|
)
|
|
except Exception:
|
|
# Fallback to plain text if markdown fails
|
|
plain = f"[{attribution}] {text}"
|
|
await self.bot.send_message(
|
|
chat_id=self.group_chat_id,
|
|
text=plain,
|
|
)
|
|
|
|
@staticmethod
|
|
def _escape_markdown(text: str) -> str:
|
|
"""Escape MarkdownV2 special characters, preserving code blocks."""
|
|
special = r"_*[]()~`>#+-=|{}.!"
|
|
result = []
|
|
i = 0
|
|
in_code_block = False
|
|
in_inline_code = False
|
|
|
|
while i < len(text):
|
|
# Check for code block
|
|
if text[i:i+3] == "```":
|
|
in_code_block = not in_code_block
|
|
result.append("```")
|
|
i += 3
|
|
continue
|
|
# Check for inline code
|
|
if text[i] == "`" and not in_code_block:
|
|
in_inline_code = not in_inline_code
|
|
result.append("`")
|
|
i += 1
|
|
continue
|
|
|
|
if not in_code_block and not in_inline_code and text[i] in special:
|
|
result.append(f"\\{text[i]}")
|
|
else:
|
|
result.append(text[i])
|
|
i += 1
|
|
|
|
return "".join(result)
|
|
|
|
async def _outbound_loop(self):
|
|
"""Poll outbound queue and send messages."""
|
|
while True:
|
|
try:
|
|
pending = self.db.get_pending_outbound()
|
|
for msg in pending:
|
|
try:
|
|
await self.send_to_group(msg["content"], msg["attribution"])
|
|
self.db.mark_outbound_sent(msg["id"])
|
|
logger.info(f"Sent outbound message {msg['id']}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to send outbound {msg['id']}: {e}")
|
|
self.db.mark_outbound_failed(msg["id"])
|
|
except Exception as e:
|
|
logger.error(f"Outbound loop error: {e}")
|
|
|
|
await asyncio.sleep(2)
|
|
|
|
async def _post_init(self, application: Application):
|
|
"""Called after bot initialization."""
|
|
self.bot = application.bot
|
|
me = await self.bot.get_me()
|
|
self._my_bot_id = me.id
|
|
logger.info(f"MCP Bridge bot started as @{me.username} (ID: {me.id})")
|
|
logger.info(f"Monitoring group chat: {self.group_chat_id}")
|
|
|
|
# Start outbound message loop
|
|
self._outbound_task = asyncio.create_task(self._outbound_loop())
|
|
|
|
async def _post_shutdown(self, application: Application):
|
|
"""Cleanup on shutdown."""
|
|
if self._outbound_task:
|
|
self._outbound_task.cancel()
|
|
try:
|
|
await self._outbound_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
def build_application(self) -> Application:
|
|
"""Build the telegram Application (but don't start it yet)."""
|
|
builder = Application.builder().token(self.token)
|
|
self.app = builder.build()
|
|
|
|
# Log ALL messages in the group (text, photos, documents, etc.)
|
|
self.app.add_handler(
|
|
MessageHandler(filters.ALL, self._log_message)
|
|
)
|
|
|
|
self.app.post_init = self._post_init
|
|
self.app.post_shutdown = self._post_shutdown
|
|
|
|
return self.app
|