Deal/archive/leadradar-legacy/backend/app/services/rules.py

"""Детерминированные правила колонок: направление, стек, слова, грейд, бюджет.

Колонка — это набор опциональных фильтров, задаёт пользователь (или ИИ при
предложении). Если текст входящего сообщения соответствует правилам — карточка
уходит в эту колонку сразу, без ML/ИИ (ML/ИИ подключаются только когда правила
не сработали). Набор фильтров может меняться: пустая группа не участвует.
"""
from __future__ import annotations

import json
import re

from ..db import store

_CUR_SYMBOLS = {"$": "USD", "€": "EUR", "₽": "RUB", "₮": "USDT", "£": "GBP", "¥": "CNY"}
_CUR_WORDS = {"usd": "USD", "eur": "EUR", "rub": "RUB", "usdt": "USDT", "gbp": "GBP", "cny": "CNY", "руб": "RUB", "долл": "USD", "бакс": "USD"}

# Грейд/уровень: тег из правил расширяется синонимами, чтобы «middle» находил
# и «mid», и «мидл», а «сеньор» находил senior и т.п.
_GRADE_ALIASES = {
    "junior": ["junior", "джун", "джуниор"],
    "middle": ["middle", "mid", "мидл"],
    "senior": ["senior", "сеньор", "сеньйор"],
    "lead": ["lead", "тимлид", "тиэмлид", "team lead", "teamlead", "tech lead", "техлид"],
    "architect": ["architect", "архитектор"],
    "intern": ["intern", "стажёр", "стажер", "trainee"],
}


def _grade_terms(tags: list[str]) -> list[str]:
    out: list[str] = []
    for t in tags:
        key = str(t).strip().lower()
        if not key:
            continue
        if key in _GRADE_ALIASES:
            out.extend(_GRADE_ALIASES[key])
        else:
            out.append(key)
    return out


# Перед матчингом правил вырезаем ссылки и markdown-ссылки: иначе фильтр ловит
# слова из трекерных хвостов/служебных строк URL (например, «desktop» в
# utm_medium=member_desktop) и в колонку попадает мусор, не имеющий отношения
# к содержанию сообщения.
_MD_URL_RE = re.compile(r"\[[^\]]*\]\([^)\s]+\)")
_RAW_URL_RE = re.compile(r"https?://[^\s<>\"']+|www\.[^\s<>\"']+")


def content_text(text: str) -> str:
    s = str(text or "")
    s = _MD_URL_RE.sub(" ", s)
    return _RAW_URL_RE.sub(" ", s)

# ищем: 1) "от 1 200 до 1 500 $", 2) "1 200–1 500 $ / 1 200$", 3) "$1 200–1 500"
# суффикс «к/К» разрешён прямо в числе: «2к», «1.5к$» (множитель в _norm_amount)
_AMT = r"\d[\d\s\u00a0]*(?:[.,]\d+)?[кkКK]?"
_RANGE = rf"({_AMT})\s*(?:[-–—]\s*({_AMT}))?"


def _norm_amount(raw: str) -> float | None:
    s = raw.replace("\u00a0", " ").replace(" ", "").replace(",", ".")
    mult = 1.0
    # «2к»/«2К»/«1.5к» — тысячи; суффикс убираем до float (иначе парс падает)
    if s and s[-1].lower() in ("k", "к"):
        mult = 1000.0
        s = s[:-1]
    try:
        v = float(s) * mult
    except ValueError:
        return None
    return v


def _cur_from_tail(text: str, m_start: int) -> str | None:
    tail = text[m_start : m_start + 12].lower().strip()
    for sym, code in _CUR_SYMBOLS.items():
        if tail.startswith(sym):
            return code
    # слово-валюта сразу после числа (с пробелом)
    for word, code in _CUR_WORDS.items():
        if tail.startswith(word):
            return code
    # валюта перед числом ($/€/₽), например "$1 200"
    head = text[max(0, m_start - 3) : m_start].strip()
    for sym, code in _CUR_SYMBOLS.items():
        if head.endswith(sym):
            return code
    return None


