webapi/bot.py

"""

webapi/bot.py — ครอบ chat_interaction() ของ chatbot_v5 เดิม โดยไม่แก้ gate / dialogue



สัญญาที่ใช้ (อ่านจาก run_testcases.py ซึ่งเทสต์ผ่าน 35/35 และ 33/33):



    chat_interaction(text, image, history, last_pids, session_id, last_vit, last_graph)

      -> (history, _state_out, refs, last_pids, session_id, last_vit, last_graph)



    history = [{"role": "user"|"assistant", "content": str | ...}, ...]  (Gradio messages)



Backend เก็บ 5 ค่าที่บอทต้องใช้ต่อเทิร์น (history, last_pids, session_id, last_vit, last_graph)

ไว้ฝั่งเซิร์ฟเวอร์ต่อ session — เบราว์เซอร์ถือแค่ session id

refs (แผง debug เดิม) ไม่ส่งไปหน้าเว็บ — เก็บแค่ Decision ไว้ใน log



การกู้บทสนทนา (เพิ่ม 4 ต.ค.)

----------------------------

SessionStore อยู่ในหน่วยความจำ จึงหายเมื่อ restart · sleep · ครบ TTL · ล้น LRU

เบราว์เซอร์เก็บประวัติไว้เองใน localStorage แล้วส่ง resume กลับมาเมื่อ session หาย

ของที่ส่งกลับมามีแค่สองอย่างและเล็กทั้งคู่



    messages      ข้อความล้วนของทั้งสองฝ่าย ไม่มีพาธรูป

    dialog_state  dict ของ DialogState (brand · codes · slots)



ไม่ส่ง last_vit / last_graph / history ดิบกลับมา เพราะ history ดิบมี {"path": ...}

ของรูป drawing ซึ่งเป็นพาธจริงในเครื่องเซิร์ฟเวอร์ — MediaRegistry มีไว้กันไม่ให้

เบราว์เซอร์เห็นพาธนั้นอยู่แล้ว ส่งออกไปคือยกเลิกการป้องกันของตัวเอง



state ที่ client ส่งมาถือเป็นข้อมูลที่เชื่อไม่ได้ ต้องผ่าน sanitize_dialog_state() ก่อน

ค่าสเปกปลอมไม่ได้อยู่แล้วเพราะเกตอ่านจาก modutech_v30_3.json ใหม่ทุกเทิร์น

สิ่งที่ปลอมได้คือรหัสรุ่นกับ slot ซึ่งทำให้คำตอบของตัวเองผิด ไม่กระทบผู้ใช้คนอื่น

"""
from __future__ import annotations

import importlib
import logging
import os
import re
import secrets
import threading
import time
from collections import OrderedDict
from dataclasses import dataclass, field
from typing import Any, Callable, Optional

log = logging.getLogger("beltbot")


# ═══════════════════════════════════════════════════════════════
# session
# ═══════════════════════════════════════════════════════════════

@dataclass
class BotSession:
    id: str
    history: list = field(default_factory=list)
    last_pids: list = field(default_factory=list)
    bot_session_id: str = ""
    last_vit: Any = None
    last_graph: Any = None
    turns: int = 0
    touched: float = field(default_factory=time.time)
    lock: threading.Lock = field(default_factory=threading.Lock, repr=False)


