microdao-daarion/crews/agromatrix_crew/doc_focus.py

"""
doc_focus.py — Doc Focus Gate helpers (v3.5 / v3.6 / v3.7).

Без залежностей від crewai/agromatrix_tools — тільки re і stdlib.
Імпортується з run.py і operator_commands.py.

Публічні функції:
  _is_doc_question(text)            → bool
  _detect_domain(text, logger)      → str
  detect_context_signals(text)      → dict
  build_mode_clarifier(text)        → str
  handle_doc_focus(sub, chat_id)    → dict
"""
from __future__ import annotations

import re
import time

# ── Тригери: повідомлення явно про документ ──────────────────────────────────
_DOC_QUESTION_RE = re.compile(
    r"звіт|документ|таблиц|xlsx|sheet|рядок|колонк|в\s+звіті|у\s+файлі|у\s+документі"
    r"|по\s+звіту|з\s+(?:цього\s+)?файлу|в\s+цьому\s+документі|по\s+документу"
    r"|з\s+документа|відкрий\s+звіт",
    re.IGNORECASE | re.UNICODE,
)
# Фінансові тригери ТІЛЬКИ якщо є прив'язка до "документу/файлу"
_DOC_FINANCIAL_RE = re.compile(
    r"(?:прибуток|витрати?|собівартість|дохід|надходж|виручк|добрив|насінн|площ|гектар|грн|грн/га)"
    r".*(?:звіт|документ|файл|xlsx)|"
    r"(?:звіт|документ|файл|xlsx).*(?:прибуток|витрати?|дохід|грн|грн/га|площ)",
    re.IGNORECASE | re.UNICODE,
)

# ── Explicit doc-токени (перемагають vision) ─────────────────────────────────
_EXPLICIT_DOC_TOKEN_RE = re.compile(
    r"по\s+звіту|у\s+файлі|в\s+файлі|у\s+документі|в\s+документі|з\s+таблиц"
    r"|у\s+звіті|в\s+звіті|по\s+документу|з\s+документ|у\s+цьому\s+(?:файлі|звіті|документі)",
    re.IGNORECASE | re.UNICODE,
)

# ── Тригери що СКАСОВУЮТЬ doc-режим ──────────────────────────────────────────
_URL_RE = re.compile(r"https?://\S+", re.IGNORECASE)
_VISION_RE = re.compile(
    r"фото|картинк|зображенн|листя|плями|шкідник|хвороба|бур'ян|бурян"
    r"|рослин|гриб|гниль|хлороз|некроз|личинк|жук|кліщ|тля",
    re.IGNORECASE | re.UNICODE,
)
_ACTION_OPS_RE = re.compile(
    r"^(?:зроби|план|внеси|зафіксуй|перевір|порахуй|додай|видали|оновни|відкрий|нагадай)",
    re.IGNORECASE | re.UNICODE,
)
_WEB_INTENT_RE = re.compile(
    r"каталог|сайт|посиланн|переглянь\s+сторінк|вивч[иі]\s+каталог|знайди\s+на\s+сайт",
    re.IGNORECASE | re.UNICODE,
)

# ── v3.6: Fact-signal — числові запити без прив'язки до "звіту" ──────────────
_FACT_UNITS_RE = re.compile(
    r"грн|uah|₴|га\b|ha\b|%|грн/га|uah/ha|тис\.?|млн\.?|\d+\s*(?:грн|га|ha|%)",
    re.IGNORECASE | re.UNICODE,
)
_FACT_WORDS_RE = re.compile(
    r"прибуток|витрати?|виручка|дохід|маржа|площа|добрива|насіння|паливо|оренда|собівартість",
    re.IGNORECASE | re.UNICODE,
)

# ── v3.7: UX-фрази для заміни ────────────────────────────────────────────────
_DOC_AWARENESS_RE = re.compile(
    r"(так,\s*пам['\u2019]ятаю|не\s+бачу\s+його|не\s+бачу\s+перед\s+собою"
    r"|мені\s+(?:не\s+)?доступний\s+документ)",
    re.IGNORECASE | re.UNICODE,
)
_VISION_INTRO_RE = re.compile(
    r"^на\s+фото\s+видно",
    re.IGNORECASE | re.UNICODE,
)


def _is_doc_question(text: str) -> bool:
    """
    Rule-based: чи питання явно про документ/звіт.
    Explicit doc-токен перемагає vision-слова (скрін таблиці + caption).
    Fail-safe: будь-яка помилка → False.
    """
    try:
        t = text.strip()
        if _URL_RE.search(t):
            return False
        if _WEB_INTENT_RE.search(t):
            return False
        if _EXPLICIT_DOC_TOKEN_RE.search(t):
            return True
        if _VISION_RE.search(t):
            return False
        if _DOC_QUESTION_RE.search(t):
            return True
        if _DOC_FINANCIAL_RE.search(t):
            return True
        return False
    except Exception:
        return False


def _detect_domain(text: str, logger=None) -> str:
    """
    Визначає домен повідомлення.
    Повертає: "doc" | "vision" | "web" | "ops" | "general"

    Пріоритети:
      URL/web > explicit_doc_token > загальні doc-тригери > vision > ops > general
    Порожній текст (caption відсутній) → "vision".
    """
    try:
        t = text.strip()
        if not t:
            return "vision"
        if _URL_RE.search(t) or _WEB_INTENT_RE.search(t):
            return "web"
        if _EXPLICIT_DOC_TOKEN_RE.search(t):
            if _VISION_RE.search(t) and logger:
                try:
                    logger.info(
                        "AGX_STEPAN_METRIC domain_override from=vision to=doc reason=explicit_doc_tokens"
                    )
                except Exception:
                    pass
            return "doc"
        if _DOC_QUESTION_RE.search(t) or _DOC_FINANCIAL_RE.search(t):
            return "doc"
        if _VISION_RE.search(t):
            return "vision"
        if _ACTION_OPS_RE.search(t):
            return "ops"
        return "general"
    except Exception:
        return "general"


