"""Government (state) fee estimation + collection for at-cost trucking services. At-cost services (IRP, IFTA, intrastate authority, ...) only collect our service fee at checkout. The real government fee is variable and must be collected separately, at cost, before we file. This module: 1. estimate_gov_fee(slug, intake) -> GovFeeEstimate Best-effort estimate of the state fee from the carrier's intake (base state, operating states, power units, weight bracket). For IRP this is an order-of-magnitude estimate ONLY (true apportioned fees are known when the state portal computes them); for fixed/near-fixed fees (IFTA license + decals, most intrastate authority filings) it is exact-ish. 2. create_gov_fee_order(...) -> child order_number Creates a CHILD compliance_orders row (service_fee_cents = 0, gov_fee_cents = estimate) linked to the parent via parent_order_number, so it flows through the existing checkout/payment-picker/webhook unchanged. 3. send_gov_fee_payment_email(...) Emails the customer a payment link (every method + correct surcharges). Design note on surcharges: the payment page itself shows ACH at 0% (Stripe ACH is 0.8% capped $5 — absorbed), card/PayPal 3%, Klarna 6%, crypto 0%. We do not hardcode those here; the existing /order page + create-session own that. """ from __future__ import annotations import json import logging import os import secrets from dataclasses import dataclass, field LOG = logging.getLogger("workers.services.gov_fee") SITE = os.getenv("PUBLIC_SITE_URL", "https://performancewest.net").rstrip("/") # IFTA: license is typically free; decals ~$0-$6 each, 1 set (2 decals) per # qualifying power unit. We bill a flat, transparent per-unit decal cost and note # it is at cost. Most base states are $0-$10 for a 1-2 truck fleet. IFTA_DECAL_FEE_PER_UNIT_CENTS = int(os.getenv("IFTA_DECAL_FEE_PER_UNIT_CENTS", "1000")) # $10/unit set, conservative # Intrastate authority: state filing fee, fixed per state. Conservative # defaults; refine per state as we confirm exact amounts. Cents. INTRASTATE_AUTHORITY_FEE_CENTS = { "SC": 0, # SC intrastate handled via federal authority; nominal/none "TX": 10000, # TxDMV intrastate "CA": 0, # MCP handled separately "FL": 5000, "GA": 5000, "_default": 5000, } # IRP apportioned registration is genuinely variable (apportioned by fleet # miles per jurisdiction + registered weight). We can only ESTIMATE here; the # exact fee comes from the base-state IRP portal at filing time. Estimate model: # per_unit_base * power_units * weight_multiplier * jurisdiction_factor # This is intentionally conservative (errs high) so a customer is never # surprised by a higher real fee; any overage is refunded / underage re-billed. IRP_PER_UNIT_BASE_CENTS = int(os.getenv("IRP_PER_UNIT_BASE_CENTS", "150000")) # $1,500/truck base IRP_WEIGHT_MULTIPLIERS = { "under_26k": 0.6, "26k_to_80k": 1.0, "over_80k": 1.4, "_default": 1.0, } # Agency / government payment-processing (card convenience) fees. Many state # portals add a card convenience fee on top of the statutory fee; we pass it # through so the customer pays the true all-in cost. Modeled as a percentage of # the government fee plus an optional flat add-on, per service. Refine per state # as confirmed. (pct_of_gov_fee, flat_cents) AGENCY_PROCESSING_FEE = { "irp-registration": (0.0, 0), # SC IRP invoice states the all-in total; no extra "ifta-application": (0.0, 0), # decals billed at cost; no card fee on a tiny amount "intrastate-authority": (0.0, 0), # filing fee paid by check/ACH; no card fee "ucr-registration": (0.0, 0), # ucr.gov includes no separate convenience fee "_default": (0.0, 0), } def agency_processing_fee_cents(slug: str, gov_fee_cents: int) -> int: """Government/agency card-convenience fee passed through to the customer.""" pct, flat = AGENCY_PROCESSING_FEE.get(slug, AGENCY_PROCESSING_FEE["_default"]) return int(round(gov_fee_cents * pct / 100.0)) + flat @dataclass class GovFeeEstimate: cents: int label: str exact: bool = False # True when the amount is fixed/known, not an estimate breakdown: list[str] = field(default_factory=list) def _int(v, default=0) -> int: try: return int(str(v).strip()) except (TypeError, ValueError): return default def estimate_gov_fee(slug: str, intake: dict) -> GovFeeEstimate: """Estimate the all-in government fee (statutory fee + any agency card/ convenience processing fee) for an at-cost trucking service.""" est = _estimate_gov_fee_base(slug, intake) proc = agency_processing_fee_cents(slug, est.cents) if proc > 0: est.breakdown.append(f"Agency processing fee: ${proc/100:.2f}") est.cents += proc est.label += f" (incl. ${proc/100:.2f} agency processing fee)" return est def _estimate_gov_fee_base(slug: str, intake: dict) -> GovFeeEstimate: """Statutory/base government fee before any agency processing surcharge.""" intake = intake or {} power_units = max(_int(intake.get("power_units"), 1), 1) base_state = (intake.get("base_state") or intake.get("address_state") or "").upper() op_states = intake.get("operating_states") or [] if isinstance(op_states, str): try: op_states = json.loads(op_states) except Exception: op_states = [s.strip() for s in op_states.split(",") if s.strip()] weight = (intake.get("gross_weight_bracket") or "_default") if slug == "ifta-application": cents = IFTA_DECAL_FEE_PER_UNIT_CENTS * power_units return GovFeeEstimate( cents=cents, label=f"IFTA license + decals ({power_units} unit set(s), {base_state or 'base state'}) — at cost", exact=False, breakdown=[f"IFTA decals: {power_units} x ${IFTA_DECAL_FEE_PER_UNIT_CENTS/100:.2f}"], ) if slug == "intrastate-authority": cents = INTRASTATE_AUTHORITY_FEE_CENTS.get(base_state, INTRASTATE_AUTHORITY_FEE_CENTS["_default"]) return GovFeeEstimate( cents=cents, label=f"{base_state or 'State'} intrastate authority filing fee — at cost", exact=base_state in INTRASTATE_AUTHORITY_FEE_CENTS, breakdown=[f"{base_state or 'State'} intrastate filing fee: ${cents/100:.2f}"], ) if slug == "irp-registration": wmult = IRP_WEIGHT_MULTIPLIERS.get(weight, IRP_WEIGHT_MULTIPLIERS["_default"]) # More operating jurisdictions => higher apportioned total. Base state + # each additional operating state adds ~12% (rough apportionment proxy). n_juris = max(len(set([base_state] + list(op_states)) - {""}), 1) jfactor = 1.0 + 0.12 * (n_juris - 1) cents = int(IRP_PER_UNIT_BASE_CENTS * power_units * wmult * jfactor) return GovFeeEstimate( cents=cents, label=(f"IRP apportioned registration ESTIMATE — {power_units} unit(s), " f"{weight.replace('_',' ')}, {n_juris} jurisdiction(s) — at cost, " f"final fee set by {base_state or 'base state'} IRP office"), exact=False, breakdown=[ f"Base: ${IRP_PER_UNIT_BASE_CENTS/100:.2f}/unit x {power_units}", f"Weight x{wmult}", f"Jurisdictions x{jfactor:.2f} ({n_juris})", f"Estimated total: ${cents/100:.2f}", ], ) return GovFeeEstimate(cents=0, label="No government fee", exact=True) def create_gov_fee_order(parent_order_number: str, slug: str, estimate: GovFeeEstimate, customer_email: str, customer_name: str, customer_phone: str = "") -> str | None: """Create (idempotently) a child gov-fee compliance order for the parent. Returns the child order_number, or None on failure. The child has service_fee_cents=0 and gov_fee_cents=estimate so it bills only the gov fee through the normal checkout flow. Re-running updates the amount of an existing unpaid child rather than creating duplicates. """ if estimate.cents <= 0: return None try: import psycopg2 conn = psycopg2.connect(os.environ.get("DATABASE_URL", "")) with conn.cursor() as cur: # Reuse an existing UNPAID gov-fee child for this parent if present. cur.execute( """SELECT order_number FROM compliance_orders WHERE parent_order_number = %s AND service_slug = %s AND payment_status = 'pending_payment' ORDER BY created_at DESC LIMIT 1""", (parent_order_number, f"{slug}-govfee"), ) row = cur.fetchone() if row: child = row[0] cur.execute( """UPDATE compliance_orders SET gov_fee_cents = %s, gov_fee_label = %s, updated_at = now() WHERE order_number = %s""", (estimate.cents, estimate.label, child), ) conn.commit() conn.close() LOG.info("[%s] Updated existing gov-fee child %s ($%.2f)", parent_order_number, child, estimate.cents / 100) return child child = "CG-" + secrets.token_hex(4).upper() cur.execute( """INSERT INTO compliance_orders (order_number, parent_order_number, service_slug, service_name, service_fee_cents, gov_fee_cents, gov_fee_label, customer_email, customer_name, customer_phone, payment_status, intake_data, intake_data_validated, fulfillment_status) VALUES (%s, %s, %s, %s, 0, %s, %s, %s, %s, %s, 'pending_payment', %s, TRUE, NULL)""", ( child, parent_order_number, f"{slug}-govfee", f"Government fee — {slug}", estimate.cents, estimate.label, customer_email, customer_name, customer_phone, json.dumps({"source": "gov-fee", "parent": parent_order_number, "breakdown": estimate.breakdown, "exact": estimate.exact}), ), ) conn.commit() conn.close() LOG.info("[%s] Created gov-fee child %s ($%.2f)", parent_order_number, child, estimate.cents / 100) return child except Exception as exc: # noqa: BLE001 LOG.error("[%s] Failed to create gov-fee child order: %s", parent_order_number, exc) return None def gov_fee_payment_url(child_order_number: str) -> str: """Public payment page for a gov-fee child order. Reuses the standard order page, which renders every payment method + surcharge and calls /api/v1/checkout/create-session with order_type=compliance.""" return f"{SITE}/order/pay?order={child_order_number}" def gov_fee_dispute_url(child_order_number: str) -> str: """Public 'this estimate looks wrong' page that opens a support ticket pre-tagged with the order so ops can correct the fee.""" return f"{SITE}/order/dispute?order={child_order_number}" def send_gov_fee_payment_email(customer_email: str, customer_name: str, service_label: str, entity_name: str, estimate: GovFeeEstimate, child_order_number: str) -> bool: """Email the customer an itemized government-fee bill + payment link, plus a link to dispute the amount if it looks wrong.""" if not customer_email: return False url = gov_fee_payment_url(child_order_number) dispute = gov_fee_dispute_url(child_order_number) amt = f"${estimate.cents / 100:,.2f}" # Itemized breakdown so the customer can check it for accuracy. items = "\n".join(f" - {line}" for line in (estimate.breakdown or [estimate.label])) qualifier = ( "This amount is the state's confirmed fee." if estimate.exact else "This is an estimate billed at cost. If the state's final fee differs we " "refund any overage or bill the small difference.") try: import smtplib from email.mime.text import MIMEText body = ( f"Hi {customer_name or 'there'},\n\n" f"Your {service_label} for {entity_name} is ready to file. The only " f"remaining step is the government/state fee, which we collect at cost.\n\n" f"Here is exactly what the {amt} covers — please review it for accuracy:\n\n" f" {estimate.label}\n" f"{items}\n" f" ----------------------------------------\n" f" Total government fee: {amt}\n\n" f"{qualifier}\n\n" f"✅ If this looks right, pay securely here (choose your method — " f"bank transfer/ACH has no processing fee):\n{url}\n\n" f"❓ If something looks wrong (wrong fleet size, states, weight, or " f"amount), tell us here and we'll fix it before you pay:\n{dispute}\n\n" f"As soon as the fee is paid we file with the state and send your " f"confirmation.\n\n" f"Order: {child_order_number}\n" f"Questions? Reply here or call (888) 411-0383.\n\n" f"Performance West Inc.\nDOT / State Motor Carrier Compliance\n" ) msg = MIMEText(body) msg["Subject"] = f"Action needed: government fee for your {service_label} ({amt})" msg["From"] = os.getenv("SMTP_FROM", "Performance West ") msg["To"] = customer_email with smtplib.SMTP(os.getenv("SMTP_HOST", "co.carrierone.com"), int(os.getenv("SMTP_PORT", "587")), timeout=30) as s: s.starttls() u, p = os.getenv("SMTP_USER", ""), os.getenv("SMTP_PASS", "") if u and p: s.login(u, p) s.sendmail("noreply@performancewest.net", [customer_email], msg.as_string()) LOG.info("Gov-fee payment email sent to %s for %s (%s)", customer_email, child_order_number, amt) return True except Exception as exc: # noqa: BLE001 LOG.warning("Failed to send gov-fee payment email to %s: %s", customer_email, exc) return False