# -*- coding: utf-8 -*-
"""SAN_PHAM.PY — đọc danh mục và giá SỐNG từ MySQL, không bao giờ đóng băng.
VÌ SAO FILE NÀY TỒN TẠI (đọc kỹ trước khi sửa)
-----------------------------------------------
Bot cũ từng chép giá ra `product_data.json` rồi ngừng cập nhật. Tới lúc
phát hiện, tệp đó chỉ còn 2/7 sản phẩm và tên sản phẩm đã đổi từ lâu — bot
trả lời sai giá cho khách suốt một thời gian dài mà không ai biết.
Quy tắc ở đây: KHÔNG CÓ tệp trung gian nào lưu giá. Mọi lần gọi hàm là một
lần (hoặc đọc cache rất ngắn) chạm vào CSDL thật. Nếu sau này có ý định
"cho nhanh, cache luôn qua đêm" — ĐỪNG, đó chính là con đường dẫn tới lỗi
cũ. Cache ở đây chỉ dài 60 GIÂY, xem lý do ở hằng số THOI_GIAN_CACHE bên
dưới.
Cách chạy tự kiểm (không cần CSDL thật):
python congcu/san_pham.py
"""
from __future__ import annotations
import logging
import os
import sys
import time
import unicodedata
from typing import Dict, List, Optional
# ---------------------------------------------------------------------
# Nhập hợp đồng dữ liệu. Chịu được cả hai cách chạy:
# - chạy như một phần của gói (python -m services.chatbot.congcu.san_pham)
# - chạy trực tiếp tệp (python congcu/san_pham.py) — khi đó relative
# import thất bại vì không có package cha, phải tự thêm sys.path.
# ---------------------------------------------------------------------
try:
from ..giao_uoc import GiaGoi, SanPham
except ImportError:
_THU_MUC_CHA = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _THU_MUC_CHA not in sys.path:
sys.path.insert(0, _THU_MUC_CHA)
from giao_uoc import GiaGoi, SanPham
# =====================================================================
# CẤU HÌNH KẾT NỐI — đọc từ .env của website, KHÔNG viết cứng mật khẩu
# =====================================================================
# Thử lần lượt: máy phát triển (Windows/XAMPP) rồi tới máy chủ thật.
DUONG_DAN_ENV_UNG_VIEN = (
'G:/Xampp/htdocs/private/.env',
'/home/admin/web/tradingauto.org/private/.env',
)
def _doc_tep_env(duong_dan: str) -> Dict[str, str]:
"""Đọc thô một tệp .env kiểu KHOA=gia_tri. Không có thư viện ngoài
nào cho việc này để tránh phụ thuộc thêm — cú pháp .env ở đây đơn
giản, tự phân tích là đủ."""
ket_qua: Dict[str, str] = {}
try:
with open(duong_dan, encoding='utf-8') as tep:
for dong in tep:
dong = dong.strip()
if not dong or dong.startswith('#') or '=' not in dong:
continue
khoa, _, gia_tri = dong.partition('=')
ket_qua[khoa.strip()] = gia_tri.strip().strip('"').strip("'")
except OSError:
# Không có tệp ở đường dẫn này — bình thường, thử đường dẫn kế tiếp.
return {}
return ket_qua
def _cau_hinh_db() -> Optional[Dict[str, str]]:
"""Gộp cấu hình: biến môi trường ĐÈ LÊN giá trị đọc từ .env, vì máy
chủ thật thường set biến môi trường trực tiếp (an toàn hơn tệp)."""
tep_env: Dict[str, str] = {}
for duong_dan in DUONG_DAN_ENV_UNG_VIEN:
tep_env = _doc_tep_env(duong_dan)
if tep_env:
break
def lay(khoa: str) -> Optional[str]:
return os.environ.get(khoa) or tep_env.get(khoa)
host, ten_db, user = lay('DB_HOST'), lay('DB_NAME'), lay('DB_USER')
mat_khau = lay('DB_PASS') or ''
if not (host and ten_db and user):
return None
return {'host': host, 'name': ten_db, 'user': user, 'pass': mat_khau}
# =====================================================================
# ĐỆM (CACHE) 60 GIÂY — CHỦ Ý NGẮN
# =====================================================================
# Vì sao 60 giây chứ không phải 0 hay vô hạn:
# - 0 giây (không cache): một cuộc hội thoại hỏi giá + hỏi lại + so
# sánh sản phẩm có thể chạm CSDL cả chục lần trong vài giây, tốn kết
# nối vô ích.
# - Vô hạn (như bot cũ): đúng là thứ đã giết bot cũ. TUYỆT ĐỐI không
# làm lại.
# - 60 giây: đổi giá trong trang quản trị thì CHẬM NHẤT một phút sau
# bot đã biết — đủ nhanh để không ai để ý, đủ dài để đỡ tải CSDL.
THOI_GIAN_CACHE_GIAY = 60
_dem: Dict[str, object] = {'thoi_diem': 0.0, 'du_lieu': None}
def mo_ta_ngan(html: str, toi_da: int = 360) -> str:
"""Đoạn ĐẦU TIÊN của product.description, bỏ thẻ HTML — đó là câu
tóm tắt tính năng chủ website tự viết cho trang sản phẩm.
Vì sao cần (23/09/2026): khách hỏi "bên bạn bán những sản phẩm nào", bot
liệt kê 8 sản phẩm nhưng SMART DCA EA và MT5 Manager chỉ có tên trống —
tầng tìm kiếm không kéo được đoạn nào về hai sản phẩm đó, và câu dẫn cấm
mô hình tự nói tính năng riêng khi không có tài liệu. Danh mục có mô tả
này đi vào MỌI lượt tư vấn (nao/tra_loi.py::_khoi_danh_muc)."""
import html as _html
import re as _re
if not html:
return ''
m = _re.search(r']*>(.*?)
', html, _re.IGNORECASE | _re.DOTALL)
doan = m.group(1) if m else html
chu = _re.sub(r'<[^>]+>', ' ', doan)
chu = _re.sub(r'\s+', ' ', _html.unescape(chu)).strip()
chu = _re.sub(r'\s+([,.;:!?…])', r'\1', chu) # thẻ bỏ đi để lại dấu cách trước dấu câu
if len(chu) > toi_da:
chu = chu[:toi_da].rsplit(' ', 1)[0].rstrip(',;:') + '…'
return chu
def _danh_sach_song_khong_cache() -> List[SanPham]:
"""Đọc thẳng từ MySQL, không qua cache. Trả [] và ghi log cảnh báo
tiếng Việt nếu bất cứ bước nào lỗi — bot thà không biết còn hơn sập."""
try:
import pymysql # import trễ: máy không cài thư viện này vẫn chạy được các phần khác
except ImportError:
logging.warning(
'Chưa cài thư viện pymysql nên không đọc được danh mục sản phẩm. '
'Cài bằng lệnh: pip install pymysql'
)
return []
cau_hinh = _cau_hinh_db()
if not cau_hinh:
logging.warning(
'Không tìm thấy cấu hình CSDL (thiếu DB_HOST/DB_NAME/DB_USER trong '
'biến môi trường hoặc tệp .env) — trả danh mục sản phẩm rỗng.'
)
return []
try:
ket_noi = pymysql.connect(
host=cau_hinh['host'],
user=cau_hinh['user'],
password=cau_hinh['pass'],
database=cau_hinh['name'],
charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor,
connect_timeout=5,
)
except Exception as loi: # noqa: BLE001 - cố tình bắt rộng, đây là biên mạng
logging.warning('Không kết nối được CSDL sản phẩm: %s', loi)
return []
try:
with ket_noi:
with ket_noi.cursor() as con_tro:
con_tro.execute(
"""
SELECT code, slug, title, title_vi, license_backend, version, description
FROM product
WHERE status = 1
-- VipTrend (8 san pham sig:, 28/09/2026) da co khoi gia gom trong
-- HoSoShop.bang_gia_them - lay ca o day thi bang gia bi lap 8 dong
AND IFNULL(license_backend, '') NOT LIKE 'sig:%%'
ORDER BY sort_order
"""
)
cac_hang_san_pham = con_tro.fetchall()
ds: List[SanPham] = []
for hang in cac_hang_san_pham:
con_tro.execute(
"""
SELECT months, price, label
FROM product_price
WHERE product_code = %s AND status = 1
ORDER BY sort_order
""",
(hang['code'],),
)
cac_goi = [
GiaGoi(
thang=int(gia['months']),
gia=float(gia['price']),
# months = 0 nghĩa là VĨNH VIỄN (đúng theo lược đồ product_price)
nhan=gia.get('label') or (
'Vĩnh viễn' if gia['months'] == 0 else f"{gia['months']} tháng"
),
)
for gia in con_tro.fetchall()
]
ds.append(
SanPham(
ma=hang['code'],
slug=hang.get('slug') or '',
ten=hang.get('title') or '',
ten_vi=hang.get('title_vi') or hang.get('title') or '',
backend=hang.get('license_backend') or '',
phien_ban=hang.get('version') or '',
cac_goi=cac_goi,
mo_ta=mo_ta_ngan(hang.get('description') or ''),
)
)
return ds
except Exception as loi: # noqa: BLE001
logging.warning('Lỗi khi đọc danh mục/giá sản phẩm từ CSDL: %s', loi)
return []
_dem_vip: Dict[str, object] = {'du_lieu': None, 'thoi_diem': 0.0}
def bang_gia_viptrend() -> Optional[tuple]:
"""Khối giá VipTrend cho HoSoShop.bang_gia_them, ĐỌC TỪ BẢNG SẢN PHẨM (chủ dự án 28/09/2026:
"đọc giá thẳng từ bảng sản phẩm") — 8 sản phẩm license_backend sig:, gói 12 tháng,
khung lấy theo kênh (stream_kenh), combo theo product.combo_voi / gia_combo. Đệm 60 giây.
Trả (tên, (dòng giá, ...), dòng gọn) hoặc None nếu đọc lỗi / chưa có sản phẩm."""
bay_gio = time.time()
if _dem_vip['du_lieu'] is not None and (bay_gio - float(_dem_vip['thoi_diem'])) < THOI_GIAN_CACHE_GIAY:
return _dem_vip['du_lieu'] # type: ignore[return-value]
try:
import pymysql
ch = _cau_hinh_db()
if not ch:
return None
with pymysql.connect(host=ch['host'], user=ch['user'], password=ch['pass'], database=ch['name'],
charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor, connect_timeout=5) as kn:
with kn.cursor() as ct:
ct.execute("""
SELECT p.title_vi, p.url_demo, p.license_backend, p.combo_voi, p.gia_combo, pp.price,
k.khung, c.title_vi AS combo_ten
FROM product p
JOIN product_price pp ON pp.product_code = p.code AND pp.months = 12 AND pp.status = 1
LEFT JOIN stream_kenh k ON CONCAT('sig:', k.ten_tin_hieu) = p.license_backend
LEFT JOIN product c ON c.license_backend = p.combo_voi AND c.status = 1
WHERE p.status = 1 AND p.license_backend LIKE 'sig:%%'
ORDER BY p.sort_order""")
hang = ct.fetchall()
if not hang:
return None
ten_ngan = lambda t: (t or '').split(' — ')[0].replace('VipTrend ', '') # noqa: E731
so = lambda v: f'{float(v):g}' # noqa: E731
dong, gon = [], []
for h in hang:
ten = ten_ngan(h['title_vi'])
them = ''
if h['gia_combo'] is not None and h['combo_ten']:
them = f", đang thuê {ten_ngan(h['combo_ten'])} còn hạn thì chỉ {so(h['gia_combo'])}"
dong.append(f"{ten} ({h['khung'] or '?'}): {so(h['price'])} credit/năm{them} — {h['url_demo']}")
gon.append(f"{ten} {so(h['price'])}")
dong += ['Chỉ thuê theo năm, không có gói theo tháng.',
'Thuê trực tiếp trên web: /thue-bot — key hiện ngay ở Đơn hàng; một key dùng chung cho mọi con VIP khách thuê.',
'Chạy bằng EA VipTrend (tải về sau khi mua) hoặc ngay trong MT5 Manager miễn phí, tab VIPTREND — không cần cài EA.']
kq = ('VipTrend — bot vàng XAU/USD chạy sẵn trên MT5 (thuê theo năm, 1 kênh = 1 tài khoản MT5)', tuple(dong),
'VipTrend (thuê theo năm): ' + ', '.join(gon) + ' credit')
_dem_vip['du_lieu'], _dem_vip['thoi_diem'] = kq, bay_gio
return kq
except Exception as loi: # noqa: BLE001
logging.warning('Không đọc được giá VipTrend từ CSDL: %s', loi)
return _dem_vip['du_lieu'] # type: ignore[return-value]
def danh_sach_san_pham() -> List[SanPham]:
"""Danh mục sản phẩm đang mở bán, kèm giá — đọc từ CSDL, đệm 60 giây."""
bay_gio = time.time()
if _dem['du_lieu'] is not None and (bay_gio - float(_dem['thoi_diem'])) < THOI_GIAN_CACHE_GIAY:
return _dem['du_lieu'] # type: ignore[return-value]
ds = _danh_sach_song_khong_cache()
if ds:
# Chỉ ghi đè cache khi đọc THÀNH CÔNG. Đọc lỗi thì không đụng vào
# cache cũ — để lần gọi sau còn cơ hội thử lại CSDL ngay, thay vì
# bị một cache rỗng "khoá" trong 60 giây tiếp theo.
_dem['du_lieu'] = ds
_dem['thoi_diem'] = bay_gio
return ds
# =====================================================================
# TÌM KIẾM — chịu được gõ thiếu dấu, gõ tên cũ
# =====================================================================
def _bo_dau(chuoi: str) -> str:
"""Chuẩn hoá về chữ thường, bỏ dấu tiếng Việt, để so khớp khoan dung.
Xử lý riêng đ/Đ vì Unicode NFD không tách nó thành d + dấu phụ."""
if not chuoi:
return ''
chuoi = chuoi.replace('đ', 'd').replace('Đ', 'D')
chuoi = unicodedata.normalize('NFD', chuoi)
chuoi = ''.join(ky_tu for ky_tu in chuoi if unicodedata.category(ky_tu) != 'Mn')
return chuoi.lower().strip()
# Tên cũ / cách gọi dân dã khách hay gõ -> mã sản phẩm thật (product.code).
# Khoá đã được chuẩn hoá bỏ dấu sẵn, vì _bo_dau() cũng áp lên câu khách gõ.
ANH_XA_TEN_CU: Dict[str, str] = {
'chep lenh': 'signaltrader',
'copy signal': 'signaltrader',
'phat tin': 'signalbroadcaster',
'quang dca': 'dcabot',
'smart dca': 'dcabot',
}
def tim_san_pham(tu_khoa: str) -> Optional[SanPham]:
"""Tìm một sản phẩm theo code/slug/title/title_vi, khoan dung dấu và
tên cũ. Trả None nếu không khớp gì — gọi nơi khác nên chuyển sang
`cau_khong_ro()` trong khuon_mau.py, KHÔNG được đoán bừa."""
if not tu_khoa:
return None
khoa = _bo_dau(tu_khoa)
ds = danh_sach_san_pham()
if not ds:
return None
# 1) Tên cũ / dân dã — khớp trước vì đây là ý định rõ ràng nhất.
for ten_cu, ma in ANH_XA_TEN_CU.items():
if ten_cu in khoa:
for sp in ds:
if sp.ma == ma:
return sp
# 2) Khớp đúng nguyên mã hoặc slug.
for sp in ds:
if khoa == _bo_dau(sp.ma) or khoa == _bo_dau(sp.slug):
return sp
# 3) Tên hiển thị (VI/EN) chứa từ khoá hoặc ngược lại — khách có thể
# gõ cả câu ("giá con dca bot bao nhiêu") hoặc chỉ viết tắt.
for sp in ds:
ten_vi, ten_en = _bo_dau(sp.ten_vi), _bo_dau(sp.ten)
if (ten_vi and (khoa in ten_vi or ten_vi in khoa)) or (
ten_en and (khoa in ten_en or ten_en in khoa)
):
return sp
# 4) Mã/slug là một phần của câu khách gõ (vd: "cho em hỏi giá dcabot đi").
# CẢNH BÁO: khớp DÀI TRƯỚC NGẮN SAU như giao_uoc.py đã dặn, để
# tránh lỗi cũ 'dca' khớp nhầm vào 'broa-dca-ster'.
for sp in sorted(ds, key=lambda s: -len(s.ma)):
if _bo_dau(sp.ma) in khoa or _bo_dau(sp.slug) in khoa:
return sp
return None
def gia_theo_goi(ma_san_pham: str) -> List[GiaGoi]:
"""Danh sách gói giá của một sản phẩm theo mã. [] nếu không có/không tìm thấy."""
for sp in danh_sach_san_pham():
if sp.ma == ma_san_pham:
return sp.cac_goi
return []
# =====================================================================
# TỰ KIỂM — không cần CSDL thật
# =====================================================================
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
print('=== Tự kiểm san_pham.py (máy này không có pymysql / không nối được CSDL) ===')
print()
ds = danh_sach_san_pham()
if not ds:
print(
'(Danh mục rỗng — ĐÚNG như kỳ vọng trên máy này: thiếu pymysql hoặc\n'
' không nối được CSDL. Cảnh báo tiếng Việt đã in ở log phía trên, và\n'
' quan trọng nhất — chương trình KHÔNG bị sập.)'
)
else:
for sp in ds:
print(f"{sp.ma:20} {sp.ten_vi:30} backend={sp.backend}")
for goi in sp.cac_goi:
print(f" - {goi.nhan}: {goi.gia} credit")
print()
print('--- Kiểm tra tìm kiếm không dấu/tên cũ vẫn chạy êm dù danh mục rỗng ---')
for tu_khoa in ['dca bot', 'chep lenh', 'smart dca', 'copy signal', 'khong ton tai']:
try:
kq = tim_san_pham(tu_khoa)
print(f" tim_san_pham({tu_khoa!r}) -> {kq.ma if kq else None}")
except Exception as loi: # noqa: BLE001
print(f" LỖI KHÔNG MONG MUỐN với {tu_khoa!r}: {loi}")