# 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)