Commit 4f52d746 authored by Vũ Hoàng Anh's avatar Vũ Hoàng Anh

feat(rules): seed 350+ fashion rules with full demographic & anchor coverage, add tests/

parent a2fe4bd9
# D:\cnf\chatbot-canifa-feedback\.agent\workflows\fashion-rules-verification.md
# Fashion Rules Verification Loop
## Mục đích
Tự động hóa việc kiểm thử hàng loạt (batch testing) các luật phối đồ (fashion rules) trên danh mục sản phẩm. Đảm bảo rằng Engine recommend các sản phẩm chính xác theo Giới tính (Gender), Dịp mặc (Occasion) và Luật kết hợp (Pairing Rules) đã được cấu hình trong database.
## Quy trình (Workflow)
**Mục tiêu:** Kiểm tra N sản phẩm (ví dụ: 100) để xác minh:
1. Sản phẩm có nhận diện đúng Giới tính và Phân loại (Product Line) không.
2. Kết quả recommend có chứa các Dịp mặc (Occasion) được phép theo luật hay không.
3. Các sản phẩm gợi ý (Target Items) có thuộc đúng Product Line quy định trong `chatbot_fashion_rules` hay không.
### Bước 1: Viết script kiểm thử tự động
Bạn có thể tự động sinh ra một file script bằng Python để fetch catalog trực tiếp thông qua `StylistEngine` và test:
- **Input:** Lấy ngẫu nhiên X sản phẩm từ catalog, đảm bảo có đủ giới tính (Nữ, Nam, Unisex, Bé Gái, Bé Trai).
- **Process:** Khởi tạo `StylistEngine`, gọi `compute_dynamic_rule_matches(code)`.
- **Verify:**
- Truy vấn luật trực tiếp từ hàm `_fetch_rules_with_reason(anchor_cat, gender)`.
- Đối chiếu kết quả recommend từ engine so với tập luật này.
- Ghi nhận: Dịp nào thiếu SP recommend (do catalog không có màu phù hợp), item nào bị recommend sai category (Vi phạm luật).
- **Output:** In ra báo cáo tóm tắt (Tổng số, Số SP pass, Số lượng lỗi if any).
### Bước 2: Chạy script và nhận report
// turbo
```bash
cd backend
python test_fashion_rules_batch.py --limit 100
```
### Bước 3: Đánh giá và Sửa lỗi
- Nếu phát hiện Occasion rỗng, nguyên nhân có thể do catalog thiếu sản phẩm có màu sắc phù hợp quy tắc phối, hoặc luật quá khắt khe.
- Nếu phát hiện Item recommend sai `product_line`, cần check lại hàm query rule của Engine hoặc việc fallback.
- Chỉnh sửa logic trong `stylist_engine.py` hoặc thêm/sửa rules trong bảng `chatbot_fashion_rules`. Lặp lại Bước 2.
This diff is collapsed.
"""
migrate_004_remaining.py — seed 27 anchor categories còn thiếu
"""
import logging, os, sys
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
backend_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
if backend_dir not in sys.path:
sys.path.insert(0, backend_dir)
TABLE = "dashboard_canifa.chatbot_fashion_rules"
RULES = [
# Áo khoác chần bông NAM
("nam", "Áo khoác chần bông", "hang_ngay","bottom","Quần jean", "Áo khoác chần bông + Quần jean nam"),
("nam", "Áo khoác chần bông", "di_choi", "bottom","Quần jean", "Áo khoác chần bông + Quần jean nam đi chơi"),
("nam", "Áo khoác chần bông", "hang_ngay","top", "Áo phông", "Áo khoác chần bông phủ Áo phông nam"),
# Áo Polo BE TRAI
("be_trai","Áo Polo", "hang_ngay","bottom","Quần soóc", "Áo Polo bé trai + Quần soóc hàng ngày"),
("be_trai","Áo Polo", "hang_ngay","bottom","Quần khaki", "Áo Polo bé trai + Quần khaki"),
("be_trai","Áo Polo", "du_lich", "bottom","Quần soóc", "Áo Polo bé trai + Quần soóc du lịch"),
# Áo Polo UNISEX
("unisex", "Áo Polo", "hang_ngay","bottom","Quần jean", "Áo Polo unisex + Quần jean"),
("unisex", "Áo Polo", "di_lam", "bottom","Quần khaki", "Áo Polo unisex + Quần khaki công sở"),
# Áo len BE TRAI
("be_trai","Áo len", "hang_ngay","bottom","Quần nỉ", "Áo len bé trai + Quần nỉ ấm"),
("be_trai","Áo len", "hang_ngay","bottom","Quần dài", "Áo len bé trai + Quần dài mùa lạnh"),
# Quần Khaki BE GAI
("be_gai", "Quần Khaki","hang_ngay","top", "Áo phông", "Quần Khaki bé gái + Áo phông"),
("be_gai", "Quần Khaki","di_choi", "top", "Áo kiểu", "Quần Khaki bé gái + Áo kiểu đi chơi"),
# Khăn / Scarf (accessories — recommend top/bottom to go with)
("unisex", "Khăn", "hang_ngay","top", "Áo phông", "Khăn + Áo phông: phụ kiện base"),
("unisex", "Khăn", "di_choi", "top", "Áo nỉ", "Khăn + Áo nỉ: mùa lạnh stylish"),
# Quần lót (underwear — minimal rules)
("nu", "Quần lót", "mac_nha", "top", "Áo phông", "Quần lót + Áo phông: lounge basic"),
("be_gai", "Quần lót", "mac_nha", "top", "Áo phông", "Quần lót bé gái + Áo phông nhà"),
# Áo khoác sợi NAM
("nam", "Áo khoác sợi","di_choi", "bottom","Quần jean", "Áo khoác sợi nam + Quần jean"),
("nam", "Áo khoác sợi","hang_ngay","top", "Áo phông", "Áo khoác sợi phủ Áo phông nam"),
# Pyjama BE TRAI
("be_trai","Pyjama", "mac_nha", "outerwear","Áo nỉ", "Pyjama bé trai + Áo nỉ mùa lạnh"),
# Áo khoác gilet chần bông (gilet)
("be_gai", "Áo khoác gilet chần bông","hang_ngay","top","Áo phông","Gilet chần bông + Áo phông bé gái"),
("be_gai", "Áo khoác gilet chần bông","hang_ngay","bottom","Quần nỉ","Gilet chần bông + Quần nỉ bé gái"),
("unisex", "Áo khoác gilet chần bông","hang_ngay","top","Áo phông", "Gilet chần bông + Áo phông unisex"),
("nu", "Áo khoác gilet chần bông","di_choi", "bottom","Quần jean","Gilet chần bông + Quần jean nữ"),
# Áo giữ nhiệt BE GAI
("be_gai", "Áo giữ nhiệt","mac_nha","bottom","Quần nỉ", "Áo giữ nhiệt + Quần nỉ bé gái ấm"),
("be_gai", "Áo giữ nhiệt","hang_ngay","outerwear","Áo nỉ", "Áo giữ nhiệt base + Áo nỉ phủ ngoài"),
# Quần giữ nhiệt NAM
("nam", "Quần giữ nhiệt","mac_nha","top","Áo giữ nhiệt","Quần giữ nhiệt + Áo giữ nhiệt base layer"),
("nam", "Quần giữ nhiệt","mac_nha","top","Áo phông", "Quần giữ nhiệt + Áo phông: thermal inside"),
# Áo nỉ UNISEX
("unisex", "Áo nỉ", "di_choi", "bottom","Quần jean", "Áo nỉ + Quần jean unisex casual"),
("unisex", "Áo nỉ", "hang_ngay","bottom","Quần dài", "Áo nỉ + Quần dài unisex chill"),
# Áo mặc nhà NAM
("nam", "Áo mặc nhà","mac_nha", "bottom","Quần mặc nhà","Áo mặc nhà + Quần mặc nhà: set nhà nam"),
("nam", "Áo mặc nhà","mac_nha", "bottom","Quần nỉ", "Áo mặc nhà + Quần nỉ nam winter home"),
# Quần váy BE GAI
("be_gai", "Quần váy", "di_choi", "top", "Áo phông", "Quần váy bé gái + Áo phông đi chơi"),
("be_gai", "Quần váy", "hang_ngay","top", "Áo kiểu", "Quần váy bé gái + Áo kiểu hàng ngày"),
# Khăn mặt (towel) — non-fashion, skip meaningful pairing; add minimal
("all", "Khăn mặt", "mac_nha", "top", "Áo mặc nhà", "Khăn mặt + Áo mặc nhà: home essentials"),
# Bộ quần áo UNISEX
("unisex", "Bộ quần áo","hang_ngay","outerwear","Áo khoác gió","Bộ quần áo unisex + Khoác nhẹ"),
# Áo khoác gilet chần bông WOMEN (đã có? thêm thêm)
("nu", "Áo khoác gilet chần bông","di_choi","top","Áo phông","Gilet chần bông nữ + Áo phông base"),
]
def run():
from common.pool_wrapper import get_pooled_connection_compat
conn = None
try:
conn = get_pooled_connection_compat()
cur = conn.cursor()
inserted = 0
for gender, anchor, occ, role, target, reason in RULES:
cur.execute(f"""
INSERT INTO {TABLE} (gender_target, anchor_category, occasion_tag, match_role, target_category, ai_reason)
VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT DO NOTHING
""", (gender, anchor, occ, role, target, reason))
if cur.rowcount > 0:
inserted += 1
conn.commit()
cur.close()
print(f"[OK] migrate_004 done: +{inserted} rules seeded ({len(RULES)} total in batch)")
except Exception as e:
if conn: conn.rollback()
print(f"[ERROR] {e}")
finally:
if conn: conn.close()
if __name__ == "__main__":
run()
"""migrate_005_final.py — seed 11 anchor categories còn lại"""
import logging, os, sys
logging.basicConfig(level=logging.INFO)
backend_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
if backend_dir not in sys.path:
sys.path.insert(0, backend_dir)
TABLE = "dashboard_canifa.chatbot_fashion_rules"
RULES = [
# Tất (socks) — minimal, recommend outfit underneath
("nam", "Tất", "the_thao","top", "Áo phông", "Tất + Áo phông: sport set hoàn chỉnh"),
("nam", "Tất", "di_lam", "top", "Áo Sơ mi", "Tất + Sơ mi: office accessory"),
("nu", "Tất", "the_thao","top", "Áo phông", "Tất thể thao + Áo phông nữ"),
("nu", "Tất", "di_lam", "top", "Blouse", "Tất + Blouse office nữ"),
("unisex","Tất", "the_thao","top", "Áo phông", "Tất + Áo phông: universal sport"),
("be_trai","Tất","the_thao","top", "Áo phông", "Tất bé trai + Áo phông thể thao"),
# Áo khoác (generic)
("nu", "Áo khoác","di_choi","bottom","Quần jean","Áo khoác nữ + Quần jean"),
("nu", "Áo khoác","du_lich","bottom","Quần jean","Áo khoác nữ + Quần jean du lịch"),
# Áo khoác gilet chần bông BE TRAI
("be_trai","Áo khoác gilet chần bông","hang_ngay","top","Áo phông","Gilet bé trai + Áo phông bên trong"),
("be_trai","Áo khoác gilet chần bông","hang_ngay","bottom","Quần nỉ","Gilet bé trai + Quần nỉ đông"),
# Khẩu trang (mask — non-fashion, add minimal)
("unisex","Khẩu trang","hang_ngay","top","Áo phông", "Khẩu trang + Áo phông: daily protection"),
# Quần lót đùi NAM
("nam", "Quần lót đùi","mac_nha","top","Áo phông", "Quần lót đùi + Áo phông: home basic nam"),
# Túi xách (bag — accessory, no forced bottom/top)
("unisex","Túi xách","di_choi","top", "Áo phông", "Túi xách + Áo phông: streetwear complete"),
("nu", "Túi xách","di_choi","top", "Áo kiểu", "Túi xách + Áo kiểu nữ: styled"),
# Áo nỉ có mũ BE GAI
("be_gai","Áo nỉ có mũ","hang_ngay","bottom","Quần leggings","Hoodie bé gái + Leggings"),
("be_gai","Áo nỉ có mũ","hang_ngay","bottom","Quần nỉ", "Hoodie bé gái + Quần nỉ ấm"),
("be_gai","Áo nỉ có mũ","di_choi", "bottom","Quần jean", "Hoodie bé gái + Quần jean đi chơi"),
# Áo mặc nhà BE TRAI
("be_trai","Áo mặc nhà","mac_nha","bottom","Quần mặc nhà","Áo mặc nhà + Quần mặc nhà bé trai"),
("be_trai","Áo mặc nhà","mac_nha","bottom","Quần nỉ", "Áo mặc nhà bé trai + Quần nỉ"),
]
def run():
from common.pool_wrapper import get_pooled_connection_compat
conn = None
try:
conn = get_pooled_connection_compat()
cur = conn.cursor()
inserted = 0
for gender, anchor, occ, role, target, reason in RULES:
cur.execute(f"""
INSERT INTO {TABLE}(gender_target,anchor_category,occasion_tag,match_role,target_category,ai_reason)
VALUES(%s,%s,%s,%s,%s,%s) ON CONFLICT DO NOTHING
""", (gender, anchor, occ, role, target, reason))
if cur.rowcount > 0:
inserted += 1
conn.commit(); cur.close()
print(f"[OK] migrate_005 done: +{inserted}/{len(RULES)} rules seeded")
except Exception as e:
if conn: conn.rollback()
print(f"[ERROR] {e}")
finally:
if conn: conn.close()
if __name__ == "__main__":
run()
"""migrate_006_edge_cases.py — seed 11 edge anchor categories còn lại"""
import os, sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
TABLE = "dashboard_canifa.chatbot_fashion_rules"
RULES = [
# Áo Sơ mi BE TRAI
("be_trai","Áo Sơ mi", "di_lam", "bottom","Quần khaki", "Sơ mi bé trai + Quần khaki đi học"),
("be_trai","Áo Sơ mi", "hang_ngay","bottom","Quần jean", "Sơ mi bé trai + Quần jean hàng ngày"),
# Áo len gilet BE GAI
("be_gai", "Áo len gilet","hang_ngay","top", "Áo phông", "Gilet len bé gái + Áo phông bên trong"),
("be_gai", "Áo len gilet","hang_ngay","bottom","Quần leggings","Gilet len bé gái + Leggings ấm áp"),
# Tất BE GAI
("be_gai","Tất", "the_thao", "top", "Áo phông", "Tất bé gái + Áo phông thể thao"),
# Áo khoác nỉ có mũ NAM (với schema tên khác)
("nam", "Áo khoác nỉ có mũ","di_choi","bottom","Quần jean","Khoác nỉ có mũ nam + Quần jean đi chơi"),
("nam", "Áo khoác nỉ có mũ","hang_ngay","top","Áo phông", "Khoác nỉ có mũ nam + Áo phông bên trong"),
# Áo khoác nỉ có mũ BE TRAI
("be_trai","Áo khoác nỉ có mũ","hang_ngay","bottom","Quần nỉ","Khoác nỉ có mũ bé trai + Quần nỉ"),
("be_trai","Áo khoác nỉ có mũ","di_choi", "bottom","Quần jean","Khoác nỉ có mũ bé trai + Quần jean"),
# Khăn mặt (others→unisex)
("all", "Khăn mặt", "mac_nha", "top", "Áo mặc nhà", "Khăn mặt + Áo mặc nhà: home essential"),
# Pyjama BE GAI
("be_gai","Pyjama", "mac_nha", "outerwear","Cardigan", "Pyjama bé gái + Cardigan ấm nhà"),
# Pyjama NU
("nu", "Pyjama", "mac_nha", "outerwear","Cardigan", "Pyjama nữ + Cardigan mùa lạnh"),
# Quần váy NU
("nu", "Quần váy", "di_choi", "top", "Áo phông", "Quần váy nữ + Áo phông đi chơi"),
("nu", "Quần váy", "di_lam", "top", "Blouse", "Quần váy nữ + Blouse văn phòng"),
# Găng tay chống nắng (sun glove — aesthetic tiny)
("unisex","Găng tay chống nắng","du_lich","top","Áo phông", "Găng tay chống nắng + Áo phông du lịch"),
("nu", "Găng tay chống nắng","du_lich","top","Áo phông", "Găng tay chống nắng + Áo phông nữ"),
# Áo khoác gilet chần bông NAM
("nam", "Áo khoác gilet chần bông","hang_ngay","top","Áo phông", "Gilet chần bông nam + Áo phông"),
("nam", "Áo khoác gilet chần bông","hang_ngay","bottom","Quần jean","Gilet nam + Quần jean"),
]
def run():
from common.pool_wrapper import get_pooled_connection_compat
conn = None
try:
conn = get_pooled_connection_compat()
cur = conn.cursor()
inserted = 0
for r in RULES:
cur.execute(f"""
INSERT INTO {TABLE}(gender_target,anchor_category,occasion_tag,match_role,target_category,ai_reason)
VALUES(%s,%s,%s,%s,%s,%s) ON CONFLICT DO NOTHING
""", r)
if cur.rowcount > 0:
inserted += 1
conn.commit(); cur.close()
print(f"[OK] migrate_006 done: +{inserted}/{len(RULES)} rules seeded")
except Exception as e:
if conn: conn.rollback()
print(f"[ERROR] {e}")
finally:
if conn: conn.close()
if __name__ == "__main__":
run()
"""migrate_007_absolute_final.py — 6 anchor categories cuối cùng"""
import os, sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
TABLE = "dashboard_canifa.chatbot_fashion_rules"
RULES = [
("be_gai","Áo lót", "mac_nha", "bottom","Quần mặc nhà", "Áo lót bé gái + Quần mặc nhà nhà"),
("be_gai","Áo lót", "mac_nha", "top", "Áo mặc nhà", "Áo lót bé gái base + Áo mặc nhà phủ"),
("nam", "Quần lót tam giác", "mac_nha", "top", "Áo phông", "Quần lót + Áo phông: home basic nam"),
("be_gai","Áo Body", "di_lam", "bottom","Chân váy", "Áo Body bé gái + Chân váy: cute look"),
("be_gai","Áo Body", "hang_ngay","bottom","Quần leggings", "Áo Body bé gái + Leggings hàng ngày"),
("be_gai","Quần leggings mặc nhà","mac_nha", "top", "Áo phông", "Leggings nhà bé gái + Áo phông"),
("be_gai","Quần leggings mặc nhà","mac_nha", "top", "Áo mặc nhà", "Leggings nhà bé gái + Áo mặc nhà"),
("be_trai","Blazer", "di_lam", "bottom","Quần khaki", "Blazer bé trai + Quần khaki đi học"),
("be_trai","Blazer", "di_lam", "top", "Áo Sơ mi", "Blazer bé trai mặc ngoài Sơ mi"),
("nu", "Áo khoác sợi", "di_choi", "bottom","Quần jean", "Áo khoác sợi nữ + Quần jean"),
("nu", "Áo khoác sợi", "hang_ngay","top", "Áo phông", "Áo khoác sợi nữ + Áo phông base"),
]
def run():
from common.pool_wrapper import get_pooled_connection_compat
conn = None
try:
conn = get_pooled_connection_compat()
cur = conn.cursor()
inserted = 0
for r in RULES:
cur.execute(f"""
INSERT INTO {TABLE}(gender_target,anchor_category,occasion_tag,match_role,target_category,ai_reason)
VALUES(%s,%s,%s,%s,%s,%s) ON CONFLICT DO NOTHING
""", r)
if cur.rowcount > 0:
inserted += 1
conn.commit(); cur.close()
print(f"[OK] migrate_007 done: +{inserted}/{len(RULES)} rules seeded")
except Exception as e:
if conn: conn.rollback()
print(f"[ERROR] {e}")
finally:
if conn: conn.close()
if __name__ == "__main__":
run()
"""migrate_008_done.py — 5 anchor categories cuối cùng (Áo Body, Bộ thể thao bé trai, ...)"""
import os, sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
TABLE = "dashboard_canifa.chatbot_fashion_rules"
RULES = [
("nu", "Áo Body", "di_lam", "bottom","Quần âu", "Áo Body nữ + Quần âu: bodysuit office"),
("nu", "Áo Body", "di_lam", "bottom","Chân váy", "Áo Body nữ + Chân váy: sleek look"),
("nu", "Áo Body", "di_choi", "bottom","Quần jean", "Áo Body nữ + Quần jean: casual chic"),
("unisex","Áo Body", "the_thao", "bottom","Quần leggings","Áo Body + Leggings: activewear"),
("nam", "Áo Body", "the_thao", "bottom","Quần thể thao","Áo Body nam + Quần thể thao gym"),
("be_trai","Bộ thể thao","the_thao","outerwear","Áo khoác gió","Bộ thể thao bé trai + Khoác gió"),
("be_trai","Bộ thể thao","the_thao","outerwear","Áo nỉ có mũ","Bộ thể thao bé trai + Hoodie"),
("unisex","Quần mặc nhà","mac_nha","top","Áo phông", "Quần mặc nhà unisex + Áo phông nhà"),
("unisex","Quần mặc nhà","mac_nha","top","Áo mặc nhà", "Quần mặc nhà unisex + Áo mặc nhà"),
]
def run():
from common.pool_wrapper import get_pooled_connection_compat
conn = None
try:
conn = get_pooled_connection_compat()
cur = conn.cursor()
inserted = 0
for r in RULES:
cur.execute(f"""
INSERT INTO {TABLE}(gender_target,anchor_category,occasion_tag,match_role,target_category,ai_reason)
VALUES(%s,%s,%s,%s,%s,%s) ON CONFLICT DO NOTHING
""", r)
if cur.rowcount > 0:
inserted += 1
conn.commit(); cur.close()
print(f"[OK] migrate_008 done: +{inserted}/{len(RULES)} rules seeded")
except Exception as e:
if conn: conn.rollback()
print(f"[ERROR] {e}")
finally:
if conn: conn.close()
if __name__ == "__main__":
run()
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from common.pool_wrapper import get_pooled_connection_compat
conn = get_pooled_connection_compat()
cur = conn.cursor()
cur.execute("SELECT anchor_category FROM chatbot_fashion_rules WHERE anchor_category LIKE '%Chân váy%'")
print("LIKE:", cur.fetchall())
cur.execute("SELECT anchor_category FROM chatbot_fashion_rules WHERE UPPER(anchor_category) = UPPER('Chân váy')")
print("UPPER:", cur.fetchall())
cur.execute("SELECT anchor_category FROM chatbot_fashion_rules WHERE LOWER(anchor_category) = LOWER('Chân váy')")
print("LOWER:", cur.fetchall())
"""
tests/test_fashion_rules_batch.py
Batch-test 100 SP ngẫu nhiên → kiểm tra fashion rule coverage & integrity.
Run:
cd backend
$env:PYTHONIOENCODING="utf-8"
.venv\Scripts\python.exe tests/test_fashion_rules_batch.py --limit 100
"""
import sys
import os
import random
import argparse
from collections import defaultdict
# ── path ──────────────────────────────────────────────────────────────────────
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from worker.stylist_engine import StylistEngine
from common.pool_wrapper import get_pooled_connection_compat
def fetch_all_anchor_categories():
"""Lấy tập hợp (anchor_category, gender_target) đã có rules trong DB."""
conn = None
try:
conn = get_pooled_connection_compat()
cur = conn.cursor()
cur.execute("""\
SELECT DISTINCT anchor_category, gender_target
FROM dashboard_canifa.chatbot_fashion_rules
WHERE gender_target != 'all'
""")
rows = cur.fetchall()
cur.close()
return {(r[0].strip().lower(), r[1].strip().lower()) for r in rows}
except Exception as e:
print(f"[DB ERROR] {e}")
return set()
finally:
if conn:
conn.close()
def run_batch_test(limit: int = 100, verbose: bool = False):
engine = StylistEngine()
catalog = engine._load_catalog()
if not catalog:
print("Catalog rong!")
return
print(f"Tong so SP trong catalog: {len(catalog)}")
# Chỉ lấy SP có product_line + gender
valid = [p for p in catalog if p.get("product_line") and p.get("gender")]
samples = random.sample(valid, min(limit, len(valid)))
print(f"Test {len(samples)} SP ngau nhien...\n")
# Anchor categories có rules trong DB (lowercase)
db_anchors = fetch_all_anchor_categories()
stats = {
"total": len(samples),
"passed": 0,
"no_rules_in_db": 0,
"empty_all_occasions": 0,
"rule_violations": 0,
}
# Thống kê anchor nào không có rules
missing_anchors: dict[str, list[str]] = defaultdict(list)
violation_details: list[str] = []
empty_details: list[str] = []
for p in samples:
code = p.get("code", "?")
anchor_cat = p.get("product_line", "")
gender_raw = p.get("gender", "")
gender_norm = engine._normalize_gender(gender_raw)
# Kiểm tra xem DB có rules cho combo này không
anchor_lower = anchor_cat.strip().lower()
has_specific = (anchor_lower, gender_norm) in db_anchors
has_all = (anchor_lower, "all") in db_anchors
if not has_specific and not has_all:
stats["no_rules_in_db"] += 1
missing_anchors[f"{anchor_cat} [{gender_raw}→{gender_norm}]"].append(code)
continue
# Gọi engine
ai_matches = engine.compute_dynamic_rule_matches(code)
db_rules = engine._fetch_rules_with_reason(anchor_cat, gender_raw)
# allowed = {occ: set(target_cat_lower)}
allowed_by_occ: dict[str, set] = defaultdict(set)
for r in db_rules:
allowed_by_occ[r["occ"]].add(r["target_cat"].strip().lower())
is_all_empty = True
violations = []
for occ, allowed_targets in allowed_by_occ.items():
occ_data = ai_matches.get(occ, {})
if not occ_data:
continue
is_all_empty = False
for role, items in occ_data.items():
for item in items:
item_cat = item.get("product_line", "").strip().lower()
if item_cat not in allowed_targets:
violations.append(
f" Dip [{occ}] role={role}: recommend '{item_cat}' | Cho phep: {allowed_targets}"
)
if is_all_empty:
stats["empty_all_occasions"] += 1
empty_details.append(f" {anchor_cat} [{gender_norm}] ({code}) -> Toan bo dip TRONG (mau khong khop)")
if violations:
stats["rule_violations"] += 1
violation_details.append(f" {anchor_cat} [{gender_norm}] ({code}):")
violation_details.extend(violations)
else:
if not is_all_empty:
stats["passed"] += 1
# ── REPORT ──────────────────────────────────────────────────────────────
sep = "=" * 55
print(sep)
print(" KET QUA KIEM THU FASHION RULES")
print(sep)
print(f"Tong test : {stats['total']}")
print(f"Passed (ok) : {stats['passed']}")
print(f"Khong co rules DB : {stats['no_rules_in_db']} <- can seed them")
print(f"Rules co, nhung trong: {stats['empty_all_occasions']} <- catalog mau chua du")
print(f"Vi pham rules : {stats['rule_violations']} <- loi engine")
print()
# Thống kê anchor thiếu rules
if missing_anchors:
print(f"--- ANCHOR CATEGORY CHUA SEED RULES ({len(missing_anchors)} loai) ---")
sorted_missing = sorted(missing_anchors.items(), key=lambda x: -len(x[1]))
for anchor, codes in sorted_missing[:20]:
print(f" {len(codes):3d} SP | {anchor}")
if len(sorted_missing) > 20:
print(f" ... va {len(sorted_missing) - 20} loai khac")
print()
if violation_details:
print(f"--- VI PHAM RULES ({stats['rule_violations']} SP) ---")
for line in violation_details[:30]:
print(line)
print()
if verbose and empty_details:
print("--- RONG DO MAU (verbose) ---")
for line in empty_details[:20]:
print(line)
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--limit", type=int, default=100)
parser.add_argument("--verbose", action="store_true")
args = parser.parse_args()
run_batch_test(args.limit, args.verbose)
import sqlite3
import os
db_path = "backend/db/memos.db"
if not os.path.exists(db_path):
print(f"❌ Database not found at {db_path}")
# Try alternate path
db_path = "backend/database/cuccu_note.db"
if not os.path.exists(db_path):
print(f"❌ Database not found at {db_path} either")
exit(1)
print(f"✅ Found database at {db_path}")
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = cursor.fetchall()
print("Tables in database:")
for table in tables:
print(f"- {table[0]}")
conn.close()
import subprocess
import json
import time
BASE_URL = "http://localhost:5100"
def run_curl(method, endpoint, data=None, token=None):
cmd = ["curl", "-s", "-X", method, f"{BASE_URL}{endpoint}"]
cmd += ["-H", "Content-Type: application/json"]
if token:
cmd += ["-H", f"Authorization: Bearer {token}"]
if data:
cmd += ["-d", json.dumps(data)]
result = subprocess.run(cmd, capture_output=True, text=True)
try:
return json.loads(result.stdout)
except:
return result.stdout
def test_api():
print("Waiting for server to be ready...")
# Simple wait-loop
for _ in range(30):
try:
res = run_curl("GET", "/api/v1/memos")
if isinstance(res, list) or "detail" in res:
print("✅ Server is UP")
break
except:
pass
time.sleep(2)
else:
print("❌ Server timed out")
return
# 1. Register
username = f"testuser_{int(time.time())}"
print(f"Testing Register with {username}...")
reg_data = {
"username": username,
"email": f"{username}@example.com",
"password": "testpassword123"
}
reg_res = run_curl("POST", "/api/v1/auth/register", data=reg_data)
print("Register Response:", json.dumps(reg_res, indent=2))
if "access_token" not in reg_res:
print("❌ Registration failed")
return
token = reg_res["access_token"]
# 2. Create Memo
print("Testing Create Memo...")
memo_data = {
"content": "Hello world from API test!",
"visibility": "PRIVATE"
}
memo_res = run_curl("POST", "/api/v1/memos", data=memo_data, token=token)
print("Create Memo Response:", json.dumps(memo_res, indent=2))
# 3. List Memos
print("Testing List Memos...")
list_res = run_curl("GET", "/api/v1/memos", token=token)
print(f"List Memos found {len(list_res)} memos")
if __name__ == "__main__":
test_api()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment