#!/usr/bin/env python3 """ Build the messy source file the ingester has to survive. Apex Instruments is fictional. Every name, figure, and contact is demo data. Produced by the Company OS spreadsheet system for djvonfrank.com. Writes: data/raw-orders.csv about 400 order rows, deliberately dirty, header row not first data/partners.csv the partner lookup the ingester joins against The mess is seeded, so the file is reproducible: run this again and you get the same bytes. Every defect below is one a real export has actually produced. """ import csv, os, random from datetime import date, timedelta HERE = os.path.dirname(os.path.abspath(__file__)) DATA = os.path.join(HERE, "data") os.makedirs(DATA, exist_ok=True) RNG = random.Random(20260930) # ---------------------------------------------------------------- catalog --- # code, canonical name, list price, standard unit cost, share of orders, qty range CATALOG = [ ("MB", "Meridian Bench", 4850.00, 2910.00, 0.14, (1, 3)), ("MP", "Meridian Pro", 7900.00, 4345.00, 0.09, (1, 2)), ("MFK", "Meridian Field Kit", 1240.00, 806.00, 0.22, (2, 8)), ("ACP", "Atlas Care Plus", 1788.00, 447.00, 0.18, (1, 12)), ("CAL", "Calibration service", 180.00, 99.00, 0.25, (1, 6)), ("TRN", "On-site training", 650.00, 429.00, 0.12, (1, 3)), ] # How each product name is actually typed into the source system. ALIASES = { "MB": ["Meridian Bench", "meridian bench", "MERIDIAN BENCH", "Meridian Bench", "Bench"], "MP": ["Meridian Pro", "meridian pro", "Meridian PRO", "Meridian Pro ", "Pro"], "MFK": ["Meridian Field Kit", "Field Kit", "meridian field kit", "Meridian Fieldkit", "MERIDIAN FIELD KIT"], "ACP": ["Atlas Care Plus", "atlas care plus", "Atlas Care+", "ATLAS CARE PLUS", "Care Plus"], "CAL": ["Calibration service", "Calibration", "calibration svc", "CALIBRATION SERVICE"], "TRN": ["On-site training", "Onsite training", "On site training", "TRAINING", "on-site training"], } PARTNERS = [ ("NLG-011", "Northlight Group", "Gulf Coast", "Tampa"), ("NLG-014", "Northlight Group", "Gulf Coast", "St. Petersburg"), ("NLG-022", "Northlight Group", "Central Florida", "Orlando"), ("NLG-027", "Northlight Group", "Central Florida", "Lakeland"), ("NLG-031", "Northlight Group", "Gulf Coast", "Sarasota"), ("NLG-036", "Northlight Group", "Gulf Coast", "Clearwater"), ("NLG-044", "Northlight Group", "Central Florida", "Brandon"), ("DIR-000", "Apex Instruments direct", "House", "Tampa"), ] PARTNER_IDS = [p[0] for p in PARTNERS] REPS = ["Jordan Vale", "Avery Kellen", "Morgan Reyes", "Sami Torres"] STATUSES = ["Shipped", "Shipped", "Shipped", "Invoiced", "Invoiced", "Open"] Q_START = date(2026, 7, 1) Q_DAYS = (date(2026, 9, 30) - Q_START).days # 91 days in the quarter # ------------------------------------------------------------ formatters ---- def fmt_date(d, style): if style == 0: return d.isoformat() # 2026-07-14 if style == 1: return "%02d/%02d/%d" % (d.month, d.day, d.year) # 07/14/2026 if style == 2: return d.strftime("%d-%b-%Y") # 14-Jul-2026 if style == 3: return "%s %d, %d" % (d.strftime("%B"), d.day, d.year) # July 14, 2026 return "%d/%d/%s" % (d.month, d.day, str(d.year)[2:]) # 7/14/26 def fmt_money(v, style): if style == 0: return "$%s" % ("{:,.2f}".format(v)) # $7,900.00 if style == 1: return "{:,.2f}".format(v) # 7,900.00 if style == 2: return "%.2f" % v # 7900.00 if style == 3: return "$%s" % ("{:,.0f}".format(v)) if v == int(v) else "$%.2f" % v return str(int(v)) if v == int(v) else "%.2f" % v # 7900 def fmt_pct(p, style): if p == 0: return RNG.choice(["0", "0%", "", "0.00"]) if style == 0: return "%d%%" % round(p * 100) # 5% if style == 1: return "%.2f" % p # 0.05 if style == 2: return "%.1f%%" % (p * 100) # 5.0% return "%.4f" % p def messy_space(s): """Trailing space, leading space, or a doubled internal space.""" k = RNG.random() if k < 0.34: return s + " " if k < 0.60: return " " + s if " " in s and k < 0.85: i = s.index(" ") return s[:i] + " " + s[i + 1:] return s + " " # --------------------------------------------------------------- rows ------- HEADERS = ["order_id", "order_date", "partner_id", "ship_city", "product", "qty", "unit_price", "discount", "ship_cost", "channel", "rep", "status", "notes"] NOTES = [ "", "", "", "", "", "Rush requested", "Ship with calibration cert", "PO on file", "Partner pickup", "Split shipment", "Backorder cleared", "Quote 2026-Q3 pricing", "Renewal", "Replaces damaged unit", ] TOTAL = 400 rows = [] used_ids = [] weights = [c[4] for c in CATALOG] for i in range(TOTAL): oid = "AP-%05d" % (30000 + i * 7 + RNG.randint(0, 4)) while oid in used_ids: oid = "AP-%05d" % (30000 + i * 7 + RNG.randint(0, 6)) used_ids.append(oid) d = Q_START + timedelta(days=RNG.randint(0, Q_DAYS)) code, name, price, cost, _, qrange = RNG.choices(CATALOG, weights=weights, k=1)[0] qty = RNG.randint(*qrange) disc = RNG.choices([0.0, 0.0, 0.0, 0.03, 0.05, 0.05, 0.08, 0.10, 0.12, 0.15], weights=[26, 14, 10, 10, 12, 8, 8, 6, 4, 2], k=1)[0] pid = RNG.choices(PARTNER_IDS, weights=[13, 15, 14, 9, 12, 11, 8, 18], k=1)[0] prow = [p for p in PARTNERS if p[0] == pid][0] city = prow[3] channel = "Direct" if pid == "DIR-000" else "Partner" ship = 0.0 if code in ("CAL", "TRN", "ACP") else round(RNG.uniform(28, 190), 2) r = { "order_id": oid, "order_date": fmt_date(d, RNG.choices([0, 1, 2, 3, 4], weights=[40, 26, 14, 12, 8], k=1)[0]), "partner_id": pid, "ship_city": city, "product": RNG.choices(ALIASES[code], weights=[52, 14, 12, 12, 10], k=1)[0] if len(ALIASES[code]) == 5 else RNG.choice(ALIASES[code]), "qty": str(qty), "unit_price": fmt_money(price, RNG.choices([0, 1, 2, 3, 4], weights=[34, 18, 16, 18, 14], k=1)[0]), "discount": fmt_pct(disc, RNG.choices([0, 1, 2, 3], weights=[46, 30, 14, 10], k=1)[0]), "ship_cost": fmt_money(ship, RNG.choice([0, 2, 4])) if ship else RNG.choice(["0", "0.00", "", "--"]), "channel": channel, "rep": RNG.choice(REPS), "status": RNG.choice(STATUSES), "notes": RNG.choice(NOTES), "_true": {"code": code, "date": d.isoformat(), "qty": qty, "price": price, "disc": disc}, } # ---- sprinkle survivable defects ------------------------------------- if RNG.random() < 0.30: r["rep"] = messy_space(r["rep"]) if RNG.random() < 0.22: r["partner_id"] = RNG.choice([pid.lower(), " " + pid, pid + " "]) if RNG.random() < 0.26: r["channel"] = RNG.choice([channel.lower(), channel.upper(), channel + " "]) if RNG.random() < 0.24: r["status"] = RNG.choice([r["status"].lower(), r["status"].upper(), " " + r["status"]]) if city == "St. Petersburg" and RNG.random() < 0.55: r["ship_city"] = RNG.choice(["St Petersburg", "st petersburg", "ST PETERSBURG"]) elif RNG.random() < 0.14: r["ship_city"] = RNG.choice([city.upper(), city.lower(), city + " "]) if RNG.random() < 0.07: r["qty"] = "%.1f" % qty # 2.0 instead of 2 if RNG.random() < 0.06: r["unit_price"] = "" # fill from the catalog if RNG.random() < 0.05: r["ship_city"] = "" # fill from the partner record if RNG.random() < 0.05: r["ship_cost"] = RNG.choice(["N/A", "--", "TBD", "n/a"]) if RNG.random() < 0.04: r["notes"] = messy_space(r["notes"]) if r["notes"] else " " rows.append(r) # ------------------------------------------------------- planted rejects ---- # Each one is a row the ingester must refuse, not quietly repair. def at(i): return rows[i] # 3 duplicate order ids (the later row is the one that gets held back) for src, dst in ((12, 141), (58, 233), (100, 318)): at(dst)["order_id"] = at(src)["order_id"] # 5 missing required fields at(37)["product"] = "" at(96)["product"] = " " at(174)["qty"] = "" at(266)["order_date"] = "" at(349)["order_id"] = "" # 2 negative quantities at(64)["qty"] = "-2" at(288)["qty"] = "-1" # 2 dates nothing can parse at(83)["order_date"] = "Q3" at(211)["order_date"] = "2026-13-45" # 2 products that are not in the catalog at(129)["product"] = "Meridian Ultra" at(304)["product"] = "Bench Riser Kit" # 3 partner ids that are not in the partner list at(45)["partner_id"] = "NLG-099" at(190)["partner_id"] = "XX-001" at(355)["partner_id"] = "NLG-052" # 2 numbers that are not numbers at(158)["qty"] = "two" at(272)["unit_price"] = "call for price" # ------------------------------------------------------------- write out ---- with open(os.path.join(DATA, "partners.csv"), "w", newline="", encoding="utf-8") as f: w = csv.writer(f) w.writerow(["partner_id", "partner_name", "region", "site_city"]) for p in PARTNERS: w.writerow(list(p)) with open(os.path.join(DATA, "raw-orders.csv"), "w", newline="", encoding="utf-8") as f: w = csv.writer(f) # Four preamble lines sit above the header, exactly as the source system emits them. w.writerow(["Apex Instruments"]) w.writerow(["Order extract, all channels"]) w.writerow(["Range: 2026-07-01 to 2026-09-30"]) w.writerow([]) w.writerow(HEADERS) for r in rows: w.writerow([r[h] for h in HEADERS]) print("wrote data/raw-orders.csv (%d data rows, header on line 5)" % len(rows)) print("wrote data/partners.csv (%d partners)" % len(PARTNERS))