372 lines
15 KiB
Python
372 lines
15 KiB
Python
import warnings
|
||
# Полностью отключаем вывод любых предупреждений в консоль
|
||
warnings.filterwarnings("ignore", category=DeprecationWarning)
|
||
warnings.filterwarnings("ignore")
|
||
|
||
import sys
|
||
import os
|
||
# Перенаправляем поток ошибок в пустоту, чтобы предупреждения не перехватывали stdio
|
||
sys.stderr = open(os.devnull, 'w')
|
||
|
||
|
||
import nntplib
|
||
import email
|
||
from email.policy import default
|
||
import re
|
||
import datetime
|
||
from email.header import decode_header
|
||
from email.utils import parsedate_to_datetime
|
||
from mcp.server.mcpserver import MCPServer
|
||
|
||
# Инициализируем сервер
|
||
mcp = MCPServer("Bionic-NNTP-Search")
|
||
|
||
NNTP_SERVER = "news.skysamara.ru"
|
||
PORT = 119
|
||
|
||
def parse_nntp_date(date_str: str) -> datetime.datetime:
|
||
"""
|
||
Конвертирует строку даты из заголовка NNTP в объект datetime с поддержкой UTC таймзоны.
|
||
"""
|
||
if not date_str:
|
||
return datetime.datetime.min.replace(tzinfo=datetime.timezone.utc)
|
||
try:
|
||
dt = parsedate_to_datetime(date_str)
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=datetime.timezone.utc)
|
||
return dt
|
||
except Exception:
|
||
return datetime.datetime.min.replace(tzinfo=datetime.timezone.utc)
|
||
|
||
def clean_body_from_quotes(body_text: str) -> str:
|
||
"""
|
||
Принимает сырой текст письма и удаляет из него все строки цитирования (>),
|
||
а также вводные маркеры цитат.
|
||
"""
|
||
if not body_text:
|
||
return ""
|
||
|
||
lines = body_text.splitlines()
|
||
clean_lines = []
|
||
|
||
intro_patterns = [
|
||
r"on\s+.*?\s+wrote\s*:",
|
||
r"в\s+письме\s+от\s+.*?\s+вы\s+писали\s*:",
|
||
r"писал\s*\(а\)\s*:",
|
||
r"---.*?original\s+message.*?---"
|
||
]
|
||
intro_regex = re.compile("|".join(intro_patterns), re.IGNORECASE)
|
||
|
||
for line in lines:
|
||
stripped_line = line.strip()
|
||
|
||
if stripped_line.startswith(">"):
|
||
continue
|
||
|
||
if intro_regex.search(stripped_line):
|
||
continue
|
||
|
||
if not stripped_line:
|
||
if clean_lines and clean_lines[-1] == "":
|
||
continue
|
||
|
||
clean_lines.append(line)
|
||
|
||
return "\n".join(clean_lines).strip()
|
||
|
||
def decode_mime_header(header_value: str) -> str:
|
||
"""Декодирует строку заголовка формата =?UTF-8?B?...?="""
|
||
if not header_value:
|
||
return ""
|
||
try:
|
||
decoded_fragments = decode_header(header_value)
|
||
parts = []
|
||
for text, encoding in decoded_fragments:
|
||
if isinstance(text, bytes):
|
||
enc = encoding if encoding else 'utf-8'
|
||
parts.append(text.decode(enc, errors='ignore'))
|
||
else:
|
||
parts.append(str(text))
|
||
return "".join(parts)
|
||
except Exception:
|
||
return header_value
|
||
|
||
|
||
@mcp.tool()
|
||
def get_skysamara_groups() -> list[str]:
|
||
"""
|
||
Получить список доступных групп на сервере news.skysamara.ru.
|
||
"""
|
||
try:
|
||
groups = []
|
||
with nntplib.NNTP(NNTP_SERVER, port=PORT, timeout=10) as s:
|
||
resp, group_list = s.list()
|
||
for group_info in group_list:
|
||
if group_info and len(group_info) > 0:
|
||
groups.append(group_info[0])
|
||
return groups
|
||
except Exception as e:
|
||
return [f"Ошибка NNTP при получении групп: {e}"]
|
||
|
||
|
||
@mcp.tool()
|
||
def search_user_by_headers(user_email: str, group_name: str, depth: int = 1000) -> list[dict]:
|
||
"""
|
||
Ищет сообщения конкретного пользователя по автору (From) в указанной группе.
|
||
"""
|
||
try:
|
||
found_messages = []
|
||
with nntplib.NNTP(NNTP_SERVER, port=PORT, timeout=10) as s:
|
||
resp, count, first, last, name = s.group(group_name)
|
||
|
||
first_num = int(first)
|
||
last_num = int(last)
|
||
|
||
start_msg = max(first_num, last_num - depth + 1)
|
||
if start_msg > last_num:
|
||
return []
|
||
|
||
resp, overviews = s.over((str(start_msg), str(last_num)))
|
||
|
||
for msg_id, over in overviews:
|
||
raw_author = over.get('from', '')
|
||
raw_subject = over.get('subject', '')
|
||
|
||
author = decode_mime_header(raw_author)
|
||
subject = decode_mime_header(raw_subject)
|
||
|
||
if user_email.lower() in author.lower():
|
||
found_messages.append({
|
||
"msg_num": str(msg_id),
|
||
"subject": subject,
|
||
"author": author
|
||
})
|
||
|
||
return found_messages
|
||
except Exception as e:
|
||
return [{"error": f"Ошибка NNTP при поиске: {e}"}]
|
||
|
||
|
||
@mcp.tool()
|
||
def fetch_messages_by_date_range(
|
||
group_name: str,
|
||
start_date_iso: str,
|
||
end_date_iso: str,
|
||
max_count: int = 100
|
||
) -> list[dict]:
|
||
"""
|
||
Извлекает сообщения из указанной группы за заданный временной интервал (YYYY-MM-DD).
|
||
"""
|
||
try:
|
||
start_dt = datetime.datetime.fromisoformat(start_date_iso).replace(tzinfo=datetime.timezone.utc)
|
||
end_dt = datetime.datetime.fromisoformat(end_date_iso).replace(tzinfo=datetime.timezone.utc)
|
||
extracted_messages = []
|
||
|
||
with nntplib.NNTP(NNTP_SERVER, port=PORT, timeout=15) as s:
|
||
resp, count, first, last, name = s.group(group_name)
|
||
first_num = int(first)
|
||
last_num = int(last)
|
||
|
||
if first_num > last_num:
|
||
return []
|
||
|
||
resp, overviews = s.over((str(first_num), str(last_num)))
|
||
|
||
for msg_id, over in reversed(overviews):
|
||
if len(extracted_messages) >= max_count:
|
||
break
|
||
|
||
raw_date = over.get('date', '')
|
||
msg_dt = parse_nntp_date(raw_date)
|
||
|
||
if msg_dt > end_dt:
|
||
continue
|
||
if msg_dt < start_dt:
|
||
break
|
||
|
||
try:
|
||
_, info = s.article(str(msg_id))
|
||
raw_bytes = b"\n".join(info.lines)
|
||
msg = email.message_from_bytes(raw_bytes, policy=default)
|
||
|
||
body_text = ""
|
||
if msg.is_multipart():
|
||
for part in msg.walk():
|
||
if part.get_content_type() == "text/plain":
|
||
payload = part.get_payload(decode=True)
|
||
if payload:
|
||
body_text += payload.decode(part.get_content_charset() or 'utf-8', errors='ignore')
|
||
else:
|
||
payload = msg.get_payload(decode=True)
|
||
if payload:
|
||
body_text = payload.decode(msg.get_content_charset() or 'utf-8', errors='ignore')
|
||
except Exception:
|
||
continue
|
||
|
||
extracted_messages.append({
|
||
"msg_num": str(msg_id),
|
||
"date": msg_dt.isoformat(),
|
||
"from": decode_mime_header(over.get('from', '[Неизвестно]')),
|
||
"subject": decode_mime_header(over.get('subject', '[Без темы]')),
|
||
"body": clean_body_from_quotes(body_text)
|
||
})
|
||
|
||
return extracted_messages
|
||
except Exception as e:
|
||
return [{"error": f"Ошибка NNTP при фильтрации по датам: {e}"}]
|
||
|
||
|
||
@mcp.tool()
|
||
def fetch_message_by_id(msg_num_or_id: str, group_name: str = None) -> dict:
|
||
"""
|
||
Получает и полностью декодирует содержимое конкретного сообщения.
|
||
"""
|
||
try:
|
||
with nntplib.NNTP(NNTP_SERVER, port=PORT, timeout=10) as s:
|
||
is_message_id = str(msg_num_or_id).startswith("<") and str(msg_num_or_id).endswith(">")
|
||
|
||
if not is_message_id:
|
||
if not group_name:
|
||
return {"error": "Для чтения сообщения по его номеру необходимо указать group_name"}
|
||
s.group(group_name)
|
||
|
||
resp, info = s.article(msg_num_or_id)
|
||
raw_bytes = b"\n".join(info.lines)
|
||
msg = email.message_from_bytes(raw_bytes, policy=default)
|
||
|
||
subject = decode_mime_header(msg.get("Subject", "[Без темы]"))
|
||
author = decode_mime_header(msg.get("From", "[Неизвестно]"))
|
||
date = msg.get("Date", "")
|
||
|
||
body_text = ""
|
||
if msg.is_multipart():
|
||
for part in msg.walk():
|
||
content_type = part.get_content_type()
|
||
content_disposition = str(part.get("Content-Disposition"))
|
||
|
||
if content_type == "text/plain" and "attachment" not in content_disposition:
|
||
payload = part.get_payload(decode=True)
|
||
if payload:
|
||
body_text += payload.decode(part.get_content_charset() or 'utf-8', errors='ignore')
|
||
else:
|
||
payload = msg.get_payload(decode=True)
|
||
if payload:
|
||
body_text = payload.decode(msg.get_content_charset() or 'utf-8', errors='ignore')
|
||
|
||
if not body_text.strip():
|
||
for part in msg.walk():
|
||
if part.get_content_type() == "text/html":
|
||
payload = part.get_payload(decode=True)
|
||
if payload:
|
||
body_text = "[HTML Content]\n" + payload.decode(part.get_content_charset() or 'utf-8', errors='ignore')
|
||
break
|
||
|
||
if not body_text.strip():
|
||
body_text = "[Тело сообщения пустое или не содержит читаемого текста]"
|
||
|
||
return {
|
||
"msg_id": msg_num_or_id,
|
||
"from": author,
|
||
"subject": subject,
|
||
"date": date,
|
||
"body": body_text
|
||
}
|
||
except Exception as e:
|
||
return {"error": f"Не удалось прочитать сообщение {msg_num_or_id}: {e}"}
|
||
|
||
|
||
@mcp.tool()
|
||
def get_user_pure_messages(group_name: str, user_email: str, depth: int = 50) -> list[dict]:
|
||
"""
|
||
Находит последние сообщения конкретного пользователя в группе и извлекает только его текст.
|
||
"""
|
||
pure_messages = []
|
||
user_email_lower = user_email.lower()
|
||
|
||
try:
|
||
with nntplib.NNTP(NNTP_SERVER, port=PORT, timeout=15) as s:
|
||
resp, count, first, last, name = s.group(group_name)
|
||
first_num = int(first)
|
||
last_num = int(last)
|
||
|
||
start_msg = max(first_num, last_num - depth + 1)
|
||
if start_msg > last_num:
|
||
return []
|
||
|
||
resp, overviews = s.over((str(start_msg), str(last_num)))
|
||
|
||
for msg_num, over in overviews:
|
||
author = decode_mime_header(over.get('from', ''))
|
||
|
||
if user_email_lower not in author.lower():
|
||
continue
|
||
|
||
try:
|
||
_, info = s.article(str(msg_num))
|
||
raw_bytes = b"\n".join(info.lines)
|
||
msg = email.message_from_bytes(raw_bytes, policy=default)
|
||
subject = decode_mime_header(msg.get("Subject", "[Без темы]"))
|
||
|
||
body_text = ""
|
||
if msg.is_multipart():
|
||
for part in msg.walk():
|
||
if part.get_content_type() == "text/plain":
|
||
payload = part.get_payload(decode=True)
|
||
if payload:
|
||
body_text += payload.decode(part.get_content_charset() or 'utf-8', errors='ignore')
|
||
else:
|
||
payload = msg.get_payload(decode=True)
|
||
if payload:
|
||
body_text = payload.decode(msg.get_content_charset() or 'utf-8', errors='ignore')
|
||
except Exception:
|
||
continue
|
||
|
||
pure_text = clean_body_from_quotes(body_text)
|
||
if pure_text:
|
||
pure_messages.append({
|
||
"msg_num": str(msg_num),
|
||
"subject": subject,
|
||
"author": author,
|
||
"pure_text": pure_text
|
||
})
|
||
|
||
return pure_messages
|
||
except Exception as e:
|
||
return [{"error": f"Ошибка при извлечении сообщений пользователя: {e}"}]
|
||
|
||
@mcp.tool()
|
||
def save_text_to_file(filename: str, content: str) -> str:
|
||
"""
|
||
Save text content into a local file. Supports absolute or relative paths.
|
||
|
||
:param filename: Path or name of the file (e.g., 'C:/Users/ysv/Documents/bionic-project/chat-kraft/ghjtrnf/test.txt')
|
||
:param content: Text to write into the file
|
||
"""
|
||
try:
|
||
import os
|
||
|
||
# Если передан просто test.txt, сохраняем в папку скрипта
|
||
# Если передан путь (абсолютный или относительный), работаем с ним
|
||
target_path = os.path.abspath(filename)
|
||
|
||
# Автоматически создаем все промежуточные папки (например, ghjtrnf), если их еще нет на диске
|
||
target_dir = os.path.dirname(target_path)
|
||
if target_dir and not os.path.exists(target_dir):
|
||
os.makedirs(target_dir, exist_ok=True)
|
||
|
||
with open(target_path, "w", encoding="utf-8") as f:
|
||
f.write(content)
|
||
|
||
return f"SUCCESS: File saved at {target_path}"
|
||
except Exception as e:
|
||
return f"ERROR: {str(e)}"
|
||
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# Для современного пакета MCPServer (MCP v2.x)
|
||
# этот параметр автоматически развернет изолированный stdio-транспорт
|
||
# без конфликтов с внутренним вводом-выводом Bionic.
|
||
mcp.run(transport="stdio")
|
||
|