"""Read-only Lightspeed R-Series data access. No third-party dependencies. Only Account, Shop, Item and Sale GET requests are implemented. OAuth is handled separately by FastMCP. All values from Lightspeed are data, never instructions. """ from __future__ import annotations import base64 import hashlib import hmac import json import re import time from datetime import date, datetime, time as daytime, timedelta, timezone from decimal import Decimal, InvalidOperation from urllib.error import HTTPError, URLError from urllib.parse import urlencode, urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener from zoneinfo import ZoneInfo API = "https://api.lightspeedapp.com/API/V3/Account" TZ = ZoneInfo("America/Toronto") class ReportError(Exception): """A safe error message suitable for a tool result.""" def numeric_id(value): value = str(value) if not re.fullmatch(r"[1-9][0-9]{0,19}", value): raise ReportError("Use a positive Lightspeed ID returned by this connector.") return value def records(document, key): value = document.get(key) if value is None: return [] if isinstance(value, dict): return [value] if isinstance(value, list) and all(isinstance(x, dict) for x in value): return value raise ReportError("Lightspeed returned an unexpected record format.") def relation(document, name, key): value = document.get(name) if value is None: return None # Missing is different from an empty relation. if not isinstance(value, dict): raise ReportError("Lightspeed returned an unexpected relation format.") return records(value, key) def truth(value): if value in (True, "true", "1", 1): return True if value in (False, "false", "0", 0): return False raise ReportError("Lightspeed omitted a required status field.") def number(value): if value is None or isinstance(value, bool): raise ReportError("A required numeric value is missing; no total was calculated.") try: result = Decimal(str(value)) except InvalidOperation: raise ReportError("Lightspeed returned an invalid numeric value.") from None if not result.is_finite(): raise ReportError("Lightspeed returned a non-finite numeric value.") return result def instant(value): try: result = datetime.fromisoformat(str(value).replace("Z", "+00:00")) if result.tzinfo is None: raise ValueError() return result.astimezone(timezone.utc) except (ValueError, TypeError): raise ReportError("Lightspeed omitted a valid completion time with timezone.") from None def date_window(start_date, end_date): try: if not all(re.fullmatch(r"\d{4}-\d{2}-\d{2}", v) for v in [start_date, end_date]): raise ValueError() start, end = date.fromisoformat(start_date), date.fromisoformat(end_date) if not 0 <= (end - start).days <= 365: raise ValueError() return (datetime.combine(start, daytime.min, TZ).astimezone(timezone.utc), datetime.combine(end + timedelta(days=1), daytime.min, TZ).astimezone(timezone.utc)) except (ValueError, TypeError, OverflowError): raise ReportError("Supply inclusive YYYY-MM-DD dates, in order, up to 366 days apart.") from None class NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None # Never send a bearer credential to a redirected URL. def get_json(url, token): request = Request(url, headers={"Authorization": "Bearer " + token, "Accept": "application/json", "User-Agent": "PawBarnReporting/0.1"}, method="GET") try: with build_opener(NoRedirect).open(request, timeout=20) as response: raw = response.read(8_000_001) if len(raw) > 8_000_000: raise ReportError("The API response is too large. No complete report was produced.") document = json.loads(raw) if not isinstance(document, dict): raise ValueError() if "Error" in document or "error" in document: raise ReportError("Lightspeed returned an error. No complete report was produced.") return document except HTTPError as exc: if exc.code in (401, 403): raise ReportError("Lightspeed denied access. Reconnect and check employee permissions.") from None if exc.code == 429: raise ReportError("Lightspeed rate limit reached. Wait briefly and retry this page.") from None raise ReportError(f"Lightspeed request failed (HTTP {exc.code}); no complete report was produced.") from None except (URLError, TimeoutError, ValueError, UnicodeError): raise ReportError("Lightspeed response unavailable or invalid. Retry; do not treat this as zero activity.") from None class Reporter: def __init__(self, account_id, cursor_key, fetch=get_json, clock=time.time): self.account = numeric_id(account_id) self.key = cursor_key.encode() if len(self.key) < 32: raise ValueError("Cursor key must be at least 32 characters") self.fetch = fetch self.clock = clock def endpoint(self, entity, item_id=None): if entity not in {"Shop", "Item", "Sale"}: raise ReportError("Unsupported resource.") suffix = "/" + numeric_id(item_id) if item_id is not None else "" return f"{API}/{self.account}/{entity}{suffix}.json" def validate_url(self, url, entity, item_id=None): expected = urlsplit(self.endpoint(entity, item_id)) actual = urlsplit(url) if (actual.scheme != "https" or actual.netloc != expected.netloc or actual.path != expected.path or actual.fragment or actual.username or actual.password): raise ReportError("Unsafe or unexpected pagination URL rejected.") return url def page(self, token, entity, params=None, url=None, item_id=None): url = url or self.endpoint(entity, item_id) + "?" + urlencode(params or {}) self.validate_url(url, entity, item_id) document = self.fetch(url, token) if not isinstance(document, dict) or "Error" in document or "error" in document: raise ReportError("Lightspeed returned an invalid page.") attributes = document.get("@attributes", {}) if not isinstance(attributes, dict): raise ReportError("Lightspeed returned invalid pagination metadata.") rows = records(document, entity) # Empty sets are valid, but an arbitrary/error JSON object is not a set. if entity not in document and str(attributes.get("count")) != "0": raise ReportError("Lightspeed did not confirm an empty result set.") next_url = attributes.get("next") if next_url is None and item_id is None: # V3 normally includes next="" at the end. Fail closed if absent. if str(attributes.get("count")) != "0": raise ReportError("Pagination metadata is missing; completeness cannot be established.") next_url = "" if next_url: if not isinstance(next_url, str): raise ReportError("Invalid pagination metadata.") self.validate_url(next_url, entity) if next_url == url: raise ReportError("Repeated pagination URL; completeness cannot be established.") return rows, next_url or "" def account_allowed(self, token): document = self.fetch(API + ".json", token) return any(str(a.get("accountID")) == self.account for a in records(document, "Account")) def encode_cursor(self, state): state = {**state, "account": self.account, "expires": int(self.clock()) + 3600} data = base64.urlsafe_b64encode(json.dumps(state, separators=(",", ":")).encode()).rstrip(b"=") signature = hmac.new(self.key, data, hashlib.sha256).hexdigest() return data.decode() + "." + signature def decode_cursor(self, cursor, query): try: if len(cursor) > 24000: raise ValueError() data, signature = cursor.split(".") expected = hmac.new(self.key, data.encode(), hashlib.sha256).hexdigest() if not hmac.compare_digest(signature, expected): raise ValueError() state = json.loads(base64.urlsafe_b64decode(data + "=" * (-len(data) % 4))) if (state["account"] != self.account or state["query"] != query or state["expires"] < self.clock()): raise ValueError() return state except (ValueError, TypeError, KeyError, UnicodeError): raise ReportError("This cursor is invalid, expired, or belongs to different filters. Restart the report.") from None def metadata(self): return {"account_id": self.account, "source": "Lightspeed Retail R-Series API V3", "retrieved_at_utc": datetime.fromtimestamp(self.clock(), timezone.utc).isoformat(), "timezone": "America/Toronto", "currency": "CAD (installation setting; verify in Lightspeed)", "live_data": True, "snapshot_warning": "Live API pages are not a frozen snapshot. Reconcile a closed period before business decisions."} def stores(self, token): rows, next_url = self.page(token, "Shop", {"limit": 100}) if next_url: raise ReportError("More than one page of shops; store list is incomplete.") return {**self.metadata(), "complete": True, "stores": [{k: r.get(k) for k in ("shopID", "name", "archived")} for r in rows]} @staticmethod def clean_item(row): result = {k: row.get(k) for k in ("itemID", "systemSku", "customSku", "upc", "ean", "description", "archived", "itemType", "categoryID", "manufacturerID")} shops = relation(row, "ItemShops", "ItemShop") result["store_inventory"] = None if shops is None else [ {k: shop.get(k) for k in ("shopID", "qoh", "reorderPoint", "reorderLevel", "backorder", "onLayaway", "onSpecialOrder", "onWorkOrder", "onTransferIn", "onTransferOut")} for shop in shops if str(shop.get("shopID")) != "0"] result["inventory_available"] = shops is not None # shopID 0 is an account summary, never another physical store. # Costs, notes, customer and employee details are deliberately not returned. return result def inventory(self, token, cursor=None, archived=False): query = {"kind": "inventory", "archived": bool(archived)} state = self.decode_cursor(cursor, query) if cursor else {"page": 0} rows, next_url = self.page(token, "Item", {"limit": 100, "sort": "itemID", "archived": str(bool(archived)).lower(), "load_relations": json.dumps(["ItemShops"])}, url=state.get("url")) output_cursor = self.encode_cursor({"query": query, "url": next_url, "page": state["page"] + 1}) if next_url else None return {**self.metadata(), "page": state["page"] + 1, "complete": not bool(next_url), "coverage": "This page only. Collect every page before making catalog-wide claims.", "archived": bool(archived), "next_cursor": output_cursor, "items": [self.clean_item(row) for row in rows]} def product(self, token, item_id): rows, _ = self.page(token, "Item", {"load_relations": json.dumps(["ItemShops"])}, item_id=item_id) if len(rows) != 1: raise ReportError("Product was not found or was not uniquely identified.") return {**self.metadata(), "item": self.clean_item(rows[0])} @staticmethod def clean_sale(row): required = ("saleID", "shopID", "completeTime", "calcSubtotal", "calcDiscount", "calcTax1", "calcTax2", "calcTotal") result = {k: row.get(k) for k in required} # Keep missing values null; do not silently turn unavailable data into zero. lines = relation(row, "SaleLines", "SaleLine") fields = ("saleLineID", "itemID", "shopID", "unitQuantity", "unitPrice", "calcSubtotal", "calcLineDiscount", "calcTransactionDiscount", "calcTax1", "calcTax2", "calcTotal") result["lines"] = None if lines is None else [{k: r.get(k) for k in fields} for r in lines] result["line_details_available"] = lines is not None return result def sales(self, token, start_date, end_date, shop_id=None, cursor=None): start, end = date_window(start_date, end_date) shop_id = numeric_id(shop_id) if shop_id is not None else None query = {"kind": "sales", "start": start_date, "end": end_date, "shop": shop_id} state = self.decode_cursor(cursor, query) if cursor else {"page": 0, "last": None, "anchor": datetime.fromtimestamp(self.clock(), timezone.utc).isoformat()} params = {"limit": 100, "sort": "-completeTime", "completed": "true", "voided": "false", "load_relations": json.dumps(["SaleLines"])} if shop_id: params["shopID"] = shop_id rows, next_url = self.page(token, "Sale", params, url=state.get("url")) previous = instant(state["last"]) if state.get("last") else None anchor = instant(state["anchor"]) matched = [] reached_start = False seen = set() for row in rows: if not truth(row.get("completed")) or truth(row.get("voided")): raise ReportError("Lightspeed did not apply the completed/voided sale filters.") if shop_id is not None and str(row.get("shopID")) != shop_id: raise ReportError("Lightspeed did not apply the store filter.") sale_id = numeric_id(row.get("saleID")) if sale_id in seen: raise ReportError("Duplicate sale on an API page; restart the report.") seen.add(sale_id) completed = instant(row.get("completeTime")) if previous is not None and completed > previous: raise ReportError("Sales were not returned in time order; completeness cannot be established.") previous = completed if completed < start: reached_start = True if start <= completed < end and completed <= anchor: matched.append(self.clean_sale(row)) complete = reached_start or not next_url output_cursor = None if complete else self.encode_cursor({"query": query, "url": next_url, "last": previous.isoformat() if previous else state.get("last"), "anchor": state["anchor"], "page": state["page"] + 1}) return {**self.metadata(), "start_date": start_date, "end_date": end_date, "start_utc": start.isoformat(), "end_exclusive_utc": end.isoformat(), "scan_started_at_utc": state["anchor"], "shop_id": shop_id, "page": state["page"] + 1, "complete": complete, "next_cursor": output_cursor, "coverage": "This page only. complete=true means the last page of the scan; earlier pages must also be retained.", "records_scanned_this_page": len(rows), "sales": matched, "definitions": "Completed, non-voided sales by completeTime. Refund signs preserved. Store is sale shopID. CAD amounts before any reinterpretation as profit."} def summary(self, token, start_date, end_date, shop_id=None, max_pages=10): if isinstance(max_pages, bool) or not 1 <= max_pages <= 20: raise ReportError("max_pages must be between 1 and 20.") cursor, seen, groups = None, set(), {} deadline = self.clock() + 45 for index in range(max_pages): if self.clock() >= deadline: raise ReportError("Report time limit reached. Use sales_page and collect every page; no total was produced.") page = self.sales(token, start_date, end_date, shop_id, cursor) for sale in page["sales"]: sid = numeric_id(sale["saleID"]) if sid in seen: raise ReportError("Duplicate sale across pages. No total was produced; restart the report.") seen.add(sid) store = numeric_id(sale["shopID"]) totals = groups.setdefault(store, {"sale_count": 0, "subtotal": Decimal(0), "discount": Decimal(0), "tax": Decimal(0), "total_including_tax": Decimal(0)}) totals["sale_count"] += 1 totals["subtotal"] += number(sale["calcSubtotal"]) totals["discount"] += number(sale["calcDiscount"]) totals["tax"] += number(sale["calcTax1"]) + number(sale["calcTax2"]) totals["total_including_tax"] += number(sale["calcTotal"]) if page["complete"]: output = [] for shop, totals in sorted(groups.items()): totals["subtotal_less_discount"] = totals["subtotal"] - totals["discount"] output.append({"shop_id": shop, **{k: str(v) if isinstance(v, Decimal) else v for k, v in totals.items()}}) return {**self.metadata(), "complete": True, "pages_scanned": index + 1, "start_date": start_date, "end_date": end_date, "sales_count": len(seen), "stores": output, "basis": "Sums of API sale header fields. subtotal_less_discount = calcSubtotal - calcDiscount. Not accounting revenue, profit, cash receipts or a native Lightspeed report.", "reconciliation_required": "Check a closed day against Lightspeed with the same store/time/tax/return rules, especially deposits, gift cards and rounding."} cursor = page["next_cursor"] raise ReportError("Page limit reached. No total was produced. Use sales_page and retain every page until complete=true.")