"""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, } @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 government/state fee for an at-cost trucking service.""" 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 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 a payment link for the government fee.""" if not customer_email: return False url = gov_fee_payment_url(child_order_number) amt = f"${estimate.cents / 100:,.2f}" qualifier = "" 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" {service_label}\n" f" Government fee: {amt}\n" f" {estimate.label}\n\n" f"Pay securely here (choose your preferred method — bank transfer/ACH " f"has no processing fee):\n{url}\n\n" f"{qualifier}\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