import re from typing import Optional NON_SOLUTION = re.compile( r"(уточнит|приложите скрин|какая ошибка|с какой проблемой" r"|нет обратной связи" r"|обратиться в службу поддержки" r"|свяжитесь с технической" r"|напишите нам|позвоните нам)", re.I ) PHONE = re.compile(r"\+?7\d{10}|8\d{10}") EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+") # Common Russian first names (vocative/именительный падеж) # followed by comma, dot, exclamation or question mark RUSSIAN_NAME = re.compile( r"\b(Анастасия|Татьяна|Михаил|Кристина|Александр|Елена" r"|Ольга|Мария|Иван|Дмитрий|Андрей|Юлия|Анна|Наталья" r"|Ирина|Екатерина|Владимир|Николай|Алексей|Сергей" r"|Максим|Роман|Виктор|Павел|Артем|Денис|Константин" r"|Галина|Людмила|Надежда|София|Вероника|Полина" r"|Алёна|Светлана|Вера|Маргарита|Лариса|Василий" r"|Виталий|Артур|Эдуард|Валентина|Зоя|Лидия)[,\.!?]", re.I ) def sanitize_pii(text: str) -> str: text = PHONE.sub("[ТЕЛЕФОН]", text) text = EMAIL.sub("[EMAIL]", text) text = RUSSIAN_NAME.sub("[ИМЯ]", text) return text def clean_text(text: str) -> str: text = re.sub(r'\[/?q\].*?\[/q\]', '', text, flags=re.DOTALL | re.IGNORECASE) text = re.sub(r'https?://\S+', '', text) text = re.sub(r'\s+', ' ', text).strip() text = sanitize_pii(text) return text SUPPORT_ROLES = {"support", "assistant", "operator", "agent"} CLIENT_ROLES = {"client", "user", "customer"} def has_real_solution(messages: list[dict]) -> Optional[str]: support_texts = [] for m in messages: if m["role"] in SUPPORT_ROLES: cleaned = clean_text(m["text"]) if cleaned: support_texts.append(cleaned) if not support_texts: return None combined = "\n".join(support_texts) if len(combined) < 100: return None last = support_texts[-1] if NON_SOLUTION.search(last): return None if last.strip().endswith("?") and len(last) < 150: return None return combined def get_client_dialogue(messages: list[dict]) -> str: texts = [] for m in messages: if m["role"] in CLIENT_ROLES: cleaned = clean_text(m["text"]) if cleaned: texts.append(cleaned) return "\n".join(texts) def build_chunk(ticket: dict) -> Optional[dict]: msgs = ticket.get("messages") or [] if not msgs: return None description = clean_text(ticket.get("description", "")) solution = has_real_solution(msgs) if not solution: return None client_msgs = get_client_dialogue(msgs) search_parts = [] if description: search_parts.append(description) if client_msgs: search_parts.append(client_msgs) search_text = "\n".join(search_parts).strip() if not search_text: return None full_text = ( f"Категория: {ticket['category']}\n" f"Проблема: {description}\n" f"Решение: {solution}" ) return { "ticket_id": ticket["ticket_id"], "client": ticket["client"], "category": ticket["category"], "search_text": search_text, "full_text": full_text, "tenant_id": None, }