bionic/mcp/nntp/nntp_mcp.py

import warnings
# Полностью отключаем вывод любых предупреждений в консоль
warnings.filterwarnings("ignore", category=DeprecationWarning)
warnings.filterwarnings("ignore")

import sys
import os
# Перенаправляем поток ошибок в пустоту, чтобы предупреждения не перехватывали stdio
sys.stderr = open(os.devnull, 'w')


import nntplib
import email
from email.policy import default
import re
import datetime
from email.header import decode_header 
from email.utils import parsedate_to_datetime
from mcp.server.mcpserver import MCPServer

# Инициализируем сервер
mcp = MCPServer("Bionic-NNTP-Search")

NNTP_SERVER = "news.skysamara.ru"
PORT = 119

def parse_nntp_date(date_str: str) -> datetime.datetime:
    """
    Конвертирует строку даты из заголовка NNTP в объект datetime с поддержкой UTC таймзоны.
    """
    if not date_str:
        return datetime.datetime.min.replace(tzinfo=datetime.timezone.utc)
    try:
        dt = parsedate_to_datetime(date_str)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=datetime.timezone.utc)
        return dt
    except Exception:
        return datetime.datetime.min.replace(tzinfo=datetime.timezone.utc)

def clean_body_from_quotes(body_text: str) -> str:
    """
    Принимает сырой текст письма и удаляет из него все строки цитирования (>),
    а также вводные маркеры цитат.
    """
    if not body_text:
        return ""
        
    lines = body_text.splitlines()
    clean_lines = []
    
    intro_patterns = [
        r"on\s+.*?\s+wrote\s*:", 
        r"в\s+письме\s+от\s+.*?\s+вы\s+писали\s*:",
        r"писал\s*\(а\)\s*:",
        r"---.*?original\s+message.*?---"
    ]
    intro_regex = re.compile("|".join(intro_patterns), re.IGNORECASE)

    for line in lines:
        stripped_line = line.strip()
        
        if stripped_line.startswith(">"):
            continue
            
        if intro_regex.search(stripped_line):
            continue
            
        if not stripped_line:
            if clean_lines and clean_lines[-1] == "":
                continue
        
        clean_lines.append(line)

    return "\n".join(clean_lines).strip()

def decode_mime_header(header_value: str) -> str:
    """Декодирует строку заголовка формата =?UTF-8?B?...?="""
    if not header_value:
        return ""
    try:
        decoded_fragments = decode_header(header_value)
        parts = []
        for text, encoding in decoded_fragments:
            if isinstance(text, bytes):
                enc = encoding if encoding else 'utf-8'
                parts.append(text.decode(enc, errors='ignore'))
            else:
                parts.append(str(text))
        return "".join(parts)
    except Exception:
        return header_value


@mcp.tool()
def get_skysamara_groups() -> list[str]:
    """
    Получить список доступных групп на сервере news.skysamara.ru.
    """
    try:
        groups = []
        with nntplib.NNTP(NNTP_SERVER, port=PORT, timeout=10) as s:
            resp, group_list = s.list()
            for group_info in group_list:
                if group_info and len(group_info) > 0:
                    groups.append(group_info[0])
        return groups
    except Exception as e:
        return [f"Ошибка NNTP при получении групп: {e}"]


@mcp.tool()
def search_user_by_headers(user_email: str, group_name: str, depth: int = 1000) -> list[dict]:
    """
    Ищет сообщения конкретного пользователя по автору (From) в указанной группе.
    """
    try:
        found_messages = []
        with nntplib.NNTP(NNTP_SERVER, port=PORT, timeout=10) as s:
            resp, count, first, last, name = s.group(group_name)
            
            first_num = int(first)
            last_num = int(last)
            
            start_msg = max(first_num, last_num - depth + 1)
            if start_msg > last_num:
                return []

            resp, overviews = s.over((str(start_msg), str(last_num)))
            
            for msg_id, over in overviews:
                raw_author = over.get('from', '')
                raw_subject = over.get('subject', '')
                
                author = decode_mime_header(raw_author)
                subject = decode_mime_header(raw_subject)
                
                if user_email.lower() in author.lower():
                    found_messages.append({
                        "msg_num": str(msg_id),
                        "subject": subject,
                        "author": author
                    })
                    
        return found_messages
    except Exception as e:
        return [{"error": f"Ошибка NNTP при поиске: {e}"}]


