diff --git a/mcp/nntp/nntp_mcp.log b/mcp/nntp/nntp_mcp.log new file mode 100644 index 0000000..e69de29 diff --git a/mcp/nntp/nntp_mcp.py b/mcp/nntp/nntp_mcp.py index c3c2956..78cec26 100644 --- a/mcp/nntp/nntp_mcp.py +++ b/mcp/nntp/nntp_mcp.py @@ -427,6 +427,72 @@ def get_latest_messages(group_name: str, count: int = 10) -> list[dict]: 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 любого сообщения из ветки (например, '') + :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__": diff --git a/mcp/nntp/nntp_mcp_state.json b/mcp/nntp/nntp_mcp_state.json new file mode 100644 index 0000000..ee8f042 --- /dev/null +++ b/mcp/nntp/nntp_mcp_state.json @@ -0,0 +1,8 @@ +{ + "news.skysamara.ru/sky.test": { + "last_seen": null, + "my_ids": [ + "<179302383962.14856.2029190571264960217@test.com>" + ] + } +} \ No newline at end of file diff --git a/mcp/nntp/nntp_replay.py b/mcp/nntp/nntp_replay.py new file mode 100644 index 0000000..0f80262 --- /dev/null +++ b/mcp/nntp/nntp_replay.py @@ -0,0 +1,412 @@ +""" +NNTP MCP-сервер: мониторинг новостных групп и ответы в тредах. +""" + +import warnings +warnings.filterwarnings("ignore") + +import sys +import os + +# stderr отправляем в лог-файл, чтобы предупреждения и трейсбэки +# не ломали stdio-транспорт, но ошибки оставались диагностируемыми +sys.stderr = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "nntp_mcp.log"), "a") + +import json +import re +import datetime +import nntplib +import email +from email.message import EmailMessage +from email.header import decode_header +from email.utils import parseaddr, formataddr, make_msgid, formatdate, parsedate_to_datetime +from email.policy import default as email_policy + +# Автоматическое переключение импорта в зависимости от окружения +try: + from mcp.server.fastmcp import FastMCP + mcp = FastMCP("nntp-monitor") + # Используем прокси-декоратор для единообразия регистрации ниже + def register_mcp_tool(func): + return mcp.tool()(func) +except ImportError: + from mcp.server.mcpserver import MCPServer + mcp = MCPServer("nntp-monitor") + def register_mcp_tool(func): + if hasattr(mcp, 'list_tools') and hasattr(mcp.list_tools, 'add_tool'): + mcp.list_tools.add_tool(func) + elif hasattr(mcp, 'register_tool'): + mcp.register_tool(func) + else: + # ИСПРАВЛЕНО: Правильный вызов декоратора со скобками для базового MCPServer + decorator = mcp.tool() + decorator(func) + return func + +STATE_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "nntp_mcp_state.json") +DEFAULT_WINDOW = 1000 +MAX_THREAD_MSGS = 40 + + +# ---------------------------------------------------------------- состояние + +def _state_key(server: str, group: str) -> str: + return f"{server.lower()}/{group.lower()}" + + +def _load_state() -> dict: + try: + with open(STATE_FILE, encoding="utf-8") as f: + return json.load(f) + except Exception: + return {} + + +def _save_state(state: dict) -> None: + tmp = STATE_FILE + ".tmp" + with open(tmp, "w", encoding="utf-8") as f: + json.dump(state, f, ensure_ascii=False, indent=1) + os.replace(tmp, STATE_FILE) + + +# ---------------------------------------------------------------- утилиты + +def _decode(header_value) -> str: + if not header_value: + return "" + try: + parts = [] + for text, enc in decode_header(str(header_value)): + if isinstance(text, bytes): + parts.append(text.decode(enc or "utf-8", errors="ignore")) + else: + parts.append(str(text)) + return "".join(parts) + except Exception: + return str(header_value) + + +def _parse_date(date_str: str) -> str: + if not date_str: + return "" + try: + dt = parsedate_to_datetime(date_str) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=datetime.timezone.utc) + return dt.isoformat() + except Exception: + return str(date_str) + + +def _clean_quotes(body_text: str) -> str: + if not body_text: + return "" + intro = re.compile( + r"on\s+.*?\s+wrote\s*:|в\s+письме\s+от.*?:|писал\s*\(а\)\s*:|" + r"---.*?original\s+message.*?---", + re.IGNORECASE, + ) + clean, prev_blank = [], False + for line in body_text.splitlines(): + stripped = line.strip() + if stripped.startswith(">"): + continue + if intro.search(stripped): + continue + if not stripped and prev_blank: + continue + prev_blank = not stripped + clean.append(line) + return "\n".join(clean).strip() + + +def _extract_body(msg) -> str: + body = "" + for part in (msg.walk() if msg.is_multipart() else [msg]): + if str(part.get("Content-Disposition", "")).lower().startswith("attachment"): + continue + if part.get_content_type() != "text/plain": + continue + payload = part.get_payload(decode=True) + if payload: + body += payload.decode(part.get_content_charset() or "utf-8", errors="ignore") + if not body.strip(): + for part in (msg.walk() if msg.is_multipart() else [msg]): + if part.get_content_type() == "text/html": + payload = part.get_payload(decode=True) + if payload: + body = "[HTML]\n" + payload.decode( + part.get_content_charset() or "utf-8", errors="ignore") + break + return body or "[Тело сообщения пустое]" + + +def _connect(server: str, port: int, user: str, password: str): + if port == 563: + return nntplib.NNTP_SSL(server, port=port, timeout=30, + user=user or None, password=password or None) + return nntplib.NNTP(server, port=port, timeout=30, + user=user or None, password=password or None) + + +def _fetch_index(s, last: int, window: int) -> list[dict]: + start = max(1, last - window + 1) + try: + _, overviews = s.over((str(start), str(last))) + except Exception: + return [] + index = [] + for num, over in overviews: + msgid = (over.get("message-id") or "").strip() or f"" + index.append({ + "num": int(num), + "msgid": msgid, + "subject": _decode(over.get("subject", "")), + "from": _decode(over.get("from", "")), + "date": _parse_date(over.get("date", "")), + "refs": [r for r in (over.get("references") or "").split() + if r.startswith("<")], + }) + index.sort(key=lambda m: m["num"]) + return index + + +def _make_root_resolver(index_by_id: dict): + resolved = {} + + def root_of(msg: dict) -> str: + chain, cur = [], msg + while True: + key = cur["msgid"] + if key in resolved: + r = resolved[key] + break + if key in chain: + r = key + break + chain.append(key) + parent_id = cur["refs"][0] if cur["refs"] else None + parent = index_by_id.get(parent_id) if parent_id else None + if parent is None: + r = parent_id or key + break + cur = parent + for k in chain: + resolved[k] = r + return r + + return root_of + + +def _thread_members(index: list[dict], target: dict) -> list[dict]: + by_id = {m["msgid"]: m for m in index} + root_of = _make_root_resolver(by_id) + root = root_of(target) + members = [m for m in index if root_of(m) == root] + return members + + +def _fetch_bodies(s, members: list[dict]) -> list[dict]: + out = [] + for m in members[-MAX_THREAD_MSGS:]: + entry = dict(m) + try: + _, info = s.article(str(m["num"])) + msg = email.message_from_bytes(b"\n".join(info.lines), policy=email_policy) + entry["body"] = _clean_quotes(_extract_body(msg)) + except Exception as e: + entry["body"] = f"[Не удалось прочитать: {e}]" + out.append(entry) + return out + + +# ---------------------------------------------------------------- инструменты + +def scan_group(server: str, group: str, port: int = 119, + user: str = "", password: str = "", + window: int = DEFAULT_WINDOW) -> dict: + """ + НАЙТИ НОВОЕ: сравнивает группу с отметкой прошлого запуска, собирает + затронутые треды целиком и возвращает их с контекстом. + """ + state = _load_state() + key = _state_key(server, group) + st = state.setdefault(key, {"last_seen": None, "my_ids": []}) + my_ids = set(st.get("my_ids", [])) + first_run = st["last_seen"] is None + + try: + with _connect(server, port, user, password) as s: + _, count, first, last, _ = s.group(group) + first, last = int(first), int(last) + + if first_run: + st["last_seen"] = max(first, last - 50) + since = st["last_seen"] + + if last <= since: + _save_state(state) + return {"status": "no_new_messages", "group": group, + "last_seen": since, "last_num": last, "threads": []} + + index = _fetch_index(s, last, window) + new_msgs = [m for m in index if m["num"] > since] + + if not new_msgs: + st["last_seen"] = last + _save_state(state) + return {"status": "no_new_messages_in_window", "group": group, + "last_num": last, "threads": []} + + threads = {} + for m in new_msgs: + members = _thread_members(index, m) + members_sorted = sorted(members, key=lambda x: x["num"]) + root = members_sorted[0]["msgid"] + t = threads.setdefault(root, {"members": {}, "new": set()}) + for mem in members_sorted: + t["members"][mem["msgid"]] = mem + t["new"].add(m["msgid"]) + + result_threads = [] + for root, t in list(threads.items())[:10]: + members = sorted(t["members"].values(), key=lambda x: x["num"]) + full = _fetch_bodies(s, members) + refs_all = {r for m in members for r in m["refs"]} + msg_ids = {m["msgid"] for m in members} + reply_to_me = bool(refs_all & my_ids) or bool(msg_ids & my_ids) + last_is_mine = members[-1]["msgid"] in my_ids + result_threads.append({ + "root_msgid": root, + "subject": full[0]["subject"] or full[-1]["subject"], + "reply_to_me": reply_to_me, + "last_is_mine": last_is_mine, + "needs_reply": bool(t["new"] - msg_ids & my_ids) and not last_is_mine, + "new_msg_nums": sorted( + m["num"] for m in members if m["msgid"] in t["new"]), + "messages": [ + {"num": m["num"], "msgid": m["msgid"], + "from": m["from"], "date": m["date"], "body": m["body"]} + for m in full + ], + }) + + st["last_seen"] = last + _save_state(state) + return {"status": "ok", "group": group, "first_run": first_run, + "last_num": last, "threads": result_threads} + + except Exception as e: + return {"status": "error", "group": group, "error": str(e)} + + +def read_thread(server: str, group: str, msg_id: str, port: int = 119, + user: str = "", password: str = "", + window: int = DEFAULT_WINDOW) -> dict: + """ + ПРОЧИТАТЬ ВЕТКУ ЦЕЛИКОМ: msg_id — Message-ID (<...>) или номер статьи. + Возвращает все сообщения треда в хронологическом порядке с телами. + """ + try: + with _connect(server, port, user, password) as s: + _, count, first, last, _ = s.group(group) + last_num = int(last) + + try: + _, info = s.article(msg_id) + probe = email.message_from_bytes(b"\n".join(info.lines), policy=email_policy) + target_id = probe.get("Message-ID", msg_id) + except Exception as e: + return {"error": f"Сообщение {msg_id} не найдено: {e}"} + + index = _fetch_index(s, last_num, window) + by_id = {m["msgid"]: m for m in index} + target = by_id.get(target_id) + if target is None: + return {"error": f"{target_id} вне окна сканирования, увеличьте window"} + + members = _thread_members(index, target) + full = _fetch_bodies(s, members) + return {"thread_root": full[0]["msgid"], + "subject": full[0]["subject"], + "messages": [{"num": m["num"], "msgid": m["msgid"], + "from": m["from"], "date": m["date"], + "body": m["body"]} for m in full]} + except Exception as e: + return {"error": str(e)} + + +def post_reply(server: str, group: str, reply_to_msg_id: str, body: str, + sender_name: str, sender_email: str, + port: int = 119, user: str = "", password: str = "") -> dict: + """ + ОТВЕТИТЬ: постит ответ на сообщение reply_to_msg_id в группу. + Автоматически проставляет Re: в Subject, In-Reply-To и References. + """ + if not re.match(r"[^@\s]+@[^@\s]+", sender_email): + return {"error": "Некорректный sender_email"} + try: + with _connect(server, port, user, password) as s: + s.group(group) + try: + _, info = s.article(reply_to_msg_id) + except Exception as e: + return {"error": f"Не найдено сообщение {reply_to_msg_id}: {e}"} + + parent = email.message_from_bytes(b"\n".join(info.lines), policy=email_policy) + parent_id = (parent.get("Message-ID") or reply_to_msg_id).strip() + parent_subject = _decode(parent.get("Subject", "")) + subject = parent_subject if re.match(r"(?i)^\s*(re|aw)\s*:", parent_subject) \ + else f"Re: {parent_subject}" + + refs = [r for r in (parent.get("References") or "").split() + if r.startswith("<")] + refs.append(parent_id) + + msg = EmailMessage() + msg["From"] = formataddr((sender_name, sender_email)) + msg["Newsgroups"] = group + msg["Subject"] = subject + msg["Date"] = formatdate(localtime=True) + msg["Message-ID"] = make_msgid(domain=sender_email.split("@")[-1]) + msg["In-Reply-To"] = parent_id + msg["References"] = " ".join(refs) + msg.set_content(body, charset="utf-8") + + data = msg.as_bytes(policy=email_policy) + resp = s.post(data.split(b"\n")) + + state = _load_state() + st = state.setdefault(_state_key(server, group), + {"last_seen": None, "my_ids": []}) + st.setdefault("my_ids", []).append(msg["Message-ID"]) + st["my_ids"] = st["my_ids"][-200:] + _save_state(state) + + return {"status": "posted", "message_id": msg["Message-ID"], + "subject": subject, "server_response": str(resp)} + except Exception as e: + return {"error": f"Ошибка отправки: {e}"} + + +def list_groups(server: str, port: int = 119, + user: str = "", password: str = "") -> list: + """Диагностика: список групп на сервере.""" + try: + with _connect(server, port, user, password) as s: + _, groups = s.list() + return [g for g in groups if g] + except Exception as e: + return [f"Ошибка: {e}"] + + +# Безопасная централизованная регистрация инструментов через прокси-функцию +scan_group = register_mcp_tool(scan_group) +read_thread = register_mcp_tool(read_thread) +post_reply = register_mcp_tool(post_reply) +list_groups = register_mcp_tool(list_groups) + + +if __name__ == "__main__": + mcp.run(transport="stdio") +