# -*- coding: utf-8 -*- """BAI_NGAY.PY — mỗi ngày tự SOẠN một bài Fanpage, gửi thẻ Telegram cho chủ shop duyệt (02/10/2026). VÌ SAO ------ Chủ dự án: "làm phần soạn bài đăng hằng ngày trên fanpage trước, chúng ta đang thiếu ở chỗ này". Từ 25/09 chỉ có 5 bài, soạn tay bằng skill dang-bai. Tệp này lo phần SOẠN tự động; phần DUYỆT + ĐĂNG giữ nguyên ở kenh/bai_dang.py (thẻ ✅ Đăng / ❌ Bỏ, trả lời vào thẻ để sửa). KHÔNG có đường nào đăng bài mà không qua nút của chủ shop. LUỒNG ----- 1. Luồng nền (10 phút một vòng): shop nào bật `bd_bat` = '1' trong chat_cau_hinh và tới giờ `bd_gio` (giờ VN) mà hôm nay chưa soạn (`bd_ngay`) -> soan_va_gui(). 2. Nhóm bài theo thứ trong tuần: shop/.py `bai_ngay['lich']` (vd thứ Hai sản phẩm, thứ Ba kiến thức…). 3. Trong nhóm, chọn NGUỒN lâu chưa dùng nhất (chat_bai_dang.nguon) -> không lặp chủ đề. 4. SỰ THẬT: chữ hiển thị trên trang nguồn (kho/trich.quet_su_kien — đúng thứ khách đọc trên web) và/hoặc mục kho tư vấn đã duyệt. Mô hình chỉ được viết theo đó; nhóm kiến thức dùng kiến thức chung của nghề, không nêu số liệu riêng của sản phẩm. 5. Kiểm sau khi soạn (bỏ markdown, link tuyệt đối, chữ cấm, độ dài) -> gửi thẻ duyệt (bai_dang.gui_duyet). Công tắc chat_cau_hinh: bd_bat ('0'), bd_gio ('8'), bd_ngay (tự ghi). CLI: python kenh/bai_ngay.py thu [nhóm] — soạn và in, KHÔNG gửi python kenh/bai_ngay.py gui [nhóm] — soạn và gửi thẻ duyệt Tự kiểm (không cần CSDL, không gọi mạng): python kenh/bai_ngay.py """ from __future__ import annotations import json import logging import os import re import sys import threading import time from datetime import date, datetime, timedelta, timezone from typing import Any, Callable, Dict, List, Optional, Tuple try: from congcu import csdl 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 congcu import csdl # type: ignore _log = logging.getLogger('chatbot.kenh.bai_ngay') CHU_KY_GIAY = 600 VN = timedelta(hours=7) TEN_NHOM = {'san_pham': 'Sản phẩm', 'kien_thuc': 'Kiến thức', 'hoi_dap': 'Hỏi đáp', 'bang_chung': 'Số liệu công khai', 'dich_vu': 'Dịch vụ'} TEN_THU = ('Thứ Hai', 'Thứ Ba', 'Thứ Tư', 'Thứ Năm', 'Thứ Sáu', 'Thứ Bảy', 'Chủ nhật') _CAM = re.compile(r'passview|x2 tài khoản|nhân đôi tài khoản|lãi chắc chắn|chắc chắn có lãi|cam kết lãi|' r'cam kết lợi nhuận|lợi nhuận ổn định mỗi tháng|không bao giờ thua|không thể thua', re.I) SO_CHU = (110, 380) # ===================================================================== # THUẦN (tự kiểm được) # ===================================================================== def _bay_gio_vn() -> datetime: return datetime.now(timezone.utc).replace(tzinfo=None) + VN def nhom_hom_nay(lich: Tuple[str, ...], ngay: date) -> str: return lich[ngay.weekday() % len(lich)] if lich else 'san_pham' def chon_nguon(ung_vien: List[Dict[str, Any]], da_dung: Dict[str, datetime], ngay: date) -> Optional[Dict[str, Any]]: """Nguồn chưa dùng (hoặc dùng lâu nhất) trước; cùng hạng thì xoay theo ngày để không luôn lấy cái đầu.""" if not ung_vien: return None n = len(ung_vien) xoay = ngay.toordinal() % n thu_tu = ung_vien[xoay:] + ung_vien[:xoay] return min(thu_tu, key=lambda x: da_dung.get(x['khoa'], datetime.min)) def thay_bien_kho(van_ban: str, lien_he: Dict[str, str], ten_mien: str, trang_bang_gia: str) -> str: """Mục kho dùng biến {zalo} {telegram} {gia:x} {ten:x} (xem public/pages/site/faq.php) -> chữ thường.""" s = str(van_ban or '') s = s.replace('{zalo}', 'Zalo ' + lien_he.get('zalo', '')).replace('{telegram}', 'Telegram ' + lien_he.get('telegram', '')) s = re.sub(r'\{gia:[^}]*\}', f'(giá xem {ten_mien}{trang_bang_gia})' if trang_bang_gia else '(giá: nhắn shop)', s) s = re.sub(r'\{ten:([^}]*)\}', r'\1', s) return s def lam_sach(nd: str, ten_mien: str, trang_goc: frozenset) -> str: """Facebook không hiểu markdown; link nội bộ phải tuyệt đối.""" s = str(nd or '').replace('\r\n', '\n').strip() s = s.replace('**', '').replace('__', '') s = re.sub(r'^#{1,6}\s+', '', s, flags=re.M) # tiêu đề markdown (hashtag #Abc không có dấu cách) s = re.sub(r'\[([^\]]+)\]\((https?://[^)\s]+)\)', r'\1: \2', s) # [chữ](link) -> chữ: link def _tuyet_doi(m: 're.Match') -> str: dau = m.group(2).strip('/').split('/')[0].split('?')[0].split('#')[0] return m.group(1) + (ten_mien + m.group(2) if dau in trang_goc else m.group(2)) s = re.sub(r'(^|[\s(:])(/[A-Za-z][A-Za-z0-9_\-./#?=&]*[A-Za-z0-9_/#=])', _tuyet_doi, s) s = re.sub(r'\n{3,}', '\n\n', s) return s def kiem(nd: str, ten_mien: str) -> List[str]: loi = [] if _CAM.search(nd): loi.append('có chữ cấm: ' + _CAM.search(nd).group(0)) so = len(re.findall(r'\w+', nd)) if not SO_CHU[0] <= so <= SO_CHU[1]: loi.append(f'độ dài {so} chữ (cần {SO_CHU[0]}–{SO_CHU[1]})') if ten_mien and ten_mien not in nd: loi.append('thiếu link ' + ten_mien) return loi def doc_json(van_ban: str) -> Dict[str, str]: s = str(van_ban or '') i, j = s.find('{'), s.rfind('}') try: d = json.loads(s[i:j + 1]) if i >= 0 and j > i else {} except ValueError: d = {} return {'chu_de': str(d.get('chu_de') or '').strip()[:150], 'noi_dung': str(d.get('noi_dung') or '').strip()} # ===================================================================== # NGUỒN SỰ THẬT # ===================================================================== def _thu_muc_pages() -> str: from kho.trich import DUONG_DAN_SITE_MAC_DINH return os.path.dirname(DUONG_DAN_SITE_MAC_DINH.rstrip('/\\')) def chu_trang(tep: str, toi_da: int = 7000) -> str: """Chữ hiển thị (te()/t()) của một trang PHP dưới public/pages, theo thứ tự trên trang, kèm các tệp con include cùng thư mục (vd intro_sections/smartdca.php). Bỏ khung chung (shell).""" from kho.trich import quet_su_kien goc = os.path.join(_thu_muc_pages(), tep) da, cau = set(), [] def _doc(duong: str, sau: int) -> None: if duong in da or sau > 2 or not os.path.isfile(duong): return da.add(duong) noi = open(duong, encoding='utf-8', errors='replace').read() cac_su_kien, _ = quet_su_kien(noi) cau.extend(x['khoa'] for x in cac_su_kien if x.get('khoa')) for m in re.finditer(r"(?:include|require)(?:_once)?\s*\(?\s*__DIR__\s*\.\s*'([^']+\.php)'", noi): if '/shell/' not in m.group(1) and '..' not in m.group(1): _doc(os.path.normpath(os.path.join(os.path.dirname(duong), m.group(1).lstrip('/'))), sau + 1) _doc(goc, 0) return '\n'.join(dict.fromkeys(cau))[:toi_da] def _muc_kho(ct, kg: str, ids=None, bo=()) -> List[Dict[str, Any]]: if ids: dau = ','.join(['%s'] * len(ids)) ct.execute(f"SELECT id, cau_hoi_chuan, tra_loi_vi FROM chat_muc WHERE khong_gian = %s AND trang_thai = 'duyet' " f"AND id IN ({dau}) ORDER BY id", (kg, *ids)) else: ct.execute("SELECT id, cau_hoi_chuan, tra_loi_vi FROM chat_muc WHERE khong_gian = %s AND trang_thai = 'duyet' " "AND CHAR_LENGTH(tra_loi_vi) >= 250 ORDER BY id", (kg,)) return [dict(r) for r in (ct.fetchall() or []) if int(r['id']) not in set(bo)] # ===================================================================== # SOẠN # ===================================================================== def _prompt(hs, nhom: str) -> str: chung = ( f'Em viết bài đăng Fanpage cho {hs.ten}. Tiếng Việt, xưng "bên mình", gọi người đọc là "bạn". Văn bản THUẦN ' '(Facebook không hiểu markdown: không **, không # đầu dòng, không [chữ](link)). 150–300 chữ.\n' 'CHỈ dùng sự thật trong phần NGUỒN; không bịa tính năng, số liệu, giá, thời hạn. Điều nguồn không nói thì không viết.\n' 'Link phải TUYỆT ĐỐI, bắt đầu bằng ' + hs.ten_mien + '. Không trùng ý với các bài gần đây (liệt kê bên dưới). ' 'Câu dẫn tới link phải nói ĐÚNG trang đó là gì (xem GHI CHÚ); đừng gọi một trang số liệu / giới thiệu là ' '"bài viết" hay "ghi chú".\n' ) kieu = { 'san_pham': ('Bài giới thiệu sản phẩm: dòng đầu là câu móc VIẾT HOA ngắn + 1 emoji; đoạn 2 nêu đúng khó khăn của ' 'người đọc; 4–7 gạch đầu dòng lợi ích, mỗi dòng bắt đầu bằng 1 emoji, cụ thể theo nguồn; nêu điều ' 'kiện / giới hạn quan trọng; kết bằng lời mời xem trang + nhắn Fanpage để được tư vấn.'), 'bang_chung': ('Bài mời xem SỐ LIỆU CÔNG KHAI: nói đúng trang đó là tài khoản chạy thật hay kết quả perday (xem GHI CHÚ), giải thích trang cho xem gì, vì sao xem số liệu trực tiếp đáng ' 'tin hơn ảnh chụp; KHÔNG ghi con số lãi / số dư nào; kết bằng link xem trực tiếp.'), 'dich_vu': ('Bài giới thiệu dịch vụ: dòng đầu móc câu VIẾT HOA + emoji; ai cần dịch vụ này; làm thế nào / quy ' 'trình (theo nguồn); kết bằng cách liên hệ / đăng ký đúng như nguồn.'), 'kien_thuc': ('Bài CHIA SẺ KIẾN THỨC giao dịch (không quảng cáo): dòng đầu là câu hỏi hoặc nhận định gây chú ý + ' 'emoji; 3–6 ý ngắn có gạch đầu dòng emoji, đúng kiến thức chung của nghề, thực tế, không hứa hẹn; ' 'chỉ MỘT câu cuối nhẹ nhàng nhắc tới bên mình kèm link (nếu có). Không nêu số liệu riêng của sản phẩm.'), 'hoi_dap': ('Bài HỎI ĐÁP từ câu khách hay hỏi: dòng đầu "❓" + câu hỏi viết lại tự nhiên; trả lời rõ ràng, chia ý ' 'gạch đầu dòng emoji, theo đúng câu trả lời trong nguồn; kết bằng mời nhắn Fanpage nếu còn thắc mắc.'), }[nhom] return (chung + kieu + '\n' + (hs.bai_ngay.get('luat') or '') + '\n3–5 hashtag ở dòng cuối.\n' 'CHỈ trả về JSON: {"chu_de": "tiêu đề ngắn cho thẻ duyệt", "noi_dung": "toàn văn bài"}') def _noi_dung_goi(nguon: Dict[str, Any], su_that: str, gan_day: List[str], ten_mien: str) -> str: link = (ten_mien + nguon['link']) if nguon.get('link') else '' d = [f"CHỦ ĐỀ: {nguon['ten']}", f"LINK CHÍNH: {link}" if link else 'LINK CHÍNH: (không có trang riêng — mời nhắn Fanpage)'] if nguon.get('them'): d.append('GHI CHÚ: ' + nguon['them']) d += ['', 'NGUỒN (sự thật được phép dùng):', su_that or '(nhóm kiến thức: dùng kiến thức chung của nghề)', '', 'BÀI GẦN ĐÂY (đừng lặp):'] + [f'- {x}' for x in gan_day] return '\n'.join(d) def soan_va_gui(kg: str, nhom: Optional[str] = None, *, gui: bool = True, ngay: Optional[date] = None, _lay_ket_noi=None, _goi: Optional[Callable[[str, str], Optional[str]]] = None, _gui_duyet=None) -> Tuple[bool, str, Optional[int]]: """Soạn MỘT bài cho shop `kg`; gui=True thì gửi thẻ duyệt. Trả (ok, thông điệp, id bài | None).""" import shop hs = shop.lay(kg) cfg = hs.bai_ngay or {} if not cfg: return False, f'Shop {kg} chưa khai bai_ngay.', None ngay = ngay or _bay_gio_vn().date() nhom = nhom or nhom_hom_nay(tuple(cfg.get('lich') or ()), ngay) nguon_nhom = (cfg.get('nguon') or {}).get(nhom) if nguon_nhom is None: return False, f'Nhóm "{nhom}" không có nguồn.', None # 1) đọc CSDL (kết nối ngắn): nguồn đã dùng, bài gần đây, mục kho def _doc(ct): ct.execute("SELECT nguon, MAX(tao_luc) AS luc FROM chat_bai_dang WHERE khong_gian = %s AND nguon IS NOT NULL " "GROUP BY nguon", (kg,)) da = {r['nguon']: r['luc'] for r in (ct.fetchall() or [])} ct.execute("SELECT chu_de, LEFT(noi_dung, 160) AS dau FROM chat_bai_dang WHERE khong_gian = %s " "ORDER BY id DESC LIMIT 8", (kg,)) gan = [f"{r['chu_de']} — {' '.join(str(r['dau']).split())}" for r in (ct.fetchall() or [])] kho = _muc_kho(ct, kg, bo=tuple(cfg.get('kho_bo') or ())) if nguon_nhom == 'kho' else [] return da, gan, kho da_dung, gan_day, kho = csdl.voi_ct(_lay_ket_noi, _doc) if nguon_nhom == 'kho': ung_vien = [{'khoa': f"kho:{r['id']}", 'ten': r['cau_hoi_chuan'], 'link': '', 'kho': (int(r['id']),)} for r in kho] else: ung_vien = list(nguon_nhom) nguon = chon_nguon(ung_vien, da_dung, ngay) if not nguon: return False, f'Nhóm "{nhom}" hết nguồn.', None # 2) sự thật: chữ trang + mục kho phan = [] if nguon.get('trang'): phan.append(chu_trang(nguon['trang'])) if nguon.get('kho'): muc = csdl.voi_ct(_lay_ket_noi, lambda ct: _muc_kho(ct, kg, ids=tuple(nguon['kho']))) phan += [f"Hỏi: {r['cau_hoi_chuan']}\nĐáp: {thay_bien_kho(r['tra_loi_vi'], hs.lien_he, hs.ten_mien, hs.trang_bang_gia)}" for r in muc] su_that = '\n\n'.join(p for p in phan if p) if nhom != 'kien_thuc' and not su_that: return False, f'Nguồn {nguon["khoa"]} không đọc được nội dung.', None # 3) gọi mô hình (KHÔNG giữ kết nối CSDL), thử lại một lần nếu bài trượt kiểm if _goi is None: from nao.compact import _goi_mo_hinh _goi = lambda h, n: _goi_mo_hinh(h, n, 'tong_hop') # noqa: E731 he_thong = _prompt(hs, nhom) goi = _noi_dung_goi(nguon, su_that, gan_day, hs.ten_mien) bai, loi = {'chu_de': '', 'noi_dung': ''}, ['chưa soạn'] for lan in range(3): # mô hình thỉnh thoảng lan man tới trần token -> không ra JSON: thử lại bai = doc_json(_goi(he_thong, goi + (f'\n\nBẢN TRƯỚC BỊ LOẠI VÌ: {"; ".join(loi)}. Viết lại, NGẮN GỌN.' if lan else '')) or '') bai['noi_dung'] = lam_sach(bai['noi_dung'], hs.ten_mien, hs.trang_goc) loi = kiem(bai['noi_dung'], hs.ten_mien) if bai['noi_dung'] else ['mô hình không trả bài'] if not loi: break if not bai['noi_dung']: return False, 'Mô hình không trả bài.', None chu_de = f"[Tự soạn · {TEN_THU[ngay.weekday()]} · {TEN_NHOM.get(nhom, nhom)}] " + (bai['chu_de'] or nguon['ten']) if loi: chu_de = '⚠️ ' + chu_de + ' — kiểm kỹ: ' + '; '.join(loi) if not gui: return True, chu_de + '\n\n' + bai['noi_dung'], None if _gui_duyet is None: from kenh.bai_dang import gui_duyet as _gui_duyet # noqa: N806 id_ = _gui_duyet(bai['noi_dung'], chu_de=chu_de[:190], anh_url=nguon.get('anh') or '', khong_gian=kg, nguon=nguon['khoa'], _lay_ket_noi=_lay_ket_noi) return True, f'Đã gửi thẻ duyệt bài #{id_} ({nguon["khoa"]}).', id_ # ===================================================================== # LUỒNG NỀN # ===================================================================== def mot_vong(_lay_ket_noi=None) -> None: import shop hom_nay = _bay_gio_vn().date() gio_vn = _bay_gio_vn().hour def _viec(ct): ct.execute("SELECT DISTINCT khong_gian FROM chat_kenh WHERE kenh = 'messenger' AND bat = 1 AND page_token <> ''") ra = [] for r in ct.fetchall() or []: kg = str(r['khong_gian']) if not shop.lay(kg).bai_ngay or shop.lay(kg).ma != kg: continue ct.execute("SELECT khoa, gia_tri FROM chat_cau_hinh WHERE khong_gian = %s AND khoa LIKE 'bd\\_%%'", (kg,)) cfg = {x['khoa']: str(x['gia_tri'] or '') for x in (ct.fetchall() or [])} try: gio = int(cfg.get('bd_gio') or 8) except ValueError: gio = 8 if cfg.get('bd_bat') == '1' and gio_vn >= gio and cfg.get('bd_ngay') != hom_nay.isoformat(): # Ghi ngày TRƯỚC khi soạn: soạn lỗi thì mai thử lại, không bao giờ soạn trùng trong ngày. ct.execute("""INSERT INTO chat_cau_hinh (khong_gian, khoa, gia_tri, cap_nhat) VALUES (%s, 'bd_ngay', %s, NOW()) ON DUPLICATE KEY UPDATE gia_tri = VALUES(gia_tri), cap_nhat = NOW()""", (kg, hom_nay.isoformat())) ra.append(kg) return ra for kg in csdl.voi_ct(_lay_ket_noi, _viec): try: ok, tb, _ = soan_va_gui(kg, _lay_ket_noi=_lay_ket_noi) (_log.info if ok else _log.warning)('Bài hằng ngày %s: %s', kg, tb.split('\n')[0]) except Exception: # noqa: BLE001 _log.exception('Soạn bài hằng ngày %s lỗi', kg) _da_chay = False _khoa = threading.Lock() def khoi_dong_nen() -> None: global _da_chay if os.environ.get('CHATBOT_BAI_NGAY', '1').strip() == '0': return with _khoa: if _da_chay: return _da_chay = True def _vong(): while True: time.sleep(CHU_KY_GIAY) try: mot_vong() except Exception: # noqa: BLE001 - luồng nền không bao giờ được chết _log.exception('Vòng bài hằng ngày lỗi.') threading.Thread(target=_vong, name='bai-ngay', daemon=True).start() _log.info('Đã bật soạn bài Fanpage hằng ngày (công tắc bd_bat / bd_gio trong chat_cau_hinh).') # ===================================================================== # CLI + TỰ KIỂM # ===================================================================== if __name__ == '__main__' and len(sys.argv) > 1: logging.basicConfig(level=logging.INFO) lenh = sys.argv[1] ok, tb, _ = soan_va_gui(os.environ.get('KHONG_GIAN', 'tradingauto'), sys.argv[2] if len(sys.argv) > 2 else None, gui=(lenh == 'gui')) print(('OK ' if ok else 'LỖI ') + tb) sys.exit(0 if ok else 1) if __name__ == '__main__': lich = ('san_pham', 'kien_thuc', 'hoi_dap', 'san_pham', 'bang_chung', 'dich_vu', 'kien_thuc') assert nhom_hom_nay(lich, date(2026, 10, 5)) == 'san_pham' and nhom_hom_nay(lich, date(2026, 10, 9)) == 'bang_chung' uv = [{'khoa': 'a'}, {'khoa': 'b'}, {'khoa': 'c'}] assert chon_nguon(uv, {'a': datetime(2026, 9, 1), 'b': datetime(2026, 9, 2)}, date(2026, 10, 2))['khoa'] == 'c' assert chon_nguon(uv, {'a': datetime(2026, 9, 3), 'b': datetime(2026, 9, 2), 'c': datetime(2026, 9, 5)}, date(2026, 10, 2))['khoa'] == 'b', 'dùng lâu nhất trước' hai = {chon_nguon(uv, {}, date(2026, 10, d))['khoa'] for d in (1, 2, 3)} assert hai == {'a', 'b', 'c'}, 'chưa dùng cái nào thì xoay theo ngày' assert chon_nguon([], {}, date(2026, 10, 2)) is None assert thay_bien_kho('Nhắn {zalo} hoặc {telegram}, giá {gia:dcabot}, sp {ten:SMART DCA}', {'zalo': '0387', 'telegram': '@q'}, 'https://t.org', '/Pricing') == \ 'Nhắn Zalo 0387 hoặc Telegram @q, giá (giá xem https://t.org/Pricing), sp SMART DCA' tg = frozenset({'toolintroduce', 'Pricing', 'pages'}) s = lam_sach('## TIÊU ĐỀ\n**Đậm** xem /toolintroduce/dcabot và [bảng giá](https://t.org/Pricing), file /etc/x\n\n\n#TradingAuto', 'https://t.org', tg) assert s == 'TIÊU ĐỀ\nĐậm xem https://t.org/toolintroduce/dcabot và bảng giá: https://t.org/Pricing, file /etc/x\n\n#TradingAuto', s assert 'có chữ cấm' in kiem('cho xin passview ' + 'chữ ' * 150 + 'https://t.org', 'https://t.org')[0] assert kiem('chữ ' * 150 + 'https://t.org', 'https://t.org') == [] assert any('độ dài' in x for x in kiem('ngắn https://t.org', 'https://t.org')) assert any('thiếu link' in x for x in kiem('chữ ' * 150, 'https://t.org')) assert 'không bảo đảm lợi nhuận' and not _CAM.search('Bot không dự đoán thị trường, không bảo đảm lợi nhuận.') assert doc_json('rác {"chu_de": "A", "noi_dung": "B"} rác') == {'chu_de': 'A', 'noi_dung': 'B'} assert doc_json('không phải json') == {'chu_de': '', 'noi_dung': ''} print('TỰ KIỂM kenh/bai_ngay.py QUA.')