@mcp.tool()
def fetch_messages_by_date_range(
    group_name: str, 
    start_date_iso: str, 
    end_date_iso: str, 
    max_count: int = 100
) -> list[dict]:
    """
    Извлекает сообщения из указанной группы за заданный временной интервал (YYYY-MM-DD).
    """
    try:
        start_dt = datetime.datetime.fromisoformat(start_date_iso).replace(tzinfo=datetime.timezone.utc)
        end_dt = datetime.datetime.fromisoformat(end_date_iso).replace(tzinfo=datetime.timezone.utc)
        extracted_messages = []
        
        with nntplib.NNTP(NNTP_SERVER, port=PORT, timeout=15) as s:
            resp, count, first, last, name = s.group(group_name)
            first_num = int(first)
            last_num = int(last)
            
            if first_num > last_num:
                return []
                
            resp, overviews = s.over((str(first_num), str(last_num)))
            
            for msg_id, over in reversed(overviews):
                if len(extracted_messages) >= max_count:
                    break
                    
                raw_date = over.get('date', '')
                msg_dt = parse_nntp_date(raw_date)
                
                if msg_dt > end_dt:
                    continue
                if msg_dt < start_dt:
                    break
                    
                try:
                    _, info = s.article(str(msg_id))
                    raw_bytes = b"\n".join(info.lines)
                    msg = email.message_from_bytes(raw_bytes, policy=default)
                    
                    body_text = ""
                    if msg.is_multipart():
                        for part in msg.walk():
                            if part.get_content_type() == "text/plain":
                                payload = part.get_payload(decode=True)
                                if payload:
                                    body_text += payload.decode(part.get_content_charset() or 'utf-8', errors='ignore')
                    else:
                        payload = msg.get_payload(decode=True)
                        if payload:
                            body_text = payload.decode(msg.get_content_charset() or 'utf-8', errors='ignore')
                except Exception:
                    continue
                
                extracted_messages.append({
                    "msg_num": str(msg_id),
                    "date": msg_dt.isoformat(),
                    "from": decode_mime_header(over.get('from', '[Неизвестно]')),
                    "subject": decode_mime_header(over.get('subject', '[Без темы]')),
                    "body": clean_body_from_quotes(body_text)
                })
                
        return extracted_messages
    except Exception as e:
        return [{"error": f"Ошибка NNTP при фильтрации по датам: {e}"}]


@mcp.tool()
def fetch_message_by_id(msg_num_or_id: str, group_name: str = None) -> dict:
    """
    Получает и полностью декодирует содержимое конкретного сообщения.
    """
    try:
        with nntplib.NNTP(NNTP_SERVER, port=PORT, timeout=10) as s:
            is_message_id = str(msg_num_or_id).startswith("<") and str(msg_num_or_id).endswith(">")
            
            if not is_message_id:
                if not group_name:
                    return {"error": "Для чтения сообщения по его номеру необходимо указать group_name"}
                s.group(group_name)
            
            resp, info = s.article(msg_num_or_id)
            raw_bytes = b"\n".join(info.lines)
            msg = email.message_from_bytes(raw_bytes, policy=default)
            
            subject = decode_mime_header(msg.get("Subject", "[Без темы]"))
            author = decode_mime_header(msg.get("From", "[Неизвестно]"))
            date = msg.get("Date", "")
            
            body_text = ""
            if msg.is_multipart():
                for part in msg.walk():
                    content_type = part.get_content_type()
                    content_disposition = str(part.get("Content-Disposition"))
                    
                    if content_type == "text/plain" and "attachment" not in content_disposition:
                        payload = part.get_payload(decode=True)
                        if payload:
                            body_text += payload.decode(part.get_content_charset() or 'utf-8', errors='ignore')
            else:
                payload = msg.get_payload(decode=True)
                if payload:
                    body_text = payload.decode(msg.get_content_charset() or 'utf-8', errors='ignore')
            
            if not body_text.strip():
                for part in msg.walk():
                    if part.get_content_type() == "text/html":
                        payload = part.get_payload(decode=True)
                        if payload:
                            body_text = "[HTML Content]\n" + payload.decode(part.get_content_charset() or 'utf-8', errors='ignore')
                            break

            if not body_text.strip():
                body_text = "[Тело сообщения пустое или не содержит читаемого текста]"

            return {
                "msg_id": msg_num_or_id,
                "from": author,
                "subject": subject,
                "date": date,
                "body": body_text
            }
    except Exception as e:
        return {"error": f"Не удалось прочитать сообщение {msg_num_or_id}: {e}"}


