#!/usr/bin/env python3 """Pricing engine replicating the Ricoh Proposal Workspace (internal). Sean's standing defaults (set 2026-10-06): - Target net profit margin: 25% -> selling = total_cost / 0.75 - Lease: 63 months, FMV, Universal lender -> monthly = selling * factor - Monthly print plan: B&W $0.014/page, color $0.08/page (customer rates) Input: parsed intake dict (from intake_watcher). Output: priced dict ready for build_proposal.save_proposal_pdf. NEVER invents numbers: missing cost/volume data -> status "needs_review" with explicit gaps; no proposal is generated until gaps are filled. """ import os, json, re, datetime BASE = os.path.dirname(os.path.abspath(__file__)) MARGIN = 0.25 TERM_MONTHS = 63 LEASE_TYPE = "FMV" BW_RATE = 0.014 COLOR_RATE = 0.08 def load_data(): d = os.path.join(BASE, "data") products = {p["model"].lower(): p for p in json.load(open(os.path.join(d, "products.json")))} accessories = {a["name"].lower(): a for a in json.load(open(os.path.join(d, "accessories.json")))} factors = json.load(open(os.path.join(d, "lease_factors.json"))) return products, accessories, factors def match_accessories(options_str, accessories): """Match free-text options against the catalog (names contain commas, so naive comma-splitting breaks them). Longest names first.""" found, gaps = [], [] remaining = " " + (options_str or "") + " " for name in sorted(accessories.keys(), key=len, reverse=True): if name in remaining.lower(): found.append(accessories[name]["name"]) remaining = remaining.lower().replace(name, " ") leftover = re.sub(r"[\s,;]+", " ", remaining).strip() if len(leftover) > 2: gaps.append("accessory price unknown for '%s'" % leftover[:60]) return found, gaps # Quote-site checkbox labels -> catalog accessories (Sean 2026-10-06 fix: # the form's generic labels never matched catalog SKUs, so every real # submission produced accessory gaps). None = genuine interest, not hardware. FORM_OPTION_MAP = { "searchable pdf / ocr": "OCR Unit Type M52", "mobile / wireless printing": "IEEE 802.11 a/b/g/n/ac Wireless Interface Type M52", "stapling / finishing": "Finisher SR3320 (1,000 sheet)", "booklet finishing": "Booklet Finisher SR3330 (1,000 sheet)", "folding": "Internal Multi-Fold Unit FD3020", "extra paper capacity": "Paper Feed Unit PB3320 (2 x 550 trays)", "fax": "G3 Interface Unit Type M52", "secure printing / card access": "Card Reader Cover Type M52", "track use by user / department": None, "compare a monochrome alternative": None, } def translate_form_options(options_str): """Map the quote site's generic option labels to catalog names. Returns (catalog_options_str, notes). Notes are internal, not gaps.""" notes = [] if not options_str: return "", notes remaining = " " + str(options_str) + " " translated = [] for label in sorted(FORM_OPTION_MAP, key=len, reverse=True): if label in remaining.lower(): cat = FORM_OPTION_MAP[label] if cat: translated.append(cat) else: notes.append("customer interest, not hardware: '%s'" % label) remaining = remaining.lower().replace(label, " ") leftover = re.sub(r"[\s,;]+", " ", remaining).strip() if leftover: translated.append(leftover) # let match_accessories try it raw return ", ".join(translated), notes # Volume-based model recommendation (the form defaults to "Please recommend # the final model", so the engine picks; Sean reviews before anything sends). RECOMMEND_LADDER = [ (6000, "IM C2510"), (15000, "IM C3510"), (30000, "IM C4510"), (60000, "IM C6010"), (float("inf"), "IM C7010"), ] def recommend_model(bw_pages, color_pages): total = (bw_pages or 0) + (color_pages or 0) for cap, model in RECOMMEND_LADDER: if total < cap: return model, total return "IM C7010", total def parse_money(s): if not s: return None m = re.search(r"[\d,]+(?:\.\d+)?", str(s)) return float(m.group(0).replace(",", "")) if m else None def parse_volume(s): """'8,000/mo', '200/day', '10-25 pages' -> monthly pages (int) or None.""" if not s: return None s = str(s).lower() nums = [float(n.replace(",", "")) for n in re.findall(r"[\d,]+", s)] if not nums: return None v = max(nums) # take the top of a range if "/day" in s or "per day" in s or "daily" in s: v *= 22 return int(v) def price_intake(intake): products, accessories, factors = load_data() gaps = [] p = {"customer": intake.get("customer", {}), "intake": intake} # --- volumes first (needed for model recommendation) --- bw_pages = parse_volume(intake.get("bwVolume")) color_pages = parse_volume(intake.get("colorVolume")) # --- equipment --- model = (intake.get("model") or "").strip() model_recommended = False model_note = "" if not model or model.lower() == "please recommend the final model": model, vol_total = recommend_model(bw_pages, color_pages) model_recommended = True model_note = "recommended by engine for ~%s pages/mo" % format(vol_total, ",d") p["model"] = model p["model_recommended"] = model_recommended p["model_note"] = model_note prod = products.get(model.lower()) if not prod: # fuzzy: try matching on model number fragments, longest match wins # (avoids "IM" beating "IM C6010" on input "Ricoh IM C6010") cands = [k for k in products if model.lower()[:6] in k or k in model.lower()] cands.sort(key=len, reverse=True) prod = products[cands[0]] if cands else None if not prod: gaps.append("model cost unknown for '%s'" % model) equip_cost = None elif not prod.get("cost"): gaps.append("model '%s' has no catalog cost - needs approved dealer cost override" % prod["model"]) equip_cost = None elif prod.get("review_flag"): gaps.append("model '%s' is flagged 'review against video' - verify cost before quoting" % prod["model"]) equip_cost = None else: equip_cost = prod["cost"] p["model"] = prod["model"] acc_list, acc_cost, acc_gaps = [], 0.0, [] catalog_options, option_notes = translate_form_options(intake.get("options", "")) p["option_notes"] = option_notes matched, match_gaps = match_accessories(catalog_options, accessories) for name in matched: ao = accessories[name.lower()] acc_list.append({"name": ao["name"], "cost": ao["price"]}) acc_cost += ao["price"] gaps.extend(match_gaps) p["accessories"] = acc_list if equip_cost is None: p.update(status="needs_review", gaps=gaps) return p total_cost = equip_cost + acc_cost selling = total_cost / (1 - MARGIN) # --- lease --- factor = None for f in factors: if f["term"] == TERM_MONTHS and f["type"].upper() == LEASE_TYPE: factor = f["factor"]; break if factor is None: gaps.append("lease factor missing for %d-mo %s" % (TERM_MONTHS, LEASE_TYPE)) p.update(status="needs_review", gaps=gaps) return p lease_monthly = selling * factor # --- print plan --- if bw_pages is None and color_pages is None: gaps.append("no monthly volume stated (bwVolume/colorVolume)") bw_maint = (bw_pages or 0) * BW_RATE color_maint = (color_pages or 0) * COLOR_RATE total_monthly = lease_monthly + bw_maint + color_maint # --- sanity bounds (council review 2026-10-06): never let an implausible # auto-priced proposal get a silent green light. Flags go to Sean's alert # and the dashboard, never to the customer. review_flags = [] if total_monthly > 3000: review_flags.append("monthly total $%s is unusually high - verify volumes before sending" % format(total_monthly, ",.2f")) if (bw_pages or 0) > 100000 or (color_pages or 0) > 50000: review_flags.append("extreme volume stated (%s B&W / %s color) - verify" % (bw_pages, color_pages)) try: users = int(re.sub(r"[^\d]", "", str(intake.get("raw", {}).get("users") or "0")) or 0) mppm = re.search(r"C(\d+)10", model or "") ppm = int(mppm.group(1)) if mppm else None if users and ppm: if users <= 5 and ppm >= 55: review_flags.append("%d users on a %d ppm system looks oversized - verify" % (users, ppm)) if users >= 50 and ppm <= 25: review_flags.append("%d users on a %d ppm system looks undersized - verify" % (users, ppm)) except Exception: pass # --- current situation --- cur_lease = parse_money(intake.get("monthlyPayment")) # current maintenance unknown from intake; leave blank unless derivable p.update( equipment_cost=round(total_cost, 2), selling_price=round(selling, 2), lease_monthly=round(lease_monthly, 2), bw_pages=bw_pages, color_pages=color_pages, bw_maint=round(bw_maint, 2), color_maint=round(color_maint, 2), total_monthly=round(total_monthly, 2), current_lease=cur_lease, current_maint=None, current_model=intake.get("currentModel"), valid_until=(datetime.date.today() + datetime.timedelta(days=30)).strftime("%B %d, %Y"), timeline={ "Contract signature by \u2026": (datetime.date.today() + datetime.timedelta(days=7)).strftime("%B %d, %Y"), "Preinstallation survey by \u2026": (datetime.date.today() + datetime.timedelta(days=14)).strftime("%B %d, %Y"), "Delivery/acceptance of equipment by \u2026": (datetime.date.today() + datetime.timedelta(days=21)).strftime("%B %d, %Y"), "System training by \u2026": (datetime.date.today() + datetime.timedelta(days=23)).strftime("%B %d, %Y"), "Account review by \u2026": (datetime.date.today() + datetime.timedelta(days=52)).strftime("%B %d, %Y"), }, margin=MARGIN, term_months=TERM_MONTHS, lease_type=LEASE_TYPE, review_flags=review_flags, ) p["status"] = "needs_review" if gaps else "ready" p["gaps"] = gaps return p if __name__ == "__main__": import sys intake = json.load(open(sys.argv[1])) print(json.dumps(price_intake(intake), indent=1))