import os, csv, time import requests from datetime import datetime, timedelta, timezone from dateutil.parser import isoparse SQUARE_TOKEN = os.getenv("SQUARE_TOKEN") # set this env var SQUARE_ENV = os.getenv("SQUARE_ENV", "production") # or "sandbox" BASE_URL = "https://connect.squareup.com" if SQUARE_ENV == "production" else "https://connect.squareupsandbox.com" OUT_DIR = os.getenv("SQ_OUT_DIR", "./square_exports") os.makedirs(OUT_DIR, exist_ok=True) HEADERS = { "Authorization": f"Bearer {SQUARE_TOKEN}", "Content-Type": "application/json", "Accept": "application/json", "Square-Version": "2025-01-23" # can be adjusted } def sq_get(path, params=None): r = requests.get(BASE_URL + path, headers=HEADERS, params=params, timeout=60) r.raise_for_status() return r.json() def sq_post(path, payload): r = requests.post(BASE_URL + path, headers=HEADERS, json=payload, timeout=60) if r.status_code >= 400: print("\n=== SQUARE ERROR ===") print("URL:", BASE_URL + path) print("STATUS:", r.status_code) print("PAYLOAD:", payload) print("BODY:", r.text) print("=== /SQUARE ERROR ===\n") r.raise_for_status() return r.json() def write_csv(filename, fieldnames, rows): fp = os.path.join(OUT_DIR, filename) with open(fp, "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=fieldnames) w.writeheader() for row in rows: w.writerow({k: row.get(k) for k in fieldnames}) print(f"Wrote {len(rows)} rows -> {fp}") def export_catalog(): # pulls ITEMS + ITEM_VARIATIONs via catalog search rows = [] cursor = None while True: payload = {"object_types": ["ITEM", "ITEM_VARIATION"], "cursor": cursor} if cursor else {"object_types": ["ITEM", "ITEM_VARIATION"]} data = sq_post("/v2/catalog/search", payload) for obj in data.get("objects", []): t = obj.get("type") base = {"catalog_object_id": obj.get("id"), "type": t, "updated_at": obj.get("updated_at"), "version": obj.get("version"), "raw_json": str(obj)} if t == "ITEM": item = obj.get("item_data", {}) or {} rows.append({ **base, "name": item.get("name"), "variation_name": "", "sku": "", "price_money": "", "currency": "" }) elif t == "ITEM_VARIATION": v = obj.get("item_variation_data", {}) or {} price = v.get("price_money") or {} rows.append({ **base, "name": (v.get("name") or ""), "variation_name": (v.get("name") or ""), "sku": v.get("sku") or "", "price_money": price.get("amount"), "currency": price.get("currency") }) cursor = data.get("cursor") if not cursor: break time.sleep(0.2) write_csv("sq_catalog_items.csv", ["catalog_object_id","type","name","variation_name","sku","price_money","currency","updated_at","version","raw_json"], rows) def export_customers(): rows = [] cursor = None while True: payload = {"cursor": cursor} if cursor else {} data = sq_post("/v2/customers/search", payload) for c in data.get("customers", []): rows.append({ "customer_id": c.get("id"), "created_at": c.get("created_at"), "updated_at": c.get("updated_at"), "given_name": c.get("given_name"), "family_name": c.get("family_name"), "email": c.get("email_address"), "phone": c.get("phone_number"), "reference_id": c.get("reference_id"), "raw_json": str(c) }) cursor = data.get("cursor") if not cursor: break time.sleep(0.2) write_csv("sq_customers.csv", ["customer_id","created_at","updated_at","given_name","family_name","email","phone","reference_id","raw_json"], rows) def get_location_ids(): data = sq_get("/v2/locations") locs = data.get("locations", []) or [] ids = [l.get("id") for l in locs if l.get("id") and (l.get("status") in (None, "ACTIVE"))] print("Found location_ids:", ids) return ids def export_orders(start_iso, end_iso, location_ids=None): orders_rows = [] items_rows = [] cursor = None query = { "filter": { "date_time_filter": {"created_at": {"start_at": start_iso, "end_at": end_iso}}, } } if not location_ids: raise ValueError("location_ids is required for this seller; pass at least one location_id.") while True: payload = {"query": query, "location_ids": location_ids} if cursor: payload["cursor"] = cursor data = sq_post("/v2/orders/search", payload) for o in data.get("orders", []): totals = o.get("total_money") or {} tax = o.get("total_tax_money") or {} disc = o.get("total_discount_money") or {} tip = o.get("total_tip_money") or {} service = o.get("total_service_charge_money") or {} source = (o.get("source") or {}).get("name") orders_rows.append({ "order_id": o.get("id"), "location_id": o.get("location_id"), "created_at": o.get("created_at"), "updated_at": o.get("updated_at"), "state": o.get("state"), "source_name": source, "customer_id": o.get("customer_id"), "total_money": totals.get("amount"), "total_tax_money": tax.get("amount"), "total_discount_money": disc.get("amount"), "total_tip_money": tip.get("amount"), "total_service_charge_money": service.get("amount"), "currency": totals.get("currency"), "raw_json": str(o) }) for li in o.get("line_items", []) or []: base_price = (li.get("base_price_money") or {}).get("amount") gross = (li.get("gross_sales_money") or {}).get("amount") total = (li.get("total_money") or {}).get("amount") tax_m = (li.get("total_tax_money") or {}).get("amount") disc_m = (li.get("total_discount_money") or {}).get("amount") items_rows.append({ "order_id": o.get("id"), "line_item_uid": li.get("uid"), "catalog_object_id": li.get("catalog_object_id"), "name": li.get("name"), "variation_name": (li.get("variation_name") or ""), "quantity": li.get("quantity"), "note": li.get("note"), "base_price_money": base_price, "gross_sales_money": gross, "total_discount_money": disc_m, "total_tax_money": tax_m, "total_money": total, "currency": (li.get("total_money") or {}).get("currency") }) cursor = data.get("cursor") if not cursor: break time.sleep(0.2) write_csv("sq_orders.csv", ["order_id","location_id","created_at","updated_at","state","source_name","customer_id", "total_money","total_tax_money","total_discount_money","total_tip_money","total_service_charge_money", "currency","raw_json"], orders_rows) write_csv("sq_order_items.csv", ["order_id","line_item_uid","catalog_object_id","name","variation_name","quantity","note", "base_price_money","gross_sales_money","total_discount_money","total_tax_money","total_money","currency"], items_rows) def export_payments(start_iso, end_iso): rows = [] cursor = None while True: params = {"begin_time": start_iso, "end_time": end_iso, "cursor": cursor} if cursor else {"begin_time": start_iso, "end_time": end_iso} data = sq_get("/v2/payments", params=params) for p in data.get("payments", []): amount = (p.get("amount_money") or {}).get("amount") tip = (p.get("tip_money") or {}).get("amount") status = p.get("status") tender = p.get("source_type") rows.append({ "payment_id": p.get("id"), "order_id": p.get("order_id"), "location_id": p.get("location_id"), "customer_id": p.get("customer_id"), "created_at": p.get("created_at"), "updated_at": p.get("updated_at"), "status": status, "source_type": tender, "amount_money": amount, "tip_money": tip, "currency": (p.get("amount_money") or {}).get("currency"), "raw_json": str(p) }) cursor = data.get("cursor") if not cursor: break time.sleep(0.2) write_csv("sq_payments.csv", ["payment_id","order_id","location_id","customer_id","created_at","updated_at","status","source_type", "amount_money","tip_money","currency","raw_json"], rows) if __name__ == "__main__": if not SQUARE_TOKEN: raise SystemExit("Set env var SQUARE_TOKEN first (his store token).") # Choose a backfill window (example: last 18 months) end = datetime.now(timezone.utc) start = end - timedelta(days=3000) # ~8 years start_iso = start.strftime("%Y-%m-%dT%H:%M:%SZ") end_iso = end.strftime("%Y-%m-%dT%H:%M:%SZ") export_catalog() export_customers() location_ids = get_location_ids() export_orders(start_iso, end_iso, location_ids=location_ids) export_payments(start_iso, end_iso)