@mcp.tool()
def get_user_pure_messages(group_name: str, user_email: str, depth: int = 50) -> list[dict]:
    """
    Находит последние сообщения конкретного пользователя в группе и извлекает только его текст.
    """
    pure_messages = []
    user_email_lower = user_email.lower()

    try:
        with nntplib.NNTP(NNTP_SERVER, port=PORT, timeout=15) as s:
            resp, count, first, last, name = s.group(group_name)
            first_num = int(first)
            last_num = int(last)
            
            start_msg = max(first_num, last_num - depth + 1)
            if start_msg > last_num:
                return []

            resp, overviews = s.over((str(start_msg), str(last_num)))

            for msg_num, over in overviews:
                author = decode_mime_header(over.get('from', ''))

                if user_email_lower not in author.lower():
                    continue

                try:
                    _, info = s.article(str(msg_num))
                    raw_bytes = b"\n".join(info.lines)
                    msg = email.message_from_bytes(raw_bytes, policy=default)
                    subject = decode_mime_header(msg.get("Subject", "[Без темы]"))

                    body_text = ""
                    if msg.is_multipart():
                        for part in msg.walk():
                            if part.get_content_type() == "text/plain":
                                payload = part.get_payload(decode=True)
                                if payload:
                                    body_text += payload.decode(part.get_content_charset() or 'utf-8', errors='ignore')
                    else:
                        payload = msg.get_payload(decode=True)
                        if payload:
                            body_text = payload.decode(msg.get_content_charset() or 'utf-8', errors='ignore')
                except Exception:
                    continue

                pure_text = clean_body_from_quotes(body_text)
                if pure_text:
                    pure_messages.append({
                        "msg_num": str(msg_num),
                        "subject": subject,
                        "author": author,
                        "pure_text": pure_text
                    })

        return pure_messages
    except Exception as e:
        return [{"error": f"Ошибка при извлечении сообщений пользователя: {e}"}]

@mcp.tool()
def save_text_to_file(filename: str, content: str) -> str:
    """
    Save text content into a local file. Supports absolute or relative paths.
    
    :param filename: Path or name of the file (e.g., 'C:/Users/ysv/Documents/bionic-project/chat-kraft/ghjtrnf/test.txt')
    :param content: Text to write into the file
    """
    try:
        import os
        
        # Если передан просто test.txt, сохраняем в папку скрипта
        # Если передан путь (абсолютный или относительный), работаем с ним
        target_path = os.path.abspath(filename)
        
        # Автоматически создаем все промежуточные папки (например, ghjtrnf), если их еще нет на диске
        target_dir = os.path.dirname(target_path)
        if target_dir and not os.path.exists(target_dir):
            os.makedirs(target_dir, exist_ok=True)
            
        with open(target_path, "w", encoding="utf-8") as f:
            f.write(content)
            
        return f"SUCCESS: File saved at {target_path}"
    except Exception as e:
        return f"ERROR: {str(e)}"


