# -*- coding: utf-8 -*- """CHỈ MỤC — dựng và đọc kho tri thức từ MỘT tệp SQLite. VÌ SAO một tệp SQLite thay vì thư mục JSON hay dịch vụ ngoài: ~500 đoạn là quy mô đọc-nhiều-ghi-hiếm điển hình — cả kho nằm gọn trong RAM của SQLite, không cần một tiến trình server riêng (khác Postgres/Milvus, vốn là một dependency vận hành nữa trên VPS vốn đã chạy web + bot). Một tệp cũng dễ sao lưu/rollback: copy một tệp là xong, không phải dump nhiều bảng. VÌ SAO ghi ra tệp TẠM rồi đổi tên đè thay vì ghi thẳng vào chi_muc.sqlite: dựng chỉ mục mất vài giây (quét 59 tệp PHP, có thể gọi cả model nhúng). Nếu tiến trình bị giết giữa chừng (deploy, OOM, Ctrl-C) trong lúc ghi thẳng, tầng tìm kiếm đang chạy sẽ đọc phải một chỉ mục DỞ DANG — thiếu nửa bảng doan, hoặc bảng có mà chưa commit xong. os.replace() là thao tác NGUYÊN TỬ trên cùng một ổ đĩa (cả POSIX lẫn Windows), nên tại mọi thời điểm chi_muc.sqlite hoặc là bản CŨ toàn vẹn, hoặc là bản MỚI toàn vẹn, không bao giờ ở giữa. """ from __future__ import annotations import hashlib import io import json import logging import os import sqlite3 import sys from array import array from datetime import datetime, timezone 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 .trich import trich_tat_ca, DUONG_DAN_SITE_MAC_DINH from .cat_doan import cat_tat_ca from .nhung import BoNhung, NhungRong from .tu_kho_tuvan import nap_tu_kho_tuvan except ImportError: sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from trich import trich_tat_ca, DUONG_DAN_SITE_MAC_DINH from cat_doan import cat_tat_ca from nhung import BoNhung, NhungRong from tu_kho_tuvan import nap_tu_kho_tuvan # Chi muc nam ngay canh ma, suy tu vi tri tep chu KHONG viet cung o dia: # /private/services/chatbot/kho/chi_muc.py -> len 2 bac la thu muc # chatbot. Viet cung duong dan Windows thi tren may chu Linux sqlite bao # 'unable to open database file' — thong bao khong nhac gi toi duong dan # nen rat de tuong la loi quyen. DUONG_DAN_MAC_DINH = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'chi_muc.sqlite') # ===================================================================== # LƯỢC ĐỒ # ===================================================================== _TAO_BANG_DOAN = """ CREATE TABLE doan ( ma TEXT PRIMARY KEY, van_ban_vi TEXT NOT NULL, van_ban_en TEXT NOT NULL DEFAULT '', san_pham TEXT, loai TEXT, duong_dan TEXT NOT NULL, tieu_de TEXT NOT NULL, nguon_tep TEXT NOT NULL, so_tu INTEGER NOT NULL DEFAULT 0, kieu TEXT, vector BLOB ) """ _TAO_BANG_SIEU_DU_LIEU = """ CREATE TABLE sieu_du_lieu ( khoa TEXT PRIMARY KEY, gia_tri TEXT NOT NULL ) """ def _vector_ra_blob(vec: List[float]) -> Optional[bytes]: """Nhét vector thẳng vào BLOB dạng float32 nhị phân. VÌ SAO float32 (array 'f') chứ không phải JSON text hay float64: một vector bge-m3 có 1024 chiều — JSON text tốn ~10-15 byte/số (dấu phẩy, dấu chấm, ký tự số), float32 nhị phân tốn đúng 4 byte/số và đọc lại KHÔNG cần parse. float64 thì thừa độ chính xác so với embedding (vốn vẫn chỉ tính bằng float32 lúc suy luận), tốn gấp đôi chỗ mà không lợi thêm gì. Rỗng ([]) thì lưu NULL, không lưu mảng rỗng — phân biệt rõ "đoạn này chưa có vector" khỏi "đoạn này có vector 0 chiều". """ if not vec: return None return array('f', vec).tobytes() def _blob_ra_vector(blob: Optional[bytes]) -> List[float]: if not blob: return [] a = array('f') a.frombytes(blob) return a.tolist() def _tinh_phien_ban(cac_doan: List[Doan]) -> str: """Hash nội dung toàn bộ chỉ mục — dùng làm 'dấu vân tay' của một lần dựng. VÌ SAO cần: tầng LLM đệm câu trả lời theo (câu hỏi, phien_ban) — chỉ mục đổi (sửa giá, thêm trang mới) thì phien_ban đổi theo, đệm cũ tự hết hiệu lực mà không cần dọn tay hay đặt TTL đoán mò. """ bam = hashlib.sha256() for d in cac_doan: # Nối theo `ma` để thứ tự luôn ổn định (không phụ thuộc thứ tự # duyệt list trong bộ nhớ), và gộp thêm văn bản để đổi NỘI DUNG # một đoạn (sửa câu chữ, không đổi mã) cũng đổi phien_ban. bam.update(d.ma.encode('utf-8')) bam.update(b'\x00') bam.update(d.van_ban_vi.encode('utf-8')) bam.update(b'\x00') bam.update(d.van_ban_en.encode('utf-8')) bam.update(b'\x01') return bam.hexdigest() def _dem_tuvan_chi_muc_cu(duong_dan: str) -> int: """Đếm số đoạn loai='tuvan' trong chỉ mục ĐANG CÓ trên đĩa. Trả 0 khi chưa có tệp, tệp hỏng, hay thiếu bảng — mọi trường hợp "không biết" đều trả 0, để KHÔNG chặn nhầm lần dựng đầu tiên. """ if not os.path.isfile(duong_dan): return 0 try: k = sqlite3.connect(duong_dan) try: return int(k.execute( "select count(*) from doan where loai='tuvan'").fetchone()[0]) finally: k.close() except Exception: # noqa: BLE001 - không biết thì coi như 0, xem docstring return 0 def _ghi_sqlite(duong_dan_tam: str, cac_doan: List[Doan], cac_vector: List[List[float]], so_chieu_vector: int, sieu_du_lieu_them: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: if os.path.exists(duong_dan_tam): os.remove(duong_dan_tam) conn = sqlite3.connect(duong_dan_tam) try: conn.execute(_TAO_BANG_DOAN) conn.execute(_TAO_BANG_SIEU_DU_LIEU) conn.executemany( """INSERT INTO doan (ma, van_ban_vi, van_ban_en, san_pham, loai, duong_dan, tieu_de, nguon_tep, so_tu, kieu, vector) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", [ (d.ma, d.van_ban_vi, d.van_ban_en, d.san_pham, d.loai, d.duong_dan, d.tieu_de, d.nguon_tep, d.so_tu, d.kieu, _vector_ra_blob(v)) for d, v in zip(cac_doan, cac_vector) ], ) phien_ban = _tinh_phien_ban(cac_doan) sieu_du_lieu = { 'dung_luc': datetime.now(timezone.utc).isoformat(), 'so_doan': len(cac_doan), 'so_chieu_vector': so_chieu_vector, 'phien_ban': phien_ban, } if sieu_du_lieu_them: sieu_du_lieu.update(sieu_du_lieu_them) conn.executemany( "INSERT INTO sieu_du_lieu (khoa, gia_tri) VALUES (?, ?)", [(k, json.dumps(v, ensure_ascii=False)) for k, v in sieu_du_lieu.items()], ) conn.commit() finally: conn.close() return sieu_du_lieu # ===================================================================== # DỰNG CHỈ MỤC # ===================================================================== def tep_chi_muc(khong_gian: Optional[str] = None) -> str: """Tệp chỉ mục của một shop (shop/): TradingAuto giữ chi_muc.sqlite, shop khác chi_muc_.sqlite.""" import shop return shop.lay(khong_gian).tep_chi_muc(os.path.dirname(DUONG_DAN_MAC_DINH)) def dung_chi_muc(bo_nhung: Optional[BoNhung] = None, duong_dan: Optional[str] = None, khong_gian: Optional[str] = None) -> Dict[str, Any]: """Trích -> cắt đoạn -> (nhúng nếu có) -> ghi SQLite (qua tệp tạm). Trả về đúng phần sieu_du_lieu vừa ghi, để nơi gọi (vd script triển khai) in log/kiểm tra ngay mà không phải mở lại tệp. """ import shop hs = shop.lay(khong_gian) duong_dan = duong_dan or tep_chi_muc(hs.ma) if bo_nhung is None: bo_nhung = NhungRong() # mặc định: chưa chọn mô hình -> chạy thuần BM25 # Nguồn web CHỈ của shop có trang web khai trong hồ sơ — shop khác không bao giờ # nhận tài liệu web của TradingAuto. if hs.thu_muc_site == '__mac_dinh__': cac_doan_web = cat_tat_ca(trich_tat_ca()) elif hs.thu_muc_site: cac_doan_web = cat_tat_ca(trich_tat_ca(hs.thu_muc_site)) else: cac_doan_web = [] # Kho tư vấn (bảng chat_muc, MySQL) là nguồn THỨ HAI, GỘP THÊM vào # nguồn web — không THAY nguồn web. Không có CSDL (máy dev chưa cấu # hình .env, hoặc CSDL tạm thời không tới được) thì vẫn phải dựng chỉ # mục êm ru từ mỗi nguồn web như trước giờ, chỉ ghi cảnh báo. try: cac_doan_tuvan = nap_tu_kho_tuvan(hs.ma) except Exception as loi: # noqa: BLE001 - không để một nguồn phụ làm hỏng cả lần dựng logging.warning( 'Không nạp được kho tư vấn (chat_muc) — chỉ mục sẽ chỉ có nguồn web. Lỗi: %s', loi) cac_doan_tuvan = [] cac_doan = cac_doan_web + cac_doan_tuvan # CHỐT CHẶN: chỉ mục rỗng là kiểu hỏng nguy hiểm nhất của cả hệ. # # Nó không nổ lỗi, không ghi log gì — vẫn in "Xong", vẫn tạo tệp # SQLite hẳn hoi. Dịch vụ khởi động bình thường, /khoe báo "song", rồi # bot trả lời khách "em chưa tìm được tài liệu" cho MỌI câu hỏi. # # Đã xảy ra thật ở lần deploy đầu: trên máy chủ thư mục web tên là # `public_html` chứ không phải `public`, nên không quét được tệp nào. # Thà dừng ồn ào ở đây còn hơn để bot sống mà vô dụng. if not cac_doan: raise RuntimeError( 'Dung chi muc ra 0 doan (ca web lan tu van) — KHONG ghi de chi muc cu.\n' ' Da quet web: %s\n' ' Kiem: thu muc do co ton tai khong, va trong do co tep .php nao khong.\n' ' Tren may chu thu muc web ten la `public_html`, o may local la `public`.' % DUONG_DAN_SITE_MAC_DINH) # CHỐT CHẶN 2: tầng tri thức ĐÃ DUYỆT biến mất trong im lặng. # # Xảy ra thật ngày 23/09/2026 trên máy chủ: `python3` ở đó KHÔNG có # pymysql, nên `nap_tu_kho_tuvan()` ném lỗi, nhánh except ở trên nuốt # lỗi thành một dòng logging.warning, rồi lần dựng ghi đè chỉ mục cũ # bằng một chỉ mục KHÔNG CÒN MỤC TƯ VẤN NÀO. Lệnh dựng vẫn in "Xong" # và báo 395 đoạn trông khoẻ mạnh — đúng 13 mục chủ nhà tự tay duyệt # (tầng 1, trả lời 0đ, ~50ms) thì bay sạch, không một tiếng động. # # Khác chốt chặn 1 ở chỗ: 0 đoạn thì dễ thấy, còn "vẫn còn 395 đoạn # nhưng mất 13 mục tốt nhất" thì không ai nhìn ra từ con số tổng. # # Vì sao SO SÁNH với chỉ mục cũ chứ không đòi "phải có > 0": trên máy # dev không cấu hình CSDL thì 0 -> 0 là BÌNH THƯỜNG, chặn cứng ở đây # sẽ làm mọi người không dựng được chỉ mục. Chỉ chặn đúng trường hợp # ĐANG CÓ mà thành KHÔNG CÒN. if not cac_doan_tuvan: _cu = _dem_tuvan_chi_muc_cu(duong_dan) if _cu > 0 and not os.environ.get('CHO_PHEP_MAT_TUVAN'): raise RuntimeError( 'Chi muc CU co %d muc tu van da duyet, lan dung NAY ra 0 ' '— KHONG ghi de.\n' ' Gan nhu chac chan la khong noi duoc CSDL, khong phai ' 'chu nha da xoa muc.\n' ' Kiem theo thu tu:\n' ' 1. python3 -c "import pymysql" — thieu thi cai: ' 'pip3 install pymysql\n' ' 2. tep .env co day du thong so CSDL khong\n' ' 3. mysql co dang chay khong\n' ' Neu chu nha THAT SU da xoa het muc tu van, chay lai voi:\n' ' CHO_PHEP_MAT_TUVAN=1 python3 kho/chi_muc.py' % _cu) if bo_nhung.so_chieu > 0 and cac_doan: cac_vector = bo_nhung.nhung([d.van_ban_vi for d in cac_doan]) else: cac_vector = [[] for _ in cac_doan] # Đếm số đoạn MANG THEO neo (đường dẫn có '#') — đo trực tiếp việc # VIỆC 1 (cắt theo neo) có tác dụng thật hay không, không phải suy # gián tiếp qua tổng số đoạn. so_doan_co_neo = sum(1 for d in cac_doan_web if '#' in d.duong_dan) duong_dan_tam = duong_dan + '.tmp' sieu_du_lieu = _ghi_sqlite( duong_dan_tam, cac_doan, cac_vector, bo_nhung.so_chieu, sieu_du_lieu_them={ 'so_doan_web': len(cac_doan_web), 'so_doan_tuvan': len(cac_doan_tuvan), 'so_doan_co_neo': so_doan_co_neo, }, ) # os.replace: nguyên tử trên cùng ổ đĩa — xem lý do ở đầu tệp. os.replace(duong_dan_tam, duong_dan) return sieu_du_lieu # ===================================================================== # ĐỌC CHỈ MỤC # ===================================================================== def doc_tat_ca(duong_dan: Optional[str] = None) -> List[Doan]: """Đọc toàn bộ bảng doan thành list[Doan]. KHÔNG trả vector — Doan (giao_uoc.py) không có trường vector, chỉ tầng tìm kiếm mới cần đọc thẳng cột BLOB đó bằng SQL của riêng nó. """ duong_dan = duong_dan or DUONG_DAN_MAC_DINH conn = sqlite3.connect(duong_dan) try: cur = conn.execute( """SELECT ma, van_ban_vi, van_ban_en, san_pham, loai, duong_dan, tieu_de, nguon_tep, so_tu, kieu FROM doan ORDER BY ma""" ) return [ Doan( ma=r[0], van_ban_vi=r[1], van_ban_en=r[2], san_pham=r[3], loai=r[4], duong_dan=r[5], tieu_de=r[6], nguon_tep=r[7], so_tu=r[8], kieu=r[9], ) for r in cur.fetchall() ] finally: conn.close() def doc_sieu_du_lieu(duong_dan: Optional[str] = None) -> Dict[str, Any]: duong_dan = duong_dan or DUONG_DAN_MAC_DINH conn = sqlite3.connect(duong_dan) try: cur = conn.execute("SELECT khoa, gia_tri FROM sieu_du_lieu") ket_qua = {} for khoa, gia_tri in cur.fetchall(): try: ket_qua[khoa] = json.loads(gia_tri) except (TypeError, ValueError): ket_qua[khoa] = gia_tri return ket_qua finally: conn.close() # ===================================================================== # CHẠY TRỰC TIẾP # ===================================================================== if __name__ == '__main__': from collections import Counter # python kho/chi_muc.py [ma_shop] — mặc định TradingAuto (như trước khi tách lõi) import sys as _sys _kg = _sys.argv[1] if len(_sys.argv) > 1 else None print('Đang dựng chỉ mục...', _kg or 'tradingauto') thong_ke = dung_chi_muc(khong_gian=_kg) print('Xong.') print() print(f"Số đoạn từ web: {thong_ke.get('so_doan_web')} | " f"Số đoạn từ kho tư vấn: {thong_ke.get('so_doan_tuvan')} | " f"Số đoạn CÓ NEO (đường dẫn có '#'): {thong_ke.get('so_doan_co_neo')} | " f"Tổng: {thong_ke.get('so_doan')}") print() for k, v in thong_ke.items(): print(f'{k}: {v}') print() print('--- Đọc lại để kiểm tra ---') cac_doan = doc_tat_ca() print(f'Số đoạn đọc lại: {len(cac_doan)}') dem_sp = Counter(d.san_pham or '(dùng chung / không rõ)' for d in cac_doan) print() print('--- Phân bố theo sản phẩm ---') for sp, sl in dem_sp.most_common(): print(f'{sp:30} {sl:5d}')