def detect_context_signals(text: str) -> dict:
    """
    v3.6: Повертає словник булевих сигналів для doc-mode gating.

    Ключі:
      has_explicit_doc_token: bool  — "по звіту", "у файлі" тощо
      has_doc_trigger:        bool  — загальні doc-тригери (звіт, документ)
      has_vision_trigger:     bool  — листя, шкідник, фото...
      has_url:                bool  — http(s)://...
      has_web_intent:         bool  — каталог, сайт...
      has_fact_signal:        bool  — числові одиниці або фін-слова
    """
    try:
        t = text.strip()
        return {
            "has_explicit_doc_token": bool(_EXPLICIT_DOC_TOKEN_RE.search(t)),
            "has_doc_trigger": bool(
                _DOC_QUESTION_RE.search(t) or _DOC_FINANCIAL_RE.search(t)
            ),
            "has_vision_trigger": bool(_VISION_RE.search(t)),
            "has_url": bool(_URL_RE.search(t)),
            "has_web_intent": bool(_WEB_INTENT_RE.search(t)),
            "has_fact_signal": bool(_FACT_UNITS_RE.search(t) or _FACT_WORDS_RE.search(t)),
        }
    except Exception:
        return {
            "has_explicit_doc_token": False, "has_doc_trigger": False,
            "has_vision_trigger": False, "has_url": False,
            "has_web_intent": False, "has_fact_signal": False,
        }


def build_mode_clarifier(text: str) -> str:
    """
    v3.6/v3.7: Одне контекстне уточнююче питання (без "!", без "будь ласка").

    URL    → "Ти про посилання чи про звіт?"
    vision → "Це про фото чи про цифри зі звіту?"
    facts  → "Це про конкретні цифри зі звіту?"
    інше   → "Йдеться про звіт чи про інше?"
    """
    try:
        t = text.strip()
        if _URL_RE.search(t):
            return "Ти про посилання чи про звіт?"
        if _VISION_RE.search(t):
            return "Це про фото чи про цифри зі звіту?"
        if _FACT_UNITS_RE.search(t) or _FACT_WORDS_RE.search(t):
            return "Це про конкретні цифри зі звіту?"
        return "Йдеться про звіт чи про інше?"
    except Exception:
        return "Йдеться про звіт чи про інше?"


def handle_doc_focus(sub: str, chat_id: str | None = None) -> dict:
    """
    /doc [on|off|status].

    /doc on     → doc_focus=True, TTL = DOC_FOCUS_TTL, cooldown скинутий
    /doc off    → doc_focus=False
    /doc status → поточний стан (focus, ttl_left, cooldown_left, active_doc_id, facts)
    """
    def _wrap(msg: str) -> dict:
        return {"ok": True, "message": msg}

    try:
        from crews.agromatrix_crew.session_context import (
            _STORE, DOC_FOCUS_TTL, is_doc_focus_active, load_session,
            is_doc_focus_cooldown_active,
        )
    except ImportError:
        return _wrap("session_context not available")

    if not chat_id:
        return _wrap("chat_id required for /doc command")

    now = time.time()

    if sub == "on":
        existing = _STORE.get(str(chat_id)) or {}
        existing["doc_focus"] = True
        existing["doc_focus_ts"] = now
        existing["doc_focus_cooldown_until"] = 0.0  # /doc on скидає cooldown
        _STORE[str(chat_id)] = existing
        doc_id = existing.get("active_doc_id") or "—"
        return _wrap(f"doc_focus=on. Документ: {str(doc_id)[:20]}. TTL={int(DOC_FOCUS_TTL)}с.")

    if sub == "off":
        existing = _STORE.get(str(chat_id)) or {}
        existing["doc_focus"] = False
        existing["doc_focus_ts"] = 0.0
        _STORE[str(chat_id)] = existing
        return _wrap("doc_focus=off. Степан відповідатиме без прив'язки до документа.")

    # status (default)
    session = load_session(str(chat_id))
    focus_active = is_doc_focus_active(session, now)
    cooldown_active = is_doc_focus_cooldown_active(session, now)
    doc_id = session.get("active_doc_id") or "—"
    doc_facts = session.get("doc_facts") or {}
    ttl_left = max(0.0, DOC_FOCUS_TTL - (now - (session.get("doc_focus_ts") or 0.0)))
    cooldown_left = max(0.0, (session.get("doc_focus_cooldown_until") or 0.0) - now)
    facts_keys = (
        ", ".join(k for k in doc_facts if k not in ("conflicts", "needs_recheck"))
        if doc_facts else "—"
    )
    cooldown_str = f" cooldown={int(cooldown_left)}с" if cooldown_active else ""
    return _wrap(
        f"doc_focus={'on' if focus_active else 'off'} "
        f"ttl_left={int(ttl_left)}с{cooldown_str} | "
        f"active_doc_id={str(doc_id)[:20]} | "
        f"facts=[{facts_keys}]"
    )