@mcp.tool()
def get_latest_messages(group_name: str, count: int = 10) -> list[dict]:
    """
    Получить заданное количество последних сообщений из указанной группы.
    
    Возвращает список словарей с полями: Msg-Num, Message-ID, Subject, Date, Name, Mail.
    Сообщения строго отсортированы от самых новых к старым.
    """
    from email.utils import parseaddr
    
    try:
        latest_messages = []
        with nntplib.NNTP(NNTP_SERVER, port=PORT, timeout=15) as s:
            # Получаем информацию о группе и номера сообщений
            resp, total_count, first, last, name = s.group(group_name)
            first_num = int(first)
            last_num = int(last)
            
            if first_num > last_num or total_count == 0:
                return []
                
            # Берем диапазон с запасом, чтобы гарантированно набрать count живых сообщений
            # (на случай, если часть постов между first и last были удалены)
            start_msg = max(first_num, last_num - (count * 2) + 1)
            
            # Получаем овервью (заголовки) сообщений в выбранном диапазоне
            resp, overviews = s.over((str(start_msg), str(last_num)))
            
            # СТРОГАЯ СОРТИРОВКА: переводим ключи в int и сортируем по убыванию (от новых к старым)
            sorted_overviews = sorted(overviews, key=lambda x: int(x[0]), reverse=True)
            
            # Обходим сообщения уже в гарантированно правильном порядке
            for msg_num, over in sorted_overviews:
                # Извлекаем message-id из овервью или генерируем запасной
                raw_msg_id = over.get('message-id', f"<{msg_num}@{NNTP_SERVER}>")
                
                raw_author = over.get('from', '')
                raw_subject = over.get('subject', '')
                raw_date = over.get('date', '')
                
                # Декодируем MIME-заголовки с помощью существующих функций проекта
                author = decode_mime_header(raw_author)
                subject = decode_mime_header(raw_subject)
                date = decode_mime_header(raw_date)
                
                # Разделяем From на Имя и Почту
                name_part, mail_part = parseaddr(author)
                
                latest_messages.append({
                    "Msg-Num": str(msg_num),  # Локальный порядковый номер в группе
                    "Message-ID": raw_msg_id, # Глобальный уникальный идентификатор
                    "Subject": subject if subject else "[Без темы]",
                    "Date": date,
                    "Name": name_part if name_part else "[Без имени]",
                    "Mail": mail_part if mail_part else "[Без почты]"
                })
                
                # Прерываем цикл, как только набрали нужное количество
                if len(latest_messages) >= count:
                    break
                    
        return latest_messages
    except Exception as e:
        return [{"error": f"Ошибка NNTP при получении последних сообщений: {e}"}]

@mcp.tool()
def get_thread_root(message_id: str) -> dict:
    """
    Находит Message-ID первого (корневого) сообщения в ветке обсуждения.
    
    :param message_id: Message-ID любого сообщения из ветки (например, '<abc@example.com>')
    :return: Словарь с полями root_message_id, subject, author или error
    """
    try:
        with nntplib.NNTP(NNTP_SERVER, port=PORT, timeout=10) as s:
            # Получаем исходное сообщение
            resp, info = s.article(message_id)
            raw_bytes = b"\n".join(info.lines)
            msg = email.message_from_bytes(raw_bytes, policy=default)
            
            # Проверяем заголовок References - там все Message-ID в цепочке
            references = msg.get("References", "").strip()
            
            if references:
                # References содержит список ID через пробел, первый - корень ветки
                ref_ids = re.findall(r'<[^>]+>', references)
                if ref_ids:
                    root_id = ref_ids[0]
                    
                    # Получаем информацию о корневом сообщении
                    try:
                        resp_root, info_root = s.article(root_id)
                        raw_root = b"\n".join(info_root.lines)
                        root_msg = email.message_from_bytes(raw_root, policy=default)
                        
                        return {
                            "root_message_id": root_id,
                            "subject": decode_mime_header(root_msg.get("Subject", "[Без темы]")),
                            "author": decode_mime_header(root_msg.get("From", "[Неизвестно]")),
                            "date": root_msg.get("Date", "")
                        }
                    except Exception:
                        # Если корневое сообщение недоступно, возвращаем хотя бы его ID
                        return {
                            "root_message_id": root_id,
                            "subject": "[Сообщение недоступно]",
                            "author": "[Неизвестно]",
                            "date": ""
                        }
            
            # Если References нет, проверяем In-Reply-To и поднимаемся рекурсивно
            in_reply_to = msg.get("In-Reply-To", "").strip()
            
            if in_reply_to:
                # Извлекаем Message-ID из In-Reply-To
                match = re.search(r'<[^>]+>', in_reply_to)
                if match:
                    parent_id = match.group(0)
                    # Рекурсивно ищем корень через родительское сообщение
                    return get_thread_root(parent_id)
            
            # Если ни References, ни In-Reply-To нет - это и есть корень
            return {
                "root_message_id": message_id,
                "subject": decode_mime_header(msg.get("Subject", "[Без темы]")),
                "author": decode_mime_header(msg.get("From", "[Неизвестно]")),
                "date": msg.get("Date", "")
            }
            
    except Exception as e:
        return {"error": f"Не удалось найти корень ветки для {message_id}: {e}"}


if __name__ == "__main__":
    # Для современного пакета MCPServer (MCP v2.x) 
    # этот параметр автоматически развернет изолированный stdio-транспорт 
    # без конфликтов с внутренним вводом-выводом Bionic.
    mcp.run(transport="stdio")