413 lines
16 KiB
Plaintext
413 lines
16 KiB
Plaintext
"""
|
||
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"<local-{num}>"
|
||
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")
|
||
|