Spaces:
Sleeping
Sleeping
File size: 3,129 Bytes
0649d3e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | import pandas as pd
import json
from pathlib import Path
import os
from dotenv import load_dotenv
# --- Load .env ---
load_dotenv()
# --- Debug biến môi trường ---
print("🔹 DEBUG: GOOGLE_API_KEY =", os.getenv("GOOGLE_API_KEY"))
print("🔹 DEBUG: GEMINI_MODEL_ID =", os.getenv("GEMINI_MODEL_ID"))
# --- Path tuyệt đối để tránh lỗi file not found ---
ROOT_DIR = Path(__file__).resolve().parent.parent.parent.parent # PINE-AI-main
DATA_DIR = ROOT_DIR / "backend" / "ai" / "data"
CSV_PATH = DATA_DIR / "test_customer.csv"
JSON_PATH = DATA_DIR / "product_collection.json"
# --- Debug check file tồn tại ---
print(f"🔹 DEBUG: CSV_PATH = {CSV_PATH}, exists = {CSV_PATH.exists()}")
print(f"🔹 DEBUG: JSON_PATH = {JSON_PATH}, exists = {JSON_PATH.exists()}")
class DataEngine:
def __init__(self, csv_path: Path = CSV_PATH, json_path: Path = JSON_PATH):
self.csv_path = Path(csv_path)
self.json_path = Path(json_path)
self.customers_df = None
self.products = {}
self._load_data()
def _load_data(self):
# 1. Load JSON sản phẩm
try:
with self.json_path.open("r", encoding="utf-8") as f:
product_list = json.load(f)
self.products = {item["id"]: item for item in product_list}
print(f"✅ [DataEngine] Đã nạp JSON gói cước: {self.json_path}")
except Exception as e:
print(f"❌ [DataEngine] Lỗi JSON ({self.json_path}): {e}")
# 2. Load CSV khách hàng
try:
self.customers_df = pd.read_csv(self.csv_path)
self.customers_df["Customer ID"] = self.customers_df["Customer ID"].astype(str)
if "id" in self.customers_df.columns:
self.customers_df["id"] = self.customers_df["id"].astype(str).str.strip()
print(f"✅ [DataEngine] Đã nạp CSV khách hàng: {self.csv_path}")
except Exception as e:
print(f"❌ [DataEngine] Lỗi CSV ({self.csv_path}): {e}")
def get_full_context(self, customer_id):
if self.customers_df is None:
return None
str_id = str(customer_id).strip()
row = self.customers_df[self.customers_df["Customer ID"] == str_id]
if row.empty:
return None
customer_data = row.iloc[0].to_dict()
# Xử lý tên gọi
gender = str(customer_data.get("Gender", "")).lower()
if "female" in gender:
display_name = "Chị"
elif "male" in gender:
display_name = "Anh"
else:
display_name = "Quý khách"
customer_data["Display_Name"] = display_name
# Lấy thông tin gói cước
current_pkg_id = str(customer_data.get("id", "")).strip()
pkg_info = self.products.get(
current_pkg_id,
{
"name": f"Gói {current_pkg_id}",
"desc": "Không có mô tả",
"price": 0,
},
)
return {
"customer": customer_data,
"current_package": pkg_info,
}
|