# -*- coding: utf-8 -*- """TU_KHO_TUVAN — biến các mục tư vấn ĐÃ DUYỆT trong MySQL thành `Doan`. Nguồn dữ liệu là bảng `chat_muc` (do một nhánh khác đang tạo) — mỗi dòng là một cặp hỏi/đáp do người soạn (không phải mô hình) viết sẵn, đã qua duyệt. Trích tương tự `kho/trich.py` trích từ mã PHP: cả hai đều lấy tri thức từ MỘT nguồn sống, không phải một bản chép tay rời rạc dễ trôi khỏi thực tế. CHỖ TRỐNG (PLACEHOLDER) — vì sao phải trung tính hoá trước khi đưa vào chỉ mục: văn bản trong `chat_muc.tra_loi_vi` / `tra_loi_en` chứa các chỗ trống kiểu `{gia:dcabot}`, `{link_tai:dcabot}`... được một module KHÁC điền lúc trả lời khách thật. Nhưng chỉ mục tìm kiếm (`kho/chi_muc.py`) đưa nguyên văn đoạn này cho mô hình ngôn ngữ ĐỌC để diễn giải câu trả lời — nếu còn `{...}` trong đó, mô hình sẽ chép nguyên cú pháp chỗ trống vào câu trả lời gửi khách (khách nhìn thấy `{gia:dcabot}` chứ không phải một con số). Vì vậy hàm `thanh_doan()` PHẢI gọi `trung_tinh_hoa()` trước khi ghép văn bản. Cách chạy tự kiểm (không cần CSDL thật): python kho/tu_kho_tuvan.py """ from __future__ import annotations import logging import os import re import sys from typing import Any, Dict, List, Optional try: from ..giao_uoc import Doan except ImportError: sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from giao_uoc import Doan try: from ..congcu import csdl except ImportError: _THU_MUC_CHATBOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if _THU_MUC_CHATBOT not in sys.path: sys.path.insert(0, _THU_MUC_CHATBOT) from congcu import csdl # ===================================================================== # ĐỌC MỤC ĐÃ DUYỆT TỪ MYSQL # ===================================================================== def doc_muc_duyet(khong_gian: str = 'tradingauto') -> List[dict]: """Các dòng `chat_muc` có trang_thai = 'duyet', đúng không gian đã cho. Trả [] và ghi logging.warning nếu thiếu CSDL / lỗi truy vấn — KHÔNG ném lỗi, để `nap_tu_kho_tuvan()` (và qua đó `dung_chi_muc()`) vẫn chạy được từ nguồn web khi không có CSDL. """ conn = csdl.ket_noi() if conn is None: # csdl.ket_noi() đã tự ghi log lý do (thiếu pymysql / thiếu cấu # hình / không nối được), không lặp lại cảnh báo ở đây. return [] try: with conn: with conn.cursor() as con_tro: con_tro.execute( """ SELECT id, khong_gian, san_pham, y_dinh, cau_hoi_chuan, cach_hoi_khac, tra_loi_vi, tra_loi_en, duong_dan_nguon, tieu_de_nguon, trang_thai, so_lan_dung FROM chat_muc WHERE trang_thai = %s AND khong_gian = %s ORDER BY id """, ('duyet', khong_gian), ) return list(con_tro.fetchall()) except Exception as loi: # noqa: BLE001 - biên CSDL, cố tình bắt rộng logging.warning('Lỗi khi đọc bảng chat_muc (kho tư vấn): %s', loi) return [] # ===================================================================== # TRUNG TÍNH HOÁ CHỖ TRỐNG # ===================================================================== # {ten_cho_trong} hoặc {ten_cho_trong:tham_so} — tham_so (vd mã sản phẩm # sau dấu ':') không dùng tới ở đây, chỉ ten_cho_trong quyết định thay gì. _MAU_CHO_TRONG = re.compile(r'\{([a-zA-Z_]+)(?::[^{}]*)?\}') _THAY_THEO_TEN: Dict[str, str] = { 'gia': 'có giá niêm yết tại /Pricing', 'bang_gia': 'có giá niêm yết tại /Pricing', 'link_tai': 'tải tại trang sản phẩm', 'ten': 'sản phẩm', 'zalo': 'kênh liên hệ của shop', 'telegram': 'kênh liên hệ của shop', } def trung_tinh_hoa(van_ban: str, nguon: str = '', lien_he: Optional[Dict[str, str]] = None) -> str: """Thay mọi chỗ trống `{...}` bằng chữ trung tính, để văn bản đưa vào chỉ mục KHÔNG còn dấu ngoặc nhọn nào (xem lý do ở đầu tệp). `lien_he` (liên hệ của shop): {zalo}/{telegram} điền SỐ THẬT thay vì chữ trung tính. Liên hệ không đổi theo giờ như giá nên điền lúc dựng chỉ mục được. Messenger 26/09/2026: mô hình chép nguyên "nhắn Zalo kênh liên hệ của shop" cho khách. `nguon` chỉ dùng để nêu rõ trong log khi gặp chỗ trống lạ — không ảnh hưởng kết quả trả về. """ if not van_ban: return van_ban def _thay(khop: 're.Match') -> str: ten_cho_trong = khop.group(1) if lien_he and lien_he.get(ten_cho_trong): return lien_he[ten_cho_trong] the_thay = _THAY_THEO_TEN.get(ten_cho_trong) if the_thay is not None: return the_thay logging.warning( 'Chỗ trống lạ không nhận ra: %r%s — đã xoá khỏi văn bản đưa vào chỉ mục.', khop.group(0), f' (nguồn: {nguon})' if nguon else '', ) return '' return _MAU_CHO_TRONG.sub(_thay, van_ban) # ===================================================================== # MỤC CSDL -> Doan # ===================================================================== def _lien_he_shop(khong_gian: Optional[str]) -> Dict[str, str]: """Liên hệ của shop sở hữu mục (shop/.py). Hỏng thì rỗng -> chữ trung tính như cũ.""" try: try: from .. import shop except ImportError: import shop # type: ignore return dict(shop.lay(khong_gian).lien_he) except Exception: # noqa: BLE001 return {} def thanh_doan(cac_muc: List[dict]) -> List[Doan]: """Mỗi dòng chat_muc -> một `Doan`. Bỏ qua (kèm cảnh báo) dòng thiếu id.""" ket_qua: List[Doan] = [] for muc in cac_muc: id_ = muc.get('id') if id_ is None: logging.warning('Bỏ qua một mục chat_muc thiếu cột id: %r', muc) continue nguon_tep = f'chat_muc#{id_}' cau_hoi_chuan = (muc.get('cau_hoi_chuan') or '').strip() cach_hoi_khac_tho = muc.get('cach_hoi_khac') or '' cac_cach_hoi = [dong.strip() for dong in cach_hoi_khac_tho.splitlines() if dong.strip()] tra_loi_vi_tho = (muc.get('tra_loi_vi') or '').strip() tra_loi_en_tho = (muc.get('tra_loi_en') or '').strip() lh = _lien_he_shop(muc.get('khong_gian')) cau_hoi_chuan_tt = trung_tinh_hoa(cau_hoi_chuan, nguon=f'{nguon_tep}.cau_hoi_chuan') cac_cach_hoi_tt = [ trung_tinh_hoa(dong, nguon=f'{nguon_tep}.cach_hoi_khac') for dong in cac_cach_hoi ] tra_loi_vi_tt = trung_tinh_hoa(tra_loi_vi_tho, nguon=f'{nguon_tep}.tra_loi_vi', lien_he=lh) tra_loi_en_tt = ( trung_tinh_hoa(tra_loi_en_tho, nguon=f'{nguon_tep}.tra_loi_en', lien_he=lh) if tra_loi_en_tho else '' ) # Ghép câu hỏi chuẩn + các cách hỏi khác + câu trả lời VI thành một # đoạn liền, mỗi phần một dòng — giữ đủ ngữ cảnh để BM25/vector # khớp được cả khi khách gõ một cách hỏi khác, không chỉ câu chuẩn. cac_phan = [p for p in ([cau_hoi_chuan_tt] + cac_cach_hoi_tt + [tra_loi_vi_tt]) if p] van_ban_vi = '\n'.join(cac_phan) san_pham = muc.get('san_pham') or None duong_dan = muc.get('duong_dan_nguon') or '/lien-he' tieu_de = muc.get('tieu_de_nguon') or cau_hoi_chuan ket_qua.append(Doan( ma='tuvan/%04d' % int(id_), van_ban_vi=van_ban_vi, van_ban_en=tra_loi_en_tt, san_pham=san_pham, loai='tuvan', duong_dan=duong_dan, tieu_de=tieu_de, nguon_tep=nguon_tep, so_tu=len(van_ban_vi.split()), )) return ket_qua def nap_tu_kho_tuvan(khong_gian: str = 'tradingauto') -> List[Doan]: """Gộp doc_muc_duyet() + thanh_doan(). [] nếu không có CSDL.""" return thanh_doan(doc_muc_duyet(khong_gian)) # ===================================================================== # TỰ KIỂM — không cần CSDL thật # ===================================================================== if __name__ == '__main__': logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s') print('=== 1) Tự kiểm trung_tinh_hoa() — mỗi loại chỗ trống một chuỗi ===') cac_truong_hop = [ 'Giá gói này {gia:dcabot} nhé anh/chị.', 'Xem {bang_gia} để biết chi tiết các gói.', '{ten:dcabot} có bản dùng thử miễn phí.', 'Anh/chị {link_tai:dcabot} về máy rồi cài theo hướng dẫn.', 'Liên hệ {zalo} nếu cần hỗ trợ gấp.', 'Hoặc nhắn {telegram} cũng được ạ.', 'Chỗ trống lạ: {khong_ro_gi_ca:xyz} không nằm trong danh sách đã biết.', ] for tc in cac_truong_hop: sau = trung_tinh_hoa(tc, nguon='tu_kiem') print(f' TRƯỚC: {tc}') print(f' SAU : {sau}') assert '{' not in sau, f'CÒN SÓT dấu ngoặc nhọn sau khi trung tính hoá: {sau!r}' print() print('=== 2) Tự kiểm thanh_doan() trên 2 mục giả lập (không cần CSDL) ===') muc_gia_lap = [ { 'id': 7, 'khong_gian': 'tradingauto', 'san_pham': 'dcabot', 'y_dinh': 'gia', 'cau_hoi_chuan': 'DCA Bot giá bao nhiêu?', 'cach_hoi_khac': 'giá dcabot bao nhiêu\nmua dca bot hết bao nhiêu tiền\n', 'tra_loi_vi': '{ten:dcabot} {gia:dcabot}. Anh/chị {link_tai:dcabot} để dùng thử.', 'tra_loi_en': '{ten:dcabot} {gia:dcabot}. You can {link_tai:dcabot} to try it.', 'duong_dan_nguon': '/toolintroduce/dcabot', 'tieu_de_nguon': 'Giới thiệu DCA Bot', 'trang_thai': 'duyet', }, { 'id': 12, 'khong_gian': 'tradingauto', 'san_pham': None, 'y_dinh': 'cam_ket', 'cau_hoi_chuan': 'Không dùng được thì liên hệ ai?', 'cach_hoi_khac': '', 'tra_loi_vi': 'Anh/chị nhắn {zalo} hoặc {telegram}, bên em hỗ trợ ngay.', 'tra_loi_en': '', 'duong_dan_nguon': '', 'tieu_de_nguon': '', 'trang_thai': 'duyet', }, ] cac_doan = thanh_doan(muc_gia_lap) for d in cac_doan: print(f' ma={d.ma} san_pham={d.san_pham} loai={d.loai} duong_dan={d.duong_dan}') print(f' tieu_de={d.tieu_de!r} nguon_tep={d.nguon_tep} so_tu={d.so_tu}') print(f' van_ban_vi={d.van_ban_vi!r}') print(f' van_ban_en={d.van_ban_en!r}') assert '{' not in d.van_ban_vi, 'CÒN SÓT { trong van_ban_vi' assert '{' not in d.van_ban_en, 'CÒN SÓT { trong van_ban_en' print() assert len(cac_doan) == 2 assert cac_doan[0].ma == 'tuvan/0007' assert cac_doan[1].duong_dan == '/lien-he', 'phải rơi về /lien-he khi duong_dan_nguon rỗng' print('OK — không còn ký tự { nào trong bất kỳ Doan nào ở trên.') print() print('=== 3) Tự kiểm nap_tu_kho_tuvan() khi KHÔNG có CSDL ===') cac_doan_that = nap_tu_kho_tuvan() print(f' Số Doan đọc từ CSDL thật: {len(cac_doan_that)} ' f'(kỳ vọng 0 trên máy không có CSDL — cảnh báo đã in ở log phía trên, ' f'chương trình KHÔNG bị sập).')