Spaces:
Sleeping
Sleeping
| 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, | |
| } | |