| """
|
| 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")
|
|
|
|
|
|
|
|
|
|
|
|
|
| @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"<[^>]+>")
|
|
|
| _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])
|
| 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
|
| RESUME_MAX_CHARS = 1200
|
| RESUME_MAX_CODES = 8
|
| RESUME_MAX_SLOTS = 12
|
|
|
|
|
| _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": "",
|
| }
|
|
|
| try:
|
| from dialogue_state import DialogState
|
| return DialogState.from_dict(out).to_dict()
|
| except Exception:
|
| return out
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 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:
|
| 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
|
|
|
|
|
|
|
|
|
|
|
|
|
| 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 = {}
|
|
|
| @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:
|
| 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:
|
| 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:
|
| 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 "")
|
|
|
