522fb7717e
- Добавлена функция sanitize_pii() в chunking.py - Удаляются номера телефонов, email-адреса, русские имена в обращениях - Фикс NON_SOLUTION убраны ложные срабатывания на штатные фразы закрытия
111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
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,
|
||
}
|