"""Wayo starter: check a local stock file without AI or extra Python packages. Save this file beside inventory.json. Run: python inventory_agent.py Keep this terminal open. Ctrl+C stops the server. This is a small starter, not a replacement for your shop's inventory system. """ import hmac import json import math import os import re import secrets from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, urlsplit INVENTORY_FILE = Path(__file__).with_name("inventory.json") MAX_REQUEST_BYTES = 65536 def words(text): return " ".join(re.findall(r"\w+", text.casefold())) def read_inventory(): items = json.loads(INVENTORY_FILE.read_text(encoding="utf-8-sig")) if not isinstance(items, list) or not items: raise ValueError("inventory.json must contain a non-empty list of products") for item in items: if not isinstance(item, dict): raise ValueError("Each product must be a JSON object") if not isinstance(item.get("name"), str) or not item["name"].strip() or len(item["name"]) > 120: raise ValueError("Each product needs a name of 1-120 characters") if type(item.get("quantity")) is not int or item["quantity"] < 0: raise ValueError("Quantity must be a whole number, zero or more") price = item.get("price") if type(price) not in (int, float) or not math.isfinite(price) or price < 0: raise ValueError("Price must be a number, zero or more") if not isinstance(item.get("currency"), str) or not re.fullmatch(r"[A-Z]{3}", item["currency"]): raise ValueError("Use a three-letter currency such as USD, EUR or EGP") aliases = item.get("aliases", []) if not isinstance(aliases, list) or not all(isinstance(a, str) and a.strip() for a in aliases): raise ValueError("Aliases must be a list of non-empty names") return items def inventory_reply(question): # Read the file again on each question: saved stock edits apply without restarting. items = read_inventory() text = words(question) catalog_requests = { "hello", "hi", "inventory", "stock", "show stock", "show inventory", "what is in stock", "what s in stock", "show all products", "list products", "hello from wayo this is a connection test", } if text in catalog_requests: matches = items else: scored = [] for item in items: names = [item["name"], *item.get("aliases", [])] lengths = [len(words(name).split()) for name in names if f" {words(name)} " in f" {text} "] if lengths: scored.append((max(lengths), item)) # Prefer "bike helmet" over the shorter "bike" alias. Ask about one item at a time. best = max((score for score, _ in scored), default=0) matches = [item for score, item in scored if score == best] if not matches: names = ", ".join(item["name"] for item in items[:10]) return { "reply": f"I don't recognize that product. I can check: {names}. Ask about one item by name.", "can_help": False, "offers": [], } shown = matches[:10] lines = [f"{item['name']}: {item['quantity']} in stock; {item['price']} {item['currency']} each." for item in shown] if len(matches) > len(shown): lines.append(f"Showing {len(shown)} of {len(matches)} products. Ask for a product by name.") return { "reply": "\n".join(lines), "can_help": True, "offers": [ { "title": item["name"], "price": item["price"], "currency": item["currency"], "per": "item", "available": item["quantity"] > 0, "notes": f"{item['quantity']} in stock", } for item in shown ], } class InventoryHandler(BaseHTTPRequestHandler): def setup(self): super().setup() self.connection.settimeout(10) def log_message(self, *_): pass # Do not log questions, request headers or the access token. def send_json(self, status, data): body = json.dumps(data, ensure_ascii=False, allow_nan=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def authorized(self): provided = self.headers.get("X-API-Key", "") if hmac.compare_digest(provided.encode(), self.server.access_token.encode()): return True self.send_json(401, {"reply": "The X-API-Key is missing or incorrect."}) return False def answer(self, question): if not isinstance(question, str) or not question.strip(): self.send_json(400, {"reply": "Send a non-empty message (or text for Webhook, q for REST)."}) return try: self.send_json(200, inventory_reply(question)) except (OSError, ValueError, TypeError, OverflowError): self.send_json( 503, {"reply": "The inventory file needs attention. Ask the business to check inventory.json."} ) def do_GET(self): url = urlsplit(self.path) if url.path == "/": self.send_json(200, {"status": "running", "next": "Use Chat API with the /chat endpoint."}) elif url.path == "/search": if self.authorized(): self.answer(parse_qs(url.query).get("q", [""])[0]) else: self.send_json(404, {"reply": "Use /chat, /webhook or /search."}) def do_POST(self): path = urlsplit(self.path).path if path not in ("/chat", "/webhook"): self.send_json(404, {"reply": "Use /chat or /webhook for POST."}) return if not self.authorized(): return try: length = int(self.headers.get("Content-Length", "0")) if length < 1 or length > MAX_REQUEST_BYTES: self.send_json(413, {"reply": "Send a JSON request between 1 and 65536 bytes."}) return data = json.loads(self.rfile.read(length)) if not isinstance(data, dict): raise ValueError("Expected an object") except (ValueError, UnicodeError): self.send_json(400, {"reply": "The request must be a JSON object."}) return self.answer(data.get("text" if path == "/webhook" else "message")) def main(): try: read_inventory() except (OSError, ValueError, TypeError, OverflowError) as error: raise SystemExit(f"Check inventory.json before starting: {error}") from error token = os.environ.get("WAYO_INVENTORY_TOKEN") or secrets.token_urlsafe(32) server = ThreadingHTTPServer(("127.0.0.1", 8080), InventoryHandler) server.access_token = token print("Inventory starter is running at http://127.0.0.1:8080", flush=True) print("Wayo connection: Chat API | endpoint: your public HTTPS address + /chat", flush=True) print("Sign-in method: API key header | Header name: X-API-Key", flush=True) print(f"Token (paste into Wayo's Key field): {token}", flush=True) print("Keep this window open. Ctrl+C stops the server.", flush=True) print("A new token is generated on restart unless WAYO_INVENTORY_TOKEN is set.", flush=True) try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() if __name__ == "__main__": main()