diff --git a/mcp/nntp/nntp_mcp.log b/mcp/nntp/nntp_mcp.log index e69de29..76e579a 100644 --- a/mcp/nntp/nntp_mcp.log +++ b/mcp/nntp/nntp_mcp.log @@ -0,0 +1,2 @@ +test + diff --git a/mcp/nntp/nntp_mcp_state.json b/mcp/nntp/nntp_mcp_state.json index ee8f042..27a9bef 100644 --- a/mcp/nntp/nntp_mcp_state.json +++ b/mcp/nntp/nntp_mcp_state.json @@ -1,8 +1,9 @@ -{ - "news.skysamara.ru/sky.test": { - "last_seen": null, - "my_ids": [ - "<179302383962.14856.2029190571264960217@test.com>" - ] - } +{ + "news.skysamara.ru/sky.test": { + "last_seen": null, + "my_ids": [ + "<179302383962.14856.2029190571264960217@test.com>", + "<179302423690.14856.15120990572980229485@test.com>" + ] + } } \ No newline at end of file diff --git a/mcp/nntp/nntp_replay — копия.txt b/mcp/nntp/nntp_replay — копия.txt new file mode 100644 index 0000000..0131ac9 --- /dev/null +++ b/mcp/nntp/nntp_replay — копия.txt @@ -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") + diff --git a/mcp/nntp/nntp_replay.py b/mcp/nntp/nntp_replay.py index 0f80262..073b8c3 100644 --- a/mcp/nntp/nntp_replay.py +++ b/mcp/nntp/nntp_replay.py @@ -1,412 +1,411 @@ -""" -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") - +""" +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: + decorator = mcp.tool() + decorator(func) + return func + +STATE_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "nntp_mcp_state.json") +DEFAULT_SERVER = "news.skysamara.run" +DEFAULT_PORT = 119 +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): + if port == 563: + return nntplib.NNTP_SSL(server, port=port, timeout=30) + return nntplib.NNTP(server, port=port, timeout=30) + + +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 = DEFAULT_SERVER, group: str = "", + port: int = DEFAULT_PORT, + 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) 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 = DEFAULT_SERVER, group: str = "", + msg_id: str = "", port: int = DEFAULT_PORT, + window: int = DEFAULT_WINDOW) -> dict: + """ + ПРОЧИТАТЬ ВЕТКУ ЦЕЛИКОМ: msg_id — Message-ID (<...>) или номер статьи. + Возвращает все сообщения треда в хронологическом порядке с телами. + """ + try: + with _connect(server, port) 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 = DEFAULT_SERVER, group: str = "", + reply_to_msg_id: str = "", body: str = "", + sender_name: str = "", sender_email: str = "", + port: int = DEFAULT_PORT) -> 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) 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 = DEFAULT_SERVER, + port: int = DEFAULT_PORT) -> list: + """Диагностика: список групп на сервере.""" + try: + with _connect(server, port) 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") \ No newline at end of file