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