233 lines
12 KiB
Plaintext
233 lines
12 KiB
Plaintext
import re
|
||
from typing import List, Optional, Set
|
||
|
||
# ------------------------------------------------------------
|
||
# Шаблоны для удаления (можно вынести в конфиг или JSON)
|
||
# ------------------------------------------------------------
|
||
# Фразы конфиденциальности / автоответы / экологические призывы (английские + русские)
|
||
DISCLAIMER_PHRASES = [
|
||
r"IMPORTANT NOTICE:\s*This e?mail.+?intended\s+only.+",
|
||
r"This e?mail (?:message )?is confidential.+?intended\s+only\s+for.+?(?=\n\s*\n|\Z)",
|
||
r"The information transmitted is intended only for the person or entity to which it is addressed",
|
||
r"If you have received this email in error, please notify the sender immediately",
|
||
r"This message contains confidential information and is intended only for the individual named",
|
||
r"Please consider the environment before printing this email",
|
||
r"This email has been scanned by the .*? antivirus",
|
||
r"Warning: Although .*? has taken reasonable precautions",
|
||
r"Sent from my iPhone",
|
||
r"Sent from my iPad",
|
||
r"Sent from my Android device",
|
||
r"Отправлено из приложения Яндекс.Почты",
|
||
r"Отправлено с iPhone",
|
||
r"Это письмо содержит конфиденциальную информацию",
|
||
r"Если вы получили это письмо по ошибке, пожалуйста, сообщите отправителю",
|
||
r"Данное сообщение может содержать сведения, составляющие коммерческую тайну",
|
||
r"Подумайте о природе, прежде чем распечатывать это письмо",
|
||
r"Автоматическое уведомление",
|
||
r"AUTO-REPLY|automatic reply|out of office",
|
||
]
|
||
|
||
# Шаблоны подписей (начало)
|
||
SIGNATURE_STARTERS = [
|
||
r"^\s*--\s*$", # стандартный разделитель подписи
|
||
r"^\s*Best regards\s*$",
|
||
r"^\s*Kind regards\s*$",
|
||
r"^\s*Sincerely\s*$",
|
||
r"^\s*Thanks\s*$",
|
||
r"^\s*Regards\s*$",
|
||
r"^\s*С уважением\s*$",
|
||
r"^\s*С наилучшими пожеланиями\s*$",
|
||
r"^\s*Спасибо\s*$",
|
||
r"^\s*Благодарю\s*$",
|
||
r"^\s*С ув.\s*$",
|
||
]
|
||
|
||
# Строки-разделители пересланных сообщений (нужно оставить только первое или удалить всё после)
|
||
FORWARD_SEPARATORS = [
|
||
r"^-{5,} Forwarded message -{5,}",
|
||
r"^-{5,} Пересылаемое сообщение -{5,}",
|
||
r"^-{5,} Пересланное сообщение -{5,}",
|
||
r"^From:.*?\nSent:.*?\nTo:.*?\nSubject:.*?\n", # Заголовок пересылки Outlook
|
||
r"^Begin forwarded message:",
|
||
r"^Начало переадресованного сообщения:",
|
||
]
|
||
|
||
# Лишние префиксы тем, которые можно упростить
|
||
RE_FWD_PREFIX = re.compile(
|
||
r"(?:RE|FW|FWD|Ответ|Пересланное|Пересылаемое|Перенаправлено|ПЕРЕСЛ)\s*:\s*",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
# ------------------------------------------------------------
|
||
# Основной класс очистки
|
||
# ------------------------------------------------------------
|
||
class EmailContextCleaner:
|
||
def __init__(
|
||
self,
|
||
remove_disclaimers: bool = True,
|
||
remove_signatures: bool = True,
|
||
collapse_repeats: bool = True,
|
||
remove_forward_chain: bool = True, # удалять цепочки пересылок (оставит только первое)
|
||
remove_empty_lines: bool = True,
|
||
remove_quote_lines: bool = True, # строки цитирования "> "
|
||
short_lines_only: int = 0, # удалять строки короче N символов (0 - нет)
|
||
max_url_length: int = 60, # обрезать URL длиннее N (0 - удалять полностью)
|
||
remove_timestamps: bool = True,
|
||
custom_phrases: Optional[List[str]] = None,
|
||
):
|
||
self.remove_disclaimers = remove_disclaimers
|
||
self.remove_signatures = remove_signatures
|
||
self.collapse_repeats = collapse_repeats
|
||
self.remove_forward_chain = remove_forward_chain
|
||
self.remove_empty_lines = remove_empty_lines
|
||
self.remove_quote_lines = remove_quote_lines
|
||
self.short_lines_only = short_lines_only
|
||
self.max_url_length = max_url_length
|
||
self.remove_timestamps = remove_timestamps
|
||
|
||
# Компиляция шаблонов
|
||
self._disclaimer_patterns = [
|
||
re.compile(p, re.IGNORECASE | re.DOTALL | re.MULTILINE)
|
||
for p in DISCLAIMER_PHRASES
|
||
]
|
||
if custom_phrases:
|
||
self._disclaimer_patterns += [
|
||
re.compile(p, re.IGNORECASE | re.DOTALL) for p in custom_phrases
|
||
]
|
||
|
||
self._signature_starters = [
|
||
re.compile(p) for p in SIGNATURE_STARTERS
|
||
]
|
||
self._forward_separators = [
|
||
re.compile(p, re.IGNORECASE | re.MULTILINE) for p in FORWARD_SEPARATORS
|
||
]
|
||
|
||
def clean(self, text: str) -> str:
|
||
if not text:
|
||
return text
|
||
|
||
# 1. Убрать дисклеймеры
|
||
if self.remove_disclaimers:
|
||
for pat in self._disclaimer_patterns:
|
||
# Заменяем на пробел, чтобы не склеить слова
|
||
text = pat.sub(" ", text)
|
||
|
||
# 2. Удалить цепочки пересылок (оставить только первую часть)
|
||
if self.remove_forward_chain:
|
||
# Ищем первый разделитель пересылки; всё после него удаляем,
|
||
# но только если перед этим уже был основной текст (чтобы не потерять единственное содержимое)
|
||
min_pos = len(text)
|
||
for pat in self._forward_separators:
|
||
match = pat.search(text)
|
||
if match:
|
||
pos = match.start()
|
||
if pos > 0 and pos < min_pos:
|
||
min_pos = pos
|
||
if min_pos > 0 and min_pos < len(text):
|
||
# Удаляем хвост, только если до разделителя было больше 50 символов осмысленного текста (не только заголовки)
|
||
before = text[:min_pos]
|
||
meaningful = re.sub(r"\s+", "", before)
|
||
if len(meaningful) > 50:
|
||
text = before.strip()
|
||
|
||
# 3. Удалить строки подписей (после "Best regards" и т.п.)
|
||
if self.remove_signatures:
|
||
lines = text.splitlines()
|
||
result_lines = []
|
||
for i, line in enumerate(lines):
|
||
stripped = line.strip()
|
||
if any(pat.match(stripped) for pat in self._signature_starters):
|
||
# Проверяем, не является ли строка частью основного текста
|
||
# Если дальше идут только пустые строки или короткие строки (типичная подпись), обрезаем
|
||
remaining = lines[i+1:]
|
||
# Подпись обычно не содержит длинных предложений без пунктуации
|
||
is_signature = all(
|
||
len(l.strip()) < 80 and not re.search(r"\b(договор|контракт|заказ|invoice|tracking|отгрузка|поставка|delivery)\b", l, re.IGNORECASE)
|
||
for l in remaining if l.strip()
|
||
)
|
||
if is_signature:
|
||
break
|
||
result_lines.append(line)
|
||
text = "\n".join(result_lines)
|
||
|
||
# 4. Удалить строки с цитированием (начинаются с ">")
|
||
if self.remove_quote_lines:
|
||
text = re.sub(r"^>.*$", "", text, flags=re.MULTILINE)
|
||
# Также удалить многоуровневое цитирование ">>"
|
||
text = re.sub(r"^(>\s*)+.*$", "", text, flags=re.MULTILINE)
|
||
|
||
# 5. Удалить очень короткие строки (если включено)
|
||
if self.short_lines_only > 0:
|
||
text = "\n".join(
|
||
line for line in text.splitlines()
|
||
if len(line.strip()) >= self.short_lines_only
|
||
)
|
||
|
||
# 6. Обработка URL
|
||
if self.max_url_length > 0 or self.max_url_length == 0:
|
||
url_pattern = re.compile(r"https?://\S+|ftp://\S+")
|
||
def replace_url(m):
|
||
url = m.group(0)
|
||
if self.max_url_length == 0:
|
||
return "" # удалить полностью
|
||
# Обрезать до max_url_length символов, сохранить протокол
|
||
if len(url) > self.max_url_length:
|
||
# Можно оставить домен + ... или просто обрезать
|
||
short = url[:self.max_url_length] + "..."
|
||
return short
|
||
return url
|
||
text = url_pattern.sub(replace_url, text)
|
||
|
||
# 7. Удалить временные метки (например, "2025-08-15 14:23 GMT+3")
|
||
if self.remove_timestamps:
|
||
text = re.sub(
|
||
r"\b\d{1,2}[./-]\d{1,2}[./-]\d{2,4}\s+\d{1,2}:\d{2}(?::\d{2})?\s*(?:AM|PM)?\s*(?:GMT[+-]\d+)?\b",
|
||
"",
|
||
text,
|
||
flags=re.IGNORECASE,
|
||
)
|
||
|
||
# 8. Удалить дублирующиеся пустые строки
|
||
if self.remove_empty_lines:
|
||
text = re.sub(r"\n\s*\n\s*\n+", "\n\n", text)
|
||
# Удалить одиночные пустые строки, если их больше двух подряд
|
||
text = re.sub(r"\n{3,}", "\n\n", text)
|
||
|
||
# 9. Схлопнуть повторяющиеся абзацы (уже было, но можно усилить)
|
||
if self.collapse_repeats:
|
||
text = self._deduplicate_paragraphs(text)
|
||
|
||
# 10. Убрать лишние пробелы в начале/конце строк
|
||
text = "\n".join(line.strip() for line in text.splitlines())
|
||
text = text.strip()
|
||
|
||
return text
|
||
|
||
def _deduplicate_paragraphs(self, text: str) -> str:
|
||
"""Удалить дословно повторяющиеся абзацы (как в RAGEngine)."""
|
||
parts = re.split(r"(\n\s*\n)", text)
|
||
seen: Set[str] = set()
|
||
out = []
|
||
for i, part in enumerate(parts):
|
||
if i % 2 == 1:
|
||
out.append(part)
|
||
continue
|
||
key = part.strip()
|
||
if len(key) < 50: # короткие фрагменты не дубликаты
|
||
out.append(part)
|
||
continue
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append(part)
|
||
return "".join(out)
|
||
|
||
# ------------------------------------------------------------
|
||
# Быстрая функция для совместимости
|
||
# ------------------------------------------------------------
|
||
_default_cleaner = EmailContextCleaner()
|
||
|
||
def clean_email_context(text: str) -> str:
|
||
"""Очищает текст письма от мусора перед отправкой в LLM."""
|
||
return _default_cleaner.clean(text)
|