class SessionStore:
    """LRU + TTL ในหน่วยความจำ · id สุ่ม 128 บิต เดาไม่ได้"""

    def __init__(self, ttl_s: int, max_sessions: int):
        self.ttl_s, self.max_sessions = ttl_s, max_sessions
        self._items: "OrderedDict[str, BotSession]" = OrderedDict()
        self._lock = threading.Lock()

    def get_or_create(self, sid: Optional[str]) -> tuple[BotSession, bool]:
        now = time.time()
        with self._lock:
            s = self._items.get(sid or "")
            if s is not None and now - s.touched <= self.ttl_s:
                s.touched = now
                self._items.move_to_end(s.id)
                return s, False
            if s is not None:
                del self._items[s.id]
            s = BotSession(id=secrets.token_urlsafe(16))
            self._items[s.id] = s
            while len(self._items) > self.max_sessions:
                self._items.popitem(last=False)
            return s, True

    def drop(self, sid: str) -> bool:
        with self._lock:
            return self._items.pop(sid, None) is not None

    def sweep(self) -> int:
        now = time.time()
        with self._lock:
            dead = [k for k, s in self._items.items() if now - s.touched > self.ttl_s]
            for k in dead:
                del self._items[k]
            return len(dead)

    def __len__(self) -> int:
        return len(self._items)


# ═══════════════════════════════════════════════════════════════
# แปลงข้อความของบอท → ข้อความสำหรับหน้าเว็บ
# ═══════════════════════════════════════════════════════════════

_IMG_SRC = re.compile(r"""<img[^>]*\bsrc\s*=\s*["']([^"']+)["']""", re.IGNORECASE)
_TAG = re.compile(r"<[^>]+>")
# รหัสภายในของ Movex เช่น movex_chain_LF820_K325 / movex_sprocket_57801 / Movex_Sprocket_880_z10_b25
_INTERNAL_ID = re.compile(r"\b[Mm]ovex_(?:chain|sprocket|Sprocket|Chain)_([A-Za-z0-9_]+)\b")


def scrub_internal_ids(text: str) -> str:
    """

    GRAPH-001: บอทหลุดรหัสภายในบางครั้ง — แก้ที่ชั้นแสดงผลเท่านั้น (ไม่แตะบอท)

    movex_chain_LF880_TAB_K325 → LF880 TAB K325 · movex_sprocket_57801 → 57801

    """
    return _INTERNAL_ID.sub(lambda m: m.group(1).replace("_", " "), text)


class MediaRegistry:
    """

    บอทส่งรูป drawing กลับเป็น <img src="path"> หรือ {"path": ...}

    เบราว์เซอร์ต้องไม่เห็น path จริง → ออก token แทน และเสิร์ฟได้เฉพาะไฟล์ใต้ MEDIA_ROOTS

    """

    def __init__(self, roots: list[str], base_dir: str):
        self.roots = [os.path.realpath(os.path.join(base_dir, r)) for r in roots]
        self.base_dir = base_dir
        self._tokens: dict[str, str] = {}
        self._by_path: dict[str, str] = {}
        self._lock = threading.Lock()

    def register(self, src: str) -> Optional[str]:
        if src.startswith(("http://", "https://", "data:image/")):
            return src                                   # ส่งต่อได้ตรง ๆ
        path = src[len("file="):] if src.startswith("file=") else src
        path = os.path.realpath(path if os.path.isabs(path) else os.path.join(self.base_dir, path))
        if not any(path == r or path.startswith(r + os.sep) for r in self.roots) or not os.path.isfile(path):
            log.warning("media ไม่อยู่ใน MEDIA_ROOTS หรือไม่มีไฟล์ — ไม่ส่งไปหน้าเว็บ: %s", src)
            return None
        with self._lock:
            tok = self._by_path.get(path)
            if tok is None:
                tok = secrets.token_urlsafe(12)
                self._tokens[tok], self._by_path[path] = path, tok
        return f"/media/{tok}"

    def resolve(self, token: str) -> Optional[str]:
        return self._tokens.get(token)


