152 lines
7.9 KiB
Python
152 lines
7.9 KiB
Python
"""Детерминированная идентификация семейства, якорей и параметров услуги.
|
||
|
||
Стабильный якорь экземпляра отделён от изменяемых параметров конфигурации:
|
||
изменение мощности, объёма ресурса или цены не должно само по себе создавать
|
||
новую услугу, тогда как разные стойки, облачные платформы и маршруты остаются
|
||
разными экземплярами. Неизвестные строки сохраняются как family `other`.
|
||
"""
|
||
import re
|
||
|
||
|
||
_HOMOGLYPHS = str.maketrans({
|
||
"A": "А", "B": "В", "C": "С", "E": "Е", "H": "Н", "K": "К",
|
||
"M": "М", "O": "О", "P": "Р", "T": "Т", "X": "Х", "Y": "У",
|
||
})
|
||
_RACK_RE = re.compile(r"\b\d[A-ZА-ЯЁ]\d[A-ZА-ЯЁ]\d{3}\b")
|
||
_NUMBER = r"(\d+(?:[.,]\d+)?)"
|
||
|
||
|
||
def _normalized(text):
|
||
"""Нормализовать регистр, сохраняя различимые английские маркеры услуг."""
|
||
return str(text or "").upper().replace("Ё", "Е")
|
||
|
||
|
||
def _decimal_text(value):
|
||
"""Представить числовой фрагмент с точкой и без незначащих нулей."""
|
||
return value.replace(",", ".").rstrip("0").rstrip(".") if "." in value or "," in value else value
|
||
|
||
|
||
def _first_capture(pattern, text):
|
||
"""Извлечь первое числовое значение, сохранив точность исходной строки."""
|
||
match = re.search(pattern, text, re.IGNORECASE)
|
||
return _decimal_text(match.group(1)) if match else None
|
||
|
||
|
||
def _service_family(text):
|
||
"""Назначить наиболее специфичное семейство по доменным словам Nubes."""
|
||
rules = (
|
||
("cloud_drive", r"VALO\s+CLOUD|ОБЛАЧН\w*\s+ДИСК"),
|
||
("rack_cabinet", r"ШКАФ\s+(?:СЕРВЕРН\w*|\d+\s*U)"),
|
||
("pdu", r"БЛОК\s+РОЗЕТОК|\bPDU\b"),
|
||
("fiber_line", r"ВОЛС|ВОЛОКН|ЛИНИ\w*\s+СВЯЗИ"),
|
||
("l2_channel", r"КАНАЛ\s+СВЯЗИ|\bL2\b|ТОЧКА\s*[-–—]?\s*ТОЧКА"),
|
||
("network_port", r"АРЕНДА\s+ПОРТА|ПОРТ\s+В\s+КОММУТАТОРЕ|CROSS\s*[-–]?\s*CONNECT|КРОСС\s*[-–]?\s*СОЕДИН"),
|
||
("ip_subnet", r"ПОДСЕТ|IPV[46]|IP\s*[-–]?\s*АДРЕС|/\s*(?:24|29|30)\b"),
|
||
("internet", r"ИНТЕРНЕТ|ТРАФИК|DDOS"),
|
||
("cloud_iaas", r"NEXT\s+GENERATION\s+CLOUD|ОБЛАЧН\w*|VCPU|VRAM|\bIAAS\b"),
|
||
("waf", r"\bWAF\b|ЗАЩИТ\w*\s+ВЕБ"),
|
||
("colocation_rack", r"СТОЙКО\s*[-–—]?\s*МЕСТ|COLOCATION|\b\d+\s*U\b"),
|
||
)
|
||
for family, pattern in rules:
|
||
if re.search(pattern, text, re.IGNORECASE):
|
||
return family
|
||
return "other"
|
||
|
||
|
||
def _extract_anchors(text, family):
|
||
"""Извлечь идентификаторы экземпляра, не перенося туда изменяемые параметры."""
|
||
anchors = set(_RACK_RE.findall(text))
|
||
if family == "fiber_line":
|
||
route = re.search(
|
||
r"(?:ОТ\s+)?(.{0,100}?)(?:\s+ДО\s+)(.{0,100})",
|
||
text,
|
||
re.IGNORECASE,
|
||
)
|
||
if route:
|
||
anchors.add("ROUTE:" + re.sub(r"\s+", " ", route.group(1) + "->" + route.group(2)).strip())
|
||
if family == "cloud_iaas":
|
||
platform = re.search(r"\b(?:AMD|INTEL)\s*\d+(?:[.,]\d+)?\s*ГГЦ|\bHIGHFREQ\b|\bSTD\b", text)
|
||
if platform:
|
||
anchors.add("PLATFORM:" + re.sub(r"\s+", "", platform.group(0)))
|
||
if family == "ip_subnet":
|
||
subnet = re.search(r"/\s*(\d{1,2})\b", text)
|
||
if subnet:
|
||
anchors.add("PREFIX:/" + subnet.group(1))
|
||
if family == "network_port":
|
||
speed = re.search(r"\b\d+(?:[.,]\d+)?\s*(?:ГБИТ/С|МБИТ/С|GBPS|MBPS)\b", text)
|
||
if speed:
|
||
anchors.add("SPEED:" + re.sub(r"\s+", "", speed.group(0)))
|
||
return "|".join(sorted(anchors))
|
||
|
||
|
||
def _extract_config(text):
|
||
"""Извлечь известные изменяемые численные параметры как точные строки."""
|
||
params = {}
|
||
patterns = {
|
||
"kw": rf"{_NUMBER}\s*КВТ\b",
|
||
"units_u": rf"(?:РАЗМЕР\s+СТОЙКИ\s*[:\-]?\s*)?(\d+)\s*U\b",
|
||
"vcpu": rf"(?:КОЛИЧЕСТВО\s*)?VCPU\s*[:\-]?\s*(\d+)",
|
||
"vram_gb": rf"(?:ОБЪ[ЕЕ]М\s*)?VRAM\s*[:\-]?\s*{_NUMBER}\s*ГБ",
|
||
"sata_gb": rf"SATA\s*[:\-]?\s*{_NUMBER}\s*ГБ",
|
||
"ssd_gb": rf"(?<!FAST\s)SSD\s*[:\-]?\s*{_NUMBER}\s*ГБ",
|
||
"fast_ssd_gb": rf"FAST\s+SSD\s*[:\-]?\s*{_NUMBER}\s*ГБ",
|
||
"users_count": rf"(?:ЧИСЛО\s+)?ПОЛЬЗОВАТЕЛ\w*\s*[:\-]?\s*(\d+)",
|
||
}
|
||
for key, pattern in patterns.items():
|
||
value = _first_capture(pattern, text)
|
||
if value is not None:
|
||
params[key] = value
|
||
speed = re.search(rf"{_NUMBER}\s*(МБИТ/С|ГБИТ/С|MBPS|GBPS)\b", text, re.IGNORECASE)
|
||
if speed:
|
||
unit = speed.group(2).upper()
|
||
params["speed_gbps" if unit in {"ГБИТ/С", "GBPS"} else "speed_mbps"] = _decimal_text(speed.group(1))
|
||
return params
|
||
|
||
|
||
def analyze_service_row(raw_name, charge_type="unknown"):
|
||
"""Вернуть семейство, anchor и извлечённые параметры для одной строки услуги."""
|
||
normalized = _normalized(raw_name)
|
||
family = _service_family(normalized)
|
||
return {
|
||
"service_family": family,
|
||
"instance_anchor": _extract_anchors(normalized, family),
|
||
"config_params": _extract_config(normalized),
|
||
"canonical_name": re.sub(r"\s+", " ", str(raw_name or "")).strip(),
|
||
"charge_type": charge_type or "unknown",
|
||
}
|
||
|
||
|
||
def _anchors_compatible(left, right):
|
||
"""Считать совместимыми одинаковые якоря либо отсутствие данных на одной стороне."""
|
||
left_set = {part for part in str(left or "").split("|") if part}
|
||
right_set = {part for part in str(right or "").split("|") if part}
|
||
return not left_set or not right_set or left_set == right_set
|
||
|
||
|
||
def match_service_item(row_info, existing_service_items, llm_client=None):
|
||
"""Сопоставить строку с существующим item без выдуманного автоматического выбора.
|
||
|
||
Если один кандидат совместим по семейству и anchor, он возвращается как
|
||
MATCHED. При нескольких кандидатах точное совпадение непустого anchor
|
||
разрешает выбор; иначе вызывающий код должен создать исключение AMBIGUOUS.
|
||
Параметр `llm_client` зарезервирован для будущего Entity Linker, но сейчас
|
||
намеренно не вызывает модель, пока это не будет отдельным проверяемым этапом.
|
||
"""
|
||
del llm_client
|
||
same_family = [
|
||
item for item in existing_service_items
|
||
if item.get("service_family") == row_info.get("service_family")
|
||
]
|
||
compatible = [
|
||
item for item in same_family
|
||
if _anchors_compatible(row_info.get("instance_anchor"), item.get("instance_anchor"))
|
||
]
|
||
if len(compatible) == 1:
|
||
return {"status": "MATCHED", "service_item": compatible[0], "candidates": compatible}
|
||
row_anchor = row_info.get("instance_anchor") or ""
|
||
exact = [item for item in compatible if item.get("instance_anchor") == row_anchor and row_anchor]
|
||
if len(exact) == 1:
|
||
return {"status": "MATCHED", "service_item": exact[0], "candidates": exact}
|
||
if compatible:
|
||
return {"status": "AMBIGUOUS", "service_item": None, "candidates": compatible}
|
||
return {"status": "NOT_FOUND", "service_item": None, "candidates": []} |