|
|
from datetime import datetime |
|
|
from pathlib import Path |
|
|
import csv |
|
|
import json |
|
|
import re |
|
|
from typing import Dict, List |
|
|
|
|
|
BASE = Path(__file__).parent |
|
|
LOGS = BASE / "logs" |
|
|
DATA = BASE / "data" |
|
|
LOGS.mkdir(exist_ok=True) |
|
|
|
|
|
CATALOG = json.loads((DATA / "catalog.json").read_text(encoding="utf-8")) |
|
|
PRICE_RULES = json.loads((DATA / "price_rules.json").read_text(encoding="utf-8")) |
|
|
IKEA_CSV = DATA / "IKEA_SA_Furniture_Web_Scrapings_sss.csv" |
|
|
SAR_TO_USD = 0.2667 |
|
|
|
|
|
|
|
|
def _to_float(value: str): |
|
|
if value is None: |
|
|
return None |
|
|
value = value.strip() |
|
|
if not value or value.lower().startswith("no "): |
|
|
return None |
|
|
try: |
|
|
return float(value) |
|
|
except ValueError: |
|
|
return None |
|
|
|
|
|
|
|
|
def _to_bool(value: str): |
|
|
if value is None: |
|
|
return None |
|
|
value = value.strip().lower() |
|
|
if value in {"true", "yes", "y", "1"}: |
|
|
return True |
|
|
if value in {"false", "no", "n", "0"}: |
|
|
return False |
|
|
return None |
|
|
|
|
|
|
|
|
def _load_ikea_catalog() -> List[Dict]: |
|
|
if not IKEA_CSV.exists(): |
|
|
fallback = BASE / "IKEA_SA_Furniture_Web_Scrapings_sss.csv" |
|
|
if not fallback.exists(): |
|
|
return [] |
|
|
target = fallback |
|
|
else: |
|
|
target = IKEA_CSV |
|
|
|
|
|
items: List[Dict] = [] |
|
|
with target.open("r", encoding="utf-8", newline="") as fh: |
|
|
reader = csv.DictReader(fh) |
|
|
for row in reader: |
|
|
item_id = (row.get("item_id") or "").strip() |
|
|
name = (row.get("name") or "").strip() |
|
|
if not item_id or not name: |
|
|
continue |
|
|
category = (row.get("category") or "").strip() |
|
|
price_sar = _to_float(row.get("price")) |
|
|
price_usd = round(price_sar * SAR_TO_USD, 2) if price_sar is not None else None |
|
|
width = _to_float(row.get("width")) |
|
|
height = _to_float(row.get("height")) |
|
|
depth = _to_float(row.get("depth")) |
|
|
short_description = (row.get("short_description") or "").strip() |
|
|
if short_description: |
|
|
short_description = re.sub(r"\s+", " ", short_description) |
|
|
other_colors = (row.get("other_colors") or "").strip() |
|
|
if other_colors.lower() in {"no", "n/a"}: |
|
|
other_colors = "" |
|
|
sellable = _to_bool(row.get("sellable_online")) |
|
|
link = (row.get("link") or "").strip() |
|
|
designer = (row.get("designer") or "").strip() |
|
|
|
|
|
searchable = " ".join( |
|
|
filter( |
|
|
None, |
|
|
[ |
|
|
item_id.lower(), |
|
|
name.lower(), |
|
|
category.lower(), |
|
|
short_description.lower(), |
|
|
other_colors.lower(), |
|
|
designer.lower(), |
|
|
], |
|
|
) |
|
|
) |
|
|
|
|
|
items.append( |
|
|
{ |
|
|
"item_id": item_id, |
|
|
"name": name, |
|
|
"category": category, |
|
|
"price_usd": price_usd, |
|
|
"price_currency": "USD" if price_usd is not None else None, |
|
|
"price_note": ( |
|
|
f"Converted from SAR at 1 SAR = {SAR_TO_USD:.4f} USD" |
|
|
if price_usd is not None |
|
|
else None |
|
|
), |
|
|
"sellable_online": sellable, |
|
|
"link": link, |
|
|
"other_colors": other_colors, |
|
|
"short_description": short_description, |
|
|
"designer": designer, |
|
|
"dimensions_cm": { |
|
|
k: v |
|
|
for k, v in {"width": width, "height": height, "depth": depth}.items() |
|
|
if v is not None |
|
|
}, |
|
|
"_search": searchable, |
|
|
} |
|
|
) |
|
|
return items |
|
|
|
|
|
|
|
|
IKEA_ITEMS = _load_ikea_catalog() |
|
|
|
|
|
|
|
|
def _copy_public_item(item: Dict) -> Dict: |
|
|
return {k: v for k, v in item.items() if not k.startswith("_")} |
|
|
|
|
|
|
|
|
def _search_ikea_items(query: str, limit: int = 5) -> List[Dict]: |
|
|
if not IKEA_ITEMS: |
|
|
return [] |
|
|
q = query.strip().lower() |
|
|
if not q: |
|
|
return [] |
|
|
words = [w for w in re.split(r"\W+", q) if w] |
|
|
scored: Dict[str, List] = {} |
|
|
for item in IKEA_ITEMS: |
|
|
score = 0 |
|
|
if q == item["item_id"].lower(): |
|
|
score += 10 |
|
|
if q in item["_search"]: |
|
|
score += 3 |
|
|
if words: |
|
|
score += sum(1 for w in words if w and w in item["_search"]) |
|
|
if score > 0: |
|
|
existing = scored.get(item["item_id"]) |
|
|
if existing is None or score > existing[0]: |
|
|
scored[item["item_id"]] = [score, item] |
|
|
if not scored: |
|
|
return [] |
|
|
top = sorted(scored.values(), key=lambda pair: (-pair[0], pair[1]["name"])) |
|
|
return [_copy_public_item(item) for _, item in top[:limit]] |
|
|
|
|
|
|
|
|
def record_customer_interest(email: str, name: str, message: str): |
|
|
entry = {"ts": datetime.utcnow().isoformat(), "email": email, "name": name, "message": message} |
|
|
out = LOGS / "leads.jsonl" |
|
|
with out.open("a", encoding="utf-8") as fh: |
|
|
fh.write(json.dumps(entry) + "\n") |
|
|
print(f"[LEAD] {entry}") |
|
|
return {"ok": True, "msg": "Thanks! We'll follow up soon."} |
|
|
|
|
|
|
|
|
def record_feedback(question: str): |
|
|
entry = {"ts": datetime.utcnow().isoformat(), "question": question} |
|
|
out = LOGS / "feedback.jsonl" |
|
|
with out.open("a", encoding="utf-8") as fh: |
|
|
fh.write(json.dumps(entry) + "\n") |
|
|
print(f"[FEEDBACK] {entry}") |
|
|
return {"ok": True, "msg": "Noted. We'll improve our answers."} |
|
|
|
|
|
|
|
|
def record_service_feedback( |
|
|
email: str, |
|
|
name: str, |
|
|
service_type: str, |
|
|
satisfaction: str, |
|
|
comments: str = "", |
|
|
): |
|
|
entry = { |
|
|
"ts": datetime.utcnow().isoformat(), |
|
|
"email": email, |
|
|
"name": name, |
|
|
"service_type": service_type, |
|
|
"satisfaction": satisfaction, |
|
|
"comments": comments or "", |
|
|
} |
|
|
out = LOGS / "service_feedback.jsonl" |
|
|
with out.open("a", encoding="utf-8") as fh: |
|
|
fh.write(json.dumps(entry) + "\n") |
|
|
print(f"[SERVICE_FEEDBACK] {entry}") |
|
|
return {"ok": True, "msg": "Thanks for the feedback! We'll share it with the team."} |
|
|
|
|
|
|
|
|
def lookup_product(query: str): |
|
|
q = (query or "").strip() |
|
|
if not q: |
|
|
return {"ok": False, "msg": "Please provide a product keyword, SKU, or IKEA item ID."} |
|
|
q_lower = q.lower() |
|
|
result = {"ok": True, "query": q} |
|
|
|
|
|
sku_match = next((item for item in CATALOG if item.get("sku", "").lower() == q_lower), None) |
|
|
if sku_match: |
|
|
result["catalog_match"] = sku_match |
|
|
|
|
|
name_hits = [ |
|
|
item |
|
|
for item in CATALOG |
|
|
if q_lower in item.get("name", "").lower() |
|
|
or any(q_lower in (opt or "").lower() for opt in item.get("color_options", [])) |
|
|
] |
|
|
if name_hits and not sku_match: |
|
|
result["catalog_results"] = name_hits |
|
|
|
|
|
category_hits = [item for item in CATALOG if item.get("category", "").lower() == q_lower] |
|
|
if category_hits: |
|
|
result["catalog_category"] = category_hits |
|
|
|
|
|
ikea_hits = _search_ikea_items(q_lower) |
|
|
if ikea_hits: |
|
|
result["ikea_results"] = ikea_hits |
|
|
|
|
|
if len(result) == 2: |
|
|
return {"ok": False, "msg": f"No products found for '{q}'."} |
|
|
return result |
|
|
|
|
|
|
|
|
def estimate_repair(issue: str, material: str = "any", size_category: str = "medium"): |
|
|
issue = issue.strip().lower() |
|
|
material = (material or "any").strip().lower() |
|
|
size = (size_category or "medium").strip().lower() |
|
|
rules = PRICE_RULES.get(issue) |
|
|
if not rules: |
|
|
return {"ok": False, "msg": f"No pricing rule for issue '{issue}'."} |
|
|
if material in rules: |
|
|
bucket = rules[material] |
|
|
elif "any" in rules: |
|
|
bucket = rules["any"] |
|
|
else: |
|
|
bucket = next(iter(rules.values())) |
|
|
if size not in bucket: |
|
|
return {"ok": False, "msg": f"Unsupported size_category '{size}'. Use small/medium/large."} |
|
|
min_p, max_p, min_d, max_d = bucket[size] |
|
|
tiers = { |
|
|
"budget": {"price": round(min_p * 0.9), "days": [min_d, max(min_d, min_d + 1)]}, |
|
|
"standard": {"price": round((min_p + max_p) / 2), "days": [min_d, max_d]}, |
|
|
"rush": {"price": round(max_p * 1.25), "days": [max(1, min_d - 1), max(1, max_d - 1)]}, |
|
|
} |
|
|
return {"ok": True, "issue": issue, "material": material, "size": size, "estimate": tiers} |
|
|
|
