"""Автономная ML-модель (наивный Байес по словам) для LeadRadar.

Хранилище — собственный DuckDB-файл (volume). Модель живёт в отдельном
контейнере и общается с основным приложением по HTTP:
  * /learn, /learn-batch  — обучение (всегда, независимо от настроек UI);
  * /predict              — предсказание (используется, только если mlEnabled);
  * /status               — готовность и статистика.
"""
from __future__ import annotations

import os
import re
import threading
import time

import duckdb

# Пороги уверенности: консервативные на старте, смягчаются по мере накопления
# опыта (см. _adaptive_margin) — ML постепенно берёт на себя больше работы.
MIN_TOTAL = 20      # суммарно примеров по всем классам, чтобы модель «включилась»
MIN_WINNER = 6      # минимум примеров у класса-победителя
MIN_WINNER_SPAM = 4
MIN_HITS = 2        # минимум различных терминов, встреченных у победителя
MARGIN = 0.9        # ln-отрыв от второго класса на старте

# Самооценка «справляется ли ML»: каждый реальный (пользовательский) обучающий
# сигнал сверяется с текущим предсказанием модели. В /status отдаётся окно
# последних решений — по нему UI подсказывает, что ИИ можно отключить.
EVAL_WINDOW = 50    # сколько последних решений показываем
EVAL_KEEP = 200     # сколько храним в БД модели

TOKEN_RE = re.compile(r"[a-zа-яё0-9@+.#]+", re.I)
# Ссылки и markdown-ссылки не должны влиять ни на обучение, ни на предсказание:
# иначе модель учит мусор из URL (utm, source, campaign…) и режет по нему заявки.
_LINK_RE = re.compile(r"https?://[^\s<>\"']+|www\.[^\s<>\"']+|\[[^\]]*\]\([^)\s]+\)")

DATA_PATH = os.getenv("ML_DATA", "./data/ml.duckdb")
_lock = threading.RLock()
_con: duckdb.DuckDBPyConnection | None = None


def _adaptive_margin(total: float) -> float:
    """Отрыв от второго класса, требуемый для самостоятельного решения.

    Чем больше примеров ML уже видела, тем ниже порог — модель набирается
    опыта и постепенно заменяет ИИ на типовых сообщениях. На старте порог
    консервативный (0.9), после ~400 примеров — 0.35.
    """
    if total >= 400:
        return 0.35
    if total >= 150:
        return 0.5
    if total >= 60:
        return 0.7
    return MARGIN


def _db() -> duckdb.DuckDBPyConnection:
    global _con
    with _lock:
        if _con is None:
            os.makedirs(os.path.dirname(DATA_PATH) or ".", exist_ok=True)
            _con = duckdb.connect(DATA_PATH)
            _con.execute(
                "CREATE TABLE IF NOT EXISTS classes (label VARCHAR PRIMARY KEY, n DOUBLE NOT NULL DEFAULT 0, updated_at BIGINT)"
            )
            _con.execute(
                "CREATE TABLE IF NOT EXISTS terms (label VARCHAR NOT NULL, term VARCHAR NOT NULL, count DOUBLE NOT NULL DEFAULT 0, "
                "PRIMARY KEY (label, term))"
            )
            _con.execute(
                "CREATE TABLE IF NOT EXISTS eval_log (created_at BIGINT NOT NULL, "
                "expected VARCHAR NOT NULL, predicted VARCHAR NOT NULL DEFAULT '', correct BOOLEAN NOT NULL)"
            )
        return _con


def tokenize(text: str) -> list[str]:
    text = _LINK_RE.sub(" ", str(text or ""))
    words = [w.lower() for w in TOKEN_RE.findall(text)]
    out: list[str] = []
    for w in words:
        if len(w) >= 3:
            out.append(w)
        if len(w) >= 6:
            out.append("~" + w[:4])
    return out


def totals() -> dict[str, float]:
    with _lock:
        rows = _db().execute("SELECT label, n FROM classes WHERE n > 0").fetchall()
    return {r[0]: float(r[1]) for r in rows}


def ready() -> bool:
    t = totals()
    total = sum(t.values())
    if total < MIN_TOTAL:
        return False
    non_spam = {k: v for k, v in t.items() if k != "spam"}
    return t.get("spam", 0) >= MIN_WINNER_SPAM and sum(non_spam.values()) >= MIN_WINNER