def extract_amounts(text: str) -> list[dict]:
    """Парсер сумм с сохранением смысла:
    - диапазон «от A до B» / «A–B» → {'from': A, 'to': B, 'cur': ...};
    - «до B» (только верхняя граница) → {'from': None, 'to': B, 'cur': ...};
    - одна сумма / «от A» без верхней границы → {'from': A, 'to': A, 'cur': ...}.
    Валюту ищем сразу после суммы (или перед ней для «$1 200»); суммы без валюты игнорируются.
    """
    out: list[dict] = []
    t = text or ""
    if not t.strip():
        return out
    occupied: list[tuple[int, int]] = []

    def _free(s: int, e: int) -> bool:
        return not any(s < oe and e > os for os, oe in occupied)

    def _add(a, b, cur: str | None, s: int, e: int) -> None:
        if cur and (a is not None or b is not None) and _free(s, e):
            out.append({"from": a, "to": b, "cur": cur})
            occupied.append((s, e))

    # 1) словесный диапазон «от A до B» (+ валюта после B)
    for m in re.finditer(r"(?i)\bот\s+(" + _AMT + r")\s+до\s+(" + _AMT + r")", t):
        a, b = _norm_amount(m.group(1)), _norm_amount(m.group(2))
        cur = _cur_from_tail(t, m.end(2))
        if a is not None and b is not None and cur:
            _add(a, b, cur, m.start(1), m.end(2))
    # 2) «до B» (без пары «от … до») — верхняя граница
    for m in re.finditer(r"(?i)\bдо\s+(" + _AMT + r")", t):
        b = _norm_amount(m.group(1))
        cur = _cur_from_tail(t, m.end(1))
        if b is not None and cur:
            _add(None, b, cur, m.start(1), m.end(1))
    # 3) «от A» без верхней границы — считаем одной суммой
    for m in re.finditer(r"(?i)\bот\s+(" + _AMT + r")", t):
        a = _norm_amount(m.group(1))
        cur = _cur_from_tail(t, m.end(1))
        if a is not None and cur:
            _add(a, a, cur, m.start(1), m.end(1))
    # 4) числовые диапазоны «A–B» / «A - B»
    for m in re.finditer(r"(?i)(" + _AMT + r")\s*(?:[-–—]|\s+до\s+)\s*(" + _AMT + r")", t):
        a, b = _norm_amount(m.group(1)), _norm_amount(m.group(2))
        cur = _cur_from_tail(t, m.end(2)) or _cur_from_tail(t, m.end(1))
        if a is not None and b is not None and cur:
            _add(a, b, cur, m.start(1), m.end(2))
    # 5) одиночные суммы с валютой (не вошедшие в конструкции выше)
    for m in re.finditer(_AMT, t):
        a = _norm_amount(m.group(0))
        cur = _cur_from_tail(t, m.end())
        if a is not None and cur:
            _add(a, a, cur, m.start(), m.end())
    return out


def _amount_in_range(amounts: list[dict], budget: dict) -> bool:
    from ..services.rates import convert_amount

    try:
        lo = float(budget.get("from")) if budget.get("from") is not None else None
        hi = float(budget.get("to")) if budget.get("to") is not None else None
    except (TypeError, ValueError):
        return False
    if lo is None and hi is None:
        return True
    target_cur = str(budget.get("cur") or "USD").upper()
    for amt in amounts:
        raw_val = amt["from"] if amt["from"] is not None else amt.get("to")
        if raw_val is None:
            continue
        try:
            val = raw_val if amt["cur"] == target_cur else convert_amount(raw_val, amt["cur"], target_cur)
        except Exception:  # noqa: BLE001
            val = None
        if val is None:
            continue
        if lo is not None and val < lo:
            continue
        if hi is not None and val > hi:
            continue
        return True
    return False


