286 lines
14 KiB
Python
286 lines
14 KiB
Python
# email_cleaner.py
|
||
import re
|
||
from typing import List, Optional, Set
|
||
|
||
# ------------------------------------------------------------
|
||
# Шаблоны конфиденциальности (русские + английские)
|
||
# ------------------------------------------------------------
|
||
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"The content of this email is confidential and intended for the recipient specified in message only\..*",
|
||
r"The content of this email is confidential and intended for the recipient specified in message only. It is strictly forbidden to share any part of this message with any third party, without a written consent of the sender. If you received this message by mistake, please reply to this message and follow with its deletion, so that we can ensure such a mistake does not occur in the future.",
|
||
# Русские варианты
|
||
r"Содержание этого письма конфиденциально и предназначено только для указанного получателя.*",
|
||
r"Данное сообщение может содержать конфиденциальную информацию.*",
|
||
r"Настоящее сообщение содержит конфиденциальную информацию.*",
|
||
r"Это сообщение содержит информацию, являющуюся коммерческой тайной.*",
|
||
r"Сообщение предназначено исключительно для указанного адресата.*",
|
||
r"Если вы не являетесь указанным получателем, пожалуйста, удалите это сообщение.*",
|
||
r"Любое распространение, копирование или использование этого сообщения.*",
|
||
r"Это письмо содержит конфиденциальную информацию.*",
|
||
# Автоответы
|
||
r"\bAUTO-REPLY\b|automatic reply|out of office",
|
||
r"Автоматическое уведомление",
|
||
]
|
||
|
||
# ------------------------------------------------------------
|
||
# Строки-разделители подписей
|
||
# ------------------------------------------------------------
|
||
SIGNATURE_STARTERS = [
|
||
r"^\s*--\s*$", # стандартный разделитель подписи
|
||
r"^\s*Thanks\s*[,.]?\s*$",
|
||
r"^\s*Thank you\s*[,.]?\s*$",
|
||
r"^\s*С уважением\s*[,.]?\s*$",
|
||
r"^\s*Best regards\s*[,.]?\s*$",
|
||
r"^\s*Kind regards\s*[,.]?\s*$",
|
||
r"^\s*Regards\s*[,.]?\s*$",
|
||
r"^\s*Здравствуйте\s*$",
|
||
r"^\s*Привет\s*$",
|
||
r"^\s*Добрый день\s*$",
|
||
r"^\s*Sincerely\s*[,.]?\s*$",
|
||
r"^\s*С наилучшими пожеланиями\s*$",
|
||
r"^\s*Спасибо\s*$",
|
||
r"^\s*Благодарю\s*$",
|
||
r"^\s*С ув\.\s*$",
|
||
r"^\s*С уважением и надеждой на сотрудничество\s*$",
|
||
]
|
||
|
||
# ------------------------------------------------------------
|
||
# Разделители цепочек пересылки (НЕ трогаем From/To/Subject внутри)
|
||
# ------------------------------------------------------------
|
||
FORWARD_SEPARATORS = [
|
||
r"^-{5,}\s*(Forwarded message|Пересланное сообщение|Пересылаемое сообщение|Переадресованное сообщение)\s*-{5,}",
|
||
r"^Begin forwarded message:",
|
||
r"^Начало переадресованного сообщения:",
|
||
r"^--- Исходное сообщение ---",
|
||
r"^-{3,}\s*Original Message\s*-{3,}",
|
||
]
|
||
|
||
# ------------------------------------------------------------
|
||
# Ключевые слова, предотвращающие удаление подписи/контактов
|
||
# ------------------------------------------------------------
|
||
TRANSPORT_KEYWORDS = [
|
||
"груз", "паллет", "вес", "адрес", "инвойс", "забор", "доставка",
|
||
"тн вэд", "контейнер", "отправка", "перевозка", "машина", "фура",
|
||
"ставка", "тариф", "расписание", "договор", "контракт", "заказ",
|
||
"invoice", "tracking", "delivery", "shipment", "cargo",
|
||
]
|
||
|
||
# ------------------------------------------------------------
|
||
# Основной класс очистки
|
||
# ------------------------------------------------------------
|
||
class EmailContextCleaner:
|
||
def __init__(
|
||
self,
|
||
remove_disclaimers: bool = True,
|
||
remove_signatures: bool = True,
|
||
remove_forward_chain: bool = True,
|
||
remove_contact_blocks: bool = True,
|
||
collapse_repeats: bool = True,
|
||
remove_empty_lines: bool = True,
|
||
remove_quote_lines: bool = True,
|
||
short_lines_only: int = 0,
|
||
max_url_length: int = 60,
|
||
remove_timestamps: bool = True,
|
||
custom_phrases: Optional[List[str]] = None,
|
||
):
|
||
self.remove_disclaimers = remove_disclaimers
|
||
self.remove_signatures = remove_signatures
|
||
self.remove_forward_chain = remove_forward_chain
|
||
self.remove_contact_blocks = remove_contact_blocks
|
||
self.collapse_repeats = collapse_repeats
|
||
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):
|
||
before = text[:min_pos]
|
||
# Удаляем хвост, только если до разделителя было осмысленного текста > 50 символов
|
||
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 any(
|
||
kw in l.lower() for kw in TRANSPORT_KEYWORDS
|
||
)
|
||
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"^(>\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 ""
|
||
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. Временные метки
|
||
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{3,}", "\n\n", text)
|
||
|
||
# 9. Дублирующиеся абзацы
|
||
if self.collapse_repeats:
|
||
text = self._deduplicate_paragraphs(text)
|
||
|
||
# 10. Блоки контактов (подписи без явного приветствия)
|
||
if self.remove_contact_blocks:
|
||
text = self._remove_contact_blocks(text)
|
||
|
||
# Финальная обрезка пробелов в строках
|
||
text = "\n".join(line.strip() for line in text.splitlines())
|
||
return text.strip()
|
||
|
||
def _deduplicate_paragraphs(self, text: str) -> str:
|
||
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)
|
||
|
||
def _remove_contact_blocks(self, text: str) -> str:
|
||
"""
|
||
Удаляет блоки, похожие на подпись: Имя Фамилия, должность, телефоны, email, сайт,
|
||
которые идут после пустой строки и не содержат ключевых слов перевозки.
|
||
"""
|
||
lines = text.splitlines()
|
||
out = []
|
||
i = 0
|
||
while i < len(lines):
|
||
line = lines[i].strip()
|
||
if self._is_contact_line(line):
|
||
# Проверим, что предыдущая строка пустая или это начало текста
|
||
if i == 0 or (i > 0 and lines[i-1].strip() == ""):
|
||
# Собираем блок, пока строки похожи на контакты или короткие и без ключевых слов
|
||
j = i
|
||
while j < len(lines):
|
||
l = lines[j].strip()
|
||
if l == "" or self._is_contact_line(l) or (len(l) < 60 and not any(
|
||
kw in l.lower() for kw in TRANSPORT_KEYWORDS
|
||
)):
|
||
j += 1
|
||
else:
|
||
break
|
||
block_text = "\n".join(lines[i:j])
|
||
# Не удаляем, если есть транспортные ключевые слова
|
||
if not any(kw in block_text.lower() for kw in TRANSPORT_KEYWORDS):
|
||
i = j
|
||
continue
|
||
out.append(line)
|
||
i += 1
|
||
return "\n".join(out)
|
||
|
||
def _is_contact_line(self, line: str) -> bool:
|
||
"""Определяет, является ли строка элементом контактной информации."""
|
||
if re.search(r"\+?\d[\d\s\(\)\-]{6,}\d", line): # телефон
|
||
return True
|
||
if re.search(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b", line): # email
|
||
return True
|
||
if re.search(r"www\.", line): # сайт
|
||
return True
|
||
if re.search(r"^\s*[A-Z][a-z]+\s+[A-Z][a-z]+\s*$", line): # Имя Фамилия (упрощённо)
|
||
return True
|
||
if re.search(r"Business Development Manager|COO|Operations dept\.|Leading Sales", line, re.IGNORECASE):
|
||
return True
|
||
return False
|
||
|
||
# ------------------------------------------------------------
|
||
# Быстрая функция для совместимости
|
||
# ------------------------------------------------------------
|
||
_default_cleaner = EmailContextCleaner()
|
||
|
||
def clean_email_context(text: str) -> str:
|
||
"""Очищает текст письма от мусора перед отправкой в LLM."""
|
||
return _default_cleaner.clean(text)
|