def _upsert_one(db, label: str, text: str, delta: float) -> None:
    """Обновление одного обучающего примера (вызывается внутри транзакции).

    Термины пишутся пакетно (executemany), а не по одному INSERT — на большой
    модели построчная вставка занимает десятки секунд и блокирует /status и
    /predict, из-за чего сервис «выглядит недоступным».
    """
    label = str(label)
    if not text or not label:
        return
    now_ms = int(time.time() * 1000)
    db.execute(
        "INSERT INTO classes(label, n, updated_at) VALUES (?, ?, ?) "
        "ON CONFLICT(label) DO UPDATE SET n = classes.n + ?, updated_at = ?",
        [label, delta, now_ms, delta, now_ms],
    )
    terms = tokenize(text)
    if terms:
        db.executemany(
            "INSERT INTO terms(label, term, count) VALUES (?, ?, ?) "
            "ON CONFLICT(label, term) DO UPDATE SET count = terms.count + ?",
            [(label, t, delta, delta) for t in terms],
        )
    if delta < 0:
        db.execute("DELETE FROM terms WHERE label = ? AND count <= 0", [label])
        db.execute("DELETE FROM classes WHERE n <= 0", [])


def learn(label: str, text: str, delta: float = 1.0) -> None:
    """Увеличить вес класса/терминов (delta>0) или «разучить» (delta<0)."""
    _maybe_eval(label, text, delta)
    with _lock:
        db = _db()
        db.execute("BEGIN")
        try:
            _upsert_one(db, label, text, delta)
            db.execute("COMMIT")
        except Exception:
            db.execute("ROLLBACK")
            raise


def learn_batch(items: list[dict]) -> int:
    """Пакетное обучение: одна транзакция + пакетные вставки терминов."""
    if not items:
        return 0
    # самооценка по реальным действиям пользователя — до применения примеров
    for it in items:
        _maybe_eval(
            str(it.get("label") or ""),
            str(it.get("text") or ""),
            float(it.get("delta", 1.0)),
        )
    with _lock:
        db = _db()
        db.execute("BEGIN")
        try:
            for it in items:
                _upsert_one(
                    db,
                    str(it.get("label") or ""),
                    str(it.get("text") or ""),
                    float(it.get("delta", 1.0)),
                )
            db.execute("COMMIT")
        except Exception:
            db.execute("ROLLBACK")
            raise
    return len(items)


# Классы типа заявки (ML учит их по ИИ-решениям/действиям, чтобы со временем
# сам определять «занятость vs разовая сделка» без вызова ИИ)
TYPE_HIRE = "t:hire"
TYPE_ORDER = "t:order"
# минимум примеров типа, чтобы ML начал выдавать тип
MIN_TYPE_WINNER = 4


def predict(text: str) -> dict:
    tokens = tokenize(text)
    t = totals()
    if not t or not tokens:
        return {"take": False, "label": None, "scores": {}, "hits": 0, "ready": ready(), "type": None}
    # Модель «включается» только с опытом: пока примеров мало (ready=False),
    # она ничего не решает и не может ошибочно удалить заявку как спам.
    if not ready():
        return {"take": False, "label": None, "scores": {}, "hits": 0, "ready": False, "type": None}
    total = sum(t.values())
    type_classes = {k: v for k, v in t.items() if k in (TYPE_HIRE, TYPE_ORDER)}

    with _lock:
        db = _db()
        scores: dict[str, float] = {}
        hits: dict[str, int] = {}
        for label, n in t.items():
            rows = db.execute("SELECT term, count FROM terms WHERE label = ? AND count > 0", [label]).fetchall()
            weights = {r[0]: float(r[1]) for r in rows}
            score = 0.0
            hit = 0
            for term in set(tokens):
                w = weights.get(term)
                if w:
                    score += 1.0 if w < 1 else (1.0 + (w - 1.0) / (w + 1.0))
                    hit += 1
            if hit:
                scores[label] = score
                hits[label] = hit

    margin = _adaptive_margin(total)
    prior = {label: n / total for label, n in t.items()}

    # ── тип заявки: только t:hire / t:order ───────────────────────────────
    type_decision = None
    if len(type_classes) >= 2:
        ranked_t = sorted(
            ((k, scores.get(k, 0.0)) for k in type_classes),
            key=lambda kv: -kv[1],
        )
        bt_label, bt_score = ranked_t[0]
        st = ranked_t[1] if len(ranked_t) > 1 else None
        bt_total = bt_score + 3.0 * prior.get(bt_label, 0.0)
        st_total = (st[1] + 3.0 * prior.get(st[0], 0.0)) if st else 0.0
        if (
            bt_score > 0
            and t.get(bt_label, 0) >= MIN_TYPE_WINNER
            and (bt_total - st_total) >= margin
        ):
            type_decision = {
                "take": True,
                "label": "hire" if bt_label == TYPE_HIRE else "order",
                "value": bt_label,
                "margin": round(margin, 2),
            }

    # ── колонка/спам: без t:* классов ─────────────────────────────────────
    regular = {k: v for k, v in t.items() if not k.startswith("t:")}
    if not regular or not scores:
        return {
            "take": False, "label": None, "scores": {}, "hits": 0, "ready": ready(),
            "type": type_decision,
        }
    ranked = sorted(((k, scores[k]) for k in regular if k in scores), key=lambda kv: -kv[1])
    if not ranked:
        return {
            "take": False, "label": None, "scores": {}, "hits": 0, "ready": ready(),
            "type": type_decision,
        }
    best_label, best_score = ranked[0]
    second = ranked[1] if len(ranked) > 1 else None
    best_total = best_score + 3.0 * prior.get(best_label, 0.0)
    second_total = (second[1] + 3.0 * prior.get(second[0], 0.0)) if second else 0.0

    is_spam = best_label == "spam"
    min_winner = MIN_WINNER_SPAM if is_spam else MIN_WINNER
    take = (
        t.get(best_label, 0) >= min_winner
        and hits[best_label] >= MIN_HITS
        and (best_total - second_total) >= margin
    )
    # термины, которые модель «узнала» в тексте у класса-победителя:
    # подсказка для структурирования карточки (стек/услуги/материалы) без ИИ
    matched_terms: list[str] = []
    if take and best_label != "spam":
        with _lock:
            db = _db()
            rows = db.execute(
                "SELECT term, count FROM terms WHERE label = ? AND count > 0", [best_label]
            ).fetchall()
        weights = {r[0]: float(r[1]) for r in rows}
        seen = set()
        for term in tokens:
            if term.startswith("~"):
                continue  # хвостовой токен (~pyth) — не нужен как подсказка стека
            if term in weights and term not in seen and term not in best_label:
                seen.add(term)
                matched_terms.append(term)
        matched_terms = sorted(seen, key=lambda x: -weights.get(x, 0))[:8]
    out_scores = {label: round(s, 3) for label, s in sorted(scores.items(), key=lambda kv: -kv[1])[:5]}
    return {
        "take": bool(take),
        "label": best_label if take else None,
        "scores": out_scores,
        "hits": hits[best_label],
        "ready": ready(),
        "margin": round(margin, 2),
        "terms": matched_terms,
        "type": type_decision,
    }