def to_web_messages(msgs: list, media: MediaRegistry, scrub: bool) -> list[dict]:
    out: list[dict] = []
    for m in msgs:
        if not isinstance(m, dict) or m.get("role") != "assistant":
            continue
        c = m.get("content")
        if isinstance(c, dict) and c.get("path"):
            url = media.register(str(c["path"]))
            if url:
                out.append({"type": "image", "url": url, "alt": c.get("alt_text") or ""})
            continue
        if isinstance(c, (list, tuple)) and c and isinstance(c[0], str) and not c[0].startswith("<"):
            url = media.register(c[0])                   # รูปแบบ tuple (path, alt) ของ Gradio รุ่นเก่า
            if url:
                out.append({"type": "image", "url": url, "alt": c[1] if len(c) > 1 else ""})
            continue
        if not isinstance(c, str):
            continue
        for src in _IMG_SRC.findall(c):
            url = media.register(src)
            if url:
                out.append({"type": "image", "url": url, "alt": ""})
        text = _TAG.sub("", c) if "<img" in c.lower() else c
        text = text.strip()
        if text:
            out.append({"type": "text", "text": scrub_internal_ids(text) if scrub else text})
    return out


_DECISION = re.compile(r"\*\*Decision:\*\*\s*(\w+)")

# ═══════════════════════════════════════════════════════════════
# กู้บทสนทนาจากประวัติที่เบราว์เซอร์เก็บไว้
# ═══════════════════════════════════════════════════════════════

RESUME_MAX_TURNS = 8          # เทิร์นล่าสุดที่เอากลับมา — Gemini ไม่ได้ใช้เกินนี้
RESUME_MAX_CHARS = 1200       # ต่อหนึ่งข้อความ กันไม่ให้ payload โตเกิน
RESUME_MAX_CODES = 8
RESUME_MAX_SLOTS = 12

# slot ที่ dialogue_state รู้จัก — ค่าอื่นทิ้ง
_SLOT_KEYS = ("wet_or_dry", "material", "line_shape", "operating_temp_c")


def rebuild_history(messages: Any) -> list:
    """ข้อความจาก localStorage → history แบบที่ chat_interaction รับได้



    คืนเฉพาะ role user/assistant ที่เป็นข้อความล้วน รูปข้ามไป

    (รูปในแชตเก่าโหลดไม่ขึ้นอยู่แล้วเพราะ token ของ MediaRegistry หมดอายุ)

    """
    out: list = []
    if not isinstance(messages, list):
        return out
    for m in messages[-(RESUME_MAX_TURNS * 2):]:
        if not isinstance(m, dict):
            continue
        role = m.get("role")
        text = m.get("text")
        if role not in ("user", "assistant") or not isinstance(text, str):
            continue
        text = text.strip()[:RESUME_MAX_CHARS]
        if text:
            out.append({"role": role, "content": text})
    return out


def sanitize_dialog_state(raw: Any, mod: Any) -> dict:
    """กรอง DialogState ที่ client ส่งมา — แต่ละแบรนด์ตรวจรหัสของตัวเอง (ชั้น 6)



    ไม่งั้น client ส่งรหัสมั่วมาแล้วบอตได้ products = [] ซึ่งดูเหมือน

    "เล่มไม่มีข้อมูล" ทั้งที่จริงคือรหัสไม่มีอยู่



    ขั้นที่นี่ทำให้ทุกแบรนด์ (ไม่รู้จักแบรนด์ไหน)

      list  เก็บเฉพาะ string ไม่เกิน RESUME_MAX_CODES · dict ไม่เกิน RESUME_MAX_SLOTS คู่

      brand ต้องเป็นแบรนด์ที่มีปลั๊กอิน · pending ไม่กู้ (บริบทเดิมหายไปแล้ว)

    ขั้นของแบรนด์: plugin.sanitize_state() — รหัสนี้มีจริงไหม slot ไหนรู้จัก

    โมดูลที่ไม่มีปลั๊กอิน (fakebot) ใช้ทางเดิม _sanitize_legacy

    """
    if not isinstance(raw, dict):
        return {}
    plugins = getattr(mod, "plugins", None)
    state_cls = getattr(mod, "DialogState", None)
    if not plugins or state_cls is None:
        return _sanitize_legacy(raw, mod)

    def bound(v):
        if isinstance(v, list):
            return [x for x in v if isinstance(x, str)][:RESUME_MAX_CODES]
        if isinstance(v, dict):
            return dict(list(v.items())[:RESUME_MAX_SLOTS])
        return v

    brand = raw.get("brand")
    out: dict = {"brand": brand if brand in plugins else None}
    for key, p in plugins.items():
        pre = f"{key}_"
        fields = {k[len(pre):]: bound(v) for k, v in raw.items() if isinstance(k, str) and k.startswith(pre)}
        for f, v in p.sanitize_state(fields).items():
            out[pre + f] = v
    out["pending"] = None
    out["pending_question"] = ""
    return state_cls.from_dict(out).to_dict()


