diff --git a/telegram-mcp-server/telegram/__init__.py b/telegram-mcp-server/telegram/__init__.py new file mode 100644 index 0000000..a8d8c64 --- /dev/null +++ b/telegram-mcp-server/telegram/__init__.py @@ -0,0 +1,29 @@ +""" +This module implements data models and functions for retrieving and sending +Telegram messages, managing chats, and working with contacts. + +The module connects to a SQLite database that stores all Telegram messages and chat data, +which is maintained by the Telegram Bridge. It also provides an HTTP client for sending +messages via the Bridge's API endpoint. + +Main features: +- Data models for messages, chats, contacts, and message context +- Database access functions for retrieving messages, chats, and contacts +- HTTP client for sending messages through the Telegram Bridge +- Helper functions for displaying formatted messages and chats +""" + +import os.path + +# Database path +MESSAGES_DB_PATH = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), '..', 'telegram-bridge', 'store', 'messages.db') +TELEGRAM_API_BASE_URL = "http://localhost:8081/api" + +# Import all components to make them available at the module level +from .models import Message, Chat, Contact, MessageContext +from .display import print_message, print_messages_list, print_chat, print_chats_list +from .api import send_message +from .database import ( + search_contacts, list_messages, get_message_context, list_chats, + get_chat, get_direct_chat_by_contact, get_contact_chats, get_last_interaction +) \ No newline at end of file diff --git a/telegram-mcp-server/telegram/api.py b/telegram-mcp-server/telegram/api.py new file mode 100644 index 0000000..2f8c039 --- /dev/null +++ b/telegram-mcp-server/telegram/api.py @@ -0,0 +1,47 @@ +""" +API client for interacting with the Telegram Bridge API. +""" + +import requests +import json +from typing import Tuple + +from . import TELEGRAM_API_BASE_URL + +def send_message(recipient: str, message: str) -> Tuple[bool, str]: + """Send a Telegram message to the specified recipient. + + Args: + recipient: The recipient - either a username (with or without @), + or a chat ID as a string or integer + message: The message text to send + + Returns: + Tuple[bool, str]: A tuple containing success status and a status message + """ + try: + # Validate input + if not recipient: + return False, "Recipient must be provided" + + url = f"{TELEGRAM_API_BASE_URL}/send" + payload = { + "recipient": recipient, + "message": message + } + + response = requests.post(url, json=payload) + + # Check if the request was successful + if response.status_code == 200: + result = response.json() + return result.get("success", False), result.get("message", "Unknown response") + else: + return False, f"Error: HTTP {response.status_code} - {response.text}" + + except requests.RequestException as e: + return False, f"Request error: {str(e)}" + except json.JSONDecodeError: + return False, f"Error parsing response: Unknown" + except Exception as e: + return False, f"Unexpected error: {str(e)}" \ No newline at end of file diff --git a/telegram-mcp-server/telegram.py b/telegram-mcp-server/telegram/database.py similarity index 69% rename from telegram-mcp-server/telegram.py rename to telegram-mcp-server/telegram/database.py index fe1c649..2a1e6bb 100644 --- a/telegram-mcp-server/telegram.py +++ b/telegram-mcp-server/telegram/database.py @@ -1,150 +1,13 @@ """ -This module implements data models and functions for retrieving and sending -Telegram messages, managing chats, and working with contacts. - -The module connects to a SQLite database that stores all Telegram messages and chat data, -which is maintained by the Telegram Bridge. It also provides an HTTP client for sending -messages via the Bridge's API endpoint. - -Main features: -- Data models for messages, chats, contacts, and message context -- Database access functions for retrieving messages, chats, and contacts -- HTTP client for sending messages through the Telegram Bridge -- Helper functions for displaying formatted messages and chats - -All database operations use parameterized queries to prevent SQL injection. +Database operations for retrieving and managing Telegram data. """ import sqlite3 -import requests -import json -import os.path from datetime import datetime -from dataclasses import dataclass -from typing import Optional, List, Tuple, Dict, Any +from typing import Optional, List, Tuple -# Database path -MESSAGES_DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'telegram-bridge', 'store', 'messages.db') -TELEGRAM_API_BASE_URL = "http://localhost:8081/api" - -@dataclass -class Message: - """ - Represents a Telegram message with all its metadata. - - Attributes: - id: Unique message identifier - chat_id: ID of the chat the message belongs to - chat_title: Title of the chat (user name, group name, etc.) - sender_name: Name of the message sender - content: Text content of the message - timestamp: Date and time when the message was sent - is_from_me: Boolean indicating if the message was sent by the user - sender_id: ID of the message sender - """ - id: int - chat_id: int - chat_title: str - sender_name: str - content: str - timestamp: datetime - is_from_me: bool - sender_id: int - -@dataclass -class Chat: - """ - Represents a Telegram chat (direct message, group, channel, etc.). - - Attributes: - id: Unique chat identifier - title: Name of the chat (user name, group name, etc.) - username: Optional Telegram username (without @) - type: Type of chat ('user', 'group', 'channel', 'supergroup') - last_message_time: Timestamp of the most recent message in the chat - """ - id: int - title: str - username: Optional[str] - type: str - last_message_time: Optional[datetime] - -@dataclass -class Contact: - """ - Represents a Telegram contact. - - Attributes: - id: Unique contact identifier - username: Optional Telegram username (without @) - name: Display name of the contact - """ - id: int - username: Optional[str] - name: str - -@dataclass -class MessageContext: - """ - Provides context around a specific message. - - Attributes: - message: The target message - before: List of messages that came before the target message - after: List of messages that came after the target message - """ - message: Message - before: List[Message] - after: List[Message] - -def print_message(message: Message, show_chat_info: bool = True) -> None: - """Print a single message with consistent formatting.""" - direction = "→" if message.is_from_me else "←" - - if show_chat_info: - print(f"[{message.timestamp:%Y-%m-%d %H:%M:%S}] {direction} Chat: {message.chat_title} (ID: {message.chat_id})") - else: - print(f"[{message.timestamp:%Y-%m-%d %H:%M:%S}] {direction}") - - print(f"From: {'Me' if message.is_from_me else message.sender_name}") - print(f"Message: {message.content}") - print("-" * 100) - -def print_messages_list(messages: List[Message], title: str = "", show_chat_info: bool = True) -> None: - """Print a list of messages with a title and consistent formatting.""" - if not messages: - print("No messages to display.") - return - - if title: - print(f"\n{title}") - print("-" * 100) - - for message in messages: - print_message(message, show_chat_info) - -def print_chat(chat: Chat) -> None: - """Print a single chat with consistent formatting.""" - print(f"Chat: {chat.title} (ID: {chat.id})") - print(f"Type: {chat.type}") - if chat.username: - print(f"Username: @{chat.username}") - if chat.last_message_time: - print(f"Last active: {chat.last_message_time:%Y-%m-%d %H:%M:%S}") - print("-" * 100) - -def print_chats_list(chats: List[Chat], title: str = "") -> None: - """Print a list of chats with a title and consistent formatting.""" - if not chats: - print("No chats to display.") - return - - if title: - print(f"\n{title}") - print("-" * 100) - - for chat in chats: - print_chat(chat) +from . import MESSAGES_DB_PATH +from .models import Message, Chat, Contact, MessageContext def search_contacts(query: str) -> List[Contact]: """Search contacts by name or username.""" @@ -579,42 +442,4 @@ def get_last_interaction(contact_id: int) -> Optional[Message]: return None finally: if 'conn' in locals(): - conn.close() - -def send_message(recipient: str, message: str) -> Tuple[bool, str]: - """Send a Telegram message to the specified recipient. - - Args: - recipient: The recipient - either a username (with or without @), - or a chat ID as a string or integer - message: The message text to send - - Returns: - Tuple[bool, str]: A tuple containing success status and a status message - """ - try: - # Validate input - if not recipient: - return False, "Recipient must be provided" - - url = f"{TELEGRAM_API_BASE_URL}/send" - payload = { - "recipient": recipient, - "message": message - } - - response = requests.post(url, json=payload) - - # Check if the request was successful - if response.status_code == 200: - result = response.json() - return result.get("success", False), result.get("message", "Unknown response") - else: - return False, f"Error: HTTP {response.status_code} - {response.text}" - - except requests.RequestException as e: - return False, f"Request error: {str(e)}" - except json.JSONDecodeError: - return False, f"Error parsing response: Unknown" - except Exception as e: - return False, f"Unexpected error: {str(e)}" \ No newline at end of file + conn.close() \ No newline at end of file diff --git a/telegram-mcp-server/telegram/models.py b/telegram-mcp-server/telegram/models.py new file mode 100644 index 0000000..ddee0e6 --- /dev/null +++ b/telegram-mcp-server/telegram/models.py @@ -0,0 +1,77 @@ +""" +Data models for Telegram entities. +""" + +from dataclasses import dataclass +from datetime import datetime +from typing import Optional, List + +@dataclass +class Message: + """ + Represents a Telegram message with all its metadata. + + Attributes: + id: Unique message identifier + chat_id: ID of the chat the message belongs to + chat_title: Title of the chat (user name, group name, etc.) + sender_name: Name of the message sender + content: Text content of the message + timestamp: Date and time when the message was sent + is_from_me: Boolean indicating if the message was sent by the user + sender_id: ID of the message sender + """ + id: int + chat_id: int + chat_title: str + sender_name: str + content: str + timestamp: datetime + is_from_me: bool + sender_id: int + +@dataclass +class Chat: + """ + Represents a Telegram chat (direct message, group, channel, etc.). + + Attributes: + id: Unique chat identifier + title: Name of the chat (user name, group name, etc.) + username: Optional Telegram username (without @) + type: Type of chat ('user', 'group', 'channel', 'supergroup') + last_message_time: Timestamp of the most recent message in the chat + """ + id: int + title: str + username: Optional[str] + type: str + last_message_time: Optional[datetime] + +@dataclass +class Contact: + """ + Represents a Telegram contact. + + Attributes: + id: Unique contact identifier + username: Optional Telegram username (without @) + name: Display name of the contact + """ + id: int + username: Optional[str] + name: str + +@dataclass +class MessageContext: + """ + Provides context around a specific message. + + Attributes: + message: The target message + before: List of messages that came before the target message + after: List of messages that came after the target message + """ + message: Message + before: List[Message] + after: List[Message] \ No newline at end of file