def _maybe_eval(label: str, text: str, delta: float) -> None:
    """Самооценка перед обучением на реальном действии пользователя.

    delta=1.0 — пользовательское действие (перенос на доску, корзина, возврат,
    ручная разметка): это «правильный ответ по карточке». Если модель уже
    включена (ready) и уверенно взяла решение — сверяем его с действием:
    совпало → в копилку верных, нет → в копилку ошибок. Гипотезы ИИ
    (delta<1), типы t:* и «разучивание» (delta<0) не оцениваются.
    """
    if delta != 1.0 or not (text or "").strip() or str(label or "").startswith("t:"):
        return
    if not ready():
        return
    pr = predict(text)
    if not pr.get("take") or not pr.get("label"):
        return  # модель не уверена — такое сообщение ушло бы ИИ, не считаем ошибкой
    correct = bool(pr["label"] == label)
    with _lock:
        db = _db()
        db.execute(
            "INSERT INTO eval_log(created_at, expected, predicted, correct) VALUES (?, ?, ?, ?)",
            [int(time.time() * 1000), str(label), str(pr["label"]), correct],
        )
        db.execute(
            f"DELETE FROM eval_log WHERE created_at < "
            f"(SELECT created_at FROM eval_log ORDER BY created_at DESC LIMIT 1 OFFSET {EVAL_KEEP})"
        )


def status() -> dict:
    t = totals()
    # окно самооценки: последние решения модели, подтверждённые действиями
    # пользователя (перенос на доску, корзина, возврат, ручная разметка)
    ev = {"count": 0, "correct": 0, "accuracy": 0.0}
    with _lock:
        rows = _db().execute(
            "SELECT count(*) AS c, coalesce(sum(CASE WHEN correct THEN 1 ELSE 0 END), 0) AS ok "
            "FROM (SELECT correct FROM eval_log ORDER BY created_at DESC LIMIT ?)",
            [EVAL_WINDOW],
        ).fetchone()
        if rows and rows[0]:
            ev["count"] = int(rows[0])
            ev["correct"] = int(rows[1])
            ev["accuracy"] = round(ev["correct"] / ev["count"], 3) if ev["count"] else 0.0
    return {
        "ready": ready(),
        "classes": {label: round(n, 2) for label, n in sorted(t.items(), key=lambda kv: -kv[1])},
        "learned": int(sum(t.values())),
        "eval": ev,
    }


def reset() -> None:
    with _lock:
        db = _db()
        db.execute("DELETE FROM terms", [])
        db.execute("DELETE FROM classes", [])
        db.execute("DELETE FROM eval_log", [])