def state_resumable(state: dict, mod: Any) -> bool:
    """มีของพอให้กู้ไหม — แบรนด์ที่รู้แล้ว หรือแบรนด์ใดมีรุ่นในความสนใจ"""
    if not state:
        return False
    plugins = getattr(mod, "plugins", None)
    if not plugins:
        return any(state.get(k) for k in ("brand", "modutech_codes", "movex_pids"))
    if state.get("brand"):
        return True
    for key, p in plugins.items():
        pre = f"{key}_"
        if p.resumable({k[len(pre):]: v for k, v in state.items() if k.startswith(pre)}):
            return True
    return False


def _sanitize_legacy(raw: Any, mod: Any) -> dict:
    """ทางเดิม — ใช้กับโมดูลที่ไม่มีปลั๊กอิน (fakebot ของ test_resume.py)

    กรอง DialogState ที่ client ส่งมา — เก็บเฉพาะที่มีจริงในฐานข้อมูล



    สองชั้น

      1. DialogState.from_dict กรองคีย์ที่คลาสไม่รู้จักออก (แก้ไว้ 1 ต.ค.)

      2. ชั้นนี้ตัดรหัสรุ่นที่ไม่มีในแคตตาล็อกและ slot ที่ไม่รู้จักออก

         ไม่งั้น client ส่งรหัสมั่วมาแล้วบอตได้ products = [] ซึ่งดูเหมือน

         "เล่มไม่มีข้อมูล" ทั้งที่จริงคือรหัสไม่มีอยู่

    """
    if not isinstance(raw, dict):
        return {}

    by_code = getattr(getattr(mod, "modutech_cat", None), "by_code", {}) or {}
    product_db = getattr(mod, "product_db", {}) or {}

    def strs(v, limit):
        return [x for x in v if isinstance(x, str)][:limit] if isinstance(v, list) else []

    codes = [c for c in strs(raw.get("modutech_codes"), RESUME_MAX_CODES) if c in by_code]
    pids = [p for p in strs(raw.get("movex_pids"), RESUME_MAX_CODES) if p in product_db]

    slots_in = raw.get("modutech_slots")
    slots: dict = {}
    if isinstance(slots_in, dict):
        for k, v in list(slots_in.items())[:RESUME_MAX_SLOTS]:
            if k in _SLOT_KEYS and isinstance(v, (str, int, float)) and not isinstance(v, bool):
                slots[k] = v

    brand = raw.get("brand")
    series = raw.get("movex_series")
    out = {
        "brand": brand if brand in ("movex", "modutech") else None,
        "modutech_codes": codes,
        "modutech_slots": slots,
        "modutech_sprocket_codes": [],      # ของเทิร์นเดียว ไม่สะสมข้ามเทิร์น
        "movex_pids": pids,
        "movex_series": series if isinstance(series, str) and len(series) <= 16 else None,
        "pending": None,                    # คำถามที่ค้างไว้ไม่กู้ — บริบทเดิมหายไปแล้ว
        "pending_question": "",
    }
    # ให้คลาสเป็นผู้ตัดสินว่าฟิลด์ไหนมีจริง เผื่อ DialogState เปลี่ยนในอนาคต
    try:
        from dialogue_state import DialogState
        return DialogState.from_dict(out).to_dict()
    except Exception:                                   # noqa: BLE001
        return out