def match_text(rules: dict, text: str) -> bool:
    """Соответствует ли текст правилам колонки. Возвращает bool.

    Колонка — это набор опциональных фильтров: направление, стек, ключевые
    слова, грейд/уровень, бюджет. Пустая группа не участвует; режим «все» —
    должны совпасть все включённые группы, «любое» — хотя бы одна.
    """
    rules = rules or {}
    lower = content_text(text).lower()
    direction = [str(x).strip().lower() for x in (rules.get("direction") or []) if str(x).strip()]
    kw = [str(x).strip().lower() for x in (rules.get("keywords") or []) if str(x).strip()]
    stack = [str(x).strip().lower() for x in (rules.get("stack") or []) if str(x).strip()]
    grade = [str(x).strip().lower() for x in (rules.get("grade") or []) if str(x).strip()]
    budget = rules.get("budget") if isinstance(rules.get("budget"), dict) else None
    enabled = [bool(kw), bool(stack), bool(direction), bool(grade), bool(budget)]

    if not any(enabled):
        return False
    grade_terms = _grade_terms(grade)
    groups = {
        "keywords": any(k in lower for k in kw) if kw else True,
        "stack": any(s in lower for s in stack) if stack else True,
        "direction": any(d in lower for d in direction) if direction else True,
        "grade": any(g in lower for g in grade_terms) if grade else True,
        "budget": (_amount_in_range(extract_amounts(text), budget) if budget else True),
    }
    mode = str(rules.get("mode") or "all").lower()
    if mode == "any":
        # «любое»: совпасть должна хотя бы одна включённая (непустая) группа.
        # Пустые группы равны True и не должны участвовать — иначе колонка
        # с одним фильтром (например «стек: WPF») ловит вообще всё.
        return any(groups[k] for k, on in zip(groups, enabled) if on and k in groups)
    # all: каждая включённая группа обязана совпасть
    return all(groups[k] for k, on in zip(groups, enabled) if on and k in groups)


def score_text(rules: dict, text: str) -> int:
    """Число совпавших фильтров (для выбора лучшей ИИ-колонки)."""
    if not text:
        return 0
    lower = content_text(text).lower()
    score = 0
    for group in ("direction", "keywords", "stack", "grade"):
        if group == "grade":
            for g in _grade_terms(rules.get(group) or []):
                if g in lower:
                    score += 1
        else:
            for w in (rules.get(group) or []):
                if str(w).lower() in lower:
                    score += 1
    return score


def excluded_terms(rules: dict | None, text: str) -> list[str]:
    """Какие слова-исключения колонки есть в тексте.

    Исключения — veto колонки: если в содержании сообщения (без ссылок и
    служебных хвостов) встречается любое из них, карточка в колонку не
    попадает, даже если все положительные условия совпали.
    """
    rules = rules or {}
    lower = content_text(text).lower()
    out: list[str] = []
    for term in rules.get("exclude") or []:
        t = str(term).strip()
        if t and t.lower() in lower:
            out.append(t)
    return out


def is_excluded(rules: dict | None, text: str) -> bool:
    return bool(excluded_terms(rules, text))


def board_accepts(board_id: str, text: str) -> bool:
    """Пропускает ли колонка этот текст.

    Колонка с активными правилами принимает только текст, прошедший её правила
    (детерминированно); колонка без правил принимает любой текст — её наполняют
    ИИ/ML/пользователь. Нужно как страховка: ИИ или ML не должны класть карточку
    в «отфильтрованную» колонку, если текст под правила не подходит. Слова-
    исключения работают как veto в любом случае.
    """
    row = store.query_one("SELECT rules FROM boards WHERE id = ?", [board_id])
    if not row:
        return False
    rules = json.loads(row["rules"] or "{}") if row["rules"] else {}
    if is_excluded(rules, text):
        return False
    if not has_active_rules(rules):
        return True
    return match_text(rules, text)


def hits(rules: dict, text: str) -> list[dict]:
    """Какие именно критерии фильтра совпали с текстом.

    Возвращает список совпадений по группам фильтра колонки:
    [{"label": "Стек", "term": "WPF"}, {"label": "Грейд", "term": "middle", "word": "mid"}, …].
    Нужно, чтобы на карточке показывать «по каким критериям она попала в колонку»
    (список совпавших условий фильтра). Ключевое слово ищется по всему тексту
    сообщения — включая стек, требования и «будет плюсом», как и наоборот.
    """
    rules = rules or {}
    lower = content_text(text).lower()
    out: list[dict] = []
    for group, label in (("direction", "Направление"), ("keywords", "Слова"), ("stack", "Стек")):
        for term in rules.get(group) or []:
            t = str(term).strip()
            if t and t.lower() in lower:
                out.append({"label": label, "term": t})
    for term in rules.get("grade") or []:
        for alias in _grade_terms([term]):
            if alias in lower:
                out.append({"label": "Грейд/уровень", "term": str(term).strip(), "word": alias})
                break
    budget = rules.get("budget") if isinstance(rules.get("budget"), dict) else None
    if budget and _amount_in_range(extract_amounts(text), budget):
        out.append({"label": "Бюджет", "term": _budget_label(budget)})
    return out


