This commit is contained in:
skysamara committed 2026-10-27 02:04:32 +03:00
1 parent 367f024cbd
commit 5c9f382a7c
4 files changed
+833 -419

No files matched your search

+2
View File
@@ -0,0 +1,2 @@
test
+8 -7
View File
@@ -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>"
]
}
}
+412
View File
@@ -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"<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")
+411 -412
View File
@@ -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"<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")
"""
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"<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 = 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")