# ═══════════════════════════════════════════════════════════════
# ปุ่มตอบเร็ว (27 ต.ค. · ผู้ใช้เลือกทาง ข: backend ส่งฟิลด์เพิ่ม)
# ═══════════════════════════════════════════════════════════════
#
# ฟิลด์ใหม่ใน response — เพิ่มอย่างเดียว ไม่เปลี่ยนสิ่งที่บอทตัดสินใจ
#     suggestions: [{"label": "½ นิ้ว", "text": "12.7 มม."}, ...]   ว่าง = ไม่มีปุ่ม
# label/text เป็นข้อความไทยที่ผู้ใช้เห็นหรือพิมพ์เองได้อยู่แล้ว ไม่มีชื่อฟิลด์ภายในหรือ path
# ถามแบรนด์ = ทำที่นี่ (แกนกลาง) · คำถามของแบรนด์ = plugin.reply_suggestions(pending, fields) ถ้ามี
# ไม่รู้จักแบรนด์ใด (test_webapi_sanitize ตรวจว่าไม่มีชื่อแบรนด์ในไฟล์นี้)

SUGGEST_MAX = 8
SUGGEST_LABEL_MAX = 40
SUGGEST_TEXT_MAX = 80
_BRAND_CONFIRM = "brand:"


def suggestions_for(state: Any, mod: Any) -> list[dict]:
    if not isinstance(state, dict):
        return []
    pending = state.get("pending")
    plugins = getattr(mod, "plugins", None)
    if not isinstance(pending, str) or not pending or not plugins:
        return []
    pairs: list = []
    if pending == "brand":
        pairs = [(f"{p.info.name} ({p.info.kind_th})", p.info.name)
                 for p in sorted(plugins.values(), key=lambda p: p.info.order)]
    elif pending.startswith(_BRAND_CONFIRM):
        pairs = [("ใช่", "ใช่"), ("ไม่ใช่", "ไม่ใช่")]
    else:
        for key, p in plugins.items():
            fn = getattr(p, "reply_suggestions", None)
            if fn is None:
                continue
            pre = f"{key}_"
            fields = {k[len(pre):]: v for k, v in state.items() if isinstance(k, str) and k.startswith(pre)}
            try:
                pairs = list(fn(pending, fields) or [])
            except Exception:                           # noqa: BLE001 — ปุ่มเป็นของเสริม พังแล้วไม่มีปุ่ม ไม่ล้มเทิร์น
                log.exception("สร้างปุ่มตอบเร็วไม่สำเร็จ")
                pairs = []
            if pairs:
                break
    out = []
    for pair in pairs[:SUGGEST_MAX]:
        if not (isinstance(pair, (tuple, list)) and len(pair) == 2):
            continue
        label, text = (str(x).strip() for x in pair)
        if label and text and len(label) <= SUGGEST_LABEL_MAX and len(text) <= SUGGEST_TEXT_MAX:
            out.append({"label": label, "text": text})
    return out


# ═══════════════════════════════════════════════════════════════
# adapter
# ═══════════════════════════════════════════════════════════════