def _budget_label(budget: dict) -> str:
    lo, hi = budget.get("from"), budget.get("to")
    cur = str(budget.get("cur") or "").upper()
    if lo is not None and hi is not None:
        return f"от {lo:g} до {hi:g} {cur}".strip()
    if hi is not None:
        return f"до {hi:g} {cur}".strip()
    if lo is not None:
        return f"от {lo:g} {cur}".strip()
    return "бюджет"


def hits_for_board(board_id: str, text: str) -> list[dict]:
    """Совпавшие критерии колонки с активными правилами; [] — правил нет."""
    row = store.query_one("SELECT rules FROM boards WHERE id = ?", [board_id])
    if not row:
        return []
    rules = json.loads(row["rules"] or "{}") if row["rules"] else {}
    if not has_active_rules(rules):
        return []
    return hits(rules, text)


def has_active_rules(rules: dict | None) -> bool:
    """Есть ли в правилах хотя бы одна реально работающая группа фильтров.

    Нужно для ML: колонка с активными правилами раскладывается только самими
    правилами (детерминированно), ML её назначать не должен — иначе в колонку
    попадает то, что под правила не подходит.
    """
    rules = rules or {}
    for key in ("direction", "keywords", "stack", "grade"):
        if any(str(x).strip() for x in (rules.get(key) or [])):
            return True
    budget = rules.get("budget")
    if isinstance(budget, dict) and (
        budget.get("from") is not None or budget.get("to") is not None
    ):
        return True
    return False


def describe(rules: dict) -> str:
    """Человекочитаемое описание правил (для обоснования и промпта)."""
    rules = rules or {}
    parts = []
    direction = rules.get("direction") or []
    stack = rules.get("stack") or []
    kw = rules.get("keywords") or []
    grade = rules.get("grade") or []
    if direction:
        parts.append("направление: " + ", ".join(str(x) for x in direction[:6]))
    if stack:
        parts.append("стек: " + ", ".join(str(x) for x in stack[:8]))
    if kw:
        parts.append("слова: " + ", ".join(str(x) for x in kw[:8]))
    if grade:
        parts.append("грейд: " + ", ".join(str(x) for x in grade[:8]))
    exc = rules.get("exclude") or []
    if exc:
        parts.append("исключено: " + ", ".join(str(x) for x in exc[:8]))
    budget = rules.get("budget") if isinstance(rules.get("budget"), dict) else None
    if budget and (budget.get("from") is not None or budget.get("to") is not None):
        lo = budget.get("from") if budget.get("from") is not None else ""
        hi = budget.get("to") if budget.get("to") is not None else ""
        parts.append(f"бюджет: {lo}–{hi} {budget.get('cur') or ''}".replace('– ', '–').replace(' –', '–'))
    if not parts:
        return "без правил (решает ИИ/ML)"
    mode = "все условия" if str(rules.get("mode") or "all").lower() != "any" else "любое из условий"
    return mode + " · " + "; ".join(parts)


def route(text: str) -> dict | None:
    """Детерминированная маршрутизация по правилам активных колонок.

    Возвращает колонку, если правила однозначно совпали (при нескольких —
    ту, где больше совпадений). Предложения ИИ в маршрутизации не участвуют.
    """
    rows = store.query(
        "SELECT * FROM boards WHERE suggested = FALSE AND rules <> '{}' AND rules <> '' ORDER BY pos"
    )
    best: dict | None = None
    best_score = 0
    for r in rows:
        rules = json.loads(r["rules"] or "{}")
        if not match_text(rules, text):
            continue
        sc = score_text(rules, text) or 1
        if sc > best_score:
            best = {"id": r["id"], "name": r["name"], "color": r["color"], "rules": rules}
            best_score = sc
    return best