fix error in gitignore
This commit is contained in:
@@ -0,0 +1 @@
|
||||
# pricing app
|
||||
@@ -0,0 +1,6 @@
|
||||
from django.apps import AppConfig
|
||||
|
||||
|
||||
class PricingConfig(AppConfig):
|
||||
default_auto_field = "django.db.models.BigAutoField"
|
||||
name = "apps.pricing"
|
||||
@@ -0,0 +1,117 @@
|
||||
"""Backend-authoritative pricing engine.
|
||||
|
||||
Produces a full, explainable breakdown and is always the source of truth —
|
||||
the frontend may display estimates but must never be trusted for financial
|
||||
values.
|
||||
"""
|
||||
from decimal import Decimal, ROUND_HALF_UP
|
||||
|
||||
CHARGE_TYPES = {"fixed", "per_kg", "per_unit", "percentage", "tiered", "fee"}
|
||||
|
||||
|
||||
class PricingEngine:
|
||||
def calculate(self, weight_kg, selected_steps, currency: str = "IRR") -> dict:
|
||||
weight = Decimal(str(weight_kg))
|
||||
lines = []
|
||||
subtotal = Decimal("0")
|
||||
|
||||
# Pass 1 — charge rules.
|
||||
for step in selected_steps:
|
||||
for rule in step.pricing_rules.filter(active=True, rule_type__in=CHARGE_TYPES).order_by("priority"):
|
||||
line = self._apply_charge_rule(rule, weight)
|
||||
if line is None:
|
||||
continue
|
||||
lines.append(line)
|
||||
subtotal += Decimal(line["amount"])
|
||||
|
||||
# Pass 2 — discounts (fixed or percentage of subtotal).
|
||||
discounts = Decimal("0")
|
||||
for step in selected_steps:
|
||||
for rule in step.pricing_rules.filter(active=True, rule_type="discount"):
|
||||
amt = Decimal(str(rule.amount))
|
||||
if (rule.config or {}).get("percentage"):
|
||||
amt = subtotal * amt / Decimal("100")
|
||||
discounts += amt
|
||||
lines.append(
|
||||
self._line(rule, weight, "discount", Decimal("0"), amt)
|
||||
)
|
||||
|
||||
# Pass 3 — taxes on (subtotal - discounts).
|
||||
taxes = Decimal("0")
|
||||
taxable = subtotal - discounts
|
||||
for step in selected_steps:
|
||||
for rule in step.pricing_rules.filter(active=True, rule_type="tax"):
|
||||
amt = taxable * Decimal(str(rule.amount)) / Decimal("100")
|
||||
taxes += amt
|
||||
lines.append(self._line(rule, weight, "tax", Decimal("0"), amt))
|
||||
|
||||
total = subtotal - discounts + taxes
|
||||
if total < 0:
|
||||
total = Decimal("0")
|
||||
|
||||
return {
|
||||
"currency": currency,
|
||||
"lines": lines,
|
||||
"subtotal": str(self._q(subtotal)),
|
||||
"discounts": str(self._q(discounts)),
|
||||
"taxes": str(self._q(taxes)),
|
||||
"total": str(self._q(total)),
|
||||
}
|
||||
|
||||
def _q(self, v: Decimal) -> Decimal:
|
||||
return v.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
||||
|
||||
def _line(self, rule, weight, kind, unit_price, amount, quantity=None) -> dict:
|
||||
return {
|
||||
"step_key": rule.step.key,
|
||||
"rule": rule.name,
|
||||
"kind": kind,
|
||||
"quantity": str(quantity if quantity is not None else weight),
|
||||
"unit_price": str(self._q(unit_price)),
|
||||
"amount": str(self._q(amount)),
|
||||
}
|
||||
|
||||
def _apply_charge_rule(self, rule, weight):
|
||||
kind = rule.rule_type
|
||||
cfg = rule.config or {}
|
||||
amount = Decimal(str(rule.amount))
|
||||
unit_price = Decimal("0")
|
||||
quantity = weight
|
||||
if kind == "fixed":
|
||||
unit_price = amount
|
||||
elif kind == "per_kg":
|
||||
unit_price = amount
|
||||
amount = amount * weight
|
||||
elif kind == "per_unit":
|
||||
units = Decimal(str(cfg.get("units", 1)))
|
||||
unit_price = amount
|
||||
amount = amount * units
|
||||
quantity = units
|
||||
elif kind == "percentage":
|
||||
base = Decimal(str(cfg.get("base", 0)))
|
||||
unit_price = amount
|
||||
amount = base * amount / Decimal("100")
|
||||
elif kind == "tiered":
|
||||
unit_price = self._tier_unit_price(cfg.get("tiers", []), weight)
|
||||
amount = unit_price * weight
|
||||
elif kind == "fee":
|
||||
unit_price = Decimal("0")
|
||||
else:
|
||||
return None
|
||||
|
||||
min_c = cfg.get("min_charge")
|
||||
max_c = cfg.get("max_charge")
|
||||
if min_c is not None and amount < Decimal(str(min_c)):
|
||||
amount = Decimal(str(min_c))
|
||||
if max_c is not None and amount > Decimal(str(max_c)):
|
||||
amount = Decimal(str(max_c))
|
||||
|
||||
return self._line(rule, weight, kind, unit_price, amount, quantity)
|
||||
|
||||
def _tier_unit_price(self, tiers, weight) -> Decimal:
|
||||
best = Decimal("0")
|
||||
for tier in tiers or []:
|
||||
low = Decimal(str(tier.get("from", 0)))
|
||||
if weight >= low:
|
||||
best = Decimal(str(tier.get("price", 0)))
|
||||
return best
|
||||
@@ -0,0 +1,35 @@
|
||||
# Generated by Django 5.2.17 on 2026-09-15 09:29
|
||||
|
||||
import django.db.models.deletion
|
||||
import uuid
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
initial = True
|
||||
|
||||
dependencies = [
|
||||
('workflows', '0001_initial'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='PricingRule',
|
||||
fields=[
|
||||
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
|
||||
('created_at', models.DateTimeField(auto_now_add=True, db_index=True)),
|
||||
('updated_at', models.DateTimeField(auto_now=True)),
|
||||
('name', models.CharField(max_length=200)),
|
||||
('rule_type', models.CharField(choices=[('fixed', 'Fixed'), ('per_kg', 'Per kilogram'), ('per_unit', 'Per unit'), ('percentage', 'Percentage'), ('tiered', 'Tiered'), ('discount', 'Discount'), ('tax', 'Tax'), ('fee', 'Additional fee')], max_length=16)),
|
||||
('amount', models.DecimalField(decimal_places=2, default=0, max_digits=18)),
|
||||
('config', models.JSONField(blank=True, default=dict)),
|
||||
('active', models.BooleanField(default=True)),
|
||||
('priority', models.IntegerField(default=0)),
|
||||
('step', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='pricing_rules', to='workflows.workflowstep')),
|
||||
],
|
||||
options={
|
||||
'ordering': ['priority'],
|
||||
},
|
||||
),
|
||||
]
|
||||
@@ -0,0 +1,33 @@
|
||||
from django.db import models
|
||||
|
||||
from apps.core.models import TimeStampedModel, UUIDModel
|
||||
|
||||
|
||||
class PricingRule(UUIDModel, TimeStampedModel):
|
||||
class RuleType(models.TextChoices):
|
||||
FIXED = "fixed", "Fixed"
|
||||
PER_KG = "per_kg", "Per kilogram"
|
||||
PER_UNIT = "per_unit", "Per unit"
|
||||
PERCENTAGE = "percentage", "Percentage"
|
||||
TIERED = "tiered", "Tiered"
|
||||
DISCOUNT = "discount", "Discount"
|
||||
TAX = "tax", "Tax"
|
||||
FEE = "fee", "Additional fee"
|
||||
|
||||
step = models.ForeignKey(
|
||||
"workflows.WorkflowStep", on_delete=models.CASCADE, related_name="pricing_rules"
|
||||
)
|
||||
name = models.CharField(max_length=200)
|
||||
rule_type = models.CharField(max_length=16, choices=RuleType.choices)
|
||||
# Base amount. Meaning depends on rule_type:
|
||||
# fixed/per_kg/per_unit -> money; percentage/discount/tax -> percent.
|
||||
amount = models.DecimalField(max_digits=18, decimal_places=2, default=0)
|
||||
config = models.JSONField(default=dict, blank=True)
|
||||
active = models.BooleanField(default=True)
|
||||
priority = models.IntegerField(default=0)
|
||||
|
||||
class Meta:
|
||||
ordering = ["priority"]
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"{self.name} ({self.rule_type})"
|
||||
@@ -0,0 +1,41 @@
|
||||
from decimal import Decimal
|
||||
|
||||
from django.test import TestCase
|
||||
|
||||
from apps.pricing.engine import PricingEngine
|
||||
from apps.pricing.models import PricingRule
|
||||
from apps.workflows.models import Workflow, WorkflowStep, WorkflowVersion
|
||||
|
||||
|
||||
class PricingEngineTests(TestCase):
|
||||
def setUp(self):
|
||||
wf = Workflow.objects.create(name="P", slug="p")
|
||||
self.version = WorkflowVersion.objects.create(workflow=wf, version=1)
|
||||
self.washing = WorkflowStep.objects.create(version=self.version, key="washing", name="Washing")
|
||||
self.drying = WorkflowStep.objects.create(version=self.version, key="drying", name="Drying")
|
||||
|
||||
def test_per_kg(self):
|
||||
PricingRule.objects.create(step=self.washing, name="Wash", rule_type="per_kg", amount=Decimal("5000"))
|
||||
result = PricingEngine().calculate(850, [self.washing])
|
||||
self.assertEqual(result["total"], "4250000.00")
|
||||
|
||||
def test_tiered(self):
|
||||
PricingRule.objects.create(step=self.washing, name="Tiered", rule_type="tiered", config={
|
||||
"tiers": [{"from": 0, "price": "5000"}, {"from": 100, "price": "4000"}, {"from": 500, "price": "3000"}],
|
||||
})
|
||||
result = PricingEngine().calculate(600, [self.washing])
|
||||
self.assertEqual(result["total"], "1800000.00")
|
||||
|
||||
def test_discount_and_tax(self):
|
||||
PricingRule.objects.create(step=self.washing, name="Wash", rule_type="per_kg", amount=Decimal("1000"))
|
||||
PricingRule.objects.create(step=self.washing, name="Disc", rule_type="discount", amount=Decimal("10"), config={"percentage": True})
|
||||
PricingRule.objects.create(step=self.washing, name="Tax", rule_type="tax", amount=Decimal("9"))
|
||||
result = PricingEngine().calculate(100, [self.washing])
|
||||
# subtotal 100000, discount 10% => 10000, taxable 90000, tax 9% => 8100, total 98100
|
||||
self.assertEqual(result["total"], "98100.00")
|
||||
|
||||
def test_multiple_steps(self):
|
||||
PricingRule.objects.create(step=self.washing, name="Wash", rule_type="per_kg", amount=Decimal("5000"))
|
||||
PricingRule.objects.create(step=self.drying, name="Dry", rule_type="per_kg", amount=Decimal("8000"))
|
||||
result = PricingEngine().calculate(100, [self.washing, self.drying])
|
||||
self.assertEqual(result["total"], "1300000.00")
|
||||
Reference in New Issue
Block a user