class BotAdapter:
    def __init__(self, module_name: str, chat_fn: Optional[Callable] = None):
        self.module_name = module_name
        self._fn = chat_fn
        self._load_lock = threading.Lock()
        self.load_error: Optional[str] = None
        self.versions: dict = {}                    # webapi/versions.collect — เก็บครั้งเดียวหลังโหลด

    @property
    def ready(self) -> bool:
        return self._fn is not None

    def load(self) -> None:
        """import chatbot_v5 ครั้งเดียว (โหลดโมเดล ViT/SigLIP/Jina + Qdrant) — ช้า จึงทำตอนเริ่ม"""
        with self._load_lock:
            if self._fn is not None:
                return
            t0 = time.time()
            try:
                mod = importlib.import_module(self.module_name)
                self._fn = getattr(mod, "chat_interaction")
                log.info("โหลด %s เสร็จใน %.1f วินาที", self.module_name, time.time() - t0)
                try:                                # อ่านอย่างเดียว · พังก็ไม่ทำให้บอทใช้ไม่ได้
                    from webapi.versions import collect
                    self.versions = collect(mod)
                except Exception:                   # noqa: BLE001
                    log.exception("เก็บข้อมูลเวอร์ชันไม่สำเร็จ")
            except Exception as e:
                self.load_error = f"{type(e).__name__}: {e}"
                log.exception("โหลดบอทไม่สำเร็จ")
                raise

    def _module(self) -> Any:
        return importlib.import_module(self.module_name)

    def seed_session(self, s: BotSession, resume: Any) -> dict:
        """ใส่ประวัติกับสถานะที่เบราว์เซอร์ส่งมาเข้า session ที่สร้างใหม่



        ไม่แก้ chatbot_v5 แม้บรรทัดเดียว — ใช้สองช่องทางที่มีอยู่แล้ว

          · history  ส่งเข้าทางพารามิเตอร์ของ chat_interaction ตามปกติ

          · state    เขียนลง _dialog_states ด้วย id ที่ขอจาก logger.new_session()

                     ซึ่งเป็นกลไกเดียวกับที่บอตใช้สร้าง id เอง (chatbot_v5 บรรทัด 833)



        คืน dict บอกว่ากู้อะไรได้บ้าง ไว้เขียน log

        """
        if self._fn is None:
            self.load()
        mod = self._module()
        info = {"messages": 0, "state": False}

        s.history = rebuild_history((resume or {}).get("messages"))
        info["messages"] = len(s.history)

        state = sanitize_dialog_state((resume or {}).get("dialog_state"), mod)
        if state_resumable(state, mod):
            sid = mod.logger.new_session()
            mod._dialog_states[sid] = state
            s.bot_session_id = sid
            info["state"] = True
        return info

    def current_state(self, s: BotSession) -> dict:
        """DialogState ปัจจุบันของ session — ส่งกลับให้เบราว์เซอร์เก็บไว้กู้รอบหน้า



        แยกเป็นเมธอดต่างหาก ไม่เปลี่ยน return ของ turn() เพราะ run_testcases.py

        ใช้สัญญาเดิมอยู่ (35/35 และ 33/33)

        """
        if not s.bot_session_id or self._fn is None:
            return {}
        try:
            return dict(self._module()._dialog_states.get(s.bot_session_id) or {})
        except Exception:                               # noqa: BLE001
            return {}

    def suggestions(self, s: BotSession) -> list[dict]:
        """ปุ่มตอบเร็วของคำถามที่บอทถามค้างใน session นี้ — ไม่มีคำถามค้าง = []"""
        if self._fn is None:
            return []
        try:
            return suggestions_for(self.current_state(s), self._module())
        except Exception:                               # noqa: BLE001
            log.exception("ปุ่มตอบเร็วผิดพลาด")
            return []

    def turn(self, s: BotSession, text: str, image) -> tuple[list, str]:
        """เรียกบอทหนึ่งเทิร์น คืน (ข้อความใหม่ของ assistant ในเทิร์นนี้, decision)"""
        if self._fn is None:
            self.load()
        before = len(s.history)
        result = self._fn(text, image, list(s.history), list(s.last_pids),
                          s.bot_session_id, s.last_vit, s.last_graph)
        history, _state_out, refs, last_pids, bot_sid, last_vit, last_graph = result
        history = list(history or [])
        new = history[before:] if len(history) >= before else history[-2:]
        s.history, s.last_pids, s.bot_session_id = history, list(last_pids or []), bot_sid or ""
        s.last_vit, s.last_graph = last_vit, last_graph
        s.turns += 1
        m = _DECISION.search(refs or "") if isinstance(refs, str) else None
        return new, (m.group(1) if m else "")